refactor: simplify web transports with Starlette

Author: frostming

docs/web-transport.md

@@ -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
 
@@ -55,8 +57,8 @@ stream; reconnect/retry is the caller's responsibility (v1 of the RFD).
 
 ## 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
@@ -65,8 +67,111 @@ 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 needs three 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 |
+| 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
 

examples/http_server.py

@@ -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())
 
 

pyproject.toml

@@ -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"]

src/acp/_transport.py

@@ -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
@@ -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()

src/acp/http/__init__.py

@@ -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]``.
 """