`agentix.serving`¶

serving ¶

Serving helpers — turn an Agent into a streaming HTTP endpoint.

Agent.stream() already yields events (answer chunks, tool activity, a final Done). To put an agent behind a web endpoint, you just need to turn those events into a wire format the browser can read incrementally. This module does exactly that, in two layers:

Dependency-free serializers — :func:event_to_dict, :func:sse_events, and :func:ndjson_events convert the stream into Server-Sent Events (the browser EventSource format) or newline-delimited JSON. No web framework required, so they work with any ASGI/WSGI stack.
A thin FastAPI/Starlette adapter — :func:sse_response / :func:ndjson_response wrap those in a StreamingResponse with the right headers. The web dependency is imported lazily, so importing agentix never requires it; install it with pip install "agentix[serving]".

For the request/response (non-streaming) side — including a run that suspends for human approval — :func:outcome_to_payload serializes an :class:~agentix.types.AgentOutcome (with any pending approvals) to a plain dict you can return as JSON. See examples/30_serving_fastapi.py.

outcome_to_payload ¶

outcome_to_payload(outcome: AgentOutcome) -> dict[str, Any]

A compact, JSON-able view of an outcome for an HTTP response.

Includes the status, answer, and usage; when the run is "suspended" it also lists the pending approvals (each with the call id to pass back to resume(decisions=…)). The full transcript is omitted — fetch it from your Store if you need it.

event_to_dict ¶

event_to_dict(event: AgentStreamEvent) -> dict[str, Any]

Convert one :class:~agentix.streaming.AgentStreamEvent to a JSON-able dict with a type discriminator the client can switch on.

format_sse ¶

format_sse(
    payload: dict[str, Any], *, event: str | None = None
) -> str

Format one payload as a Server-Sent Events record.

sse_events `async` ¶

sse_events(
    events: AsyncIterator[AgentStreamEvent],
) -> AsyncIterator[str]

Map an agent event stream to Server-Sent Events text chunks.

The SSE event: field carries the event type (answer, tool_started, tool_finished, done) so a browser EventSource can listen per type.

ndjson_events `async` ¶

ndjson_events(
    events: AsyncIterator[AgentStreamEvent],
) -> AsyncIterator[str]

Map an agent event stream to newline-delimited JSON (one object per line).

sse_response ¶

sse_response(
    events: AsyncIterator[AgentStreamEvent],
    *,
    headers: dict[str, str] | None = None,
) -> StreamingResponse

A StreamingResponse streaming an agent run as Server-Sent Events::

@app.post("/chat")
async def chat(body: ChatIn):
    return sse_response(agent.stream(body.message))

Works with FastAPI or Starlette (agentix[serving]).

ndjson_response ¶

ndjson_response(
    events: AsyncIterator[AgentStreamEvent],
    *,
    headers: dict[str, str] | None = None,
) -> StreamingResponse

A StreamingResponse streaming an agent run as newline-delimited JSON.

agentix.serving¶