Pydantic · Python
Build a research agent with Pydantic AI
Build a research agent on Pydantic AI with typed dependencies, validated structured output and unlob as a tool.
What you are building
Your process
Agent
Typed deps and output
Research model
Validated, not parsed
Tools
web_search
Gets the client from RunContext
assemble_context
Same injected client
Services
Anthropic API
claude-opus-5
unlob
Your own index
Install
pip install pydantic-ai httpxStep by step
Model the answer you want
Start from the output type. Everything else follows from it, and the model is constrained to produce something that validates — which removes an entire category of downstream parsing bugs.
models.pypython from pydantic import BaseModel, Field class Finding(BaseModel): claim: str url: str independent_sources: int = Field( description="How many separately-owned hosts carry this claim" ) class Research(BaseModel): answer: str findings: list[Finding] caveats: list[str] = Field( default_factory=list, description="Claims resting on a single source, or gaps in coverage", )Inject the client, register the tool
deps_type makes the httpx client an explicit dependency. In a test you pass a stub; in production you pass a real client with your timeouts. No global state, no monkeypatching.
agent.pypython from dataclasses import dataclass import httpx from pydantic_ai import Agent, RunContext @dataclass class Deps: client: httpx.AsyncClient api_key: str agent = Agent( "anthropic:claude-opus-5", deps_type=Deps, output_type=Research, system_prompt=( "Search before you answer. Put any claim resting on a single source " "in caveats rather than stating it plainly." ), ) @agent.tool async def web_search(ctx: RunContext[Deps], query: str, min_sources: int = 2) -> list[dict]: """Search the web. Returns metadata only, never page bodies.""" r = await ctx.deps.client.get( "https://api.unlob.com/search", headers={"x-api-key": ctx.deps.api_key}, params={"q": query, "min_independent_sources": min_sources, "limit": 8}, ) r.raise_for_status() return r.json()["results"]Test it without the network
This is the payoff for the dependency injection. Because the HTTP client arrives through deps rather than being imported, a test hands the agent a stub transport and asserts on the validated output — no network, no model spend, no mocking library.
test_agent.pypython import httpx, pytest from pydantic_ai.models.test import TestModel from agent import agent, Deps, Research @pytest.mark.asyncio async def test_flags_single_source_claims(): # A stub transport: the agent never touches the network. def handler(request: httpx.Request) -> httpx.Response: return httpx.Response(200, json={"results": [ {"url": "https://example.com/a", "title": "A", "snippet": "...", "independent_sources": 1}, ]}) transport = httpx.MockTransport(handler) async with httpx.AsyncClient(transport=transport) as client: deps = Deps(client=client, api_key="test") # TestModel exercises the tools and the output schema without # calling a real provider. with agent.override(model=TestModel()): result = await agent.run("anything", deps=deps) assert isinstance(result.output, Research)
What happens when it runs
The whole thing
Complete and runnable. Set the two environment variables and it works.
"""A research agent on Pydantic AI, backed by the unlob web index.
pip install pydantic-ai httpx
export UNLOB_API_KEY=ulb_...
export ANTHROPIC_API_KEY=sk-ant-...
"""
import asyncio
import os
from dataclasses import dataclass
import httpx
from pydantic import BaseModel, Field
from pydantic_ai import Agent, RunContext
class Finding(BaseModel):
claim: str
url: str
independent_sources: int = Field(
description="How many separately-owned hosts carry this claim"
)
class Research(BaseModel):
answer: str
findings: list[Finding]
caveats: list[str] = Field(
default_factory=list,
description="Claims resting on a single source, or gaps in coverage",
)
@dataclass
class Deps:
client: httpx.AsyncClient
api_key: str
agent = Agent(
"anthropic:claude-opus-5",
deps_type=Deps,
output_type=Research,
system_prompt=(
"Search before you answer. Put any claim resting on a single source "
"in caveats rather than stating it plainly."
),
)
@agent.tool
async def web_search(
ctx: RunContext[Deps], query: str, min_sources: int = 2
) -> list[dict]:
"""Search the web. Returns metadata only, never page bodies."""
r = await ctx.deps.client.get(
"https://api.unlob.com/search",
headers={"x-api-key": ctx.deps.api_key},
params={"q": query, "min_independent_sources": min_sources, "limit": 8},
)
r.raise_for_status()
return r.json()["results"]
@agent.tool
async def assemble_context(
ctx: RunContext[Deps], query: str, budget: int = 3000
) -> dict:
"""Build a corroborated context pack packed to a token budget."""
r = await ctx.deps.client.get(
"https://api.unlob.com/assemble_context",
headers={"x-api-key": ctx.deps.api_key},
params={"query": query, "budget": budget, "min_independent_sources": 2},
)
r.raise_for_status()
return r.json()
async def main() -> None:
async with httpx.AsyncClient(timeout=30) as client:
deps = Deps(client=client, api_key=os.environ["UNLOB_API_KEY"])
result = await agent.run(
"What changed in the EU AI Act GPAI rules in 2026?", deps=deps
)
print(result.output.answer)
for f in result.output.findings:
print(f" [{f.independent_sources}] {f.url}")
for c in result.output.caveats:
print(f" caveat: {c}")
if __name__ == "__main__":
asyncio.run(main())What will go wrong
The failures that cost an afternoon rather than a minute, because they produce something that looks like it is working.
Validation failures are retried, not raised. The model gets the error and another attempt, which is usually what you want — but a badly-specified output type turns into a quiet retry loop that costs real money. Watch the retry count.
Field descriptions are prompt. `independent_sources` gets populated correctly because the Field description explains what it means; drop it and the model invents a number.
deps_type must match what you pass to run(). The mismatch is caught by your type checker and not at runtime, so run mypy or pyright — that is most of the point of using this framework.
Use the provider-prefixed model string (`anthropic:claude-opus-5`). A bare model name resolves against the default provider and produces a confusing auth error.
Where this agent can go next
Every result the agent receives is a node in the coverage graph. These are the tools you can add to it without leaving Pydantic AI — each is one more entry in the same tools list.
Frequently asked questions
Does structured output make the agent less capable?
It constrains the shape, not the reasoning. The trade worth knowing about is that a rigid schema on a genuinely open-ended question forces the model to fit an answer into boxes that do not suit it — which is why the `caveats` list exists in the model above.
How do I test an agent built this way?
Pass a stub client through deps. Because the HTTP client is injected rather than imported, a test can return fixed search results and assert on the structured output without touching the network or the model provider.
You need a key to run this
10,000 requests a month on the free tier, no card. Enough to build the agent and put a real evaluation set through it.