Skip to content

Build a question-answering agent

An agent that answers questions about your data needs two things your application already governs: a way to read the data, and a way to reach the model. Written by hand, both turn into parallel machinery — a table of tool schemas mirroring your operations, and an outbound call nobody reviewed. This recipe assembles them out of what is already there: the agent-tool bridge for the first, a declared egress route for the second. The loop that joins them stays yours, and the answer can stream back over the same route.

The runnable version lives at examples/recipes/query_agent/ and runs fully in-process on the mock — the provider is answered by a registered handler, so there is no API key and no network.

The aggregate, and what may be queried

A Notes document aggregate, plus the one thing that makes it safe to point an agent at: a query policy. The read model has three fields; only two of them may be filtered, one may be sorted, and one may be grouped.

class Note(Document):
    title: str
    category: str
    body: str


class NoteRead(ReadDocument):
    title: str
    category: str
    body: str


class CreateNote(CreateDocumentCmd):
    title: str
    category: str
    body: str


NOTES = StrKeyNamespace(prefix="notes")

notes_spec = DocumentSpec(
    name="notes",
    read=NoteRead,
    write={"domain": Note, "create_cmd": CreateNote},
    # The allow-set the agent is told about and held to: `body` is readable but not a
    # field anything may filter, sort or group by.
    query_policy=QueryFieldPolicy(
        filterable={"title", "category"},
        sortable={"title"},
        aggregatable={"category"},
    ),
)


def notes_registry() -> FrozenOperationRegistry:
    return build_document_registry(notes_spec, ns=NOTES).freeze()

The policy is not decoration. It is the allow-set the list operation enforces, and — as the next section shows — the one the agent is told about.

The palette is the agent's whole world

operation_tools projects the operations you name into tool definitions. The allowlist is required, so an agent's capabilities are a decision in your wiring rather than a consequence of what happens to be registered:

def answering_tools(registry: FrozenOperationRegistry) -> OperationToolset:
    """The agent's whole world: list notes, and count them by category."""

    return operation_tools(
        registry,
        include=[
            NOTES.key(DocumentKernelOp.LIST),
            NOTES.key(DocumentKernelOp.AGG_LIST),
        ],
    )

Two read tools, and no write tool anywhere. An agent that only answers questions cannot create, update or delete a note — not because a check denies it, but because there is nothing in the palette to call.

Each definition carries a name (notes__list: providers refuse the . in notes.list), a JSON Schema taken from the operation's own input DTO, and a description. For a filter-accepting operation that description also spells out the filter surface:

List documents by filters and sorts (offset pagination). Filterable fields — $eq, $ilike, $in, $like, $neq, $nin, $null, $regex on every field; string: category, title. Sortable by: title. Aggregatable by: category.

The operators every field shares come first, then fields are grouped by the rest of their operator set, so each list is stated once, not once per field. Each field stays under its own type: number (also $gt, $gte, $lt, $lte): price; temporal (as number): created_at on a wider model.

That sentence is the read model's policy, not its field list — body is readable and absent from it. It is the same sentence the MCP surface hands an external agent, from the same source, so the two cannot tell one model two different stories. A model that reads it gets the field names and the operators right on its first attempt instead of learning them one refusal at a time.

The hop to the model is declared egress

The question and the rows retrieved for it leave your trust boundary when you call a provider. Declare that route as a service and say so:

class ModelArgs(BaseModel):
    question: str
    tools: list[dict[str, Any]]
    stream: bool = False


class ModelToolCall(BaseModel):
    id: str
    name: str
    input: dict[str, Any]


class ModelReply(BaseModel):
    text: str = ""
    tool_call: ModelToolCall | None = None


model_messages = HttpOperationSpec(
    name="messages",
    method="POST",
    path="/v1/messages",
    args_type=ModelArgs,
    return_type=ModelReply,
)

model_service = HttpServiceSpec(name="model", operations={"messages": model_messages})


