agui-claude-adapter

AG-UI Claude Adapter

A transport-agnostic protocol adapter bridging AG-UI (Agent-User Interaction Protocol) ↔ Claude Agent SDK (Python, drives the claude CLI as a subprocess).

frontend (AG-UI events)  ⇄  Adapter (this package)  ⇄  ClaudeSDKClient  ⇄  claude CLI

Design principles

• Stateless. The Adapter holds no cross-run mutable state — no client cache, no session manager, no generated IDs. Each process_run() call is self-contained.
• No opinion on session identity. thread_id / run_id are taken verbatim from the caller; resume/session behavior is the caller's decision via sdk_options.
• Converters are standalone, first-class citizens. MessageConverter (AG-UI → SDK) and EventBuilder (SDK → AG-UI) can be used independently.
• Transport-agnostic tool bridge. Frontend-defined tools are bridged through an in-process MCP server — no dependency on any web framework.

---

Install

pip install agui-claude-adapter            # core converters + adapter
pip install agui-claude-adapter[tools]     # + frontend-tool bridge (mcp)

Prerequisite: claude CLI on PATH (npm install -g @anthropic-ai/claude-code).

---

Usage paths

This package gives you three levels of control. Pick whichever matches your architecture.

Path 1 — Standalone converters (no Adapter needed)

You own the ClaudeSDKClient lifecycle. Call the two converters directly wherever you need them.

from claude_agent_sdk import ClaudeSDKClient, ClaudeAgentOptions
from agui_claude_adapter import MessageConverter, EventBuilder

async def my_agent_turn(run_input):
    client = ClaudeSDKClient(options=my_options)
    await client.connect()
    try:
        # 1. AG-UI messages → SDK format
        sdk_messages = MessageConverter.agui_to_sdk(run_input.messages)
        await client.query(sdk_messages[0]["content"])

        # 2. SDK output → AG-UI events
        builder = EventBuilder()
        async for event in builder.process(
            client.receive_response(),
            thread_id=run_input.thread_id,
            run_id=run_input.run_id,
        ):
            yield event   # each is an ag_ui.core.Event subclass
    finally:
        await client.disconnect()

Use this path when:

• You already manage your own ClaudeSDKClient pool / per-thread workers.
• You need fine-grained control over connect / query / disconnect (e.g. multi-turn with a persistent subprocess — see the demo for a worked example).
• You want minimal overhead — no extra layers between you and the SDK.

Path 2 — Adapter.process_run() (batteries-included)

The Adapter is a stateless convenience: it creates a fresh ClaudeSDKClient per call, connects, sends, streams, and disconnects — all for you.

from agui_claude_adapter import Adapter

adapter = Adapter()
async for event in adapter.process_run(run_input):
    print(event.type, event)

You can pass per-run sdk_options (resume, model, system prompt, …) — the adapter forwards them unchanged:

from claude_agent_sdk import ClaudeAgentOptions

opts = ClaudeAgentOptions(
    include_partial_messages=True,
    max_turns=10,
    permission_mode="bypassPermissions",
    resume="cli-session-id-from-last-turn",   # business-layer decision
)
async for event in adapter.process_run(run_input, sdk_options=opts):
    ...

Use this path when:

• You want a single-turn, fire-and-forget agent call.
• You don't need persistent subprocesses across turns (each process_run is isolated).

Path 3 — Business-layer orchestration (demo's pattern)

For multi-turn conversations with memory, you keep a long-lived ClaudeSDKClient (one CLI subprocess per thread) and drive the converters against it directly — just like Path 1, but with a persistent client.

The demo (demo/server.py) implements this with a _ThreadWorker per thread: the worker owns one ClaudeSDKClient inside a dedicated asyncio.Task, request tasks submit messages via a queue, and the two standalone converters do the rest. This pattern keeps the core adapter stateless while the business layer owns session identity and client lifecycle.

See demo/server.py for the full implementation — it is extensively commented and serves as the reference for this pattern.

---

Working with multimodal messages (images, documents)

AG-UI represents multimodal content as an array of content parts. The converter maps them to Anthropic's native format automatically.

Frontend to AG-UI UserMessage

The frontend sends an AG-UI content-part array:

{
  "message": [
    {"type": "text",  "text": "Look at this image"},
    {"type": "image", "source": {"type": "data", "value": "<base64>", "mimeType": "image/png"}}
  ]
}

What the converter does

MessageConverter.agui_to_sdk maps each part:

The SDK routes the converted content to the CLI, which delivers it to the model. Text-only paths are unchanged.

Single vs multiple messages

• One multimodal message: goes through the async-iterable path (the string fast-path is guarded so lists don't cause TypeError).
• Multiple messages (full history): already uses the async-iterable path — multimodal content works the same way.

---

Frontend-defined tools

When the model calls a tool that the frontend executes (not a built-in CLI tool), you need a bridge between the CLI's MCP handler and the frontend's result:

CLI calls tool  →  MCP call_tool  →  bridge.wait_for(name, args)  →  BLOCK
Frontend runs   →  POST /tool_result                            →  bridge.resolve(name, args, result)
                                                                     UNBLOCK → return to CLI

Using Adapter.process_run with tools

async for event in adapter.process_run(
    run_input,
    tools=run_input.tools,  # list of ag_ui.core.Tool
):
    ...

When tools is provided the adapter wires up an in-process MCP server (FrontendToolBridge + FrontendToolRegistry + build_frontend_tool_mcp_server) internally and tears it down at run end.

Using the tool bridge directly (Path 1 / Path 3)

from agui_claude_adapter import (
    FrontendToolBridge,
    FrontendToolRegistry,
    build_frontend_tool_mcp_server,
)

bridge = FrontendToolBridge()
registry = FrontendToolRegistry(agui_tools)
mcp_server = build_frontend_tool_mcp_server(bridge, registry)

sdk_options.mcp_servers = {
    "frontend-tools": {
        "type": "sdk",
        "name": "frontend-tools",
        "instance": mcp_server,
    }
}

When the frontend returns a tool result, deliver it through the bridge:

# In your POST /api/tool_result handler (or WebSocket, or …)
matched = bridge.resolve(tool_name, arguments, result)
# matched == True  → a waiting call_tool handler was unblocked
# matched == False → result buffered (arrived before call_tool), or stale POST

Important: The MCP call_tool handler never learns the Anthropic tool_use_id, so the bridge correlates by a (tool_name, arguments) fingerprint instead. The frontend must send tool_name + arguments (not just tool_call_id) when reporting a result. Both sides canonicalise with json.dumps(sort_keys=True) inside Python, so key order and JS/Python serialization differences don't cause mismatches.

---

Exception handling

Adapter.process_run classifies exceptions into typed AdapterError subclasses and emits a RunErrorEvent rather than raising:

When using the standalone converters (Path 1), you handle SDK exceptions yourself — the converters only do format translation.

---

Package reference

---

Run the demo

# Backend (requires claude CLI)
uv run python demo/server.py       # → http://127.0.0.1:8000

# Frontend
cd demo/frontend && npm run dev    # → http://localhost:5173 (proxies /api → :8000)

The demo is a full-stack reference: multi-turn chat (persistent _ThreadWorker per thread), file upload with image → Anthropic conversion, and generative-UI tools (setBackgroundColor, showChoiceCard, flightBookingWizard).

---

Development

uv run pytest tests/                          # full suite (102 tests)
uv run pytest tests/test_adapter.py -v        # single file

asyncio_mode = "auto" is set — no @pytest.mark.asyncio needed on async tests.

---

Documentation