Skip to content

Web frontends (AI SDK and AG-UI)

pyox-agents speaks the stream formats of two frontend toolkits, so a chat UI built with them can talk to your agent directly:

Frontend Python module Stream
Vercel AI SDK 6 and 7 (useChat, assistant-ui) pyox_agents.ai_sdk run.ai_sdk()
AG-UI 1.0 (CopilotKit, HttpAgent) pyox_agents.ag_ui run.ag_ui(thread_id, run_id)

The streams are made in Rust, straight from the run's events: Python only passes the finished text to your web server. The modules read the requests in Python, where they are easy to change.

A server for both

--8 < --"examples/web_server.py"

stream(agent, body, **options) reads the request, runs the agent and returns the stream. options go to agent.stream (deps=..., text_chunks=...).

Use stream(), not run()

Both streams come from a streamed run: agent.stream(...) or agent.stream_sync(...). await agent.run(...) returns only the finished RunFinished, which has no ai_sdk() or ag_ui().

run = agent.stream(prompt)                    # yes
return StreamingResponse(run.ai_sdk(), headers=ai_sdk.HEADERS)

result = await agent.run(prompt)              # no: AttributeError
result.ai_sdk()

To send the whole answer in one response instead of streaming it, join the stream: "".join(agent.stream_sync(prompt).ai_sdk()).

The pieces, for more control

stream is three steps. Do them yourself to check who is asking, read your own fields from the request, or keep the conversation on the server:

@app.post("/api/chat")
async def chat(request: Request, user: User = Depends(current_user)) -> StreamingResponse:
    chat = ai_sdk.parse(await request.body())  # chat.prompt, chat.history, chat.answers, chat.body
    run = agent.stream(chat.prompt, conversation_id=f"{user.id}:{chat.chat_id}", answers=chat.answers, deps=user)
    return StreamingResponse(run.ai_sdk(), headers=ai_sdk.HEADERS)

ag_ui.parse returns the same fields (prompt, history, answers, frontend_tools), plus thread_id, run_id, state, context and forwarded_props.

Where the conversation lives. By default it comes from the browser with each request (history). With conversation_id=, the agent's history store keeps it instead and the browser's copy is ignored: only the new prompt and the answers are used. Keep it on the server when you can't trust the browser with it, and for tools that ask more than once or use ctx.step (their earlier answers are not in the browser's copy).

Tools that wait for a person

Approvals, questions and frontend tools use each toolkit's own way of waiting:

Waiting call In the stream The browser replies with
approval tool-approval-request (approval id = call id) and a data-pyox-request part addToolApprovalResponse({id, approved, reason})
ctx.ask question a data-pyox-request part addToolOutput({toolCallId, output})
frontend tool a data-pyox-request part addToolOutput({toolCallId, output}) (or onToolCall)

Every waiting call gets a data-pyox-request part. It has the call's id as its id, and {toolCallId, toolName, waiting: true, kind, key, data, metadata} as its data: data is what the tool asked, metadata what it left for your UI. When the call goes on, the part is replaced with waiting: false. Send the reply with sendAutomaticallyWhen: lastAssistantMessageIsCompleteWithApprovalResponses (or your own check), and the next request carries it.

Waiting call In the stream The client replies with
approval RUN_FINISHED outcome interrupt, reason approval, with a responseSchema resume payload {approved, editedArgs?, reason?}
ctx.ask question RUN_FINISHED outcome interrupt, reason input, message = the question resume payload: the answer
frontend tool (the request's tools) RUN_FINISHED outcome success, with pendingToolCallIds a tool message with the result

Each interrupt's id is the tool call's id. A cancelled resume denies the call.

Results of frontend tools are not sent back to the frontend: it made them.

Shared state and subagents

The run's state is the client's: ag_ui.stream and ai_sdk.stream run the agent with it, and each change a tool makes is sent at once. Subagents stream inside the tool call that runs them.

Run event In the stream
StateChanged a data-state part (id state) with the whole state; it replaces the last one
a subagent's work a data-subagent part (id: the tool call's id), sent once per batch of events

The data-state part stays in the message and comes back with the next request: ai_sdk.parse reads the newest one as ChatRequest.state. The data-subagent part's data is {toolCallId, name, task, status, result, parts}: status is running, waiting, done or error, and parts are the subagent's own text and dynamic-tool parts, ready for your message renderer.

When a subagent's tool waits for the user, the subagent's tool call waits: its data-pyox-request has the request's own id (call_0/call_1) in requestId, and an approval's approvalId is that id. Answer it with addToolApprovalResponse or addToolOutput on the subagent's tool call, as usual.

Run event In the stream
StateChanged STATE_DELTA: a JSON Patch on the state the client sent
SubagentStarted SUBAGENT_STARTED (parentToolCallId: the tool call running it)
a subagent's events the usual events, with subagentRunId
SubagentFinished SUBAGENT_FINISHED (outcome suspended while it waits), or SUBAGENT_ERROR

A subagent's approvals and questions are interrupts of the run with subagentRunId, resumed like any other. ag_ui.parse leaves the subagent's messages (the ones with subagentRunId) out of the history.

A subagent that waits for the user needs the conversation on the server (conversation_id=): its own conversation is kept in the history, and the browser's copy does not have it.

Errors

A failed run (the API is down, too many turns) ends the stream with the toolkit's error: an error chunk for the AI SDK, RUN_ERROR (with the error's class name as code) for AG-UI. Other exceptions (bugs) are raised as usual.

Not supported (yet)

  • AG-UI reasoning events.
  • AI SDK 5 and older: use AI SDK 6 or newer.
  • File parts must be data: or https:// URLs, so a browser can't make your server read a local path.