unlob Docs
Browse documentation

Tool use, directly

Hand the tools to the Messages API yourself, when you want to see every call.

When to do it this way

The MCP connector is less code. Define the tools yourself when you need to inspect or filter a result before Claude sees it, add your own caching or retry policy, run a tool conditionally on something outside the conversation, or log every call for audit.

Generate the tool definitions

Do not hand-write them. The tool definitions the MCP server advertises are the tool definitions the Messages API wants, minus one field name:

import httpx

def unlob_tools() -> list[dict]:
    """Fetch unlob's tool definitions and shape them for the Messages API.

    Needs no key — the descriptions of the API are open, so an agent can learn what
    the service does before it has one.
    """
    spec = httpx.get("https://api.unlob.com/mcp/tools.json", timeout=10).json()
    return [
        {
            "name": t["name"],
            "description": t["description"],
            "input_schema": t["inputSchema"],
        }
        for t in spec["tools"]
    ]

Fetch once at startup and cache it. The descriptions are written for a model to read — they say when to reach for each tool, not just what it takes — so passing them through verbatim is doing real work for you.

Trim to what your agent needs. Eleven tools is a lot of schema in every request:

KEEP = {"web_search", "corroborate", "assemble_context", "get_document"}
tools = [t for t in unlob_tools() if t["name"] in KEEP]

Execute them

Every tool maps to a GET with the same parameter names, so one dispatcher covers all eleven:

import os, httpx

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

# The two tools whose REST path is not just /<tool name>.
_PATHS = {"web_search": "/search", "get_document": "/doc"}

def call_unlob(name: str, args: dict) -> str:
    if name == "get_document":
        r = _api.get(f"/doc/{args['id']}")
    else:
        params = {("q" if k == "query" else k): v for k, v in args.items()}
        # A URL query string cannot carry an array; the MCP schema allows both.
        params = {k: ",".join(v) if isinstance(v, list) else v for k, v in params.items()}
        r = _api.get(_PATHS.get(name, f"/{name}"), params=params)

    if r.status_code == 429 and "retry-after" not in r.headers:
        # Hard-capped monthly quota. Say so plainly — a model told only "429" will
        # sensibly retry, and retrying will not help until the period rolls over.
        return "unlob: monthly quota exhausted. Do not retry; tell the user."
    r.raise_for_status()
    return r.text

The loop

The tool runner drives it for you:

import anthropic

client = anthropic.Anthropic()

runner = client.beta.messages.tool_runner(
    model="claude-opus-5",
    max_tokens=16000,
    thinking={"type": "adaptive"},
    tools=tools,
    messages=[{"role": "user", "content": question}],
)

Or write it yourself, which is worth doing if the point was to see every call:

messages = [{"role": "user", "content": question}]

while True:
    response = client.messages.create(
        model="claude-opus-5",
        max_tokens=16000,
        thinking={"type": "adaptive"},
        tools=tools,
        messages=messages,
    )
    messages.append({"role": "assistant", "content": response.content})

    if response.stop_reason != "tool_use":
        break

    # Claude may emit several tool_use blocks in one message. Return ALL of the
    # results in a SINGLE user message — splitting them across messages teaches
    # Claude to stop making parallel calls.
    results = []
    for block in response.content:
        if block.type != "tool_use":
            continue
        try:
            output = call_unlob(block.name, block.input)
            results.append({
                "type": "tool_result",
                "tool_use_id": block.id,
                "content": output,
            })
        except Exception as e:
            # A failed tool still needs a result block. Dropping it leaves the
            # turn malformed.
            results.append({
                "type": "tool_result",
                "tool_use_id": block.id,
                "content": str(e),
                "is_error": True,
            })
    messages.append({"role": "user", "content": results})

TypeScript

import Anthropic from "@anthropic-ai/sdk";

const client = new Anthropic();

const spec = await (await fetch("https://api.unlob.com/mcp/tools.json")).json();
const tools: Anthropic.Tool[] = spec.tools.map((t: any) => ({
  name: t.name,
  description: t.description,
  input_schema: t.inputSchema,
}));

const response = await client.messages.create({
  model: "claude-opus-5",
  max_tokens: 16000,
  thinking: { type: "adaptive" },
  tools,
  messages: [{ role: "user", content: question }],
});

The one filter worth setting by default

params.setdefault("min_independent_sources", 2)
params.setdefault("collapse", "story")

Applied in your dispatcher rather than left to the model, these two do most of the work of keeping a context window honest: the first drops single-source claims, the second stops the same syndicated article arriving five times. Neither costs an extra request.

Set them as defaults the model can override, not as hard-coded values — sometimes a single-source document is exactly what you are looking for.

Handling partial

body = r.json()
if body.get("partial"):
    return "WARNING: incomplete — part of the corpus was unreachable.\n" + r.text

A model handed a short result with no explanation will report it as “nothing found”. This is the one place where a line of your own text in the tool result is worth more than the raw JSON.

Next