def model_wiring() -> dict[str, HttpServiceConfig]:
    """How the provider hop is wired in a deployment that really calls one.

    The question and the rows retrieved for it leave the trust boundary on this route, so
    it says so — and saying so requires accepting it. Declare the first without the second
    and wiring fails closed with `http_egress_unacknowledged` rather than shipping the
    data and mentioning it in a comment.
    """

    return {
        "model": HttpServiceConfig(
            base_url="https://api.provider.example",
            egress_sensitive=True,
            acknowledge_data_egress=True,
        )
    }

Two fields, deliberately separate: egress_sensitive is a fact about the data the route carries, acknowledge_data_egress is an operator accepting it. Declare the first without the second and wiring fails closed with http_egress_unacknowledged rather than shipping the data and noting it in a comment. Calls through the route also tag their span forze.egress.sensitive, so sensitive egress is queryable in observability.

This is a governance marker and a conscious-choice gate — nothing inspects a payload. And it covers the hops you declare: a bare HttpClient handed to a provider's own SDK has no service config and therefore nothing to gate. Declaring the hop as a service is what puts it inside the marker, and the SDK can still make the call, through a transport over the service.

The example itself never reaches a transport — the handler below answers the service call in-process — so this config is what a deployment wires rather than something the run exercises. Its test constructs it, and constructs it without the acknowledgement to see the refusal, which is the part worth pinning.

The loop is yours

One turn: ask the model, run the tool it picked, hand back what came out.

async def answer(question: str, *, ctx: ExecutionContext, registry: FrozenOperationRegistry) -> str:
    """One turn: ask the model, run the tool it picked, hand back what came out.

    A production loop keeps going — feeding the result back, letting the model ask for
    another tool, stopping when it answers in prose. The shape does not change: the
    palette is projected once, and every call the model makes returns here.
    """

    tools = answering_tools(registry)
    # The operation, not its name: what comes back is `ModelReply` because that is the
    # `return_type` this spec declares. A name would type-check just as well and hand back a
    # bare `BaseModel`, leaving the fields below unreachable.
    reply = await ctx.http.service(model_service).invoke(
        model_messages,
        ModelArgs(
            question=question,
            tools=[
                {
                    "name": tool.name,
                    "description": tool.description,
                    "input_schema": tool.input_schema,
                }
                for tool in tools.defs
            ],
        ),
    )

    if reply.tool_call is None:
        return reply.text

    result = await dispatch_tool_use(
        ToolUse(
            id=reply.tool_call.id,
            name=reply.tool_call.name,
            input=reply.tool_call.input,
        ),
        ctx=ctx,
        tools=tools,
    )

    return _render(result)


def _render(result: ToolResult) -> str:
    """What the next turn would be told — the rows, or the error the model can act on."""

    if result.is_error:
        return f"tool refused: {result.content}"

    hits: list[Any] = result.content["hits"] if isinstance(result.content, dict) else []

    return "; ".join(str(hit) for hit in hits)

A production loop keeps going — feed the result back, let the model ask for another tool, stop when it answers in prose. Nothing about that shape is the framework's business, and the bridge does not try to own it: it dispatches one call at a time and returns.

invoke is handed the operation rather than its name, and that is what makes reply a ModelReply here: the declaration carries args_type and return_type, so both reach your own variables. Naming the operation "messages" works too and hands back a bare BaseModel — the right form when the operation comes from config and there is no declaration to point at, and the wrong one when you meant to read a field off the answer.

dispatch_tool_use runs the call through run_operation on the context you pass, so tenancy, permissions, the deadline and audit apply exactly as they would to an HTTP request for the same operation. A governed failure — a denied permission, a field outside the policy — comes back as is_error with the code, which is something the model can act on. An infrastructure failure propagates to your loop instead, because retrying a dropped connection is not the model's job.

The answer streams over the same hop

A model that answers in prose usually streams it, a token at a time. stream takes the operation invoke takes and yields the response body as it arrives, so the streamed answer leaves through the declared route like everything else:

