refactor: update examples for basic echo

Author: Chojan Shang

examples/agent.py

@@ -1,48 +1,97 @@
 import asyncio
+from typing import Any
 
 from acp import (
     Agent,
     AgentSideConnection,
     AuthenticateRequest,
+    AuthenticateResponse,
     CancelNotification,
     InitializeRequest,
     InitializeResponse,
-    LoadSessionRequest,
     NewSessionRequest,
     NewSessionResponse,
     PromptRequest,
     PromptResponse,
+    SessionNotification,
+    SetSessionModeRequest,
+    SetSessionModeResponse,
     stdio_streams,
+    PROTOCOL_VERSION,
 )
+from acp.schema import ContentBlock1, SessionUpdate2
 
 
-class EchoAgent(Agent):
+class ExampleAgent(Agent):
+    def __init__(self, conn: AgentSideConnection) -> None:
+        self._conn = conn
+        self._next_session_id = 0
+
     async def initialize(self, params: InitializeRequest) -> InitializeResponse:
-        # Avoid serializer warnings by omitting defaults
-        return InitializeResponse(protocolVersion=params.protocolVersion)
+        return InitializeResponse(protocolVersion=PROTOCOL_VERSION, agentCapabilities=None, authMethods=[])
 
-    async def newSession(self, params: NewSessionRequest) -> NewSessionResponse:
-        return NewSessionResponse(sessionId="sess-1")
+    async def authenticate(self, params: AuthenticateRequest) -> AuthenticateResponse | None:  # noqa: ARG002
+        return {}
 
-    async def loadSession(self, params: LoadSessionRequest) -> None:
-        return None
+    async def newSession(self, params: NewSessionRequest) -> NewSessionResponse:  # noqa: ARG002
+        session_id = f"sess-{self._next_session_id}"
+        self._next_session_id += 1
+        return NewSessionResponse(sessionId=session_id)
 
-    async def authenticate(self, params: AuthenticateRequest) -> None:
+    async def loadSession(self, params):  # type: ignore[override]
         return None
 
+    async def setSessionMode(self, params: SetSessionModeRequest) -> SetSessionModeResponse | None:  # noqa: ARG002
+        return {}
+
     async def prompt(self, params: PromptRequest) -> PromptResponse:
-        # Normally you'd stream updates via sessionUpdate
+        # Stream a couple of agent message chunks, then end the turn
+        # 1) Prefix
+        await self._conn.sessionUpdate(
+            SessionNotification(
+                sessionId=params.sessionId,
+                update=SessionUpdate2(
+                    sessionUpdate="agent_message_chunk",
+                    content=ContentBlock1(type="text", text="Client sent: "),
+                ),
+            )
+        )
+        # 2) Echo text blocks
+        for block in params.prompt:
+            if isinstance(block, dict):
+                # tolerate raw dicts
+                if block.get("type") == "text":
+                    text = str(block.get("text", ""))
+                else:
+                    text = f"<{block.get('type', 'content')}>"
+            else:
+                # pydantic model ContentBlock1
+                text = getattr(block, "text", "<content>")
+            await self._conn.sessionUpdate(
+                SessionNotification(
+                    sessionId=params.sessionId,
+                    update=SessionUpdate2(
+                        sessionUpdate="agent_message_chunk",
+                        content=ContentBlock1(type="text", text=text),
+                    ),
+                )
+            )
         return PromptResponse(stopReason="end_turn")
 
-    async def cancel(self, params: CancelNotification) -> None:
+    async def cancel(self, params: CancelNotification) -> None:  # noqa: ARG002
+        return None
+
+    async def extMethod(self, method: str, params: dict) -> dict:  # noqa: ARG002
+        return {"example": "response"}
+
+    async def extNotification(self, method: str, params: dict) -> None:  # noqa: ARG002
         return None
 
 
 async def main() -> None:
     reader, writer = await stdio_streams()
     # For an agent process, local writes go to client stdin (writer=stdout)
-    AgentSideConnection(lambda _conn: EchoAgent(), writer, reader)
-    # Keep running; in a real agent you would await tasks or add your own loop
+    AgentSideConnection(lambda conn: ExampleAgent(conn), writer, reader)
     await asyncio.Event().wait()
 
 

examples/client.py

@@ -1,6 +1,7 @@
 import asyncio
 import os
 import sys
