Skip to content

AG-UI

Available · pip install "agent-adapter[ag-ui]" · ag-ui-protocol>=1.0,<2 (protocol 1.0)

AG-UI is the event protocol between an agent and its user interface (CopilotKit and others). This bridge streams one of your agent functions as an AG-UI run: each tool call becomes TOOL_CALL_* events, a call that waits for approval ends the run with an interrupt, and the next run's resume entry goes to your ApprovalPort. The SSE output plugs into Starlette and FastAPI.

Frameworks that ship their own AG-UI adapter (Pydantic AI, LangGraph, ADK, MAF, CrewAI) already carry our tools inside them. Use this bridge when your app runs our executor itself.

Names

Name (agent_adapter.integrations.ag_ui) What it is
AGUIBridge(executor, *, port=None, interrupts=None, max_events=10_000, trace=True) runs agents; holds open interrupts. port decides resumes (default: the executor's port)
bridge.stream(run_input, agent, *, principal="") async iterator of AG-UI event models
bridge.sse(run_input, agent, *, principal="") the same, as text/event-stream frames
AGUIRun what agent(run) gets: call, text, custom, messages(), executor, ctx, resumed
agui_endpoint(bridge, agent, *, authenticate, max_body_bytes=1 MiB, allowed_hosts=None) a Starlette/FastAPI POST handler; 421 for a Host not served (default: local names when authenticate=anonymous, else any)
anonymous an authenticate that admits everyone as "" (local use only)
parse_input(body) / to_messages(run_input) a bounded RunAgentInput; its conversation as Messages
InterruptTable(ttl=900, max_interrupts=1024, max_per_principal=64) the open interrupts (in memory, one process); a caller over its cap drops its own oldest, never another's
INTEGRATION the integration object; export_tool gives an AG-UI Tool descriptor

Run, pause, resume

def support_lead_approves(request: ApprovalRequest) -> bool:
    # The remote "approved: true" only asks; this port decides who may approve what.
    resume = request.metadata.get("resume_payload") or {}
    return request.metadata.get("principal") == "lead-7" and resume.get("approved") is True


# The executor parks confirm calls; the bridge's port decides each resume request.
executor = Executor(registry, approval=DeferredApproval(MemoryApprovalStore()))
bridge = AGUIBridge(executor, port=CallbackApproval(support_lead_approves))
async def agent(run: AGUIRun) -> Any:
    if run.input.resume:  # a resumed run: the bridge already ran or denied the paused calls
        result = run.resumed[0]
        run.text(f"refund {'done' if result.ok else 'refused'}")
        return result.value
    total = await run.call("order_total", {"order_id": "o-7"})
    await run.call("refund", {"order_id": "o-7", "amount": total.value})  # pauses
    return None


first = {"threadId": "t-1", "runId": "r-1", "messages": []}
events = [e async for e in bridge.stream(first, agent, principal="lead-7")]
print([e.type.value for e in events])
interrupt = events[-1].outcome.interrupts[0]
refund_call = [e for e in events if e.type.value == "TOOL_CALL_START"][-1]
print(interrupt.reason, interrupt.tool_call_id == refund_call.tool_call_id)

answer = {"interruptId": interrupt.id, "status": "resolved", "payload": {"approved": True}}
second = {"threadId": "t-1", "runId": "r-2", "messages": [], "resume": [answer]}
events = [e async for e in bridge.stream(second, agent, principal="lead-7")]
print([e.type.value for e in events])
print(events[-1].result)

sequenceDiagram
  participant UI as AG-UI client
  participant B as AGUIBridge
  participant A as agent(run)
  participant Ex as Executor
  participant P as bridge port
  UI->>B: RunAgentInput (run r-1)
  B->>A: agent(run)
  A->>Ex: run.call("refund", ...)
  Ex-->>UI: TOOL_CALL_START / ARGS / END
  Ex-->>B: tool.pending (executor's port deferred)
  B-->>UI: RUN_FINISHED outcome=interrupt (id, toolCallId)
  UI->>B: RunAgentInput (run r-2) resume=[{interruptId, payload: {approved: true}}]
  B->>P: ApprovalRequest (+ principal, thread, payload in metadata)
  P-->>B: decision
  Note over B: approved: one-use token bound to run, call id, tool, argument digest
  B->>Ex: run_call(same call id, ctx.approval = token's decision)
  Ex-->>UI: TOOL_CALL_RESULT
  B->>A: agent(run) with run.resumed
  B-->>UI: RUN_FINISHED

Events

Executor activity AG-UI event
a tool call starts (hook before_tool) TOOL_CALL_START, TOOL_CALL_ARGS (JSON of the validated arguments), TOOL_CALL_END
it returns (hook after_tool) TOOL_CALL_RESULT (content: the value as JSON text)
tool.failed TOOL_CALL_RESULT with {"error": {"kind", "message"}}
tool.pending no result; the run ends with an interrupt for that call
run.text(...) / run.custom(...) TEXT_MESSAGE_START/CONTENT/END / CUSTOM
any other trace event (trace=True) CUSTOM named agent_adapter.trace, data without content keys
agent returns value RUN_FINISHED with result (or the interrupt outcome)
agent raises RUN_ERROR with code = error kind; foreign errors show only their type

run.call and run_tool_loop(model, registry, run.messages(), executor=run.executor, ctx=run.ctx) both stream this way. Past max_events, trace events are dropped first. Closing the stream (a client disconnect) cancels the agent.

Approval

Resume entry What happens
status: "cancelled", or a payload other than {"approved": true} denied without asking; TOOL_CALL_RESULT with approval-denied
{"approved": true} a request to the bridge's port; only its decision counts
port approves the call runs once, with an adapter-issued one-use token; port edits apply
port defers (ApprovalPending) the call stays paused; a new interrupt is returned
unknown, expired, answered before, other thread or other principal ignored with a warning and a CUSTOM agent_adapter.resume_ignored; nothing runs

A client's editedArgs are passed to the port in metadata["resume_payload"]; they never apply by themselves. With the default port (a DeferredApproval), a remote "approved" does nothing until someone decides the record in the store, for example over the HTTP server.

Serve it

TOKENS = {"s3cret-token": "lead-7"}  # load from your secret store, never from source


async def authenticate(request: Any) -> str | None:
    header = request.headers.get("authorization", "")
    scheme, _, token = header.partition(" ")
    for known, principal in TOKENS.items():
        if scheme.lower() == "bearer" and hmac.compare_digest(token, known):
            return principal
    return None  # 401


endpoint = agui_endpoint(AGUIBridge(Executor(registry)), agent, authenticate=authenticate)


def starlette_app() -> Any:
    from starlette.applications import Starlette
    from starlette.routing import Route

    return Starlette(routes=[Route("/agui", endpoint, methods=["POST"])])


def fastapi_app() -> Any:
    from fastapi import FastAPI

    app = FastAPI()
    app.add_route("/agui", endpoint, methods=["POST"])
    return app
Response When
401 authenticate returned None
415 / 413 / 422 not application/json / body over max_body_bytes / invalid RunAgentInput
200 text/event-stream the run's events, one data: <json> frame each

Public host checklist: authenticate every request (never anonymous); serve over TLS; give each user their own principal (interrupts are bound to it); put a database-backed store behind DeferredApproval and run one bridge process, or pin a thread to one process, since interrupts live in memory.

Limits

  • Bounds: 1 MiB body, 1,000 messages, 64 resume entries, ids of 1 to 128 characters.
  • Interrupts expire after 15 minutes. An interrupt left out of a resume stays open.
  • Frontend tools in RunAgentInput.tools are not run here; only registry tools are.

Files

File Role
integrations/ag_ui/bridge.py AGUIBridge: the run driver, outcomes, interrupts
integrations/ag_ui/run.py AGUIRun: the agent's handle
integrations/ag_ui/_events.py hook + sink: executor activity to AG-UI events
integrations/ag_ui/_resume.py resume entries to port decisions and tokens
integrations/ag_ui/_interrupts.py PendingCall, InterruptTable
integrations/ag_ui/_input.py parse_input, to_messages, bounds
integrations/ag_ui/_channel.py the per-run, thread-safe event queue
integrations/ag_ui/starlette.py agui_endpoint, anonymous
integrations/ag_ui/_support.py INTEGRATION, version guard, SSE encoding