Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
141 changes: 136 additions & 5 deletions docs/web-transport.md
Original file line number Diff line number Diff line change
Expand Up @@ -21,7 +21,9 @@ Both reuse the existing JSON-RPC message format and ACP lifecycle
pip install "agent-client-protocol[http]"
```

This pulls in `httpx[http2]` (HTTP/2 + SSE consumption) and `websockets`.
This pulls in `httpx[http2]` (HTTP/2 + SSE consumption), `websockets`, and
`starlette` (the server application). The core SDK and stdio transport do not
require these optional dependencies.

## Client

Expand Down Expand Up @@ -53,10 +55,35 @@ header, then opens the connection-scoped SSE stream. When a new `sessionId`
appears it opens that session-scoped stream too. A single SSE attempt is made per
stream; reconnect/retry is the caller's responsibility (v1 of the RFD).

### Loading an existing session

Both HTTP and WebSocket support `load_session()` when the agent advertises
`loadSession` and implements session persistence:

```python
init = await conn.initialize(protocol_version=1)
if init.agent_capabilities.load_session:
await conn.load_session(session_id="saved-session-id", cwd="/workspace", mcp_servers=[])
await conn.prompt(session_id="saved-session-id", prompt=[...])
```

For HTTP, history replay and the load response use the connection SSE stream.
The client correlates the response with the session ID from the load request,
then opens the session SSE stream for further prompts and agent callbacks. This
also works with an empty history: a load response need not contain `sessionId`.
Replay is consumed as it arrives, so histories larger than the SSE buffer do not
wait for a session stream to open. WebSocket uses its existing bidirectional
connection for both replay and subsequent messages.

A failed load returns its JSON-RPC error on the connection stream and can be
retried. The server removes streams provisioned only for failed loads, while
preserving established sessions and overlapping loads. It does not change the
agent's load response or automatically enable the agent's `loadSession` capability.

## Server

The server core is framework-agnostic; a thin ASGI adapter bridges it to your
web framework:
The server uses Starlette for HTTP requests, responses, routing, streaming,
WebSocket handling, and application lifespan:

```python
from acp.http.asgi import create_asgi_app
Expand All @@ -65,8 +92,112 @@ from acp.http.asgi import create_asgi_app
app = create_asgi_app(lambda conn: MyAgent())
```

`app` is a standard ASGI 3.0 application handling `POST`/`GET`/`DELETE` and
WebSocket upgrades on the ACP endpoint.
`app` is a `starlette.applications.Starlette` instance handling
`POST`/`GET`/`DELETE` and WebSocket upgrades at `/acp` by default. Set the
keyword-only `path` argument to use a different endpoint for both transports:

```python
app = create_asgi_app(lambda conn: MyAgent(), path="/rpc")
```

Other paths do not serve ACP. Starlette supplies
`Request`, `JSONResponse`, `StreamingResponse`, and `WebSocket`; the SDK keeps
ACP connection and session routing.

### Mounting in another application

Mount the app at the desired prefix. The parent must enter the child lifespan
so HTTP connections are cleaned up during shutdown (mounted application
lifespans are not run automatically):

```python
from contextlib import asynccontextmanager
from starlette.applications import Starlette
from starlette.routing import Mount

acp_app = create_asgi_app(lambda conn: MyAgent())

@asynccontextmanager
async def lifespan(app):
async with acp_app.router.lifespan_context(acp_app):
yield

app = Starlette(routes=[Mount("/agents", app=acp_app)], lifespan=lifespan)
# Connect to /agents/acp using either HTTP or WebSocket.
```

With `path="/rpc"`, the mounted endpoint is `/agents/rpc`. Use `path="/"` to
serve ACP at the mount root (`/agents/`).

### How the server fits together

Start reading at `acp/http/asgi.py`. It creates Starlette routes, passes parsed
HTTP requests to `AcpServer`, and binds WebSockets in `acp/ws/server.py`. Both use the existing
`AgentSideConnection` and its message-level `Transport` interface:

```text
HTTP POST → _HttpTransport incoming queue → AgentSideConnection → agent
HTTP GET ← StreamingResponse ← SSE buffer ← _HttpTransport.send() ← agent output