+from typing import Optional
 
 from acp import (
     Client,
@@ -9,64 +10,66 @@
     InitializeRequest,
     NewSessionRequest,
     PromptRequest,
-    ReadTextFileRequest,
-    ReadTextFileResponse,
-    RequestPermissionRequest,
-    RequestPermissionResponse,
     SessionNotification,
-    WriteTextFileRequest,
-    stdio_streams,
 )
 
 
-class MinimalClient(Client):
-    async def writeTextFile(self, params: WriteTextFileRequest) -> None:
-        print(f"write {params.path}", file=sys.stderr)
-
-    async def readTextFile(self, params: ReadTextFileRequest) -> ReadTextFileResponse:
-        return ReadTextFileResponse(content="example")
+class ExampleClient(Client):
+    async def sessionUpdate(self, params: SessionNotification) -> None:
+        update = params.update
+        kind = getattr(update, "sessionUpdate", None) if not isinstance(update, dict) else update.get("sessionUpdate")
+        if kind == "agent_message_chunk":
+            # Handle both dict and model shapes
+            content = update["content"] if isinstance(update, dict) else getattr(update, "content", None)
+            text = content.get("text") if isinstance(content, dict) else getattr(content, "text", "<content>")
+            print(f"| Agent: {text}")
 
-    async def requestPermission(self, params: RequestPermissionRequest) -> RequestPermissionResponse:
-        return RequestPermissionResponse.model_validate({"outcome": {"outcome": "selected", "optionId": "allow"}})
 
-    async def sessionUpdate(self, params: SessionNotification) -> None:
-        print(f"session update: {params}", file=sys.stderr)
+async def interactive_loop(conn: ClientSideConnection, session_id: str) -> None:
+    loop = asyncio.get_running_loop()
+    while True:
+        try:
+            line = await loop.run_in_executor(None, lambda: input("> "))
+        except EOFError:
+            break
+        if not line:
+            continue
+        try:
+            await conn.prompt(PromptRequest(sessionId=session_id, prompt=[{"type": "text", "text": line}]))
+        except Exception as e:  # noqa: BLE001
+            print(f"error: {e}", file=sys.stderr)
 
-    # Optional terminal methods (not implemented in this minimal client)
-    async def createTerminal(self, params) -> None:
-        pass
 
-    async def terminalOutput(self, params) -> None:
-        pass
+async def main(argv: list[str]) -> int:
+    if len(argv) < 2:
+        print("Usage: python examples/client.py AGENT_PROGRAM [ARGS...]", file=sys.stderr)
+        return 2
 
-    async def releaseTerminal(self, params) -> None:
-        pass
+    # Spawn agent subprocess
+    proc = await asyncio.create_subprocess_exec(
+        sys.executable,
+        *argv[1:],
+        stdin=asyncio.subprocess.PIPE,
+        stdout=asyncio.subprocess.PIPE,
+    )
+    assert proc.stdin and proc.stdout
 
-    async def waitForTerminalExit(self, params) -> None:
-        pass
+    # Connect to agent stdio
+    conn = ClientSideConnection(lambda _agent: ExampleClient(), proc.stdin, proc.stdout)
 
-    async def killTerminal(self, params) -> None:
-        pass
+    # Initialize and create session
+    await conn.initialize(InitializeRequest(protocolVersion=PROTOCOL_VERSION, clientCapabilities=None))
+    new_sess = await conn.newSession(NewSessionRequest(mcpServers=[], cwd=os.getcwd()))
 
+    # Run REPL until EOF
+    await interactive_loop(conn, new_sess.sessionId)
 
-async def main() -> None:
-    reader, writer = await stdio_streams()
-    client_conn = ClientSideConnection(lambda _agent: MinimalClient(), writer, reader)
-    # 1) initialize
-    resp = await client_conn.initialize(InitializeRequest(protocolVersion=PROTOCOL_VERSION))
-    print(f"Initialized with protocol version: {resp.protocolVersion}", file=sys.stderr)
-    # 2) new session
-    new_sess = await client_conn.newSession(NewSessionRequest(mcpServers=[], cwd=os.getcwd()))
-    # 3) prompt
-    await client_conn.prompt(
-        PromptRequest(
-            sessionId=new_sess.sessionId,
-            prompt=[{"type": "text", "text": "Hello from client"}],
-        )
-    )
-    # Small grace period to allow duplex messages to flush
-    await asyncio.sleep(0.2)
+    try:
+        proc.terminate()
+    except ProcessLookupError:
+        pass
+    return 0
 
 
 if __name__ == "__main__":
