loom.ai.fastapi

HTTP surface of the AI pillar.

The only subpackage of loom.ai that imports FastAPI and Starlette, so import loom.ai keeps working on a base installation. Mounting mirrors loom.rest.fastapi.sql.bind_sql_endpoints(): routes are added to an existing application, and nothing here returns a router.

class loom.ai.fastapi.AgentJSONResponse(*args, **kwargs)[source]

Bases: MsgspecJSONResponse

Agent response encoded once by the module-level agent encoder.

Example:

return AgentJSONResponse(content=result_payload(result))
Parameters:

  • args (Any)

  • kwargs (Any)

Return type:

Any

render(content)[source]

Encode content to JSON bytes in a single pass.

Parameters:

content (object) – Value to serialise, typically the mapping result_payload() or error_response() built.

Returns:

The UTF-8 encoded JSON body.

Return type:

bytes

loom.ai.fastapi.bind_agent_endpoints(app, *, runtime, config, authenticator=None, observability_runtime=None, prefix='/agents')[source]

Mount /run, /stream and /health for every opted-in agent.

Only agents present in ai.endpoints with enabled and a named auth are reachable; every other compiled agent exposes no HTTP surface at all. Each mount is announced with a WARNING carrying its security state.

Parameters:

  • app (fastapi.FastAPI) – FastAPI application to mount the routes on.

  • runtime (AgentRuntime) – Entered runtime serving the agents.

  • config (AiConfig) – Parsed ai: section.

  • authenticator (Authenticator | None) – Mechanism authenticating callers of the application. None is only acceptable for agents declaring allow_anonymous.

  • observability_runtime (ObservabilityRuntime | None) – Runtime emitting one span per run, over both /run and /stream. This surface is the single owner of that span: AgentRuntime emits none.

  • prefix (str) – Path prefix the routes are mounted under.

Raises:

ConfigError – When an agent opts into HTTP without a usable authenticator and without allow_anonymous.

Return type:

None

Example:

bind_agent_endpoints(app, runtime=runtime, config=ai_config,
                     authenticator=authenticator)
loom.ai.fastapi.encode_sse_event(event)[source]

Encode one agent event as a single SSE frame.

The event’s name travels on the event: line, so the union’s type tag never appears inside the data payload.

Parameters:

event (TextDeltaEvent | ToolCallEvent | ToolResultEvent | ErrorEvent | FinalEvent) – Event to encode.

Returns:

The frame event: <name>\ndata: <json>\n\n.

Raises:

KeyError – When the event is not one of the five contract members — widening the union without widening this map is a contract break, not a silent pass-through.

Return type:

bytes

Example:

frame = encode_sse_event(TextDeltaEvent(text="Demand rose "))
loom.ai.fastapi.stream_sse(events, *, heartbeat_ms)[source]

Encode an agent event stream as SSE frames, with heartbeats.

Composition of this module’s encoder with the transport-wide heartbeat race of with_heartbeats(): encoding is what the HTTP contract owns, the race is not.

Parameters:
Returns:

The encoded SSE frames, ending at the single terminal frame and interleaved with heartbeat comment frames.

Return type:

AsyncIterator[bytes]

Example:

async for frame in stream_sse(events, heartbeat_ms=15000):
    ...