async def stream_answer(question: str, *, ctx: ExecutionContext) -> AsyncGenerator[str]:
    """The model's prose answer as it is generated, over the same declared route.

    `stream` sends the request `invoke` would send and yields the body as it arrives. The
    events are lines and a network splits them wherever it likes, so lines are cut from a
    buffer rather than from each chunk.
    """

    chunks = ctx.http.service(model_service).stream(
        model_messages,
        ModelArgs(question=question, tools=[], stream=True),
    )
    buffer = b""

    async with aclosing(chunks):
        async for chunk in chunks:
            *lines, buffer = (buffer + chunk).split(b"\n")

            for raw in lines:
                line = raw.removesuffix(b"\r")  # a provider may end lines with CRLF

                if not line.startswith(b"data:"):
                    continue

                # The space after the colon is optional, and only one is stripped.
                data = line.removeprefix(b"data:").removeprefix(b" ")

                if data != b"[DONE]":
                    yield data.decode()

The request is the one invoke sends: the acknowledgement gates it, the credential and the deadline budget ride on it as it opens, and its span carries forze.egress.sensitive. A rejected request raises before the first token, with the error invoke would raise. Server-sent events are lines, and a network cuts chunks wherever it likes, so lines come out of a buffer rather than out of each chunk. aclosing closes the connection as soon as you stop reading — on an early return as much as at the end.

What stands in for the provider

The example registers a handler for the model service. It keyword-matches the question, and answers a streaming request with events cut mid-line:

def canned_model(args: ModelArgs | None) -> ModelReply:
    """Stands in for the provider: picks a tool from the palette it was given.

    A real model reads the descriptions — which carry the filterable fields and their
    operators, so it does not have to guess — and answers with a tool call in its own
    format, which the loop adapts to a `ToolUse`. This one keyword-matches, so the example
    is deterministic.
    """

    question = args.question if args is not None else ""
    names = [tool["name"] for tool in (args.tools if args is not None else [])]

    if "how many" in question.lower():
        return ModelReply(
            tool_call=ModelToolCall(
                id="call-1",
                name=next(name for name in names if name.endswith("__agg_list")),
                input={
                    "aggregates": {
                        "$groups": {"category": "category"},
                        "$computed": {"n": {"$count": None}},
                    }
                },
            )
        )

    return ModelReply(
        tool_call=ModelToolCall(
            id="call-2",
            name=next(name for name in names if name.endswith("__list")),
            input={"filters": {"$values": {"category": "postgres"}}, "sorts": {"title": "asc"}},
        )
    )


def canned_stream() -> list[bytes]:
    """Stands in for the provider's streamed answer: server-sent events, a token per event.

    The second chunk starts mid-event, as a network is free to cut it.
    """

    return [
        b"data: Two\n\ndata:  no",
        b"tes\n\ndata:  mention postgres.\n\ndata: [DONE]\n\n",
    ]

Only the provider's answer is stubbed. The palette it chooses from, the dispatch its choice triggers, and the governance around both are the real thing — which is what lets a test drive the whole recipe:

uv run python -m examples.recipes.query_agent.app
uv run pytest tests/unit/test_examples/test_query_agent.py

Swapping in a real provider means pointing the service at the provider's real base URL. Then either describe its endpoint as an HttpOperationSpec, writing the request and response models and parsing the provider's events yourself, or keep the provider's SDK. To keep the SDK, give its httpx.AsyncClient the transport from http_service_transport(ctx, spec) — see Hand a vendor SDK the service. The SDK's requests then carry the same acknowledgement, egress tag, credential and deadline budget, and a request that leaves the service's base URL is refused. The loop, the palette and the egress declaration stay as they are either way.

Notes

  • Without a query policy every read field is advertised, bookkeeping columns included — around 600 characters for a two-field model, in every tool the palette carries. A policy keeps the description small as well as the surface narrow, which is the same reason to write one twice over. To narrow only the palette and leave HTTP the spec's allow-set, pass query_policies to operation_tools — see Agent tools.

  • The filter grammar is repeated in every filter tool's schema unless the palette is built with shared_filter_grammar=True, which moves it to tools.filter_grammar for the system prompt — see Agent tools.

  • A read-only palette is the default. operation_tools refuses a command operation unless you ask for a command-capable toolset, so an agent that must act is a visible decision in the diff.

  • Delivering the streamed answer to a user is a transport concern, not an agent one. The realtime recipes — delivery to offline users, tenant-sharded realtime — apply unchanged to the tokens stream_answer yields.
  • An operation whose read model is marked sensitive cannot be projected at all, so credential material has no path into a tool palette.