### Install Agent Client Protocol SDK

Source: https://github.com/agentclientprotocol/python-sdk/blob/main/docs/index.md

Install the SDK using pip or uv. Ensure you have Python installed.

```bash
pip install agent-client-protocol
```

```bash
# or
uv add agent-client-protocol
```

--------------------------------

### Install Project Dependencies

Source: https://github.com/agentclientprotocol/python-sdk/blob/main/CONTRIBUTING.md

Run this command to bootstrap the project tooling, including setting up uv, syncing dependencies, and installing pre-commit hooks.

```bash
make install
```

--------------------------------

### Install agent-client-protocol

Source: https://github.com/agentclientprotocol/python-sdk/blob/main/README.md

Install the agent-client-protocol package using pip or uv.

```bash
pip install agent-client-protocol
```

```bash
uv add agent-client-protocol
```

--------------------------------

### Drive Gemini CLI as ACP Agent (CLI Examples)

Source: https://context7.com/agentclientprotocol/python-sdk/llms.txt

Examples of driving the Gemini CLI as an ACP agent using command-line arguments. Options include auto-approving permissions, selecting models, enabling sandbox mode, overriding binary paths, and enabling debug output.

```bash
# Auto-approve all permissions, use default model
python examples/gemini.py --yolo
```

```bash
# Select a specific model and enable Gemini sandbox
python examples/gemini.py --model gemini-1.5-pro --sandbox
```

```bash
# Override CLI binary path and enable debug output
ACP_GEMINI_BIN=/opt/gemini/bin/gemini python examples/gemini.py --debug
```

--------------------------------

### Update client connection wiring

Source: https://github.com/agentclientprotocol/python-sdk/blob/main/docs/migration-guide-0.7.md

Use the `connect_to_agent` helper function instead of `ClientSideConnection` for connecting clients. This simplifies the connection setup process.

```python
# Old pattern:
conn = ClientSideConnection(lambda conn: MyClient(), proc.stdin, proc.stdout)

# New pattern:
conn = connect_to_agent(MyClient(), proc.stdin, proc.stdout)
```

--------------------------------

### Implement ACP Agent with Basic Methods

Source: https://context7.com/agentclientprotocol/python-sdk/llms.txt

Provides a basic implementation of the `acp.interfaces.Agent` Protocol. This example shows how to handle connection, initialization, authentication, new/load sessions, and prompts, including signaling tool calls.

```python
import asyncio
from typing import Any
from acp import (
    PROTOCOL_VERSION, Agent, AuthenticateResponse, InitializeResponse,
    LoadSessionResponse, NewSessionResponse, PromptResponse, run_agent,
    text_block, update_agent_message, start_tool_call, update_tool_call,
    tool_content,
)
from acp.interfaces import Client
from acp.schema import AgentCapabilities, Implementation, ClientCapabilities

class MyAgent(Agent):
    _conn: Client
    _next_id = 0

    def on_connect(self, conn: Client) -> None:
        self._conn = conn

    async def initialize(self, protocol_version, client_capabilities=None, client_info=None, **kwargs) -> InitializeResponse:
        return InitializeResponse(
            protocol_version=PROTOCOL_VERSION,
            agent_capabilities=AgentCapabilities(),
            agent_info=Implementation(name="my-agent", title="My Agent", version="1.0.0"),
        )

    async def authenticate(self, method_id, **kwargs) -> AuthenticateResponse | None:
        return AuthenticateResponse()  # accept all auth methods

    async def new_session(self, cwd, mcp_servers, **kwargs) -> NewSessionResponse:
        self._next_id += 1
        return NewSessionResponse(session_id=str(self._next_id))

    async def load_session(self, cwd, session_id, mcp_servers=None, **kwargs) -> LoadSessionResponse | None:
        return LoadSessionResponse()

    async def prompt(self, prompt, session_id, **kwargs) -> PromptResponse:
        # Signal a tool call, do work, stream output, finish
        call_id = f"call-{session_id}"
        await self._conn.session_update(session_id, start_tool_call(call_id, "Process request", kind="read", status="pending"))
        # … do actual work here …
        await self._conn.session_update(session_id, update_tool_call(call_id, status="completed",
                                                                      content=[tool_content(text_block("Done."))]))
        await self._conn.session_update(session_id, update_agent_message(text_block("All done!")))
        return PromptResponse(stop_reason="end_turn")

    async def cancel(self, session_id, **kwargs) -> None:
        pass  # cancel in-flight operations here

    async def ext_method(self, method, params) -> dict[str, Any]:
        return {}

    async def ext_notification(self, method, params) -> None:
        pass

asyncio.run(run_agent(MyAgent()))

```

--------------------------------

### Start a File Read Tool Call

Source: https://context7.com/agentclientprotocol/python-sdk/llms.txt

A convenience wrapper to start a tool call specifically for reading a file. It automatically sets the kind to 'read' and includes the file path in the locations.

```python
read_start = start_read_tool_call("call-002", "Read README.md", "/project/README.md")
```

--------------------------------

### Start a File Edit Tool Call

Source: https://context7.com/agentclientprotocol/python-sdk/llms.txt

A convenience wrapper to start a tool call for editing a file. It takes the file path and the new content as arguments.

```python
edit_start = start_edit_tool_call(
    "call-003", "Update config",
    "/project/config.json",
    content='{"debug": true}',
)
```

--------------------------------

### Update agent connection wiring

Source: https://github.com/agentclientprotocol/python-sdk/blob/main/docs/migration-guide-0.7.md

Replace manual `AgentSideConnection` setup with the `run_agent` helper function for simpler agent execution. If the agent runs over stdio, `run_agent` will handle stream acquisition.

```python
# Old pattern:
conn = AgentSideConnection(lambda conn: Agent(), writer, reader)
await asyncio.Event().wait()  # keep running

# New pattern:
await run_agent(MyAgent(), input_stream=writer, output_stream=reader)
```

