Server

A2UIServer wraps a plain autogen.beta.Agent and is a ready-to-serve Starlette ASGI app that speaks canonical A2UI over HTTP. It mirrors autogen.beta.a2a.A2AServer: hold the agent, configure A2UI with flat kwargs, pick a transport= for the wire encoding, and bring your own uvicorn (or any ASGI server) to run the A2UIServer instance directly.

A2UI's wire is transport-agnostic, so one deployment serves one frontend over one transport:

• RestTransport(encoding="sse") — Server-Sent Events (text/event-stream), suited to browser EventSource clients and reconnection.
• RestTransport(encoding="jsonl") — canonical A2UI NDJSON (application/x-ndjson), suited to generic A2UI clients that consume the native JSON Lines wire.
• AgUiTransport() — AG-UI events for CopilotKit's renderer (see Serving over AG-UI below).

The transports live in autogen.beta.a2ui.transports.

Minimal Server#

The A2UIServer instance is itself the ASGI app — hand it straight to uvicorn:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17

import uvicorn

from autogen.beta import Agent
from autogen.beta.a2ui import A2UIServer
from autogen.beta.a2ui.transports import RestTransport
from autogen.beta.config import AnthropicConfig

agent = Agent(name="ui_agent", config=AnthropicConfig(model="claude-sonnet-4-6"))

app = A2UIServer(
    agent,
    transport=RestTransport(encoding="jsonl"),  # POST /a2ui  →  A2UI NDJSON
    protocol_version="v0.9",
)

if __name__ == "__main__":
    uvicorn.run(app, host="127.0.0.1", port=8000)

The turn is served at POST /a2ui by default; pass RestTransport(encoding="jsonl", path="/chat") to change it. Swap encoding="jsonl" for encoding="sse" to stream Server-Sent Events instead.

The Request Contract#

The adapter speaks a minimal, dependency-free JSON body (no ag-ui / a2a-sdk types) so any client that can POST JSON can drive the agent:

{
  "messages":  [{"role": "user", "content": "show a booking form"}],
  "variables": {"locale": "en"},
  "a2ui":      [{"version": "v0.9", "action": {"name": "confirm", "...": "..."}}],
  "a2uiClientCapabilities": {"v0.9": {"supportedCatalogIds": ["..."]}}
}

A body that is not valid JSON, not a JSON object, or has a mistyped field returns 400 with an {"error": ...} payload. A failure during the turn (after the 200 has been sent) surfaces as a final event: error (SSE) or {"error": ...} (NDJSON) frame and is logged — the connection is not torn down silently.

Response Frames#

Each turn streams the conversational prose first (when present), then one frame per validated A2UI message.

{"text": "Here is your booking form."}
{"version":"v0.9","createSurface":{"surfaceId":"s1","catalogId":"https://a2ui.org/..."}}
{"version":"v0.9","updateComponents":{"surfaceId":"s1","components":[...]}}

event: text
data: {"text": "Here is your booking form."}

data: {"version":"v0.9","createSurface":{"surfaceId":"s1","catalogId":"https://a2ui.org/..."}}

event: done
data: {}

When the agent answers in plain prose (or validation degrades after exhausting retries), only the prose frame is emitted — there are no A2UI message frames.

Driving It From a Client#

Any HTTP client works. Here is a full round-trip with httpx, reading the NDJSON stream line by line:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20

import json

import httpx

async def ask_ui(text: str) -> tuple[str, list[dict]]:
    prose, messages = "", []
    async with httpx.AsyncClient(base_url="http://127.0.0.1:8000") as client:
        async with client.stream("POST", "/a2ui", json={"messages": [{"role": "user", "content": text}]}) as resp:
            resp.raise_for_status()
            async for line in resp.aiter_lines():
                if not line:
                    continue
                frame = json.loads(line)
                if "text" in frame:
                    prose = frame["text"]
                elif "error" in frame:
                    raise RuntimeError(frame["error"])
                else:
                    messages.append(frame)  # a canonical A2UI message
    return prose, messages

Because the server is stateless, a follow-up turn just resends the prior messages plus the new one:

1
2
3
4
5
6
7

json={
    "messages": [
        {"role": "user", "content": "show a booking form"},
        {"role": "assistant", "content": "Here is your booking form."},
        {"role": "user", "content": "make it for 4 people"},
    ]
}

Serving over AG-UI (CopilotKit)#

Swap RestTransport for AgUiTransport to serve the same agent as AG-UI events — the protocol CopilotKit's @copilotkit/a2ui-renderer consumes. Nothing about the agent changes; only the wire encoding does:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13

import uvicorn

from autogen.beta import Agent
from autogen.beta.a2ui import A2UIServer
from autogen.beta.a2ui.transports import AgUiTransport
from autogen.beta.config import AnthropicConfig

agent = Agent(name="ui_agent", config=AnthropicConfig(model="claude-sonnet-4-6"))

app = A2UIServer(agent, transport=AgUiTransport())  # POST /  →  AG-UI SSE events

if __name__ == "__main__":
    uvicorn.run(app, host="127.0.0.1", port=8000)

The turn is served at POST / by default (pass AgUiTransport(path="/agent") to change it) and the request body is an AG-UI RunAgentInput. Each validated A2UI message reaches the renderer as an ACTIVITY_SNAPSHOT event with activityType="a2ui-surface" and the operations under content["a2ui_operations"]; the prose arrives as TEXT_MESSAGE_CHUNK. A CopilotKit button click rides forwardedProps.a2uiAction and is rewritten into the next turn — the same click round-trip as REST (see Actions).