unlob Docs
Console
Browse documentation

Pydantic AI

A typed toolset from the MCP server, or a typed tool of your own.

As an MCP server

import os
from pydantic_ai import Agent
from pydantic_ai.mcp import MCPServerStreamableHTTP

unlob = MCPServerStreamableHTTP(
    "https://api.unlob.com/mcp",
    headers={"x-api-key": os.environ["UNLOB_API_KEY"]},
)

agent = Agent(
    "anthropic:claude-opus-5",
    toolsets=[unlob],
    system_prompt=(
        "Search with unlob. Use assemble_context for open questions and web_search "
        "for specific ones. Call corroborate before stating anything as established."
    ),
)

async def main():
    async with agent:
        result = await agent.run("What is corroborated about the Acme acquisition?")
        print(result.output)

async with agent opens the MCP connection for the run and closes it after. Without it, each call reconnects.

A typed tool

Pydantic AI’s argument model becomes the tool schema, which makes constraints — enums, ranges — part of what the model is told rather than something you validate afterwards.

import os, httpx
from typing import Literal
from pydantic import BaseModel, Field
from pydantic_ai import Agent, RunContext

class Hit(BaseModel):
    title: str
    url: str
    text: str
    independent_sources: int = Field(
        0, description="Distinct sites asserting this. 1 means a single source."
    )

class SearchResult(BaseModel):
    incomplete: bool = Field(
        False,
        description="True when part of the corpus was unreachable. The answer is "
                    "short because of a failure, not because there is nothing more.",
    )
    hits: list[Hit]

_client = httpx.AsyncClient(
    base_url="https://api.unlob.com",
    headers={"x-api-key": os.environ["UNLOB_API_KEY"]},
    timeout=20,
)

agent = Agent("anthropic:claude-opus-5")

@agent.tool
async def web_search(
    ctx: RunContext[None],
    query: str,
    mode: Literal["keyword", "semantic", "hybrid"] = "hybrid",
    limit: int = 5,
) -> SearchResult:
    """Search the web for passages. Returns passage text, not links."""
    r = await _client.get("/search", params={
        "q": query,
        "mode": mode,
        "limit": min(limit, 20),
        "min_independent_sources": 2,
        "collapse": "story",
    })
    r.raise_for_status()
    body = r.json()
    return SearchResult(
        incomplete=bool(body.get("partial")),
        hits=[
            Hit(
                title=h["title"],
                url=h["url"],
                text=h["snippet"],
                independent_sources=h.get("independent_sources", 0),
            )
            for h in body["results"]
        ],
    )

Putting incomplete in the return model rather than raising is deliberate: partial is a 200, and an exception would hide a result that is real but short.

Structured research output

Where Pydantic AI earns its keep is the output type. Ask for citations as a schema and you get them as a schema:

class Finding(BaseModel):
    claim: str
    sources: list[str] = Field(description="URLs supporting this claim")
    independently_corroborated: bool

class Research(BaseModel):
    summary: str
    findings: list[Finding]

agent = Agent("anthropic:claude-opus-5", output_type=Research, toolsets=[unlob])

independently_corroborated maps directly onto what corroborate returns, so the model has something real to fill it from rather than a guess.

Next