feat(http): serve v2 over Streamable HTTP and WebSocket

The Streamable HTTP + WebSocket transports predate the experimental v2
runtime and were hard-wired to the v1 AgentSideConnection and its method
table. This completes the port so the transports are protocol-agnostic,
matching the TypeScript reference (AcpServer takes an AgentConnector):

- AcpServer / handle_websocket accept an AgentProtocolRouter, which
  negotiates v1 or v2 from the first initialize and attaches the right
  runtime.
- AgentProtocolConnection gains a public listen() so stdio, HTTP and
  WebSocket drive it uniformly.
- session/resume joins session/load as a replay method, so v2 resume
  routes replay over the connection stream exactly like v1 load.
- v2 AgentSideConnection.attach now calls on_connect, matching __init__
  and the v1 attach (agents like crow-cli v2 that learn their connection
  from on_connect would otherwise never receive it).

Tests: v2 over HTTP, v2 over WebSocket, v2 resume replay, and a
dual-stack test serving v1 and v2 clients on one port.

Session-Id: enchanted-spectral-skunk-of-ampleness

Author: odellus

src/acp/experimental/negotiation.py

@@ -136,6 +136,9 @@ def __init__(self, connection: Connection) -> None:
     async def _listen(self) -> None:
         await self._connection.main_loop()
 
+    async def listen(self) -> None:
+        await self._listen()
+
     async def close(self) -> None:
         await self._connection.close()
 

src/acp/experimental/v2/agent.py

