feat: Auto-generate agent/client methods based on the schema

.pre-commit-config.yaml

@@ -21,4 +21,3 @@ repos:
         args: [ --exit-non-zero-on-fix ]
         exclude: ^src/acp/(meta|schema)\.py$
       - id: ruff-format
-        exclude: ^src/acp/(meta|schema)\.py$

Makefile

@@ -8,6 +8,8 @@ install: ## Install the virtual environment and install the pre-commit hooks
 gen-all: ## Generate all code from schema
 	@echo "🚀 Generating all code"
 	@uv run scripts/gen_all.py
+	@uv run ruff check --fix
+	@uv run ruff format .
 
 .PHONY: check
 check: ## Run code quality tools.

docs/quickstart.md

@@ -76,27 +76,26 @@ from pathlib import Path
 
 from acp import spawn_agent_process, text_block
 from acp.interfaces import Client
-from acp.schema import InitializeRequest, NewSessionRequest, PromptRequest, SessionNotification
 
 
 class SimpleClient(Client):
-    async def requestPermission(self, params):  # pragma: no cover - minimal stub
+    async def request_permission(
+        self, options, session_id, tool_call, **kwargs: Any
+    )
         return {"outcome": {"outcome": "cancelled"}}
 
-    async def sessionUpdate(self, params: SessionNotification) -> None:
-        print("update:", params.session_id, params.update)
+    async def session_update(self, session_id, update, **kwargs):
+        print("update:", session_id, update)
 
 
 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):
-        await conn.initialize(InitializeRequest(protocol_version=1))
-        session = await conn.newSession(NewSessionRequest(cwd=str(script.parent), mcp_servers=[]))
+        await conn.initialize(protocol_version=1)
+        session = await conn.new_session(cwd=str(script.parent), mcp_servers=[])
         await conn.prompt(
-            PromptRequest(
-                session_id=session.session_id,
-                prompt=[text_block("Hello from spawn!")],
-            )
+            session_id=session.session_id,
+            prompt=[text_block("Hello from spawn!")],
         )
 
 asyncio.run(main())
@@ -111,12 +110,12 @@ _Swap the echo demo for your own `Agent` subclass._
 Create your own agent by subclassing `acp.Agent`. The pattern mirrors the echo example:
 
 ```python
-from acp import Agent, PromptRequest, PromptResponse
+from acp import Agent, PromptResponse
 
 
 class MyAgent(Agent):
-    async def prompt(self, params: PromptRequest) -> PromptResponse:
-        # inspect params.prompt, stream updates, then finish the turn
+    async def prompt(self, prompt, session_id, **kwargs) -> PromptResponse:
+        # inspect prompt, stream updates, then finish the turn
         return PromptResponse(stop_reason="end_turn")
 ```
 

examples/agent.py

@@ -5,26 +5,31 @@
 from acp import (
     Agent,
     AgentSideConnection,
-    AuthenticateRequest,
     AuthenticateResponse,
-    CancelNotification,
-    InitializeRequest,
     InitializeResponse,
-    LoadSessionRequest,
     LoadSessionResponse,
-    NewSessionRequest,
     NewSessionResponse,
-    PromptRequest,
     PromptResponse,
-    SetSessionModeRequest,
     SetSessionModeResponse,
-    session_notification,
     stdio_streams,
     text_block,
     update_agent_message,
     PROTOCOL_VERSION,
 )
-from acp.schema import AgentCapabilities, AgentMessageChunk, Implementation
+from acp.schema import (
+    AgentCapabilities,
+    AgentMessageChunk,
+    AudioContentBlock,
+    ClientCapabilities,
+    EmbeddedResourceContentBlock,
+    HttpMcpServer,
+    ImageContentBlock,
+    Implementation,
+    ResourceContentBlock,
+    SseMcpServer,
+    StdioMcpServer,
+    TextContentBlock,
+)
 
 
 class ExampleAgent(Agent):
@@ -35,54 +40,75 @@ def __init__(self, conn: AgentSideConnection) -> None:
 
     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.sessionUpdate(session_notification(session_id, update))
-
-    async def initialize(self, params: InitializeRequest) -> InitializeResponse:  # noqa: ARG002
+        await self._conn.session_update(session_id, update)
+
+    async def initialize(
+        self,
+        protocol_version: int,
+        client_capabilities: ClientCapabilities | None = None,
+        client_info: Implementation | None = None,
+        **kwargs: Any,
+    ) -> InitializeResponse:
         logging.info("Received initialize request")
         return InitializeResponse(
             protocol_version=PROTOCOL_VERSION,
             agent_capabilities=AgentCapabilities(),
             agent_info=Implementation(name="example-agent", title="Example Agent", version="0.1.0"),
         )
 
-    async def authenticate(self, params: AuthenticateRequest) -> AuthenticateResponse | None:  # noqa: ARG002
-        logging.info("Received authenticate request %s", params.method_id)
+    async def authenticate(self, method_id: str, **kwargs: Any) -> AuthenticateResponse | None:
+        logging.info("Received authenticate request %s", method_id)
         return AuthenticateResponse()
 
