claude-code-acp-py（上）：ACP 概念與 Client 實作

import asyncio
from claude_code_acp import ClaudeClient

async def main():
    client = ClaudeClient(cwd=".", system_prompt="You are a helpful assistant.")

    @client.on_text
    async def handle_text(text: str):
        print(text, end="", flush=True)

    @client.on_tool_start
    async def handle_tool_start(tool_id: str, name: str, input: dict):
        print(f"\n[Tool] {name}")

    @client.on_tool_end
    async def handle_tool_end(tool_id: str, status: str, output):
        icon = "ok" if status == "completed" else "fail"
        print(f" [{icon}]")

    @client.on_permission
    async def handle_permission(name: str, input: dict) -> bool:
        print(f"Permission requested: {name}")
        return True  # 或者詢問使用者

    @client.on_complete
    async def handle_complete():
        print("\n--- Done ---")

    response = await client.query("Create a hello.py file that prints Hello World")
    print(f"\nFull response: {response}")

asyncio.run(main())

# client.py 的 EventHandler 內部
if current_len == 0:
    # 第一個 chunk，直接輸出
    client._text_buffer = text
    if client.events.on_text:
        await client.events.on_text(text)
elif text == client._text_buffer:
    # 完全重複，跳過
    pass
elif text.startswith(client._text_buffer):
    # 累積式更新 - text 擴展了 buffer，只輸出新的部分
    new_part = text[current_len:]
    if new_part:
        client._text_buffer = text
        if client.events.on_text:
            await client.events.on_text(new_part)
else:
    # 新的增量 chunk - 直接附加到 buffer
    client._text_buffer += text
    if client.events.on_text:
        await client.events.on_text(text)

async def request_permission(self, **kwargs: Any) -> dict:
    tool_call = kwargs.get("tool_call", {})
    name = tool_call.get("title", "Unknown")
    raw_input = tool_call.get("raw_input", {})

    approved = True
    if client.events.on_permission:
        approved = await client.events.on_permission(name, raw_input)

    if approved:
        return {"outcome": {"outcome": "selected", "option_id": "allow"}}
    return {"outcome": {"outcome": "selected", "option_id": "reject"}}

# 初始化時可帶 MCP server 和 system prompt
client = ClaudeClient(cwd=".", mcp_servers=[...], system_prompt="...")

# 手動建立 session（query 會自動建立）
session_id = await client.start_session()

# 設定權限模式
await client.set_mode("acceptEdits")  # 或 "default", "plan", "bypassPermissions"

async def main():
    client = ClaudeClient(cwd=".")
    await client.set_mode("bypassPermissions")

    @client.on_text
    async def on_text(text):
        print(text, end="")

    await client.query("Create a complete Flask app with tests")

import asyncio
from claude_code_acp import AcpClient

async def main():
    client = AcpClient(command="claude-code-acp")

    @client.on_text
    async def handle_text(text: str):
        print(text, end="", flush=True)

    @client.on_tool_start
    async def handle_tool(tool_id: str, name: str, input: dict):
        print(f"\n[Tool] {name}")

    @client.on_permission
    async def handle_permission(name: str, input: dict, options: list) -> str:
        """回傳 option_id: 'allow', 'reject', 或 'allow_always'"""
        print(f"Permission: {name}")
        return "allow"

    @client.on_complete
    async def handle_complete():
        print("\n--- Done ---")

    async with client:
        response = await client.prompt("What files are here?")

asyncio.run(main())

from claude_code_acp import AcpClient

# Claude（本套件）
claude = AcpClient(command="claude-code-acp")

# Gemini CLI
gemini = AcpClient(command="gemini", args=["--experimental-acp"])

# TypeScript 版本
ts_claude = AcpClient(command="npx", args=["@zed-industries/claude-code-acp"])

# 自訂 ACP agent
custom = AcpClient(command="my-custom-agent")

client = AcpClient(
    command="claude-code-acp",
    cwd="/tmp",
    mcp_servers=[{
        "name": "nanobanana",
        "command": "uvx",
        "args": ["nanobanana"],
        "env": {"GEMINI_API_KEY": "your-key"},
    }],
)

# acp_client.py 的 new_session 方法
mcp_servers_acp = []
for srv in self.mcp_servers:
    env_vars = []
    if "env" in srv and srv["env"]:
        for k, v in srv["env"].items():
            env_vars.append(EnvVariable(name=k, value=v))
    mcp_servers_acp.append(
        McpServerStdio(
            name=srv.get("name", "mcp"),
            command=srv.get("command", ""),
            args=srv.get("args", []),
            env=env_vars,
        )
    )

@client.on_file_read
async def handle_read(path: str) -> str | None:
    """回傳內容以覆蓋，或 None 讓正常讀取進行。"""
    print(f"Reading: {path}")
    return None

@client.on_file_write
async def handle_write(path: str, content: str) -> bool:
    """回傳 True 允許，False 阻擋。"""
    print(f"Writing: {path}")
    return input("Allow? [y/N]: ").lower() == "y"

async def write_text_file(self, path: str, content: str, **kwargs) -> None:
    # 先詢問 handler 是否要阻擋
    if client.events.on_file_write:
        allowed = await client.events.on_file_write(path, content)
        if not allowed:
            logger.info(f"File write blocked by handler: {path}")
            return

    # 實際寫入檔案
    file_path = Path(path)
    file_path.parent.mkdir(parents=True, exist_ok=True)
    file_path.write_text(content, encoding="utf-8")

async def read_text_file(self, path: str, **kwargs) -> dict:
    # 先詢問 handler 是否要覆蓋內容
    if client.events.on_file_read:
        override = await client.events.on_file_read(path)
        if override is not None:
            return {"content": override}

    # 實際讀取檔案
    file_path = Path(path)
    if not file_path.exists():
        return {"content": "", "error": f"File not found: {path}"}
    content = file_path.read_text(encoding="utf-8")
    return {"content": content}

@client.on_terminal_create
async def handle_terminal(command: str, cwd: str) -> bool:
    """回傳 True 允許，False 阻擋。"""
    print(f"Command: {command} in {cwd}")
    return input("Allow? [y/N]: ").lower() == "y"

@client.on_terminal_output
async def handle_output(terminal_id: str, output: str) -> None:
    print(output, end="")

@dataclass
class TerminalProcess:
    """代表一個活躍的 terminal process。"""
    process: asyncio.subprocess.Process
    command: str
    cwd: str
    output_buffer: list[str]
    exit_code: int | None = None

import asyncio
from claude_code_acp import AcpClient

async def ask(agent_name, command, args, prompt):
    client = AcpClient(command=command, args=args)

    @client.on_text
    async def on_text(text):
        pass  # 靜默收集

    async with client:
        response = await client.prompt(prompt)
        print(f"{agent_name}: {response[:100]}...")

async def main():
    await asyncio.gather(
        ask("Claude", "claude-code-acp", [], "What is ACP?"),
        ask("Gemini", "gemini", ["--experimental-acp"], "What is ACP?"),
    )

asyncio.run(main())