@@ -84,6 +84,8 @@ def attach(
         self._state = InitializationState()
         self._conn = connection
         agent = agent_factory(self)
+        if on_connect := getattr(agent, "on_connect", None):
+            on_connect(self)
         router = _AgentRouter(agent, self._state)
         return self, router
 

src/acp/http/asgi.py

@@ -10,6 +10,7 @@
 from collections.abc import AsyncIterator
 from contextlib import asynccontextmanager
 from functools import partial
+from typing import TYPE_CHECKING
 
 from starlette.applications import Starlette
 from starlette.requests import Request
@@ -21,10 +22,13 @@
 from .protocol import ACP_ENDPOINT_PATH, CONNECTION_ID_HEADER, CONTENT_TYPE_SSE, SESSION_ID_HEADER
 from .server import AcpServer, AgentFactory
 
+if TYPE_CHECKING:
+    from ..experimental.negotiation import AgentProtocolRouter
+
 __all__ = ["create_asgi_app"]
 
 
-def create_asgi_app(agent_factory: AgentFactory, *, path: str = ACP_ENDPOINT_PATH) -> Starlette:
+def create_asgi_app(agent_factory: AgentFactory | AgentProtocolRouter, *, path: str = ACP_ENDPOINT_PATH) -> Starlette:
     """Create a Starlette app with one agent instance per connection.
 
     The app handles POST/GET/DELETE and WebSocket at ``path`` (default: /acp).

src/acp/http/server.py

@@ -45,6 +45,7 @@
 if TYPE_CHECKING:
     from collections.abc import AsyncGenerator
 
+    from ..experimental.negotiation import AgentProtocolConnection, AgentProtocolRouter
     from ..interfaces import Agent
 
 __all__ = [
@@ -224,9 +225,17 @@ class AcpServer:
             ``AgentSideConnection`` to produce a per-connection ``Agent``.
     """
 
-    def __init__(self, agent_factory: AgentFactory) -> None:
+    def __init__(self, agent_factory: AgentFactory | AgentProtocolRouter) -> None:
         self.agent_factory = agent_factory
-        self._connections: dict[str, tuple[AgentSideConnection, _HttpTransport]] = {}
+        self._connections: dict[str, tuple[AgentSideConnection | AgentProtocolConnection, _HttpTransport]] = {}
+
+    def _connect(self, transport: _HttpTransport) -> AgentSideConnection | AgentProtocolConnection:
+        """Bind one agent (v1) or protocol router (v1/v2) to a transport."""
+        from ..experimental.negotiation import AgentProtocolRouter
+
+        if isinstance(self.agent_factory, AgentProtocolRouter):
+            return self.agent_factory.connect(transport)
+        return AgentSideConnection(self.agent_factory, transport)
 
     # -- POST ---------------------------------------------------------------
 
@@ -268,7 +277,7 @@ async def handle_post(
     async def _handle_initialize(self, message: dict[str, Any]) -> Response:
         connection_id = uuid.uuid4().hex
         transport = _HttpTransport(message.get("id"))
-        conn = AgentSideConnection(self.agent_factory, transport)
+        conn = self._connect(transport)
         self._connections[connection_id] = (conn, transport)
         # Deliver initialize to the agent and await its response so we can return
         # the 200 body synchronously (initialize is the one blocking POST). If the

src/acp/ws/server.py

@@ -13,6 +13,7 @@
 from ..http.protocol import CONNECTION_ID_HEADER
 
 if TYPE_CHECKING:
+    from ..experimental.negotiation import AgentProtocolRouter
     from ..http.server import AgentFactory
 
 __all__ = ["handle_websocket"]
@@ -53,11 +54,16 @@ async def close(self) -> None:
             await self._ws.close()
 
 
-async def handle_websocket(agent_factory: AgentFactory, websocket: WebSocket) -> None:
+async def handle_websocket(agent_factory: AgentFactory | AgentProtocolRouter, websocket: WebSocket) -> None:
     """Run one agent for the lifetime of the socket; disconnect cancels its work."""
     await websocket.accept(headers=[(CONNECTION_ID_HEADER.lower().encode(), uuid.uuid4().hex.encode())])
     transport = _WebSocketTransport(websocket)
-    conn = AgentSideConnection(agent_factory, transport, listening=False)
+    from ..experimental.negotiation import AgentProtocolRouter
+
+    if isinstance(agent_factory, AgentProtocolRouter):
+        conn = agent_factory.connect(transport, listening=False)
+    else:
+        conn = AgentSideConnection(agent_factory, transport, listening=False)
     try:
         await conn.listen()
     finally:

tests/http/test_v2_loopback.py

@@ -0,0 +1,226 @@
+"""v2 end-to-end loopback over the version-agnostic HTTP/WS transports.
+
+The v2 runtime is served through an :class:`AgentProtocolRouter` instead of a
+v1 ``AgentFactory``; the ASGI server must negotiate v2 from ``initialize`` and
+route ``session/update`` traffic (and ``session/resume`` replay) the same way it
+already routes v1's ``session/load``.
+"""
+
+from __future__ import annotations
+
+import asyncio
+from typing import Any
+
+import pytest
+
+import acp
+from acp.experimental import AgentProtocolRouter, v2
+from acp.http.asgi import create_asgi_app
+from acp.http.client import create_http_stream
+from acp.ws.client import create_websocket_stream
+from tests.conftest import TestAgent, TestClient
+
+
+class RoutedV2Agent:
+    def __init__(self) -> None:
+        self.connection: v2.AgentSideConnection | None = None
+
+    def on_connect(self, connection: v2.AgentSideConnection) -> None:
+        self.connection = connection
+
+    async def initialize(
+        self,
+        protocol_version: int,
+        info: v2.schema.Implementation,
+        capabilities: v2.schema.ClientCapabilities | None = None,
+        **kwargs: Any,
+    ) -> v2.schema.InitializeResponse:
+        return v2.schema.InitializeResponse(
+            protocol_version=v2.PROTOCOL_VERSION,
+            info=v2.schema.Implementation(name="v2-agent", version="1.0.0"),
+        )
+
+    async def new_session(
+        self,
+        cwd: str,
+        additional_directories: list[str] | None = None,
+        mcp_servers: list[Any] | None = None,
+        **kwargs: Any,
+    ) -> v2.schema.NewSessionResponse:
+        return v2.schema.NewSessionResponse(session_id="sess-v2")
+
+    async def prompt(
+        self,
+        session_id: str,
+        prompt: list[Any],
+        **kwargs: Any,
+    ) -> v2.schema.PromptResponse:
+        assert self.connection is not None
+        await self.connection.session_update(
+            session_id=session_id,
+            update=v2.schema.AgentMessageChunk(
+                message_id="msg-1",
+                content=v2.schema.TextContentBlock(text="hello-v2"),
+            ),
+        )
+        return v2.schema.PromptResponse(message_id="user-msg-1")
+
+    async def resume_session(
+        self,
+        session_id: str,
+        cwd: str,
+        additional_directories: list[str] | None = None,
+        mcp_servers: list[Any] | None = None,
+        replay_from: Any = None,
+        **kwargs: Any,
+    ) -> v2.schema.ResumeSessionResponse:
+        assert self.connection is not None
+        for i in range(3):
+            await self.connection.session_update(
+                session_id=session_id,
+                update=v2.schema.AgentMessageChunk(
+                    message_id=f"history-{i}",
+                    content=v2.schema.TextContentBlock(text=f"history-{i}"),
+                ),
+            )
+        return v2.schema.ResumeSessionResponse()
+
+
+class CapturingClient:
+    def __init__(self) -> None:
+        self.updates: list[tuple[str, Any]] = []
+
+    async def session_update(self, session_id: str, update: Any, **kwargs: Any) -> None:
+        self.updates.append((session_id, update))
+
+
+def _make_app(router: AgentProtocolRouter) -> Any:
+    return create_asgi_app(router)
+
+
+def _v2_agent_factory(connection: v2.AgentSideConnection) -> RoutedV2Agent:
+    return RoutedV2Agent()
+
+
+def _router() -> AgentProtocolRouter:
+    return AgentProtocolRouter(v2=_v2_agent_factory)
+
+
+async def _v2_client(transport: Any) -> tuple[v2.ClientSideConnection, CapturingClient]:
+    client = CapturingClient()
+    conn = v2.connect_to_agent(client, transport)
+    return conn, client
+
+
+@pytest.mark.asyncio
+async def test_v2_http_prompt_streams_update(serve_asgi) -> None:
+    server = await serve_asgi(_make_app(_router()))
+    transport = create_http_stream(server.http_url)
+    conn, client = await _v2_client(transport)
+    try:
+        init = await asyncio.wait_for(
+            conn.initialize(
+                protocol_version=v2.PROTOCOL_VERSION,
+                info=v2.schema.Implementation(name="v2-client", version="1.0.0"),
+            ),
+            timeout=10,
+        )
+        assert init.protocol_version == v2.PROTOCOL_VERSION
+        new = await asyncio.wait_for(conn.new_session(cwd="."), timeout=10)
+        assert new.session_id == "sess-v2"
+        result = await asyncio.wait_for(
+            conn.prompt(session_id=new.session_id, prompt=[v2.schema.TextContentBlock(text="hi")]),
+            timeout=10,
+        )
+        assert result.message_id == "user-msg-1"
+        await asyncio.sleep(0.2)
+        assert client.updates, "expected a session/update over SSE"
+        assert client.updates[0][0] == "sess-v2"
+        assert client.updates[0][1].content.text == "hello-v2"
+    finally:
+        await conn.close()
+        await transport.close()
+
+
+@pytest.mark.asyncio
+async def test_v2_websocket_prompt_streams_update(serve_asgi) -> None:
+    server = await serve_asgi(_make_app(_router()))
+    transport = await create_websocket_stream(server.ws_url)
+    conn, client = await _v2_client(transport)
+    try:
+        await asyncio.wait_for(
+            conn.initialize(
+                protocol_version=v2.PROTOCOL_VERSION,
+                info=v2.schema.Implementation(name="v2-client", version="1.0.0"),
+            ),
+            timeout=10,
+        )
+        new = await asyncio.wait_for(conn.new_session(cwd="."), timeout=10)
+        await asyncio.wait_for(
+            conn.prompt(session_id=new.session_id, prompt=[v2.schema.TextContentBlock(text="hi")]),
+            timeout=10,
+        )
+        await asyncio.sleep(0.2)
+        assert client.updates, "expected a session/update over WS"
+        assert client.updates[0][1].content.text == "hello-v2"
+    finally:
+        await conn.close()
+        await transport.close()
+
+
+@pytest.mark.asyncio
+async def test_v2_http_resume_replays_history(serve_asgi) -> None:
+    server = await serve_asgi(_make_app(_router()))
+    transport = create_http_stream(server.http_url)
+    conn, client = await _v2_client(transport)
+    try:
+        await asyncio.wait_for(
+            conn.initialize(
+                protocol_version=v2.PROTOCOL_VERSION,
+                info=v2.schema.Implementation(name="v2-client", version="1.0.0"),
+            ),
+            timeout=10,
+        )
+        loaded = await asyncio.wait_for(
+            conn.resume_session(session_id="saved-session", cwd="/"),
+            timeout=10,
+        )
+        assert loaded == v2.schema.ResumeSessionResponse()
+        assert [update.content.text for _, update in client.updates] == [f"history-{i}" for i in range(3)]
+    finally:
+        await conn.close()
+        await transport.close()
+
+
+@pytest.mark.asyncio
+async def test_dual_stack_http_serves_v1_and_v2(serve_asgi) -> None:
+    router = AgentProtocolRouter(v1=lambda conn: TestAgent(), v2=_v2_agent_factory)
+    server = await serve_asgi(create_asgi_app(router))
+
+    v1_transport = create_http_stream(server.http_url)
+    v1_conn = acp.connect_to_agent(TestClient(), v1_transport)
+    try:
+        init = await asyncio.wait_for(v1_conn.initialize(protocol_version=acp.PROTOCOL_VERSION), timeout=10)
+        assert init.protocol_version == acp.PROTOCOL_VERSION
+        new = await asyncio.wait_for(v1_conn.new_session(cwd="."), timeout=10)
+        assert new.session_id == "test-session-123"
+    finally:
+        await v1_conn.close()
+        await v1_transport.close()
+
+    v2_transport = create_http_stream(server.http_url)
+    v2_conn, _ = await _v2_client(v2_transport)
+    try:
+        init = await asyncio.wait_for(
+            v2_conn.initialize(
+                protocol_version=v2.PROTOCOL_VERSION,
+                info=v2.schema.Implementation(name="v2-client", version="1.0.0"),
+            ),
+            timeout=10,
+        )
+        assert init.protocol_version == v2.PROTOCOL_VERSION
+        new = await asyncio.wait_for(v2_conn.new_session(cwd="."), timeout=10)
+        assert new.session_id == "sess-v2"
+    finally:
+        await v2_conn.close()
+        await v2_transport.close()