Tutorial: a research agent¶

Available · pip install agent-adapter (no extra) · about 15 minutes

You build a small research agent end to end: tools from plain functions, retrieval from three sources at once, a model loop, a human approval before it publishes, and a cost report. Everything runs offline: a scripted function stands in for the model.

File	Does
`examples/tutorial_research.py`	the whole agent; `python examples/tutorial_research.py` prints each step
`tests/docs/test_tutorial_examples.py`	runs it and checks the output

flowchart LR
  Q["question"] --> L["run_tool_loop"]
  L --> M["model (scripted)"]
  M -- "3 tool calls" --> P["run_many: web, papers, news<br/>at once"]
  P --> M
  M -- "publish_report" --> A{"approval<br/>(DeferredApproval)"}
  A -- "pending" --> R["reviewer decides<br/>(store.decide)"]
  R --> L2["run_tool_loop again<br/>same ctx, paused messages"]
  L2 --> D["answer"] --> C["ledger.summary"]

1. Tools from functions¶

@registry.tool(kind="retrieval", parallel_safe=True, timeout=10, retries=2)
async def search_web(query: str, limit: int = 3) -> list[str]:
    """Search the web for a query."""
    return [f"web: {query} #{n}" for n in range(1, limit + 1)]


@registry.tool(kind="retrieval", parallel_safe=True, timeout=10, retries=2)
async def search_papers(query: str) -> list[str]:
    """Search papers for a query."""
    return [f"paper: {query} survey"]


def legacy_news_lookup(topic: str, region: str = "us") -> list[str]:
    """Existing code you do not want to edit: news headlines for a topic."""
    return [f"news ({region}): {topic} rules change"]


registry.wrap(legacy_news_lookup, name="search_news", kind="retrieval", parallel_safe=True)


@registry.tool(confirm=True, effects="additive")
def publish_report(title: str, body: str) -> str:
    """Publish a report to the team wiki (needs approval)."""
    PUBLISHED.append(title)
    return f"published {title!r} ({len(body)} chars)"

Line	Means
`@registry.tool(...)`	the signature and docstring become the schema the model sees
`parallel_safe=True`	calls may share the run; without it a call runs alone
`timeout=10, retries=2`	each attempt is cut at 10 s; failures retry with backoff
`registry.wrap(legacy_news_lookup, name=...)`	existing code becomes a tool without edits
`confirm=True`	the executor asks the approval port before every call

2. One executor¶

store, ledger = MemoryApprovalStore(), Ledger(prices=PRICES)
executor = Executor(registry, approval=DeferredApproval(store), ledger=ledger)
ctx = executor.context(run_id="research-1")

The executor runs every call: validation, approval, timeout, retries, ledger. The store holds approvals until someone decides. The context names the run (research-1): guards, approvals and the cost report key on it.

3. Parallel retrieval¶

def retrieval_calls(query: str, prefix: str = "") -> list[ToolCall]:
    """One call per source; call ids must be unique within a run."""
    return [
        ToolCall(name="search_web", arguments={"query": query, "limit": 2}, id=f"{prefix}web"),
        ToolCall(name="search_papers", arguments={"query": query}, id=f"{prefix}papers"),
        ToolCall(name="search_news", arguments={"topic": query}, id=f"{prefix}news"),
    ]


async def retrieve(executor: Executor, query: str, ctx: ToolContext) -> list[str]:
    """Run every source at once; a failed source is skipped, not fatal."""
    results = await executor.run_many(retrieval_calls(query), ctx, concurrency=3)
    return [hit for result in results if result.ok for hit in result.value]

run_many starts the three searches together (up to concurrency) and returns ToolResults in input order. A failed source is a result with ok=False, not an exception, so one dead source does not sink the answer.

Output: ['web: EV battery #1', 'web: EV battery #2', 'paper: EV battery survey', 'news (us): EV battery rules change']

4. The model loop¶

async def fake_model(request: LLMRequest) -> dict[str, Any]:
    """Stands in for a real model: search, then publish, then answer."""
    last = request.messages[-1]
    usage = Usage(input_tokens=2_000, output_tokens=200)
    if last.role == "user":
        calls = tuple(retrieval_calls(last.content, prefix="loop-"))
        return {"tool_calls": calls, "finish_reason": "tool_calls", "usage": usage}
    if last.name != "publish_report":
        hits = [m.content for m in request.messages if m.role == "tool"]
        args = {"title": "EV battery", "body": "\n".join(hits)}
        call = ToolCall(name="publish_report", arguments=args, id="publish")
        return {"tool_calls": (call,), "finish_reason": "tool_calls", "usage": usage}
    return {"text": f"Done: {last.content}", "usage": usage}


MODEL = CallableModel(fake_model, provider="host", model="research-v1")
PRICES = Prices.load(
    {
        "version": "demo",
        "models": {"research-v1": {"provider": "host", "input": 1.0, "output": 4.0}},
    }
)

Swap CallableModel(fake_model, ...) for OpenAIModel, AnthropicModel or GeminiModel from agent_adapter.models; nothing else changes. The price table makes the ledger price each model call.

question = (Message(role="user", content="EV battery"),)
paused = await run_tool_loop(MODEL, registry, question, executor=executor, ctx=ctx)
print(paused.finish, PUBLISHED)  # the model asked to publish: the call waits

The loop calls the model, runs its tool calls (the three searches at once), feeds the results back, and stops when the model answers or a call waits for approval. Here it asks to publish, so finish is pending and nothing was published.

5. Approve, edit, resume¶

[waiting] = await store.list(run_id="research-1", status="pending")
print(waiting.request.tool, sorted(waiting.request.arguments))  # what a reviewer sees
edit = {**waiting.request.arguments, "title": "EV battery (draft)"}
await store.decide(waiting.approval_id, Decision(approved=True, edited_args=edit))
done = await run_tool_loop(MODEL, registry, paused.messages, executor=executor, ctx=ctx)
print(done.finish, done.final.text, PUBLISHED)  # ran once, with the edited title

Step	What happens
`store.list(status="pending")`	the reviewer sees the tool and its arguments
`store.decide(id, Decision(approved=True, edited_args=...))`	approve with a new title; edits are validated before the tool runs
`run_tool_loop(..., paused.messages, ctx=ctx)`	the same run resumes: the waiting call runs first, once, then the model answers

Resuming from paused.messages a second time does not publish again: the decision was used once, so the call waits for a new one. Decision(approved=False) turns the call into an approval-denied error the model sees.

6. Cost report¶

summary = ledger.summary("research-1")
print(f"${summary.total_usd:.4f}", {kind: s.calls for kind, s in summary.by_kind.items()})
print(json.dumps({stage: round(s.cost_usd, 4) for stage, s in summary.by_stage.items()}))

Output: $0.0084 {'tool': 7, 'llm': 3}. Three model calls at $0.0028 each. Seven tool calls: three searches in step 3, three in the loop, one publish. By stage, the model's cost sits under tool_loop and each tool under its own name.

Next steps¶

To	Read
cap spend per run	`BudgetHook(max_usd)` in Hooks
serve the agent over HTTP	FastAPI or HTTP server
a fixed pipeline instead of a model loop	Flows
more on approval ports and stores	Approval
prompt caching, response caching	Cache