--------------------------------

### Start a Generic Tool Call

Source: https://context7.com/agentclientprotocol/python-sdk/llms.txt

Initiates a tool call with a specific ID, name, kind, status, and raw input. Use this for general-purpose tool invocations.

```python
start = start_tool_call(
    "call-001", "Search codebase",
    kind="read", status="pending",
    raw_input={"query": "Agent class"},
)
```

--------------------------------

### Programmatically spawn an agent process

Source: https://github.com/agentclientprotocol/python-sdk/blob/main/docs/quickstart.md

Use the spawn_agent_process helper to manage the child process, wire its stdio into ACP framing, and handle lifecycle management. This example demonstrates initializing a connection, creating a new session, and sending a prompt.

```python
import asyncio
import sys
from pathlib import Path
from typing import Any

from acp import spawn_agent_process, text_block
from acp.interfaces import Client


class SimpleClient(Client):
    async def request_permission(
        self,
        options,
        session_id,
        tool_call,
        **kwargs: Any
    ):
        return {"outcome": {"outcome": "cancelled"}}

    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(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(
            session_id=session.session_id,
            prompt=[text_block("Hello from spawn!")],
        )

asyncio.run(main())
```

--------------------------------

### Configure Zed to connect to the Echo Agent

Source: https://github.com/agentclientprotocol/python-sdk/blob/main/docs/quickstart.md

Add an Agent Server entry in Zed's settings.json to point it to the Python echo agent script. This example shows configuration for both direct Python execution and using uv.

```json
{
  "agent_servers": {
    "Echo Agent (Python)": {
      "type": "custom",
      "command": "/abs/path/to/python",
      "args": [
        "/abs/path/to/agentclientprotocol/python-sdk/examples/echo_agent.py"
      ]
    }
  }
}
```

```json
{
  "agent_servers": {
    "Echo Agent (Python)": {
      "type": "custom",
      "command": "uv",
      "args": [
        "run",
        "/abs/path/to/agentclientprotocol/python-sdk/examples/echo_agent.py"
      ],
    }
  }
}
```

--------------------------------

### Get Default Permission Options

Source: https://github.com/agentclientprotocol/python-sdk/blob/main/docs/contrib.md

Use default_permission_options() to expose the canonical Approve / Approve for session / Reject option triple when you need to customize permission prompts.

```python
from acp.contrib import permissions

options = permissions.default_permission_options()
print(f'Approve: {options.approve_label}')
print(f'Reject: {options.reject_label}')
```

--------------------------------

### run_agent(agent)

Source: https://context7.com/agentclientprotocol/python-sdk/llms.txt

Starts an ACP agent by wrapping an Agent implementation in a JSON-RPC loop that reads from stdin and writes to stdout. This function should be called within an asyncio.run() entrypoint.

```APIDOC
## run_agent(agent) 

### Description
Wraps an `Agent` implementation in a JSON-RPC loop that reads from stdin and writes to stdout. Call this inside an `asyncio.run()` entrypoint to make the script spawn-able by any ACP client.

### Method
```python
async def run_agent(agent: Agent)
```

### Parameters
* **agent** (`Agent`) - An instance of an `Agent` subclass implementing the ACP agent logic.

### Request Example
```python
import asyncio
from acp import Agent, run_agent

class MyAgent(Agent):
    # ... agent implementation ...
    pass

async def main():
    await run_agent(MyAgent())

if __name__ == "__main__":
    asyncio.run(main())
```

### Response
This function runs indefinitely, processing agent messages, until the agent process is terminated.
```

--------------------------------

### Tool call helpers

Source: https://context7.com/agentclientprotocol/python-sdk/llms.txt

Builders for the two-phase tool call flow, including starting an operation and updating its progress.

```APIDOC
## Tool call helpers — `start_tool_call`, `start_read_tool_call`, `start_edit_tool_call`, `update_tool_call`, `tool_content`, `tool_diff_content`, `tool_terminal_ref`

Builders for the two-phase tool call flow: start an operation with `start_tool_call` (yields `ToolCallStart`), then update it with `update_tool_call` (yields `ToolCallProgress`).

```python
from acp import (
    start_tool_call, start_read_tool_call, start_edit_tool_call,
    update_tool_call, text_block, tool_content, tool_diff_content, tool_terminal_ref,
)
```
```

--------------------------------

### Start a Tool Call with ToolCallTracker

Source: https://github.com/agentclientprotocol/python-sdk/blob/main/docs/contrib.md

Use ToolCallTracker to emit canonical ToolCallStart updates and maintain an in-memory view of tool calls. Keep one tracker near the agent event loop.

```python
from acp.contrib import tool_calls

tracker = tool_calls.ToolCallTracker()

# Start a new tool call
tracker.start(tool_call_id='my-call-id', name='my-tool', arguments={'arg1': 'value1'})
```

--------------------------------

### Programmatic Gemini CLI Bridge with ACP

Source: https://context7.com/agentclientprotocol/python-sdk/llms.txt

Programmatic equivalent of driving the Gemini CLI as an ACP agent using Python. This example demonstrates launching the Gemini CLI subprocess and connecting to it using the ACP Python SDK, including initializing the connection with client capabilities.

