ACP 详解（选修）

agents:
  - id: claude
    name: Claude Code
    command: npx
    args: [-y, '@agentclientprotocol/claude-agent-acp@0.101.0']
  - id: kimi
    name: Kimi CLI
    command: npx
    args: [-y, '@moonshot-ai/kimi-code@2.5.2', acp]
  - id: opencode
    name: OpenCode
    command: opencode
    args: [acp]

#!/usr/bin/env python3
"""回声 Agent：一个不接模型的 ACP Agent，把你说的话写进 echo.txt。运行：python3 echo_agent.py"""
import json
import os
import sys

sys.stdin.reconfigure(encoding="utf-8")
sys.stdout.reconfigure(encoding="utf-8")
sessions = {}   # sessionId -> 工作目录
next_id = 0     # Agent 自己发出的请求的编号，和 Client 的编号各数各的

# ① 收发：一行一条 JSON。stdout 只能写协议消息，日志写 stderr
def send(msg):
    line = json.dumps({"jsonrpc": "2.0", **msg}, ensure_ascii=False, separators=(",", ":"))
    print(line, flush=True)

def read():                      # 读到 EOF 返回 None，空行跳过
    for line in sys.stdin:
        if line.strip():
            return json.loads(line)
    return None

def update(session_id, upd):
    send({"method": "session/update", "params": {"sessionId": session_id, "update": upd}})

def say(session_id, text):
    update(session_id, {"sessionUpdate": "agent_message_chunk",
                        "content": {"type": "text", "text": text}})

# ② 反过来问 Client：发请求，然后一直读，直到读到同一个 id 的响应
def ask(method, params):
    global next_id
    my_id = next_id
    next_id += 1
    send({"id": my_id, "method": method, "params": params})
    while True:
        msg = read()
        if msg is None:
            sys.exit(0)
        if msg.get("id") == my_id and "method" not in msg:
            return msg.get("result", {})
        log = json.dumps(msg, ensure_ascii=False, separators=(",", ":"))
        print("等回复时收到，先忽略：" + log, file=sys.stderr)

# ③ 一轮对话：先回话，再申请写文件，按用户的决定办
def prompt(params):
    sid = params["sessionId"]
    text = "".join(b.get("text", "") for b in params["prompt"] if b["type"] == "text")
    path = os.path.join(sessions[sid], "echo.txt")
    say(sid, "收到：" + text + "\n")
    update(sid, {"sessionUpdate": "tool_call", "toolCallId": "call_1", "title": "把这句话写进 echo.txt",
                 "kind": "edit", "status": "pending", "locations": [{"path": path}]})
    answer = ask("session/request_permission", {
        "sessionId": sid,
        "toolCall": {"toolCallId": "call_1"},
        "options": [{"optionId": "allow", "name": "允许", "kind": "allow_once"},
                    {"optionId": "reject", "name": "拒绝", "kind": "reject_once"}]})
    outcome = answer.get("outcome", {})
    if outcome.get("outcome") == "cancelled":
        return {"stopReason": "cancelled"}
    if outcome.get("optionId") == "allow":
        with open(path, "w", encoding="utf-8") as f:
            f.write(text + "\n")
        update(sid, {"sessionUpdate": "tool_call_update", "toolCallId": "call_1", "status": "completed"})
        say(sid, "写好了。")
    else:
        update(sid, {"sessionUpdate": "tool_call_update", "toolCallId": "call_1", "status": "failed"})
        say(sid, "你没同意，我就不写了。")
    return {"stopReason": "end_turn"}

# ④ 分发：按 method 决定怎么回，不认识的返回 None
def handle(msg):
    method, params = msg.get("method"), msg.get("params", {})
    if method == "initialize":
        return {"protocolVersion": 1, "agentCapabilities": {},
                "agentInfo": {"name": "echo-agent", "title": "回声 Agent", "version": "0.1.0"},
                "authMethods": []}
    if method == "session/new":
        sid = "sess_" + str(len(sessions) + 1)
        sessions[sid] = params["cwd"]
        return {"sessionId": sid}
    if method == "session/prompt":
        return prompt(params)
    return None

# ⑤ 主循环：读到 EOF（Client 关掉了 stdin）就退出
while (msg := read()) is not None:
    if "id" not in msg:          # 通知（比如 session/cancel）不用回
        continue
    result = handle(msg)
    if result is None:
        send({"id": msg["id"], "error": {"code": -32601, "message": "Method not found"}})
    else:
        send({"id": msg["id"], "result": result})

python3 echo_agent.py

{"jsonrpc":"2.0","id":0,"method":"initialize","params":{"protocolVersion":1,"clientCapabilities":{},"clientInfo":{"name":"hand","version":"0.0.1"}}}

{"jsonrpc":"2.0","id":0,"result":{"protocolVersion":1,"agentCapabilities":{},"agentInfo":{"name":"echo-agent","title":"回声 Agent","version":"0.1.0"},"authMethods":[]}}

mkdir -p /tmp/acp-demo

{"jsonrpc":"2.0","id":1,"method":"session/new","params":{"cwd":"/tmp/acp-demo","mcpServers":[]}}

{"jsonrpc":"2.0","id":1,"result":{"sessionId":"sess_1"}}

{"jsonrpc":"2.0","id":2,"method":"session/prompt","params":{"sessionId":"sess_1","prompt":[{"type":"text","text":"你好，ACP"}]}}

{"jsonrpc":"2.0","method":"session/update","params":{"sessionId":"sess_1","update":{"sessionUpdate":"agent_message_chunk","content":{"type":"text","text":"收到：你好，ACP\n"}}}}
{"jsonrpc":"2.0","method":"session/update","params":{"sessionId":"sess_1","update":{"sessionUpdate":"tool_call","toolCallId":"call_1","title":"把这句话写进 echo.txt","kind":"edit","status":"pending","locations":[{"path":"/tmp/acp-demo/echo.txt"}]}}}
{"jsonrpc":"2.0","id":0,"method":"session/request_permission","params":{"sessionId":"sess_1","toolCall":{"toolCallId":"call_1"},"options":[{"optionId":"allow","name":"允许","kind":"allow_once"},{"optionId":"reject","name":"拒绝","kind":"reject_once"}]}}

{"jsonrpc":"2.0","id":0,"result":{"outcome":{"outcome":"selected","optionId":"allow"}}}

{"jsonrpc":"2.0","method":"session/update","params":{"sessionId":"sess_1","update":{"sessionUpdate":"tool_call_update","toolCallId":"call_1","status":"completed"}}}
{"jsonrpc":"2.0","method":"session/update","params":{"sessionId":"sess_1","update":{"sessionUpdate":"agent_message_chunk","content":{"type":"text","text":"写好了。"}}}}
{"jsonrpc":"2.0","id":2,"result":{"stopReason":"end_turn"}}

{"jsonrpc":"2.0","method":"session/update","params":{"sessionId":"sess_1","update":{"sessionUpdate":"tool_call_update","toolCallId":"call_1","status":"failed"}}}
{"jsonrpc":"2.0","method":"session/update","params":{"sessionId":"sess_1","update":{"sessionUpdate":"agent_message_chunk","content":{"type":"text","text":"你没同意，我就不写了。"}}}}
{"jsonrpc":"2.0","id":2,"result":{"stopReason":"end_turn"}}