-    asyncio.run(main())
+    raise SystemExit(asyncio.run(main(sys.argv)))

examples/duet.py

@@ -1,41 +1,10 @@
 import asyncio
-import contextlib
-import json
 import os
-import signal
 import sys
 from pathlib import Path
 
 
-async def _relay(reader: asyncio.StreamReader, writer: asyncio.StreamWriter, tag: str):
-    try:
-        while True:
-            line = await reader.readline()
-            if not line:
-                break
-            # Mirror to the other end unchanged
-            writer.write(line)
-            try:
-                await writer.drain()
-            except ConnectionError:
-                break
-            # Try to pretty-print the JSON-RPC message for visibility
-            try:
-                obj = json.loads(line.decode("utf-8", errors="replace"))
-                pretty = json.dumps(obj, ensure_ascii=False, indent=2)
-                print(f"[{tag}] {pretty}", file=sys.stderr)
-            except Exception:
-                # Non-JSON (shouldn't happen on the protocol stream)
-                print(f"[{tag}] {line!r}", file=sys.stderr)
-    finally:
-        try:
-            writer.close()
-            await writer.wait_closed()
-        except Exception:
-            pass
-
-
-async def main() -> None:
+async def main() -> int:
     root = Path(__file__).resolve().parent
     agent_path = str(root / "agent.py")
     client_path = str(root / "client.py")
@@ -45,68 +14,15 @@ async def main() -> None:
     src_dir = str((root.parent / "src").resolve())
     env["PYTHONPATH"] = src_dir + os.pathsep + env.get("PYTHONPATH", "")
 
-    agent = await asyncio.create_subprocess_exec(
-        sys.executable,
-        agent_path,
-        stdin=asyncio.subprocess.PIPE,
-        stdout=asyncio.subprocess.PIPE,
-        stderr=sys.stderr,
-        env=env,
-    )
-    client = await asyncio.create_subprocess_exec(
+    # Run the client and let it spawn the agent, wiring stdio automatically.
+    proc = await asyncio.create_subprocess_exec(
         sys.executable,
         client_path,
-        stdin=asyncio.subprocess.PIPE,
-        stdout=asyncio.subprocess.PIPE,
-        stderr=sys.stderr,
+        agent_path,
         env=env,
     )
-
-    assert agent.stdout and agent.stdin and client.stdout and client.stdin
-
-    # Wire: agent.stdout -> client.stdin, client.stdout -> agent.stdin
-    t1 = asyncio.create_task(_relay(agent.stdout, client.stdin, "agent→client"))
-    t2 = asyncio.create_task(_relay(client.stdout, agent.stdin, "client→agent"))
-
-    # Handle shutdown
-    stop = asyncio.Event()
-
-    def _on_sigint(*_):
-        stop.set()
-
-    loop = asyncio.get_running_loop()
-    try:
-        loop.add_signal_handler(signal.SIGINT, _on_sigint)
-        loop.add_signal_handler(signal.SIGTERM, _on_sigint)
-    except NotImplementedError:
-        pass
-
-    done, _ = await asyncio.wait(
-        {
-            t1,
-            t2,
-            asyncio.create_task(agent.wait()),
-            asyncio.create_task(client.wait()),
-            asyncio.create_task(stop.wait()),
-        },
-        return_when=asyncio.FIRST_COMPLETED,
-    )
-
-    # Teardown
-    for proc in (agent, client):
-        if proc.returncode is None:
-            with contextlib.suppress(ProcessLookupError):
-                proc.terminate()
-            try:
-                await asyncio.wait_for(proc.wait(), 2)
-            except asyncio.TimeoutError:
-                with contextlib.suppress(ProcessLookupError):
-                    proc.kill()
-    for task in (t1, t2):
-        task.cancel()
-        with contextlib.suppress(asyncio.CancelledError):
-            await task
+    return await proc.wait()
 
 
 if __name__ == "__main__":
-    asyncio.run(main())
+    raise SystemExit(asyncio.run(main()))