```python
# Programmatic equivalent
import asyncio, os, sys
from acp import PROTOCOL_VERSION, connect_to_agent, text_block
from acp.schema import ClientCapabilities, FileSystemCapabilities

# Launching Gemini CLI as an ACP agent
async def main():
    import asyncio.subprocess as sp
    proc = await asyncio.create_subprocess_exec(
        "gemini", "--experimental-acp", "--model", "gemini-1.5-pro",
        stdin=sp.PIPE, stdout=sp.PIPE,
    )
    # Import the GeminiClient from examples/gemini.py for a full-featured client
    # or supply a custom Client implementation
    from examples.gemini import GeminiClient
    client = GeminiClient(auto_approve=True)
    conn = connect_to_agent(client, proc.stdin, proc.stdout)
    resp = await conn.initialize(
        protocol_version=PROTOCOL_VERSION,
        client_capabilities=ClientCapabilities(
            fs=FileSystemCapabilities(read_text_file=True, write_text_file=True),
            terminal=True,
        ),
    )
    print(f"Connected (protocol v{resp.protocol_version})")
    session = await conn.new_session(cwd=os.getcwd(), mcp_servers=[])
    await conn.prompt(session_id=session.session_id, prompt=[text_block("Summarise the README")])

asyncio.run(main())
```

--------------------------------

### Get Sanitized Subprocess Environment with default_environment

Source: https://context7.com/agentclientprotocol/python-sdk/llms.txt

Obtain a minimal environment dictionary suitable for sandboxed subprocesses. This function provides platform-adaptive essential variables like PATH and HOME.

```python
from acp.transports import default_environment, spawn_stdio_transport
import asyncio, sys

env = default_environment()
print(list(env.keys()))
# POSIX: ['HOME', 'LOGNAME', 'PATH', 'SHELL', 'TERM', 'USER']
# Windows: ['APPDATA', 'HOMEDRIVE', 'HOMEPATH', 'LOCALAPPDATA', 'PATH', ...]

# Merge with agent-specific variables before spawning
async def main():
    merged = {**env, "MYAPP_DEBUG": "1", "MYAPP_LOG_LEVEL": "info"}
    async with spawn_stdio_transport(sys.executable, "examples/echo_agent.py", env=merged) as (reader, writer, proc):
        print("Agent pid:", proc.pid)

asyncio.run(main())

```

--------------------------------

### Tool Call Management

Source: https://context7.com/agentclientprotocol/python-sdk/llms.txt

Functions for initiating, updating, and managing tool calls within an agent's workflow. This includes starting new tool calls, updating their status and content, and formatting output.

```APIDOC
## Tool Call Management Functions

These functions are used to manage the lifecycle and content of tool calls initiated by the agent.

### `start_tool_call`

Initiates a new tool call.

```python
start = start_tool_call(
    "call-001", "Search codebase",
    kind="read", status="pending",
    raw_input={"query": "Agent class"},
)
```

### `start_read_tool_call`

A convenience wrapper for starting tool calls with the `read` kind.

```python
read_start = start_read_tool_call("call-002", "Read README.md", "/project/README.md")
# ToolCallStart with kind='read', locations=[ToolCallLocation(path='/project/README.md')]
```

### `start_edit_tool_call`

A convenience wrapper for starting tool calls with the `edit` kind.

```python
edit_start = start_edit_tool_call(
    "call-003", "Update config",
    "/project/config.json",
    content='{"debug": true}',
)
```

### `update_tool_call`

Updates the status and content of an existing tool call.

```python
finish = update_tool_call(
    "call-001",
    status="completed",
    content=[tool_content(text_block("Found 12 matches."))],
    raw_output={"matches": 12},
)

edit_finish = update_tool_call("call-003", status="completed", content=[diff])

terminal_finish = update_tool_call("call-004", status="completed", content=[terminal_ref])
```

### `tool_content`

Formats content for a tool call, such as text blocks or diffs.

```python
content=[tool_content(text_block("Found 12 matches."))]
```

### `tool_diff_content`

Generates diff content for an edit result.

```python
diff = tool_diff_content(
    "/project/config.json",
    new_text='{"debug": true}',
    old_text='{"debug": false}',
)
```

### `tool_terminal_ref`

Creates a reference to a terminal.

```python
terminal_ref = tool_terminal_ref("term-1")
```
```

--------------------------------

### Get Immutable SessionSnapshot

Source: https://github.com/agentclientprotocol/python-sdk/blob/main/docs/contrib.md

Call snapshot() on a SessionAccumulator to get an immutable Pydantic model containing the plan, current mode, commands, and ordered message history. Render tool calls and user messages from this snapshot.

```python
snapshot = accumulator.snapshot()

# Access data from the snapshot
plan = snapshot.plan
user_messages = snapshot.user_messages
```

--------------------------------

### `spawn_agent_process(client, command, *args)`

Source: https://context7.com/agentclientprotocol/python-sdk/llms.txt

An async context manager that launches an agent script as a subprocess, wires its stdio into ACP framing, and returns a ready-to-use `ClientSideConnection`. Handles graceful shutdown when the block exits.

```APIDOC
## `spawn_agent_process(client, command, *args)` — Spawn an agent subprocess

An async context manager that launches an agent script as a subprocess, wires its stdio into ACP framing, and returns a ready-to-use `ClientSideConnection`. Handles graceful shutdown when the block exits.

```python
import asyncio
import sys
from pathlib import Path
from typing import Any

from acp import PROTOCOL_VERSION, spawn_agent_process, text_block
from acp.interfaces import Client
from acp.schema import AgentMessageChunk, TextContentBlock

class SimpleClient(Client):
    async def request_permission(self, options, session_id, tool_call, **kwargs):
        from acp import RequestError
        raise RequestError.method_not_found("session/request_permission")

    async def session_update(self, session_id, update, **kwargs) -> None:
        if isinstance(update, AgentMessageChunk) and isinstance(update.content, TextContentBlock):
            print(f"[{session_id}] {update.content.text}")

    async def write_text_file(self, content, path, session_id, **kwargs): ...
    async def read_text_file(self, path, session_id, **kwargs): ...
    async def create_terminal(self, command, session_id, **kwargs): ...
    async def terminal_output(self, session_id, terminal_id, **kwargs): ...
    async def release_terminal(self, session_id, terminal_id, **kwargs): ...
    async def wait_for_terminal_exit(self, session_id, terminal_id, **kwargs): ...
    async def kill_terminal(self, session_id, terminal_id, **kwargs): ...
    async def ext_method(self, method, params): return {}
    async def ext_notification(self, method, params): ...

