feat: Expose new APIs for running agent and connect with client

Signed-off-by: Frost Ming <me@frostming.com>

Author: frostming

docs/quickstart.md

@@ -90,7 +90,7 @@ class SimpleClient(Client):
 
 async def main() -> None:
     script = Path("examples/echo_agent.py")
-    async with spawn_agent_process(lambda _agent: SimpleClient(), sys.executable, str(script)) as (conn, _proc):
+    async with spawn_agent_process(SimpleClient(), sys.executable, str(script)) as (conn, _proc):
         await conn.initialize(protocol_version=1)
         session = await conn.new_session(cwd=str(script.parent), mcp_servers=[])
         await conn.prompt(
@@ -119,7 +119,7 @@ class MyAgent(Agent):
         return PromptResponse(stop_reason="end_turn")
 ```
 
-Hook it up with `AgentSideConnection` inside an async entrypoint and wire it to your client. Refer to:
+Run it with `run_agent()` inside an async entrypoint and wire it to your client. Refer to:
 
 - [`examples/echo_agent.py`](https://github.com/agentclientprotocol/python-sdk/blob/main/examples/echo_agent.py) for the smallest streaming agent
 - [`examples/agent.py`](https://github.com/agentclientprotocol/python-sdk/blob/main/examples/agent.py) for an implementation that negotiates capabilities and streams richer updates

examples/agent.py

@@ -11,11 +11,12 @@
     NewSessionResponse,
     PromptResponse,
     SetSessionModeResponse,
-    stdio_streams,
+    run_agent,
     text_block,
     update_agent_message,
     PROTOCOL_VERSION,
 )
+from acp.interfaces import Client
 from acp.schema import (
     AgentCapabilities,
     AgentMessageChunk,
@@ -33,11 +34,15 @@
 
 
 class ExampleAgent(Agent):
-    def __init__(self, conn: AgentSideConnection) -> None:
-        self._conn = conn
+    _conn: Client
+
+    def __init__(self) -> None:
         self._next_session_id = 0
         self._sessions: set[str] = set()
 
+    def on_connect(self, conn: Client) -> None:
+        self._conn = conn
+
     async def _send_agent_message(self, session_id: str, content: Any) -> None:
         update = content if isinstance(content, AgentMessageChunk) else update_agent_message(content)
         await self._conn.session_update(session_id, update)
@@ -114,9 +119,7 @@ async def ext_notification(self, method: str, params: dict[str, Any]) -> None:
 
 async def main() -> None:
     logging.basicConfig(level=logging.INFO)
-    reader, writer = await stdio_streams()
-    AgentSideConnection(ExampleAgent, writer, reader)
-    await asyncio.Event().wait()
+    await run_agent(ExampleAgent())
 
 
 if __name__ == "__main__":

examples/client.py

@@ -10,9 +10,7 @@
 from acp import (
     Client,
     ClientSideConnection,
-    InitializeRequest,
-    NewSessionRequest,
-    PromptRequest,
+    connect_to_agent,
     RequestError,
     text_block,
     PROTOCOL_VERSION,
@@ -190,7 +188,7 @@ async def main(argv: list[str]) -> int:
         return 1
 
     client_impl = ExampleClient()
-    conn = ClientSideConnection(lambda _agent: client_impl, proc.stdin, proc.stdout)
+    conn = connect_to_agent(client_impl, proc.stdin, proc.stdout)
 
     await conn.initialize(
         protocol_version=PROTOCOL_VERSION,

examples/duet.py

@@ -30,7 +30,7 @@ async def main() -> int:
     client_module = _load_client_module(root / "client.py")
     client = client_module.ExampleClient()
 
-    async with spawn_agent_process(lambda _agent: client, sys.executable, str(agent_path), env=env) as (
+    async with spawn_agent_process(client, sys.executable, str(agent_path), env=env) as (
         conn,
         process,
     ):

examples/echo_agent.py

@@ -8,10 +8,11 @@
     InitializeResponse,
     NewSessionResponse,
     PromptResponse,
-    stdio_streams,
+    run_agent,
     text_block,
     update_agent_message,
 )
+from acp.interfaces import Client
 from acp.schema import (
     AudioContentBlock,
     ClientCapabilities,
@@ -27,7 +28,9 @@
 
 
 class EchoAgent(Agent):
-    def __init__(self, conn: AgentSideConnection) -> None:
+    _conn: Client
+
+    def on_connect(self, conn: Client) -> None:
         self._conn = conn
 
     async def initialize(
@@ -67,9 +70,7 @@ async def prompt(
 
 
 async def main() -> None:
-    reader, writer = await stdio_streams()
-    AgentSideConnection(EchoAgent, writer, reader)
-    await asyncio.Event().wait()
+    await run_agent(EchoAgent())
 
 
 if __name__ == "__main__":

examples/gemini.py

@@ -13,11 +13,12 @@
 
 from acp import (
     Client,
-    ClientSideConnection,
+    connect_to_agent,
     PROTOCOL_VERSION,
     RequestError,
     text_block,
 )
+from acp.core import ClientSideConnection
 from acp.schema import (
     AgentMessageChunk,
     AgentPlanUpdate,
@@ -316,7 +317,7 @@ async def run(argv: list[str]) -> int:
         return 1
 
     client_impl = GeminiClient(auto_approve=args.yolo)
-    conn = ClientSideConnection(lambda _agent: client_impl, proc.stdin, proc.stdout)
+    conn = connect_to_agent(client_impl, proc.stdin, proc.stdout)
 
     try:
         init_resp = await conn.initialize(

src/acp/__init__.py

@@ -1,10 +1,12 @@
+from typing import Any
+
 from .core import (
     Agent,
-    AgentSideConnection,
     Client,
-    ClientSideConnection,
     RequestError,
     TerminalHandle,
+    connect_to_agent,
+    run_agent,
 )
 from .helpers import (
     audio_block,
@@ -73,6 +75,19 @@
 from .stdio import spawn_agent_process, spawn_client_process, spawn_stdio_connection, stdio_streams
 from .transports import default_environment, spawn_stdio_transport
 
+_DEPRECATED_NAMES = [
+    (
+        "AgentSideConnection",
+        "acp.core:AgentSideConnection",
+        "Using `AgentSideConnection` directly is deprecated, please use `acp.run_agent` instead.",
+    ),
+    (
+        "ClientSideConnection",
+        "acp.core:ClientSideConnection",
+        "Using `ClientSideConnection` directly is deprecated, please use `acp.connect_to_agent` instead.",
+    ),
+]
+
 __all__ = [  # noqa: RUF022
     # constants
     "PROTOCOL_VERSION",
@@ -113,8 +128,8 @@
     "ReleaseTerminalRequest",
     "ReleaseTerminalResponse",
     # core
-    "AgentSideConnection",
-    "ClientSideConnection",
+    "run_agent",
+    "connect_to_agent",
     "RequestError",
     "Agent",
     "Client",
@@ -151,3 +166,16 @@
     "start_edit_tool_call",
     "update_tool_call",
 ]
+
+
+def __getattr__(name: str) -> Any:
+    import warnings
+    from importlib import import_module
+
+    for deprecated_name, new_path, warning in _DEPRECATED_NAMES:
+        if name == deprecated_name:
+            warnings.warn(warning, DeprecationWarning, stacklevel=2)
+            module_name, attr_name = new_path.split(":")
+            module = import_module(module_name)
+            return getattr(module, attr_name)
+    raise AttributeError(f"module {__name__} has no attribute {name}")  # noqa: TRY003

src/acp/agent/connection.py

@@ -2,10 +2,10 @@
 
 import asyncio
 from collections.abc import Callable
-from typing import Any
+from typing import Any, cast, final
 
 from ..connection import Connection
-from ..interfaces import Agent
+from ..interfaces import Agent, Client
 from ..meta import CLIENT_METHODS
 from ..schema import (
     AgentMessageChunk,
@@ -38,28 +38,37 @@
     WriteTextFileResponse,
 )
 from ..terminal import TerminalHandle
-from ..utils import notify_model, param_model, request_model, request_optional_model
+from ..utils import compatible_class, notify_model, param_model, request_model, request_optional_model
 from .router import build_agent_router
 
 __all__ = ["AgentSideConnection"]
 _AGENT_CONNECTION_ERROR = "AgentSideConnection requires asyncio StreamWriter/StreamReader"
 
 
+@final
+@compatible_class
 class AgentSideConnection:
     """Agent-side connection wrapper that dispatches JSON-RPC messages to a Client implementation."""
 
     def __init__(
         self,
-        to_agent: Callable[[AgentSideConnection], Agent],
+        to_agent: Callable[[Client], Agent] | Agent,
         input_stream: Any,
         output_stream: Any,
+        listening: bool = False,
         **connection_kwargs: Any,
     ) -> None:
-        agent = to_agent(self)
-        handler = build_agent_router(agent)
+        agent = to_agent(cast(Client, self)) if callable(to_agent) else to_agent
         if not isinstance(input_stream, asyncio.StreamWriter) or not isinstance(output_stream, asyncio.StreamReader):
             raise TypeError(_AGENT_CONNECTION_ERROR)
-        self._conn = Connection(handler, input_stream, output_stream, **connection_kwargs)
+        handler = build_agent_router(agent)  # type: ignore[arg-type]
+        self._conn = Connection(handler, input_stream, output_stream, listening=listening, **connection_kwargs)
+        if on_connect := getattr(agent, "on_connect", None):
+            on_connect(self)
+
+    async def listen(self) -> None:
+        """Start listening for incoming messages."""
+        await self._conn.main_loop()
 
     @param_model(SessionNotification)
     async def session_update(

src/acp/client/connection.py

@@ -2,7 +2,7 @@
 
 import asyncio
 from collections.abc import Callable
-from typing import Any
+from typing import Any, cast, final
 
 from ..connection import Connection
 from ..interfaces import Agent, Client
@@ -34,24 +34,32 @@
     StdioMcpServer,
     TextContentBlock,
 )
-from ..utils import notify_model, param_model, request_model, request_model_from_dict
+from ..utils import compatible_class, notify_model, param_model, request_model, request_model_from_dict
 from .router import build_client_router
 
 __all__ = ["ClientSideConnection"]
 _CLIENT_CONNECTION_ERROR = "ClientSideConnection requires asyncio StreamWriter/StreamReader"
 
 
+@final
+@compatible_class
 class ClientSideConnection:
     """Client-side connection wrapper that dispatches JSON-RPC messages to an Agent implementation."""
 
     def __init__(
-        self, to_client: Callable[[Agent], Client], input_stream: Any, output_stream: Any, **connection_kwargs: Any
+        self,
+        to_client: Callable[[Agent], Client] | Client,
+        input_stream: Any,
+        output_stream: Any,
+        **connection_kwargs: Any,
     ) -> None:
         if not isinstance(input_stream, asyncio.StreamWriter) or not isinstance(output_stream, asyncio.StreamReader):
             raise TypeError(_CLIENT_CONNECTION_ERROR)
-        client = to_client(self)
-        handler = build_client_router(client)
+        client = to_client(cast(Agent, self)) if callable(to_client) else to_client
+        handler = build_client_router(client)  # type: ignore[arg-type]
         self._conn = Connection(handler, input_stream, output_stream, **connection_kwargs)
+        if on_connect := getattr(client, "on_connect", None):
+            on_connect(self)
 
     @param_model(InitializeRequest)
     async def initialize(

src/acp/connection.py

@@ -72,6 +72,7 @@ def __init__(
         dispatcher_factory: DispatcherFactory | None = None,
         sender_factory: SenderFactory | None = None,
         observers: list[StreamObserver] | None = None,
+        listening: bool = True,
     ) -> None:
         self._handler = handler
         self._writer = writer
@@ -83,11 +84,14 @@ def __init__(
         self._queue = queue or InMemoryMessageQueue()
         self._closed = False
         self._sender = (sender_factory or self._default_sender_factory)(self._writer, self._tasks)
-        self._recv_task = self._tasks.create(
-            self._receive_loop(),
-            name="acp.Connection.receive",
-            on_error=self._on_receive_error,
-        )
+        if listening:
+            self._recv_task = self._tasks.create(
+                self._receive_loop(),
+                name="acp.Connection.receive",
+                on_error=self._on_receive_error,
+            )
+        else:
+            self._recv_task = None
         dispatcher_factory = dispatcher_factory or self._default_dispatcher_factory
         self._dispatcher = dispatcher_factory(
             self._queue,
@@ -109,6 +113,14 @@ async def close(self) -> None:
         await self._tasks.shutdown()
         self._state.reject_all_outgoing(ConnectionError("Connection closed"))
 
+    async def main_loop(self) -> None:
+        try:
+            await self._receive_loop()
+        except Exception as exc:
+            logging.exception("Connection main loop failed", exc_info=exc)
+            self._on_receive_error(None, exc)  # type: ignore[arg-type]
+            raise
+
     async def __aenter__(self) -> Connection:
         return self