Starlette WebSocket ↔ _WebSocketTransport ↔ AgentSideConnection ↔ agent
```

For HTTP, `AcpServer` owns a dictionary of active connections. Each connection
has one incoming queue and one SSE buffer per stream. The incoming queue lets
POST return `202` while the agent handles the request. Output goes directly to
the relevant SSE buffer; there is no intermediate transport pair or pump task.

HTTP output follows these routing rules:

| Message | Destination | Why |
| --- | --- | --- |
| `initialize` response | POST body, via one Future | Establishes the connection before GET streams open |
| Response containing a new `sessionId` | Connection SSE stream | The client needs the ID before it can open the session stream |
| `session/load` replay and response | Connection SSE stream | Replay precedes the response; the client gets the session ID from the original request |
| Other messages | Session SSE stream when known, otherwise connection stream | Responses use their request's recorded session; requests/notifications carry `sessionId` |

`OutboundStream` retains a bounded buffer, backpressure, and close handling.
Idle SSE streams emit keepalives. These support slow readers, streams that open
after messages arrive, and orderly teardown. `DELETE` and server shutdown close
the HTTP connections and cancel their agent work.

WebSocket already provides one bidirectional stream. Its transport adapts
Starlette's socket to JSON-RPC messages; it needs no HTTP connection registry,
session routing, SSE buffers, or multiplex mode. The ASGI handler owns the agent
connection and closes it on socket disconnect or handler cancellation.

### Simplification experiment

The original 80 HTTP, WebSocket, and RPC tests passed after each ablation.
The Starlette migration also passes these behaviors; assertions now inspect
Starlette response objects and WebSocket tests use the framework's socket:

| Stage | Removed | Lines across the three server files |
| --- | --- | ---: |
| Baseline | — | 715 |
| First ablation | `ConnectionRegistry`, WebSocket multiplex mode, WebSocket pump tasks, forwarding-only ASGI method | 647 |
| Second ablation | HTTP memory transport pair and pump, `ConnectionState`, generic response-waiter map | 581 |
| Starlette migration | Custom ASGI app, request/header parsing, response encoding, WebSocket state tracking, `PostResult` | 483 |

This measures structural simplification and regression coverage, not throughput
or latency. Additional tests cover interrupted initialization, closing a full
SSE buffer, WebSocket cancellation/disconnect, invalid frames, and session routing
of concurrent success/error responses.

`create_asgi_app(agent_factory, *, path="/acp")` returns a Starlette application.
The default route is now `/acp`, replacing the earlier catch-all route.
`AcpServer.handle_post()` and `handle_delete()` return
Starlette responses (`status_code`, byte `body`, and case-insensitive `headers`).
`open_stream()` and `close()` keep their signatures.
The experimental `AcpAsgiApp` and `PostResult` wrappers were removed, along with
`ConnectionRegistry`, `ConnectionState`, `AcpServer.registry`, and
`create_websocket_connection()`. Direct WebSocket integrations now use
`handle_websocket(agent_factory, websocket)` with a Starlette `WebSocket`.
WebSocket lifetimes belong to their ASGI handlers; `AcpServer.close()` manages
HTTP connections.

The migration adds checks for HTTP error statuses, unsupported methods, mounting,
lifespan cleanup, and reopening an SSE stream after disconnect.

### HTTP/2 server requirement

Expand Down
2 changes: 1 addition & 1 deletion examples/http_server.py
Original file line number Diff line number Diff line change
Expand Up @@ -59,7 +59,7 @@ async def prompt(self, session_id: str, prompt: list[Any], **kwargs: Any) -> Pro
return PromptResponse(stop_reason="end_turn")


# One agent instance per connection.
# A Starlette application with one agent instance per connection.
app = create_asgi_app(lambda conn: EchoAgent())


Expand Down
3 changes: 2 additions & 1 deletion pyproject.toml
Original file line number Diff line number Diff line change
Expand Up @@ -48,12 +48,13 @@ dev = [
"httpx[http2]>=0.27",
"websockets>=12.0",
"uvicorn>=0.30",
"starlette>=0.49.3",
]

[project.optional-dependencies]
logfire = ["logfire>=0.14", "opentelemetry-sdk>=1.28.0"]
# Experimental remote transports (Streamable HTTP + WebSocket), client + server.
http = ["httpx[http2]>=0.27", "websockets>=12.0"]
http = ["httpx[http2]>=0.27", "websockets>=12.0", "starlette>=0.49.3"]

[build-system]
requires = ["pdm-backend"]
Expand Down
7 changes: 2 additions & 5 deletions src/acp/_transport.py
Original file line number Diff line number Diff line change
Expand Up @@ -8,8 +8,7 @@
The existing stdio path is re-expressed on top of this seam via
:class:`NdjsonTransport`, which wraps the current byte-stream framing so there
is **zero behaviour change** for stdio users. :func:`memory_transport_pair`
gives two linked in-memory transports, used by the HTTP/WS server to bind an
``AgentSideConnection`` to its message pump.
gives two linked in-memory transports for in-process connections and tests.
"""

from __future__ import annotations
Expand Down Expand Up @@ -140,9 +139,7 @@ def memory_transport_pair() -> tuple[Transport, Transport]:
"""Return two linked in-memory transports.

A message ``send`` on one end becomes available via ``receive`` on the
other. Closing an end enqueues an EOF (``None``) for its peer. This mirrors
the ``TransformStream`` pair the TypeScript SDK uses to bind a server-side
connection to its HTTP/WS message pump.
other. Closing an end enqueues an EOF (``None``) for its peer.
"""
a_to_b: asyncio.Queue[dict[str, Any] | None] = asyncio.Queue()
b_to_a: asyncio.Queue[dict[str, Any] | None] = asyncio.Queue()
Expand Down
2 changes: 1 addition & 1 deletion src/acp/http/__init__.py
Original file line number Diff line number Diff line change
@@ -1,7 +1,7 @@
"""Streamable HTTP transport for ACP (experimental).

Public exports are import-guarded: the heavy client/server implementations pull
in optional dependencies (``httpx[http2]``). Importing a symbol without the
in optional dependencies (``httpx[http2]`` and ``starlette``). Importing a symbol without the
extra installed raises a friendly ``ImportError`` pointing at
``pip install agent-client-protocol[http]``.
"""
Expand Down
Loading
Loading