async def main() -> None:
    script = Path("examples/echo_agent.py")
    async with spawn_agent_process(SimpleClient(), sys.executable, str(script)) as (conn, proc):
        await conn.initialize(protocol_version=PROTOCOL_VERSION)
        session = await conn.new_session(cwd=str(script.parent), mcp_servers=[])
        await conn.prompt(
            session_id=session.session_id,
            prompt=[text_block("Ping from spawn_agent_process!")],
        )
        # Output → [<session_id>] Ping from spawn_agent_process!
        print(f"Agent exited with code: {proc.returncode}")

asyncio.run(main())
```
```

--------------------------------

### Run Project Checks

Source: https://github.com/agentclientprotocol/python-sdk/blob/main/README.md

Execute formatting, linting, type analysis, and dependency checks for the project.

```bash
make check
```

--------------------------------

### Spawn Agent Process with Python SDK

Source: https://context7.com/agentclientprotocol/python-sdk/llms.txt

Launches an agent script as a subprocess, handling stdio framing and graceful shutdown. Requires a `Client` implementation to interact with the agent.

```python
import asyncio
import sys
from pathlib import Path
from typing import Any

from acp import PROTOCOL_VERSION, spawn_agent_process, text_block
from acp.interfaces import Client
from acp.schema import AgentMessageChunk, TextContentBlock

class SimpleClient(Client):
    async def request_permission(self, options, session_id, tool_call, **kwargs):
        from acp import RequestError
        raise RequestError.method_not_found("session/request_permission")

    async def session_update(self, session_id, update, **kwargs) -> None:
        if isinstance(update, AgentMessageChunk) and isinstance(update.content, TextContentBlock):
            print(f"[{session_id}] {update.content.text}")

    async def write_text_file(self, content, path, session_id, **kwargs): ...
    async def read_text_file(self, path, session_id, **kwargs): ...
    async def create_terminal(self, command, session_id, **kwargs): ...
    async def terminal_output(self, session_id, terminal_id, **kwargs): ...
    async def release_terminal(self, session_id, terminal_id, **kwargs): ...
    async def wait_for_terminal_exit(self, session_id, terminal_id, **kwargs): ...
    async def kill_terminal(self, session_id, terminal_id, **kwargs): ...
    async def ext_method(self, method, params): return {}
    async def ext_notification(self, method, params): ...

async def main() -> None:
    script = Path("examples/echo_agent.py")
    async with spawn_agent_process(SimpleClient(), sys.executable, str(script)) as (conn, proc):
        await conn.initialize(protocol_version=PROTOCOL_VERSION)
        session = await conn.new_session(cwd=str(script.parent), mcp_servers=[])
        await conn.prompt(
            session_id=session.session_id,
            prompt=[text_block("Ping from spawn_agent_process!")],
        )
        # Output → [<session_id>] Ping from spawn_agent_process!
        print(f"Agent exited with code: {proc.returncode}")

asyncio.run(main())
```

--------------------------------

### Run the Echo Agent

Source: https://github.com/agentclientprotocol/python-sdk/blob/main/docs/quickstart.md

Launch the provided streaming agent so clients have something to communicate with. This agent streams text blocks back to any ACP client.

```bash
python examples/echo_agent.py
```

--------------------------------

### `spawn_client_process(agent, command, *args)`

Source: https://context7.com/agentclientprotocol/python-sdk/llms.txt

The mirror of `spawn_agent_process`: launches an ACP client as a subprocess and returns an `AgentSideConnection` so Python code can drive the client programmatically.

```APIDOC
## `spawn_client_process(agent, command, *args)` — Spawn a client subprocess

The mirror of `spawn_agent_process`: launches an ACP client as a subprocess and returns an `AgentSideConnection` so Python code can drive the client programmatically.

```python
import asyncio
import sys
from pathlib import Path
from typing import Any

from acp import PROTOCOL_VERSION, Agent, InitializeResponse, NewSessionResponse, PromptResponse
from acp import run_agent, spawn_client_process, text_block, update_agent_message
from acp.interfaces import Client

class InlineAgent(Agent):
    _conn: Client

    def on_connect(self, conn: Client) -> None:
        self._conn = conn

    async def initialize(self, protocol_version, **kwargs) -> InitializeResponse:
        return InitializeResponse(protocol_version=protocol_version)

    async def new_session(self, cwd, mcp_servers, **kwargs) -> NewSessionResponse:
        return NewSessionResponse(session_id="session-0")

    async def prompt(self, prompt, session_id, **kwargs) -> PromptResponse:
        await self._conn.session_update(
            session_id=session_id,
            update=update_agent_message(text_block("Response from InlineAgent")),
        )
        return PromptResponse(stop_reason="end_turn")

    async def cancel(self, session_id, **kwargs) -> None: ...
    async def ext_method(self, method, params): return {}
    async def ext_notification(self, method, params): ...

async def main():
    client_script = Path("examples/client.py")
    async with spawn_client_process(InlineAgent(), sys.executable, str(client_script)) as (conn, proc):
        # conn is an AgentSideConnection; the client subprocess drives it
        print("Client process running, pid:", proc.pid)

asyncio.run(main())
```
```

--------------------------------

### Apply SessionNotification to SessionAccumulator

Source: https://github.com/agentclientprotocol/python-sdk/blob/main/docs/contrib.md

Use SessionAccumulator to reconcile tool call and update payloads, even if the start event arrives late. Create one accumulator per UI controller and feed raw SessionNotification events into it.

```python
from acp.contrib import session_state

accumulator = session_state.SessionAccumulator()

# Apply a notification
accumulator.apply(notification)
```

--------------------------------

### Commit and Push Changes

Source: https://github.com/agentclientprotocol/python-sdk/blob/main/CONTRIBUTING.md