-    async def newSession(self, params: NewSessionRequest) -> NewSessionResponse:  # noqa: ARG002
+    async def new_session(
+        self, cwd: str, mcp_servers: list[HttpMcpServer | SseMcpServer | StdioMcpServer], **kwargs: Any
+    ) -> NewSessionResponse:
         logging.info("Received new session request")
         session_id = str(self._next_session_id)
         self._next_session_id += 1
         self._sessions.add(session_id)
         return NewSessionResponse(session_id=session_id, modes=None)
 
-    async def loadSession(self, params: LoadSessionRequest) -> LoadSessionResponse | None:  # noqa: ARG002
-        logging.info("Received load session request %s", params.session_id)
-        self._sessions.add(params.session_id)
+    async def load_session(
+        self, cwd: str, mcp_servers: list[HttpMcpServer | SseMcpServer | StdioMcpServer], session_id: str, **kwargs: Any
+    ) -> LoadSessionResponse | None:
+        logging.info("Received load session request %s", session_id)
+        self._sessions.add(session_id)
         return LoadSessionResponse()
 
-    async def setSessionMode(self, params: SetSessionModeRequest) -> SetSessionModeResponse | None:  # noqa: ARG002
-        logging.info("Received set session mode request %s -> %s", params.session_id, params.mode_id)
+    async def set_session_mode(self, mode_id: str, session_id: str, **kwargs: Any) -> SetSessionModeResponse | None:
+        logging.info("Received set session mode request %s -> %s", session_id, mode_id)
         return SetSessionModeResponse()
 
-    async def prompt(self, params: PromptRequest) -> PromptResponse:
-        logging.info("Received prompt request for session %s", params.session_id)
-        if params.session_id not in self._sessions:
-            self._sessions.add(params.session_id)
-
-        await self._send_agent_message(params.session_id, text_block("Client sent:"))
-        for block in params.prompt:
-            await self._send_agent_message(params.session_id, block)
+    async def prompt(
+        self,
+        prompt: list[
+            TextContentBlock
+            | ImageContentBlock
+            | AudioContentBlock
+            | ResourceContentBlock
+            | EmbeddedResourceContentBlock
+        ],
+        session_id: str,
+        **kwargs: Any,
+    ) -> PromptResponse:
+        logging.info("Received prompt request for session %s", session_id)
+        if session_id not in self._sessions:
+            self._sessions.add(session_id)
+
+        await self._send_agent_message(session_id, text_block("Client sent:"))
+        for block in prompt:
+            await self._send_agent_message(session_id, block)
         return PromptResponse(stop_reason="end_turn")
 
-    async def cancel(self, params: CancelNotification) -> None:  # noqa: ARG002
-        logging.info("Received cancel notification for session %s", params.session_id)
+    async def cancel(self, session_id: str, **kwargs: Any) -> None:
+        logging.info("Received cancel notification for session %s", session_id)
 
-    async def extMethod(self, method: str, params: dict) -> dict:  # noqa: ARG002
+    async def ext_method(self, method: str, params: dict[str, Any]) -> dict[str, Any]:
         logging.info("Received extension method call: %s", method)
         return {"example": "response"}
 
-    async def extNotification(self, method: str, params: dict) -> None:  # noqa: ARG002
+    async def ext_notification(self, method: str, params: dict[str, Any]) -> None:
         logging.info("Received extension notification: %s", method)
 
 

examples/client.py

@@ -5,6 +5,7 @@
 import os
 import sys
 from pathlib import Path
+from typing import Any
 
 from acp import (
     Client,
@@ -13,49 +14,98 @@
     NewSessionRequest,
     PromptRequest,
     RequestError,
-    SessionNotification,
     text_block,
     PROTOCOL_VERSION,
 )
 from acp.schema import (
     AgentMessageChunk,
+    AgentPlanUpdate,
+    AgentThoughtChunk,
     AudioContentBlock,
+    AvailableCommandsUpdate,
     ClientCapabilities,
+    CreateTerminalResponse,
+    CurrentModeUpdate,
     EmbeddedResourceContentBlock,
+    EnvVariable,
     ImageContentBlock,
     Implementation,
+    KillTerminalCommandResponse,
+    PermissionOption,
+    ReadTextFileResponse,
+    ReleaseTerminalResponse,
+    RequestPermissionResponse,
     ResourceContentBlock,
+    TerminalOutputResponse,
     TextContentBlock,
+    ToolCall,
+    ToolCallProgress,
+    ToolCallStart,
+    UserMessageChunk,
+    WaitForTerminalExitResponse,
+    WriteTextFileResponse,
 )
 
 
 class ExampleClient(Client):
-    async def requestPermission(self, params):  # type: ignore[override]
+    async def request_permission(
+        self, options: list[PermissionOption], session_id: str, tool_call: ToolCall, **kwargs: Any
+    ) -> RequestPermissionResponse:
         raise RequestError.method_not_found("session/request_permission")
 
