[WIP] RFD-based ACP HTTP and WS web transport implementation

AGENTS.md

@@ -7,6 +7,7 @@ Use this page as the quick orientation for the Python SDK repo. It mirrors the t
 | Path | Why it exists |
 | --- | --- |
 | `src/acp/` | Runtime package: agent/client bases, transports, helpers, schema bindings, contrib utilities |
+| `src/acp/http/`, `src/acp/ws/` | Experimental remote transports (Streamable HTTP + WebSocket), client + server; opt-in via the `[http]` extra |
 | `schema/` | Upstream JSON schema sources (regenerate with `make gen-all`) |
 | `examples/` | Runnable scripts such as `echo_agent.py`, `client.py`, `gemini.py`, `duet.py` |
 | `tests/` | Pytest suite, including optional Gemini smoke tests in `tests/test_gemini_example.py` |

docs/web-transport.md

@@ -0,0 +1,108 @@
+# Web Transport (Streamable HTTP & WebSocket)
+
+> **Experimental.** The remote web transports are experimental and may change.
+> They ship as an optional extra and are import-guarded.
+
+The SDK can run ACP over two remote connectivity profiles in addition to stdio:
+
+- **Streamable HTTP** — `POST` for client→server messages, long-lived `GET` SSE
+  streams for server→client messages (one connection-scoped stream plus one per
+  session), and `DELETE` to terminate. `initialize` returns `200 OK` with a JSON
+  body; all other POSTs return `202 Accepted`. **Requires HTTP/2.**
+- **WebSocket** — a `GET` upgrade on the same endpoint carrying full-duplex
+  JSON-RPC text frames.
+
+Both reuse the existing JSON-RPC message format and ACP lifecycle
+(`initialize` → session methods → close).
+
+## Installation
+
+```bash
+pip install "agent-client-protocol[http]"
+```
+
+This pulls in `httpx[http2]` (HTTP/2 + SSE consumption) and `websockets`.
+
+## Client
+
+Both transports produce a message-level `Transport` that plugs into the existing
+`connect_to_agent`:
+
+```python
+from acp import connect_to_agent
+from acp.http import create_http_stream
+from acp.ws import create_websocket_stream
+
+# Streamable HTTP
+transport = create_http_stream("http://localhost:8000/acp")
+conn = connect_to_agent(my_client, transport)
+
+# ...or WebSocket
+transport = await create_websocket_stream("ws://localhost:8000/acp")
+conn = connect_to_agent(my_client, transport)
+
+init = await conn.initialize(protocol_version=1)
+session = await conn.new_session(cwd="/tmp", mcp_servers=[])
+await conn.prompt(session_id=session.session_id, prompt=[...])
+await conn.close()
+await transport.close()
+```
+
+The client sends `initialize` first, reads the `Acp-Connection-Id` response
+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).
+
+## Server
+
+The server core is framework-agnostic; a thin ASGI adapter bridges it to your
+web framework:
+
+```python
+from acp.http.asgi import create_asgi_app
+
+# One agent instance is created per connection.
+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.
+
+### HTTP/2 server requirement
+
+> ⚠️ **Uvicorn does not serve HTTP/2.** For a spec-compliant Streamable HTTP
+> server, run an HTTP/2-capable ASGI server (**Hypercorn**, Daphne, Granian) or
+> terminate HTTP/2 at a proxy. The WebSocket profile works on Uvicorn.
+
+```python
+import asyncio
+import hypercorn.asyncio
+from hypercorn.config import Config
+
+config = Config()
+config.bind = ["localhost:8000"]
+config.alpn_protocols = ["h2", "http/1.1"]
+asyncio.run(hypercorn.asyncio.serve(app, config))
+```
+
+## Examples
+
+- [`examples/http_server.py`](https://github.com/agentclientprotocol/python-sdk/blob/main/examples/http_server.py) — serve an agent over HTTP + WS (Hypercorn).
+- [`examples/http_client.py`](https://github.com/agentclientprotocol/python-sdk/blob/main/examples/http_client.py) — connect over Streamable HTTP.
+- [`examples/ws_client.py`](https://github.com/agentclientprotocol/python-sdk/blob/main/examples/ws_client.py) — connect over WebSocket.
+
+## Identity model
+
+- `Acp-Connection-Id` (HTTP header) — returned by `initialize`; required on all
+  post-initialize HTTP requests and every GET stream.
+- `Acp-Session-Id` (HTTP header) — required on session-scoped POSTs and the
+  session-scoped GET stream.
+- `sessionId` (JSON-RPC field) — carried in params/results and used to route
+  messages to the correct stream.
+
+## Not yet supported (deferred to a future revision)
+
+- `Last-Event-ID` / SSE resumability and message sequencing.
+- Client-side automatic reconnect/backoff.
+- Batch JSON-RPC (the server returns `501`).
+- `Acp-Protocol-Version` header enforcement.

examples/http_client.py

@@ -0,0 +1,54 @@
+# /// script
+# requires-python = ">=3.10,<3.15"
+# dependencies = [
+#     "agent-client-protocol[http]",
+# ]
+# ///
+"""Connect to a remote ACP agent over Streamable HTTP (experimental).
+
+Start the server first (``uv run examples/http_server.py``), then run this.
+"""
+
+import asyncio
+from typing import Any
+
+from acp import connect_to_agent, text_block
+from acp.http import create_http_stream
+from acp.interfaces import Client
+
+
+class ExampleClient(Client):
+    async def request_permission(self, session_id: str, tool_call: Any, options: Any, **kwargs: Any) -> Any:
+        # Auto-allow the first option.
+        return {"outcome": {"outcome": "selected", "optionId": options[0]["optionId"]}}
+
+    async def session_update(self, session_id: str, update: Any, **kwargs: Any) -> None:
+        content = getattr(update, "content", None)
+        text = getattr(content, "text", None) if content is not None else None
+        if text:
+            print(f"<< {text}")
+
+    async def write_text_file(self, *args: Any, **kwargs: Any) -> None:
+        return None
+
+    async def read_text_file(self, *args: Any, **kwargs: Any) -> Any:
+        return {"content": ""}
+
+
+async def main() -> None:
+    transport = create_http_stream("http://localhost:8000/acp")
+    conn = connect_to_agent(ExampleClient(), transport)
+    try:
+        init = await conn.initialize(protocol_version=1)
+        print(f"initialized (protocol v{init.protocol_version})")
+        session = await conn.new_session(cwd=".", mcp_servers=[])
+        print(f"session: {session.session_id}")
+        result = await conn.prompt(session_id=session.session_id, prompt=[text_block("hello over http")])
+        print(f"stop reason: {result.stop_reason}")
+    finally:
+        await conn.close()
+        await transport.close()
+
+
+if __name__ == "__main__":
+    asyncio.run(main())

examples/http_server.py

@@ -0,0 +1,81 @@
+# /// script
+# requires-python = ">=3.10,<3.15"
+# dependencies = [
+#     "agent-client-protocol[http]",
+#     "hypercorn>=0.17",
+# ]
+# ///
+"""Serve an ACP agent over Streamable HTTP + WebSocket (experimental).
+
+Run with an HTTP/2-capable ASGI server for spec-compliant Streamable HTTP.  This
+example uses Hypercorn; Uvicorn works for WebSocket but does not serve HTTP/2.
+
+    uv run examples/http_server.py
+    # then, in another terminal:
+    uv run examples/http_client.py
+    uv run examples/ws_client.py
+"""
+
+import asyncio
+from typing import Any
+from uuid import uuid4
+
+from acp import (
+    Agent,
+    InitializeResponse,
+    NewSessionResponse,
+    PromptResponse,
+    text_block,
+    update_agent_message,
+)
+from acp.http.asgi import create_asgi_app
+from acp.interfaces import Client
+from acp.schema import ClientCapabilities, Implementation
+
+
+class EchoAgent(Agent):
+    _conn: Client
+
+    def on_connect(self, conn: Client) -> None:
+        self._conn = conn
+
+    async def initialize(
+        self,
+        protocol_version: int,
+        client_capabilities: ClientCapabilities | None = None,
+        client_info: Implementation | None = None,
+        **kwargs: Any,
+    ) -> InitializeResponse:
+        return InitializeResponse(protocol_version=protocol_version)
+
+    async def new_session(self, cwd: str = "", **kwargs: Any) -> NewSessionResponse:
+        return NewSessionResponse(session_id=uuid4().hex)
+
+    async def prompt(self, session_id: str, prompt: list[Any], **kwargs: Any) -> PromptResponse:
+        for block in prompt:
+            text = block.get("text", "") if isinstance(block, dict) else getattr(block, "text", "")
+            await self._conn.session_update(
+                session_id=session_id,
+                update=update_agent_message(text_block(f"echo: {text}")),
+            )
+        return PromptResponse(stop_reason="end_turn")
+
+
+# One agent instance per connection.
+app = create_asgi_app(lambda conn: EchoAgent())
+
+
+async def main() -> None:
+    import hypercorn.asyncio
+    from hypercorn.config import Config
+
+    config = Config()
+    config.bind = ["localhost:8000"]
+    # Enable HTTP/2 (Streamable HTTP requires it). Hypercorn negotiates h2c/h2.
+    config.alpn_protocols = ["h2", "http/1.1"]
+    print("Serving ACP agent on http://localhost:8000/acp (HTTP + WS)")
+    await hypercorn.asyncio.serve(app, config)
+
+
+if __name__ == "__main__":
+    asyncio.run(main())

examples/ws_client.py

@@ -0,0 +1,53 @@
+# /// script
+# requires-python = ">=3.10,<3.15"
+# dependencies = [
+#     "agent-client-protocol[http]",
+# ]
+# ///
+"""Connect to a remote ACP agent over WebSocket (experimental).
+
+Start the server first (``uv run examples/http_server.py``), then run this.
+"""
+
+import asyncio
+from typing import Any
+
+from acp import connect_to_agent, text_block
+from acp.interfaces import Client
+from acp.ws import create_websocket_stream
+
+
+class ExampleClient(Client):
+    async def request_permission(self, session_id: str, tool_call: Any, options: Any, **kwargs: Any) -> Any:
+        return {"outcome": {"outcome": "selected", "optionId": options[0]["optionId"]}}
+
+    async def session_update(self, session_id: str, update: Any, **kwargs: Any) -> None:
+        content = getattr(update, "content", None)
+        text = getattr(content, "text", None) if content is not None else None
+        if text:
+            print(f"<< {text}")
+
+    async def write_text_file(self, *args: Any, **kwargs: Any) -> None:
+        return None
+
+    async def read_text_file(self, *args: Any, **kwargs: Any) -> Any:
+        return {"content": ""}
+
+
+async def main() -> None:
+    transport = await create_websocket_stream("ws://localhost:8000/acp")
+    conn = connect_to_agent(ExampleClient(), transport)
+    try:
+        init = await conn.initialize(protocol_version=1)
+        print(f"initialized (protocol v{init.protocol_version})")
+        session = await conn.new_session(cwd=".", mcp_servers=[])
+        print(f"session: {session.session_id}")
+        result = await conn.prompt(session_id=session.session_id, prompt=[text_block("hello over websocket")])
+        print(f"stop reason: {result.stop_reason}")
+    finally:
+        await conn.close()
+        await transport.close()
+
+
+if __name__ == "__main__":
+    asyncio.run(main())

mkdocs.yml

@@ -11,6 +11,7 @@ nav:
   - Home: index.md
   - Quick Start: quickstart.md
   - Use Cases: use-cases.md
+  - Web Transport (HTTP/WS): web-transport.md
   - Experimental Contrib: contrib.md
   - Releasing: releasing.md
   - 0.11 Migration Guide: migration-guide-0.11.md

pyproject.toml

@@ -44,10 +44,15 @@ dev = [
     "mkdocstrings[python]>=0.26.1",
     "python-dotenv>=1.1.1",
     "prek>=0.2.17",
+    "httpx[http2]>=0.27",
+    "websockets>=12.0",
+    "uvicorn>=0.30",
 ]
 
 [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"]
 
 [build-system]
 requires = ["pdm-backend"]

src/acp/_cookies.py

@@ -0,0 +1,53 @@
+"""In-memory cookie store for the WebSocket handshake.
+
+The HTTP client relies on ``httpx``'s built-in cookie jar for session affinity,
+but the WebSocket handshake needs a small, explicit store to collect
+``Set-Cookie`` headers from the upgrade response and echo them back as a
+``Cookie`` request header for the socket lifetime.
+
+This is intentionally minimal: it stores name→value pairs without attribute
+parsing (domain/path/expiry), matching the affinity-only use case in the RFD.
+"""
+
+from __future__ import annotations
+
+__all__ = ["MemoryAcpCookieStore"]
+
+
+class MemoryAcpCookieStore:
+    """A tiny name→value cookie store keyed by cookie name."""
+
+    def __init__(self) -> None:
+        self._cookies: dict[str, str] = {}
+
+    def store_set_cookie(self, header_value: str) -> None:
+        """Ingest a single ``Set-Cookie`` header value.
+
+        Only the leading ``name=value`` pair is retained; cookie attributes
+        (``; Path=/``, ``; HttpOnly`` etc.) are ignored.
+        """
+        first = header_value.split(";", 1)[0].strip()
+        if not first or "=" not in first:
+            return
+        name, _, value = first.partition("=")
+        name = name.strip()
+        if name:
+            self._cookies[name] = value.strip()
+
+    def store_set_cookies(self, header_values: list[str]) -> None:
+        """Ingest multiple ``Set-Cookie`` header values."""
+        for value in header_values:
+            self.store_set_cookie(value)
+
+    def cookie_header(self) -> str | None:
+        """Render the stored cookies as a ``Cookie`` request header value."""
+        if not self._cookies:
+            return None
+        return "; ".join(f"{name}={value}" for name, value in self._cookies.items())
+
+    def clear(self) -> None:
+        """Drop all stored cookies."""
+        self._cookies.clear()
+
+    def __len__(self) -> int:
+        return len(self._cookies)