Stage your changes, commit them with a Conventional Commits formatted message, and then push the branch to your origin.

```bash
git commit -m "feat: add better tool call helper"
```

```bash
git push origin <branch>
```

--------------------------------

### Connect Client to Agent with Streams

Source: https://context7.com/agentclientprotocol/python-sdk/llms.txt

Use this function when you have already spawned the agent process and hold its stdin/stdout handles. It creates a ClientSideConnection wrapping the provided asyncio streams.

```python
import asyncio
import os
from typing import Any

from acp import PROTOCOL_VERSION, Client, RequestError, connect_to_agent, text_block
from acp.schema import (
    AgentMessageChunk, ClientCapabilities, Implementation,
    PermissionOption, RequestPermissionResponse, TextContentBlock, ToolCall,
    WriteTextFileResponse,
)

class MinimalClient(Client):
    async def request_permission(self, options, session_id, tool_call, **kwargs):
        raise RequestError.method_not_found("session/request_permission")

    async def write_text_file(self, content, path, session_id, **kwargs):
        raise RequestError.method_not_found("fs/write_text_file")

    async def read_text_file(self, path, session_id, **kwargs):
        raise RequestError.method_not_found("fs/read_text_file")

    async def session_update(self, session_id, update, **kwargs) -> None:
        if isinstance(update, AgentMessageChunk) and isinstance(update.content, TextContentBlock):
            print(f"Agent: {update.content.text}")

    async def create_terminal(self, command, session_id, **kwargs):
        raise RequestError.method_not_found("terminal/create")

    async def terminal_output(self, session_id, terminal_id, **kwargs):
        raise RequestError.method_not_found("terminal/output")

    async def release_terminal(self, session_id, terminal_id, **kwargs):
        raise RequestError.method_not_found("terminal/release")

    async def wait_for_terminal_exit(self, session_id, terminal_id, **kwargs):
        raise RequestError.method_not_found("terminal/wait_for_exit")

    async def kill_terminal(self, session_id, terminal_id, **kwargs):
        raise RequestError.method_not_found("terminal/kill")

    async def ext_method(self, method, params):
        raise RequestError.method_not_found(method)

    async def ext_notification(self, method, params) -> None:
        pass

async def main():
    import asyncio.subprocess as sp
    proc = await asyncio.create_subprocess_exec(
        "python", "examples/echo_agent.py",
        stdin=sp.PIPE, stdout=sp.PIPE,
    )
    conn = connect_to_agent(MinimalClient(), proc.stdin, proc.stdout)
    await conn.initialize(
        protocol_version=PROTOCOL_VERSION,
        client_capabilities=ClientCapabilities(),
        client_info=Implementation(name="my-client", title="My Client", version="0.1.0"),
    )
    session = await conn.new_session(mcp_servers=[], cwd=os.getcwd())
    await conn.prompt(session_id=session.session_id, prompt=[text_block("Hello!")])
    # Output → Agent: Hello!
    proc.terminate()

asyncio.run(main())

```

--------------------------------

### Run Gemini CLI Integration

Source: https://github.com/agentclientprotocol/python-sdk/blob/main/docs/quickstart.md

Execute the Gemini CLI integration script. Use flags like `--yolo` for auto-approval or specify the model and sandbox mode for different testing scenarios. This demonstrates permission flows.

```bash
python examples/gemini.py --yolo                # auto-approve permission prompts
```

```bash
python examples/gemini.py --sandbox --model gemini-1.5-pro
```

--------------------------------

### connect_to_agent

Source: https://context7.com/agentclientprotocol/python-sdk/llms.txt

Creates a ClientSideConnection by wrapping asyncio streams. This is useful when you have already spawned the agent process and have access to its stdin/stdout handles.

```APIDOC
## `connect_to_agent(client, input_stream, output_stream)`

### Description
Creates a `ClientSideConnection` wrapping the given asyncio streams. Use this when you have already spawned the agent process yourself and hold its stdin/stdout handles.

### Parameters
- **client**: An instance of a class that inherits from `acp.Client`.
- **input_stream**: An asyncio stream for writing data to the agent (e.g., agent's stdin).
- **output_stream**: An asyncio stream for reading data from the agent (e.g., agent's stdout).

### Example Usage
```python
import asyncio
import os
from acp import connect_to_agent, PROTOCOL_VERSION, ClientCapabilities, Implementation, text_block

class MinimalClient(Client):
    # ... (implementation of MinimalClient as shown in the source)
    pass

async def main():
    import asyncio.subprocess as sp
    proc = await asyncio.create_subprocess_exec(
        "python", "examples/echo_agent.py",
        stdin=sp.PIPE, stdout=sp.PIPE,
    )
    conn = connect_to_agent(MinimalClient(), proc.stdin, proc.stdout)
    await conn.initialize(
        protocol_version=PROTOCOL_VERSION,
        client_capabilities=ClientCapabilities(),
        client_info=Implementation(name="my-client", title="My Client", version="0.1.0"),
    )
    session = await conn.new_session(mcp_servers=[], cwd=os.getcwd())
    await conn.prompt(session_id=session.session_id, prompt=[text_block("Hello!")])
    proc.terminate()

asyncio.run(main())
```
```

--------------------------------

### Request Permission with PermissionBroker

Source: https://github.com/agentclientprotocol/python-sdk/blob/main/docs/contrib.md

Use PermissionBroker.request_for() to wrap requestPermission RPCs, reusing tracker state and defaulting to Approve/Approve for session/Reject options. Customize options using default_permission_options().

```python
from acp.contrib import permissions

broker = permissions.PermissionBroker(tracker=tracker) # Optionally pass tracker

# Request permission for a tool call
await broker.request_for(tool_call_id='my-call-id', content='Are you sure?')

# Customize options
custom_options = permissions.default_permission_options()
custom_options.approve_label = 'Yes, proceed'
await broker.request_for(tool_call_id='my-call-id', content='Custom prompt?', options=custom_options)
```

