Skip to content
unlob

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

Your side of the API. unlob is one HTTPS service among the others — nothing here is specific to how our index works, so this agent survives a change of framework.

Install

Dependenciesbash
pip install pydantic-ai httpx

Step by step

  1. 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",
        )
  2. 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"]
  3. 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

YouAgentClaudeunlobrun(question, deps=deps)prompt + output schematool callGET /searchhits + independent_sourcestool resultstructured outputvalidate against Researchresult.output
One turn of the loop. Search returns metadata only, so the agent spends a few hundred tokens deciding rather than tens of thousands reading.

The whole thing

Complete and runnable. Set the two environment variables and it works.

research_agent.py — completepython
"""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.

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.

relatedNear-duplicatescorroborateIndependent sourcesauthoritiesAuthoritative hostsdossierThe entity briefpathA second documentassemble_contextA packed context setA search resultwhat your agent has
The two marked in blue are the tools this tutorial wires up. The other four are the same shape.

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.