Keenable Search
KeenableSearch gives an agent web research tools backed by the
Keenable search API: search that returns a short excerpt
of each hit, and full-page retrieval for digging into a specific URL.
While Pydantic AI Harness is on 0.x releases, the API may change between minor releases; when it does, deprecation warnings and release-note migration guidance tell you (or your agent) exactly how to upgrade. See the version policy.
Web research needs two tools that have to agree with each other: search, to survey what exists, and retrieval, to read the pages worth reading. Wiring them together means budgeting what each returns, or the agent either sees titles with no substance or drowns in full pages it was going to discard.
Getting there usually also means an account, an API key, and another vendor SDK in the dependency tree before the first query runs.
KeenableSearch bundles the plumbing into a single
capability: the two tools, their output
budgets, and short research guidance in the system prompt. It needs neither a
key nor an install to run.
Nothing to install and nothing to configure: Keenable’s endpoints are keyless
by default, and this capability talks to them over httpx2, which
pydantic-ai-harness already depends on. The examples use an Anthropic model,
so set ANTHROPIC_API_KEY for them.
from pydantic_ai import Agent
from pydantic_ai_harness import KeenableSearch
agent = Agent('anthropic:claude-sonnet-5-5', capabilities=[KeenableSearch()])
result = agent.run_sync('What changed in the EU AI Act this year? Cite your sources.')
print(result.output)
The capability adds two tools:
web_search(query): the matching pages, each with title, URL, and a short excerpt. Titles are cut at 300 characters, and a result whose URL is longer than 2,048 characters or has whitespace inside is left out.get_page(url): one page as markdown.
Both return a ToolReturn whose
metadata['sources'] lists the URLs and titles behind the result, so an
application can render citations from
ToolReturnPart.metadata without
parsing the text.
from pydantic_ai import Agent
from pydantic_ai_harness import KeenableSearch
agent = Agent(
'anthropic:claude-sonnet-5-5',
capabilities=[
KeenableSearch(
num_results=10,
max_snippet_chars=800,
max_page_chars=20_000,
)
],
)
| Field | Default | What it does |
|---|---|---|
num_results | 5 | How many results web_search returns, from 1 to 50. |
max_snippet_chars | 500 | Excerpt budget per search result. |
max_page_chars | 10_000 | Page budget for get_page; longer pages are truncated and marked, with the marker counted against the budget. A budget too small to hold the marker ends the page with … instead. |
guidance | None | Replaces the default research instructions, including their untrusted-content rule; '' contributes none. |
client | None | A KeenableClient to use instead of the default HTTP client. |
max_snippet_chars exists because Keenable’s snippets run to thousands of
characters, far more than the excerpt a typical search API returns. Left
unbounded, a single web_search would spend the context window
on pages the agent has not chosen to read yet; get_page is how it opts into
one.
Keenable is keyless by default, so nothing is required to get started. An API
key raises rate limits; keyed calls draw on your Keenable credits, and searches
run in Keenable’s default pro mode:
export KEENABLE_API_KEY='keen_...'
Set KEENABLE_API_URL to point at a different Keenable deployment. It must be
https, except against loopback for local development, and carry no query or
fragment: the endpoint path is appended to it.
To configure either explicitly rather than through the environment, pass a client:
from pydantic_ai_harness.keenable import HttpKeenableClient, KeenableSearch
KeenableSearch(client=HttpKeenableClient(api_key='keen_...'))
Any object satisfying the KeenableClient protocol works, which is also how
tests substitute a fake.
To reuse connections, or to send requests through a proxy, custom TLS, or a
retrying transport, pass your own httpx2.AsyncClient as
HttpKeenableClient(http_client=...). Its timeouts apply alongside the
30-second request deadline, and it stays open for you to close. Without one,
each request opens and closes its own client. Requests never follow redirects,
whatever the client is configured to do, so the API key stays with the
configured host.
For a proxy, a self-hosted deployment, or a KeenableClient of your own, this
is what the default client sends. search posts a JSON body
{"query": ..., "max_results": ...} to /v1/search, or to
/v1/search/public when there is no key, and returns the response’s results
list; the toolset passes num_results as max_results. fetch sends
GET /v1/fetch, or /v1/fetch/public when keyless, with the page address in
the url query parameter, and returns the response object; the toolset reads
its content, title, and url. Every request carries an X-Keenable-Title: Pydantic AI Harness header and adds X-API-Key when a key is set. Each request must finish
within 30 seconds, and a response body over 10 MiB is abandoned before parsing.
Any HTTP error status other than 401, 402, and 403 becomes
ModelRetry, as do network failures, a
request past its deadline, and a 2xx whose body is too large, not JSON, or a
search response without its results list, so the model can wait, rephrase,
or try another URL rather than abort the run. A
get_page call that comes back with no readable content is a ModelRetry too. get_page reads Keenable’s indexed copy
of a page, and Keenable documents that fetching a URL it has not indexed fails;
the model sees that failure as a ModelRetry as well. A 401, 402, or
403 propagates as httpx2.HTTPStatusError: a rejected key or an exhausted
account is configuration the model cannot fix.
Two instances register the same tool names and share the default id
(keenable_search), so two that differ raise an error. To run a second setup in
one agent, give it a distinct id and wrap it in core’s PrefixTools
capability, which puts a prefix in front of its tool names.
Core WebSearch, ExaSearch, and YouSearch name their tools the same way —
web_search, plus get_page for the two harness ones — and an agent cannot
have two tools with the same name, so adding one of them next to
KeenableSearch fails on the agent’s first run. Give an agent one web search
capability, or wrap the extra one in PrefixTools to rename its tools.
To use the provider’s native search where there is one and Keenable’s
elsewhere, pass its tool to core WebSearch instead of adding the capability.
This adds only web_search, not get_page, so the agent needs another tool to
read the pages it finds.
from pydantic_ai.capabilities import WebSearch
from pydantic_ai_harness import KeenableSearch
WebSearch(local=KeenableSearch().web_search_tool())
Under durable execution, each Keenable request is recorded, so a recovered run reuses the result instead of making the request again. Temporal and Prefect record each tool call in its own activity or task. DBOS runs function tools in workflow code, so there the request runs as its own step.
The records are named after the capability’s id, which defaults to
keenable_search, so durable execution needs no configuration. Changing an
id renames the records, which in-flight runs then cannot find.
KeenableSearch works with Pydantic AI’s
agent spec, so you can declare it in a config
file instead of Python:
# agent.yaml
model: anthropic:claude-sonnet-5-5
capabilities:
- KeenableSearch:
num_results: 10
from pydantic_ai import Agent
from pydantic_ai_harness import KeenableSearch
agent = Agent.from_file('agent.yaml', custom_capability_types=[KeenableSearch])
The client field is not spec-serializable; spec-loaded instances always
build the default HttpKeenableClient, which reads KEENABLE_API_KEY and
KEENABLE_API_URL from the environment.
KeenableSearch emits no spans of its own. Core’s tool-call spans cover each
search and page read.
Bases: AbstractCapability[AgentDepsT]
Web research for agents, backed by the Keenable search API.
Adds two tools: web_search, which returns search results with a short
excerpt of each page, and get_page, which retrieves the full text of a
specific URL as markdown.
from pydantic_ai import Agent
from pydantic_ai_harness import KeenableSearch
agent = Agent('anthropic:claude-sonnet-4-6', capabilities=[KeenableSearch()])
It needs no API key and no extra install: Keenable’s public endpoints are
keyless, and the capability talks to them over httpx2, which
pydantic-ai-harness already depends on. Set KEENABLE_API_KEY (or pass a
configured client) to raise rate limits.
Each tool’s Keenable request runs as a durable operation, so under durable execution a recovered run reuses the recorded result instead of making the request again.
Number of results web_search returns per query, from 1 to 50.
Sent to Keenable as max_results, the API’s own limit.
Type: int Default: 5
Maximum characters of excerpt web_search returns per result.
Keenable’s snippets run to thousands of characters, far more than the
excerpt a typical search API returns, so leaving this unbounded would spend
the context window on pages the agent has not chosen to read yet.
get_page is how the agent opts into a full page.
Type: int Default: 500
Maximum characters of page text get_page returns.
Longer pages are truncated and marked, so the model can tell that the page
continued past what it was shown. The marker counts against this budget,
and a budget too small to hold it ends the page with … instead.
Type: int Default: 10000
Custom research guidance for the system prompt.
Leave as None for the default guidance, or set '' to contribute no
instructions at all. Custom guidance replaces the default entirely,
including its rule that fetched pages and search results are untrusted
data, so consider stating that rule in your own guidance.
Type: str | None Default: None
Keenable client to use; when None, an HttpKeenableClient is built.
Any object satisfying the KeenableClient protocol works: use it to pass
an API key explicitly, point at a different base URL, or substitute a fake
in tests.
Type: KeenableClient | None Default: None
Stable identity for durable execution, which records each Keenable request under it.
Type: str | None Default: 'keenable_search'
def __post_init__() -> None
Validate the output budgets, so a bad one fails here and not mid-run.
def get_instructions() -> AgentInstructions[AgentDepsT] | None
Static research guidance: search wide, read the promising pages in full, cite URLs.
A non-None guidance replaces the default; '' disables instructions
entirely.
AgentInstructions[AgentDepsT] | None
@classmethod
def combine(
cls,
capabilities: Sequence[AbstractCapability[AgentDepsT]],
) -> AbstractCapability[AgentDepsT]
Two under one id are one configuration stated twice; two that disagree raise rather than merge.
AbstractCapability[AgentDepsT]
def get_toolset() -> KeenableSearchToolset[AgentDepsT]
Build the toolset providing web_search and get_page.
KeenableSearchToolset[AgentDepsT]
def web_search_tool() -> Tool[AgentDepsT]
Keenable’s web_search on its own, configured from this capability, to pass as WebSearch(local=...).
Tool[AgentDepsT]
@classmethod
def from_spec(
cls,
*,
num_results: int = 5,
max_snippet_chars: int = 500,
max_page_chars: int = 10000,
guidance: str | None = None,
id: str | None = 'keenable_search',
description: str | None = None,
defer_loading: bool = False,
) -> KeenableSearch[AgentDepsT]
Construct the capability from serializable spec options.
The client field is not spec-serializable, so spec-loaded instances
always build the default HttpKeenableClient.
KeenableSearch[AgentDepsT]
Bases: FunctionToolset[AgentDepsT]
Gives an agent web research tools backed by the Keenable API.
web_search surveys the web and returns results with a short excerpt of
each page, and get_page retrieves one specific URL as markdown.
Each tool returns a ToolReturn whose return_value is the text the model
sees and whose metadata['sources'] lists the result URLs and titles
(KeenableSource dicts), so applications can render citations from
ToolReturnPart.metadata without parsing the text.
web_search asks Keenable for num_results results and trims each excerpt
to max_snippet_chars, and get_page truncates at max_page_chars with a
marker so the model knows the page continued; a budget too small to hold
the marker ends the page with … instead. All three bounds are validated
here as well as by KeenableSearch.
KeenableSearch passes operations so that each tool’s Keenable request
runs as one of its durable operations, whose result durable execution
records instead of making the request again on recovery.
def __init__(
*,
client: KeenableClient | None,
num_results: int,
max_snippet_chars: int,
max_page_chars: int,
id: str | None = None,
operations: KeenableSearchOperations | None = None,
) -> None
Configure the tools.
client : KeenableClient | None
The Keenable client the tools call. None uses an
HttpKeenableClient configured from the KEENABLE_API_KEY and
KEENABLE_API_URL environment variables.
num_results : int
Number of results web_search asks Keenable for, from
1 to 50.
max_snippet_chars : int
Maximum characters of excerpt web_search
returns per result.
max_page_chars : int
Maximum characters of page text get_page returns,
truncation marker included.
An optional unique ID for the toolset, which durable execution needs to identify it.
operations : KeenableSearchOperations | None Default: None
The durable operations each tool’s Keenable request
runs through. None makes the request directly.
ValueError— If a budget is not a positive integer, ornum_resultsis above 50.
@async
def web_search(query: str) -> ToolReturn[str]
Search the web and return matching pages, each with a short excerpt.
ToolReturn[str] — The matching pages, each with title, URL, and an excerpt.
query : str
The search query. Natural-language questions and keyword queries both work.
@async
def get_page(url: str) -> ToolReturn[str]
Retrieve one web page and return its content as markdown.
ToolReturn[str] — The page content as markdown, truncated to the configured budget.
url : str
The absolute URL of the page to read.
The default KeenableClient, talking to the Keenable API over httpx2.
Keenable needs no credentials: with no API key the public endpoints are used, which is why this capability works out of the box. An API key raises rate limits.
def __init__(
*,
api_key: str | None = None,
base_url: str | None = None,
http_client: httpx2.AsyncClient | None = None,
) -> None
Configure the client.
The Keenable API key. Defaults to the KEENABLE_API_KEY
environment variable; with neither, the keyless public
endpoints are used.
The API base URL. Defaults to the KEENABLE_API_URL
environment variable, then to the hosted API. It must be
https, or plain http against loopback, with no query,
fragment, or credentials.
http_client : httpx2.AsyncClient | None Default: None
The httpx2.AsyncClient to send every request through,
to reuse connections or to configure a proxy, TLS, or a retrying
transport. Its own timeouts apply, alongside the 30-second
deadline on each request. It stays open; closing it is up to
you. Defaults to a new client for each request, closed when the
request ends.
@async
def search(query: str, *, max_results: int) -> list[dict[str, object]]
Search the web via POST /v1/search, or /v1/search/public when keyless.
@async
def fetch(url: str) -> dict[str, object]
Retrieve a page via GET /v1/fetch, or /v1/fetch/public when keyless.
Bases: Protocol
The subset of the Keenable API that KeenableSearchToolset calls.
Any object with these two methods can back the toolset. Pass one via
KeenableSearch.client to configure authentication or the base URL
explicitly, or to substitute a fake in tests.
@async
def search(query: str, *, max_results: int) -> Sequence[Mapping[str, object]]
Search the web and return up to max_results raw result objects.
Sequence[Mapping[str, object]]
@async
def fetch(url: str) -> Mapping[str, object]
Retrieve one page and return the raw response payload.
Bases: TypedDict
One source behind a tool result, carried in ToolReturn.metadata['sources'].