--------------------------------

### Tool Call Helper Functions

Source: https://context7.com/agentclientprotocol/python-sdk/llms.txt

These builders facilitate the two-phase tool call flow. Use `start_tool_call` (or its variants) to initiate an operation, which yields `ToolCallStart`. Subsequently, use `update_tool_call` to send progress updates, yielding `ToolCallProgress`.

```python
from acp import (
    start_tool_call, start_read_tool_call, start_edit_tool_call,
    update_tool_call, text_block, tool_content, tool_diff_content, tool_terminal_ref,
)
```

--------------------------------

### `stdio_streams(limit)`

Source: https://context7.com/agentclientprotocol/python-sdk/llms.txt

Create cross-platform asyncio stdio streams. Returns `(asyncio.StreamReader, asyncio.StreamWriter)` wired to `sys.stdin` / `sys.stdout`. On Windows it uses a background thread feeder; on POSIX it uses `loop.connect_read_pipe`. Pass `limit` to increase the buffer for multimodal payloads (the default in `run_agent` is 50 MB).

```APIDOC
## `stdio_streams(limit)` — Create cross-platform asyncio stdio streams

Returns `(asyncio.StreamReader, asyncio.StreamWriter)` wired to `sys.stdin` / `sys.stdout`. On Windows it uses a background thread feeder; on POSIX it uses `loop.connect_read_pipe`. Pass `limit` to increase the buffer for multimodal payloads (the default in `run_agent` is 50 MB).

```python
import asyncio
from acp.stdio import stdio_streams

async def raw_stdio_example():
    reader, writer = await stdio_streams(limit=50 * 1024 * 1024)  # 50 MB buffer
    # Use reader/writer directly with acp.connection.Connection for custom transports
    line = await reader.readline()
    writer.write(line)  # echo back
    await writer.drain()

asyncio.run(raw_stdio_example())
```
```

--------------------------------

### Spawn Client Process with Python SDK

Source: https://context7.com/agentclientprotocol/python-sdk/llms.txt

Launches an ACP client as a subprocess, returning an `AgentSideConnection` for programmatic control. Requires an `Agent` implementation to handle client-side logic.

```python
import asyncio
import sys
from pathlib import Path
from typing import Any

from acp import PROTOCOL_VERSION, Agent, InitializeResponse, NewSessionResponse, PromptResponse
from acp import run_agent, spawn_client_process, text_block, update_agent_message
from acp.interfaces import Client

class InlineAgent(Agent):
    _conn: Client

    def on_connect(self, conn: Client) -> None:
        self._conn = conn

    async def initialize(self, protocol_version, **kwargs) -> InitializeResponse:
        return InitializeResponse(protocol_version=protocol_version)

    async def new_session(self, cwd, mcp_servers, **kwargs) -> NewSessionResponse:
        return NewSessionResponse(session_id="session-0")

    async def prompt(self, prompt, session_id, **kwargs) -> PromptResponse:
        await self._conn.session_update(
            session_id=session_id,
            update=update_agent_message(text_block("Response from InlineAgent")),
        )
        return PromptResponse(stop_reason="end_turn")

    async def cancel(self, session_id, **kwargs) -> None: ...
    async def ext_method(self, method, params): return {}
    async def ext_notification(self, method, params): ...

async def main():
    client_script = Path("examples/client.py")
    async with spawn_client_process(InlineAgent(), sys.executable, str(client_script)) as (conn, proc):
        # conn is an AgentSideConnection; the client subprocess drives it
        print("Client process running, pid:", proc.pid)

asyncio.run(main())
```

--------------------------------

### Spawn Subprocess with Stdio Transport

Source: https://context7.com/agentclientprotocol/python-sdk/llms.txt

Use this async context manager to spawn a subprocess with stdin and stdout pipes. It handles defensive shutdown, closing stdin, waiting for graceful exit, and escalating to terminate/kill if necessary. Ensure the `shutdown_timeout` is appropriate for your subprocess.

```python
import asyncio
import sys
from acp.transports import spawn_stdio_transport

async def main():
    async with spawn_stdio_transport(
        sys.executable, "examples/echo_agent.py",
        shutdown_timeout=5.0,
    ) as (reader, writer, proc):
        # Write a raw newline-delimited JSON-RPC message
        import json
        msg = json.dumps({"jsonrpc": "2.0", "id": 1, "method": "initialize",
                          "params": {"protocolVersion": 1}})
        writer.write((msg + "\n").encode())
        await writer.drain()
        response_line = await reader.readline()
        print("Raw response:", response_line.decode().strip())
    # Process is cleanly terminated after the block

asyncio.run(main())
```

--------------------------------

### Cross-Python Environment Testing

Source: https://github.com/agentclientprotocol/python-sdk/blob/main/CONTRIBUTING.md

Use 'tox' to run tests across different Python versions, mirroring the CI environment's test matrix.

```bash
tox
```

--------------------------------

### Create Cross-Platform Asyncio Stdio Streams

Source: https://context7.com/agentclientprotocol/python-sdk/llms.txt

Provides `asyncio.StreamReader` and `asyncio.StreamWriter` for `sys.stdin` and `sys.stdout`. Supports a configurable buffer limit for multimodal payloads.

```python
import asyncio
from acp.stdio import stdio_streams

async def raw_stdio_example():
    reader, writer = await stdio_streams(limit=50 * 1024 * 1024)  # 50 MB buffer
    # Use reader/writer directly with acp.connection.Connection for custom transports
    line = await reader.readline()
    writer.write(line)  # echo back
    await writer.drain()

asyncio.run(raw_stdio_example())
```

--------------------------------

### Run an ACP agent over stdio

Source: https://context7.com/agentclientprotocol/python-sdk/llms.txt

