Metadata-Version: 2.4
Name: agui-claude-adapter
Version: 0.1.0
Summary: AG-UI ↔ Claude Agent SDK protocol adapter
Author: Far-Away-Flipped
License: MIT
Keywords: ag-ui,claude,agent,protocol,adapter
Classifier: Development Status :: 3 - Alpha
Classifier: Intended Audience :: Developers
Classifier: License :: OSI Approved :: MIT License
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Classifier: Topic :: Software Development :: Libraries :: Python Modules
Requires-Python: >=3.12
Description-Content-Type: text/markdown
Requires-Dist: claude-agent-sdk===0.1.48
Requires-Dist: ag-ui-protocol>=0.1.0
Provides-Extra: tools
Requires-Dist: mcp>=1.0.0; extra == "tools"
Provides-Extra: dev
Requires-Dist: pytest>=8.0.0; extra == "dev"
Requires-Dist: pytest-asyncio>=0.24.0; extra == "dev"
Provides-Extra: demo
Requires-Dist: fastapi>=0.115.0; extra == "demo"
Requires-Dist: uvicorn>=0.30.0; extra == "demo"
Requires-Dist: sse-starlette>=2.0.0; extra == "demo"

# 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

```bash
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.

```python
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.

```python
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:

```python
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:

```json
{
  "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:

| AG-UI input | Anthropic output |
|---|---|
| `{"type":"text","text":"..."}` | `{"type":"text","text":"..."}` |
| `{"type":"image","source":{"type":"data","value":"...","mimeType":"image/png"}}` | `{"type":"image","source":{"type":"base64","media_type":"image/png","data":"..."}}` |
| `{"type":"document","source":...}` | Same mapping (`data`→`base64`, `mimeType`→`media_type`, `value`→`data`) |

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

```python
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)

```python
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:

```python
# 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:

| Exception | Error code |
|---|---|
| `AdapterConnectionError` | `CONNECTION_ERROR` |
| `AdapterAuthError` | `AUTH_ERROR` |
| `AdapterTimeoutError` | `TIMEOUT_ERROR` |
| `AdapterRateLimitError` | `RATE_LIMIT_ERROR` |
| `AdapterInvalidRequestError` | `INVALID_REQUEST` |
| `AdapterServerError` | `SERVER_ERROR` |
| Other / unknown | `UNKNOWN_ERROR` |

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

---

## Package reference

| Component | Import | Description |
|---|---|---|
| `MessageConverter` | `agui_claude_adapter` | Static methods: `agui_to_sdk(messages) → list[dict]`, `sdk_to_agui_message(sdk_msg) → AssistantMessage` |
| `EventBuilder` | `agui_claude_adapter` | `process(sdk_stream, *, thread_id, run_id) → AsyncIterator[AGUIEvent]` |
| `Adapter` | `agui_claude_adapter` | `process_run(input, *, sdk_options=None, tools=None) → AsyncIterator[AGUIEvent]` |
| `FrontendToolBridge` | `agui_claude_adapter` | `wait_for(name, args, *, timeout)`, `resolve(name, args, result) → bool`, `cancel_all()` |
| `FrontendToolRegistry` | `agui_claude_adapter` | `replace(tools)`, `snapshot()`, `names()` |
| `build_frontend_tool_mcp_server` | `agui_claude_adapter` | `(bridge, registry, *, server_name, call_timeout) → McpServer` |
| `ToolConverter` | `agui_claude_adapter` | AG-UI Tool ↔ OpenAI/Anthropic schema conversion utilities |

---

## Run the demo

```bash
# 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

```bash
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

| Document | Purpose |
|---|---|
| `docs/KNOWN_ISSUES.md` | Prioritized defect log (all core-adapter issues fixed as of 2026-07-09) |
| `docs/stateless-adapter-plan.md` | Design rationale for the stateless refactor |
| `docs/tool-bridge-core-plan.md` | Transport-agnostic tool-bridge architecture |
| `docs/python-packaging-guide.md` | Step-by-step Python packaging & publishing tutorial |
