OpenAI Agents SDK

Nothing here depends on which model you run — unlob is a search API, and the Agents SDK supports several providers. These examples use the SDK’s own conventions.

As an MCP server

import os
from agents import Agent, Runner
from agents.mcp import MCPServerStreamableHttp

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

agent = Agent(
    name="Researcher",
    instructions=(
        "Search with unlob. Use assemble_context for open questions and web_search "
        "for specific ones. Call corroborate before stating anything as established "
        "fact, and report how many independent sources carry it."
    ),
    mcp_servers=[unlob],
)

async with unlob:
    result = await Runner.run(agent, "What is corroborated about the Acme acquisition?")
    print(result.final_output)

The async with matters: it opens the connection once for the run rather than per call, and closes it afterwards.

tool_filter is worth using. Eleven tools is a lot of schema, and an agent that only researches needs four:

from agents.mcp import create_static_tool_filter

unlob = MCPServerStreamableHttp(
    name="unlob",
    params={...},
    tool_filter=create_static_tool_filter(
        allowed_tool_names=["web_search", "corroborate", "dossier", "assemble_context"],
    ),
)

As a function tool

import os, httpx
from agents import Agent, function_tool

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

@function_tool
async def web_search(query: str, vertical: str | None = None, limit: int = 5) -> str:
    """Search the web for passages. Returns passage text, not links.

    Args:
        query: Supports AND, OR, -exclude and "exact phrase".
        vertical: e.g. code, science. Omit to let the router choose.
        limit: Maximum results, 1-20.
    """
    params = {
        "q": query,
        "limit": min(limit, 20),
        # Applied here rather than left to the model: drop single-source claims,
        # and fold near-duplicates.
        "min_independent_sources": 2,
        "collapse": "story",
    }
    if vertical:
        params["vertical"] = vertical

    r = await _client.get("/search", params=params)
    if r.status_code == 429 and "retry-after" not in r.headers:
        return "Monthly quota exhausted. Do not retry; tell the user."
    r.raise_for_status()
    body = r.json()

    header = "WARNING: incomplete — part of the corpus was unreachable.\n\n" if body.get("partial") else ""
    return header + "\n\n".join(
        f"{h['title']} — {h['url']} ({h.get('independent_sources', 0)} independent sources)\n{h['snippet']}"
        for h in body["results"]
    ) or "No results found."

agent = Agent(name="Researcher", tools=[web_search])

The docstring is the tool description the model reads, so the second sentence — “returns passage text, not links” — earns its place: it stops the agent following up with a fetch it does not need.

Next

• MCP tools
• Search recipes