Wraps an Agent implementation in a JSON-RPC loop for stdio communication. Call this within an asyncio.run() entrypoint to make the script spawnable by ACP clients.

```python
import asyncio
from typing import Any
from uuid import uuid4

from acp import Agent, InitializeResponse, NewSessionResponse, PromptResponse, run_agent, text_block, update_agent_message
from acp.interfaces import Client
from acp.schema import ClientCapabilities, HttpMcpServer, Implementation, McpServerStdio, SseMcpServer, TextContentBlock

class EchoAgent(Agent):
    _conn: Client

    def on_connect(self, conn: Client) -> None:
        # Store the connection so prompt() can push session updates back
        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, mcp_servers: list[HttpMcpServer | SseMcpServer | McpServerStdio], **kwargs: Any
    ) -> NewSessionResponse:
        return NewSessionResponse(session_id=uuid4().hex)

    async def prompt(self, prompt, session_id: str, **kwargs: Any) -> PromptResponse:
        for block in prompt:
            text = getattr(block, "text", "")
            chunk = update_agent_message(text_block(text))
            await self._conn.session_update(session_id=session_id, update=chunk)
        return PromptResponse(stop_reason="end_turn")

async def main() -> None:
    await run_agent(EchoAgent())

if __name__ == "__main__":
    asyncio.run(main())
# Run: python echo_agent.py
# The agent now speaks ACP over stdin/stdout.
```

--------------------------------

### Run Code Quality and Tests

Source: https://github.com/agentclientprotocol/python-sdk/blob/main/CONTRIBUTING.md

Execute the 'check' target for formatting, linting, type analysis, and dependency checks. Use the 'test' target to run pytest and doctests.

```bash
make check
```

```bash
make test
```

--------------------------------

### Run Project Tests

Source: https://github.com/agentclientprotocol/python-sdk/blob/main/README.md

Execute pytest, including doctests, within the managed virtual environment.

```bash
make test
```

--------------------------------

### default_environment

Source: https://context7.com/agentclientprotocol/python-sdk/llms.txt

Returns a minimal `dict[str, str]` containing only the essential environment variables (platform-adaptive: `PATH`, `HOME`, etc.) following MCP best-practice for sandboxed subprocess spawning. This function is useful for securely launching subprocesses.

```APIDOC
## `default_environment()` — sanitised subprocess environment

Returns a minimal `dict[str, str]` containing only the essential environment variables (platform-adaptive: `PATH`, `HOME`, etc.) following MCP best-practice for sandboxed subprocess spawning.

```python
from acp.transports import default_environment, spawn_stdio_transport
import asyncio, sys

env = default_environment()
print(list(env.keys()))
# POSIX: ['HOME', 'LOGNAME', 'PATH', 'SHELL', 'TERM', 'USER']
# Windows: ['APPDATA', 'HOMEDRIVE', 'HOMEPATH', 'LOCALAPPDATA', 'PATH', ...]

# Merge with agent-specific variables before spawning
async def main():
    merged = {**env, "MYAPP_DEBUG": "1", "MYAPP_LOG_LEVEL": "info"}
    async with spawn_stdio_transport(sys.executable, "examples/echo_agent.py", env=merged) as (reader, writer, proc):
        print("Agent pid:", proc.pid)

asyncio.run(main())
```
```

--------------------------------

### Update client interface method implementation

Source: https://github.com/agentclientprotocol/python-sdk/blob/main/docs/migration-guide-0.7.md

Rename client methods to snake_case and adjust parameters to receive explicit schema fields. Use `**kwargs` for any additional extension data.

```python
class RecordingClient(Client):
    async def request_permission(self, options, session_id, tool_call, **kwargs):
        ...
```

--------------------------------

### PermissionBroker

Source: https://context7.com/agentclientprotocol/python-sdk/llms.txt

Integrates with ToolCallTracker to issue `requestPermission` RPCs using a consistent three-option default (Approve / Approve for session / Reject). It simplifies the process of requesting user permissions for agent actions.

```APIDOC
## `PermissionBroker` — coordinate user-approval requests

`acp.contrib.permissions.PermissionBroker` integrates with `ToolCallTracker` to issue `requestPermission` RPCs using a consistent three-option default (Approve / Approve for session / Reject).

```python
import asyncio
from acp.contrib.permissions import PermissionBroker, default_permission_options
from acp.contrib.tool_calls import ToolCallTracker
from acp.schema import RequestPermissionRequest, RequestPermissionResponse, AllowedOutcome

tracker = ToolCallTracker()

async def mock_requester(req: RequestPermissionRequest) -> RequestPermissionResponse:
    # In a real agent this would call conn.request_permission(...)
    print(f"Permission requested for tool: {req.tool_call.title}")
    for opt in req.options:
        print(f"  - {opt.name} ({opt.kind})")
    # Auto-approve for demo
    allow_opt = next(o for o in req.options if o.kind == "allow_once")
    return RequestPermissionResponse(outcome=AllowedOutcome(option_id=allow_opt.option_id, outcome="selected"))

# Start a tracked tool call first
tracker.start("write-op-1", title="Write /etc/hosts", kind="edit",
              status="pending", raw_input={"path": "/etc/hosts"})

broker = PermissionBroker(
    session_id="session-xyz",
    requester=mock_requester,
    tracker=tracker,
    default_options=list(default_permission_options()),
)

async def main():
    response = await broker.request_for(
        "write-op-1",
        description="The agent wants to modify /etc/hosts to add a custom hostname.",
    )
    print(f"User outcome: {response.outcome.outcome}")
    # Output:
    # Permission requested for tool: Write /etc/hosts
    #   - Approve (allow_once)
    #   - Approve for session (allow_always)
    #   - Reject (reject_once)
    # User outcome: selected

asyncio.run(main())
```
```

--------------------------------

### Configure Zed Editor Agent Servers

Source: https://context7.com/agentclientprotocol/python-sdk/llms.txt