-    async def writeTextFile(self, params):  # type: ignore[override]
+    async def write_text_file(
+        self, content: str, path: str, session_id: str, **kwargs: Any
+    ) -> WriteTextFileResponse | None:
         raise RequestError.method_not_found("fs/write_text_file")
 
-    async def readTextFile(self, params):  # type: ignore[override]
+    async def read_text_file(
+        self, path: str, session_id: str, limit: int | None = None, line: int | None = None, **kwargs: Any
+    ) -> ReadTextFileResponse:
         raise RequestError.method_not_found("fs/read_text_file")
 
-    async def createTerminal(self, params):  # type: ignore[override]
+    async def create_terminal(
+        self,
+        command: str,
+        session_id: str,
+        args: list[str] | None = None,
+        cwd: str | None = None,
+        env: list[EnvVariable] | None = None,
+        output_byte_limit: int | None = None,
+        **kwargs: Any,
+    ) -> CreateTerminalResponse:
         raise RequestError.method_not_found("terminal/create")
 
-    async def terminalOutput(self, params):  # type: ignore[override]
+    async def terminal_output(self, session_id: str, terminal_id: str, **kwargs: Any) -> TerminalOutputResponse:
         raise RequestError.method_not_found("terminal/output")
 
-    async def releaseTerminal(self, params):  # type: ignore[override]
+    async def release_terminal(
+        self, session_id: str, terminal_id: str, **kwargs: Any
+    ) -> ReleaseTerminalResponse | None:
         raise RequestError.method_not_found("terminal/release")
 
-    async def waitForTerminalExit(self, params):  # type: ignore[override]
+    async def wait_for_terminal_exit(
+        self, session_id: str, terminal_id: str, **kwargs: Any
+    ) -> WaitForTerminalExitResponse:
         raise RequestError.method_not_found("terminal/wait_for_exit")
 
-    async def killTerminal(self, params):  # type: ignore[override]
+    async def kill_terminal(
+        self, session_id: str, terminal_id: str, **kwargs: Any
+    ) -> KillTerminalCommandResponse | None:
         raise RequestError.method_not_found("terminal/kill")
 
-    async def sessionUpdate(self, params: SessionNotification) -> None:
-        update = params.update
+    async def session_update(
+        self,
+        session_id: str,
+        update: UserMessageChunk
+        | AgentMessageChunk
+        | AgentThoughtChunk
+        | ToolCallStart
+        | ToolCallProgress
+        | AgentPlanUpdate
+        | AvailableCommandsUpdate
+        | CurrentModeUpdate,
+        **kwargs: Any,
+    ) -> None:
         if not isinstance(update, AgentMessageChunk):
             return
 
@@ -76,10 +126,10 @@ async def sessionUpdate(self, params: SessionNotification) -> None:
 
         print(f"| Agent: {text}")
 
-    async def extMethod(self, method: str, params: dict) -> dict:  # noqa: ARG002
+    async def ext_method(self, method: str, params: dict) -> dict:
         raise RequestError.method_not_found(method)
 
-    async def extNotification(self, method: str, params: dict) -> None:  # noqa: ARG002
+    async def ext_notification(self, method: str, params: dict) -> None:
         raise RequestError.method_not_found(method)
 
 
@@ -103,10 +153,8 @@ async def interactive_loop(conn: ClientSideConnection, session_id: str) -> None:
 
         try:
             await conn.prompt(
-                PromptRequest(
-                    session_id=session_id,
-                    prompt=[text_block(line)],
-                )
+                session_id=session_id,
+                prompt=[text_block(line)],
             )
         except Exception as exc:  # noqa: BLE001
             logging.error("Prompt failed: %s", exc)
@@ -145,13 +193,11 @@ async def main(argv: list[str]) -> int:
     conn = ClientSideConnection(lambda _agent: client_impl, proc.stdin, proc.stdout)
 
     await conn.initialize(
-        InitializeRequest(
-            protocol_version=PROTOCOL_VERSION,
-            client_capabilities=ClientCapabilities(),
-            client_info=Implementation(name="example-client", title="Example Client", version="0.1.0"),
-        )
+        protocol_version=PROTOCOL_VERSION,
+        client_capabilities=ClientCapabilities(),
+        client_info=Implementation(name="example-client", title="Example Client", version="0.1.0"),
     )
-    session = await conn.newSession(NewSessionRequest(mcp_servers=[], cwd=os.getcwd()))
+    session = await conn.new_session(mcp_servers=[], cwd=os.getcwd())
 
     await interactive_loop(conn, session.session_id)
 

examples/duet.py

@@ -34,9 +34,9 @@ async def main() -> int:
         conn,
         process,
     ):
-        await conn.initialize(InitializeRequest(protocolVersion=PROTOCOL_VERSION, clientCapabilities=None))
-        session = await conn.newSession(NewSessionRequest(mcpServers=[], cwd=str(root)))
-        await client_module.interactive_loop(conn, session.sessionId)
+        await conn.initialize(protocol_version=PROTOCOL_VERSION, client_capabilities=None)
+        session = await conn.new_session(mcp_servers=[], cwd=str(root))
+        await client_module.interactive_loop(conn, session.session_id)
 
     return process.returncode or 0