Register custom agent scripts in Zed's settings.json to enable the editor to spawn them as ACP agent servers. Supports direct Python execution or using a Python environment manager like 'uv'.

```json
{
  "agent_servers": {
    "My Python Agent": {
      "type": "custom",
      "command": "/usr/bin/python3",
      "args": ["/abs/path/to/my_agent.py"]
    },
    "My uv Agent": {
      "type": "custom",
      "command": "uv",
      "args": ["run", "/abs/path/to/my_agent.py"]
    }
  }
}
```

--------------------------------

### Coordinate User Approval with PermissionBroker

Source: https://context7.com/agentclientprotocol/python-sdk/llms.txt

Integrate PermissionBroker with ToolCallTracker to issue `requestPermission` RPCs. It supports a consistent three-option default: Approve, Approve for session, and Reject.

```python
import asyncio
from acp.contrib.permissions import PermissionBroker, default_permission_options
from acp.contrib.tool_calls import ToolCallTracker
from acp.schema import RequestPermissionRequest, RequestPermissionResponse, AllowedOutcome

tracker = ToolCallTracker()

async def mock_requester(req: RequestPermissionRequest) -> RequestPermissionResponse:
    # In a real agent this would call conn.request_permission(...)
    print(f"Permission requested for tool: {req.tool_call.title}")
    for opt in req.options:
        print(f"  - {opt.name} ({opt.kind})")
    # Auto-approve for demo
    allow_opt = next(o for o in req.options if o.kind == "allow_once")
    return RequestPermissionResponse(outcome=AllowedOutcome(option_id=allow_opt.option_id, outcome="selected"))

# Start a tracked tool call first
tracker.start("write-op-1", title="Write /etc/hosts", kind="edit",
              status="pending", raw_input={"path": "/etc/hosts"})

broker = PermissionBroker(
    session_id="session-xyz",
    requester=mock_requester,
    tracker=tracker,
    default_options=list(default_permission_options()),
)

async def main():
    response = await broker.request_for(
        "write-op-1",
        description="The agent wants to modify /etc/hosts to add a custom hostname.",
    )
    print(f"User outcome: {response.outcome.outcome}")
    # Output:
    # Permission requested for tool: Write /etc/hosts
    #   - Approve (allow_once)
    #   - Approve for session (allow_always)
    #   - Reject (reject_once)
    # User outcome: selected

asyncio.run(main())

```

--------------------------------

### Run Tests with Gemini Integration

Source: https://github.com/agentclientprotocol/python-sdk/blob/main/CONTRIBUTING.md

Optionally, enable Gemini tests by setting the ACP_ENABLE_GEMINI_TESTS environment variable before running the test suite, provided you have the Gemini CLI available.

```bash
ACP_ENABLE_GEMINI_TESTS=1 make test
```

--------------------------------

### Clean Build Artifacts

Source: https://github.com/agentclientprotocol/python-sdk/blob/main/docs/releasing.md

Use this command to remove any previously built artifacts, providing a clean slate before regenerating or rebuilding components. This is useful when encountering issues with stale build data.

```bash
make clean
```

--------------------------------

### Session Update Helper Functions

Source: https://context7.com/agentclientprotocol/python-sdk/llms.txt

These helpers construct `session/update` notification payloads. The `update_*_text` variants are convenient shortcuts for streaming string content directly. Use `session_notification` to wrap any update for raw dispatch.

```python
from acp import (
    text_block, image_block,
    update_agent_message, update_agent_message_text,
    update_agent_thought, update_agent_thought_text,
    update_user_message_text,
    update_plan, plan_entry,
)
from acp.helpers import session_notification

# Stream a text chunk to the client
chunk = update_agent_message_text("Thinking about your question…")
# AgentMessageChunk(session_update='agent_message_chunk', content=TextContentBlock(...))

# Stream an image chunk
img_chunk = update_agent_message(image_block("BASE64==", "image/png"))

# Internal thought (not shown to user by default in most clients)
thought = update_agent_thought_text("The user probably means X.")

# Reflect back the user's own message
user_echo = update_user_message_text("Echo of user input")

# Publish a plan with three prioritised entries
plan = update_plan([
    plan_entry("Gather requirements", priority="high", status="completed"),
    plan_entry("Draft solution",       priority="high", status="in_progress"),
    plan_entry("Write tests",          priority="medium", status="pending"),
])
# AgentPlanUpdate(session_update='plan', entries=[...])

# Wrap any update in a SessionNotification for raw dispatch
notif = session_notification("session-abc123", chunk)
```

--------------------------------

### spawn_stdio_transport

Source: https://context7.com/agentclientprotocol/python-sdk/llms.txt

Low-level subprocess stdio context manager that spawns a subprocess with stdin=PIPE, stdout=PIPE and yields (StreamReader, StreamWriter, Process). It applies defensive shutdown.

```APIDOC
## `spawn_stdio_transport(command, *args)` — Low-level subprocess stdio context manager

Async context manager that spawns a subprocess with `stdin=PIPE, stdout=PIPE` and yields `(StreamReader, StreamWriter, Process)`. Applies defensive shutdown: closes stdin, waits for graceful exit, then escalates to `terminate` and `kill`.

```python
import asyncio
import sys
from acp.transports import spawn_stdio_transport

async def main():
    async with spawn_stdio_transport(
        sys.executable, "examples/echo_agent.py",
        shutdown_timeout=5.0,
    ) as (reader, writer, proc):
        # Write a raw newline-delimited JSON-RPC message
        import json
        msg = json.dumps({"jsonrpc": "2.0", "id": 1, "method": "initialize",
                          "params": {"protocolVersion": 1}})
        writer.write((msg + "\n").encode())
        await writer.drain()
        response_line = await reader.readline()
        print("Raw response:", response_line.decode().strip())
    # Process is cleanly terminated after the block

asyncio.run(main())
```
```