Acp Dev Guide

ACP (Agent Client Protocol) 开发指南

1. 协议概述

ACP 是一个标准化协议，定义了 Client（代码编辑器/IDE）与 Agent（AI 编码助手）之间的通信方式。类似于 LSP 对语言服务器的标准化作用，ACP 让任何兼容的 Agent 都能接入任何兼容的 Client。

协议特性：

• 基于 JSON-RPC 2.0
• 传输层：stdio（主要）、Streamable HTTP（草案）
• 消息类型：Method（请求-响应）、Notification（单向通知）
• 所有文件路径必须是绝对路径，行号从 1 开始

角色定义：

2. 官方文档与资源

2.1 核心文档

2.2 生态系统

3. SDK 获取与安装

3.1 TypeScript SDK

npm install @agentclientprotocol/sdk

核心类：

3.2 Python SDK

pip install agent-client-protocol
# 或
uv add agent-client-protocol

核心类/函数：

3.3 其他 SDK

4. 协议方法完整参考

4.1 Agent 实现的方法（Client → Agent）

必选方法

可选方法

通知（Client → Agent，无响应）

4.2 Client 实现的方法（Agent → Client）

必选方法

可选方法（取决于 Client 声明的能力）

通知（Agent → Client，无响应）

4.3 Capability 协商

Client Capabilities（Client 在 initialize 请求中声明）：

interface ClientCapabilities {
  fs?: {
    readTextFile?: boolean;   // 允许 Agent 读文件
    writeTextFile?: boolean;  // 允许 Agent 写文件
  };
  terminal?: boolean;         // 允许 Agent 创建/管理终端
}

Agent Capabilities（Agent 在 initialize 响应中声明）：

interface AgentCapabilities {
  loadSession?: boolean;        // 支持恢复会话
  promptCapabilities?: {
    image?: boolean;            // 支持图片内容
    audio?: boolean;            // 支持音频内容
    embeddedContext?: boolean;  // 支持嵌入式上下文
  };
  mcp?: {
    http?: boolean;             // 支持 HTTP 传输的 MCP
    sse?: boolean;              // 支持 SSE 传输的 MCP（已弃用）
  };
}

5. 消息流完整生命周期

5.1 初始化

Client → Agent:  initialize { protocolVersion, clientCapabilities, clientInfo }
Agent → Client:  initialize response { protocolVersion, agentCapabilities, agentInfo, authMethods }

5.2 创建会话

Client → Agent:  session/new { cwd, mcpServers }
Agent → Client:  response { sessionId }

5.3 Prompt Turn 循环

Client → Agent:  session/prompt { sessionId, prompt: ContentBlock[] }

  ┌─ Agent 处理循环 ─────────────────────────────────┐
  │                                                    │
  │  Agent → Client: session/update (plan)             │
  │  Agent → Client: session/update (agent_message)    │
  │                                                    │
  │  if tool call:                                     │
  │    Agent → Client: session/update (tool_call)      │
  │    Agent → Client: request_permission (可选)        │
  │    Client → Agent: permission response             │
  │    Agent → Client: session/update (in_progress)    │
  │    Agent → Client: session/update (completed)      │
  │    → 将结果送回 LLM，继续循环                         │
  │                                                    │
  │  if cancelled:                                     │
  │    Client → Agent: session/cancel                  │
  │    Agent → Client: prompt response (cancelled)     │
  │                                                    │
  └────────────────────────────────────────────────────┘

Agent → Client:  session/prompt response { stopReason }

StopReason 类型：

5.4 session/update 通知类型

6. 实现 ACP Agent（TypeScript）

6.1 最小 Agent

import { AgentSideConnection, ndJsonStream, PROTOCOL_VERSION } from '@agentclientprotocol/sdk';
import type { Agent, StopReason } from '@agentclientprotocol/sdk';

const stream = ndJsonStream(process.stdin, process.stdout);

const connection = new AgentSideConnection(
  (conn): Agent => ({
    initialize: async (params) => ({
      protocolVersion: PROTOCOL_VERSION,
      agentCapabilities: {},
      agentInfo: { name: 'my-agent', version: '1.0.0' },
      authMethods: [],
    }),

    newSession: async (params) => ({
      sessionId: `sess_${Date.now()}`,
    }),

    prompt: async (params) => {
      const text = params.prompt
        .filter((b): b is { type: 'text'; text: string } => b.type === 'text')
        .map((b) => b.text)
        .join('\n');

      await conn.sessionUpdate({
        sessionId: params.sessionId,
        update: {
          sessionUpdate: 'agent_message_chunk',
          content: { type: 'text', text: `Echo: ${text}` },
        },
      });

      return { stopReason: 'end_turn' as StopReason };
    },
  }),
  stream,
);

await connection.closed;

6.2 使用 Client 能力（读写文件、终端）

prompt: async (params) => {
  // 读文件（仅当 Client 声明了 fs.readTextFile）
  const file = await conn.readTextFile({
    sessionId: params.sessionId,
    path: '/absolute/path/to/file.ts',
  });

  // 写文件（仅当 Client 声明了 fs.writeTextFile）
  await conn.writeTextFile({
    sessionId: params.sessionId,
    path: '/absolute/path/to/output.ts',
    content: '// generated code',
  });

  // 创建终端（仅当 Client 声明了 terminal）
  const terminal = await conn.createTerminal({
    sessionId: params.sessionId,
    command: 'npm',
    args: ['test'],
    cwd: '/project/root',
  });
  const output = await terminal.waitForExit();

  // 请求用户授权
  const permission = await conn.requestPermission({
    sessionId: params.sessionId,
    toolCall: { toolCallId: 'call_001' },
    options: [
      { optionId: 'allow', name: 'Allow', kind: 'allow_once' },
      { optionId: 'reject', name: 'Reject', kind: 'reject_once' },
    ],
  });

  return { stopReason: 'end_turn' };
}

7. 实现 ACP Agent（Python）

7.1 最小 Agent

import asyncio
from acp import Agent, PromptResponse, run_agent

class EchoAgent(Agent):
    async def initialize(self, protocol_version, client_capabilities, **kw):
        return {
            "protocolVersion": 1,
            "agentCapabilities": {},
            "agentInfo": {"name": "echo-agent", "version": "1.0.0"},
            "authMethods": [],
        }

    async def new_session(self, cwd, mcp_servers=None, **kw):
        return {"sessionId": f"sess_{id(self)}"}

    async def prompt(self, prompt, session_id, **kw) -> PromptResponse:
        text = " ".join(b["text"] for b in prompt if b.get("type") == "text")
        await self.connection.session_update(
            session_id=session_id,
            update={
                "sessionUpdate": "agent_message_chunk",
                "content": {"type": "text", "text": f"Echo: {text}"},
            },
        )
        return PromptResponse(stop_reason="end_turn")

if __name__ == "__main__":
    asyncio.run(run_agent(EchoAgent()))

7.2 作为 Client 启动 Agent

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

class MyClient(Client):
    async def request_permission(self, options, session_id, tool_call, **kw):
        return {"outcome": {"outcome": "selected", "optionId": options[0]["optionId"]}}

    async def session_update(self, session_id, update, **kw):
        print(f"[{session_id}] {update}")

async def main():
    async with spawn_agent_process(
        MyClient(), "gemini", "--experimental-acp"
    ) as (conn, proc):
        await conn.initialize(protocol_version=1)
        session = await conn.new_session(cwd="/my/project", mcp_servers=[])
        result = await conn.prompt(
            session_id=session.session_id,
            prompt=[text_block("Explain this codebase")],
        )
        print(f"Stop reason: {result.stop_reason}")

asyncio.run(main())

8. 实现 ACP Client（TypeScript）

AgentDispatch 的 ClientNode 就是一个 ACP Client 的实现。核心步骤：

import { ClientSideConnection, ndJsonStream, PROTOCOL_VERSION } from '@agentclientprotocol/sdk';
import { spawn } from 'node:child_process';

// 1. 启动 Agent 子进程
const proc = spawn('gemini', ['--experimental-acp'], {
  stdio: ['pipe', 'pipe', 'pipe'],
  cwd: '/project/root',
});

// 2. 建立 ACP 连接
const stream = ndJsonStream(proc.stdout, proc.stdin);
const connection = new ClientSideConnection(
  (_agent) => ({
    // Client 实现的方法
    requestPermission: async (params) => ({
      outcome: { outcome: 'selected', optionId: params.options[0].optionId },
    }),
    readTextFile: async (params) => ({
      content: await fs.readFile(params.path, 'utf-8'),
    }),
    writeTextFile: async (params) => {
      await fs.writeFile(params.path, params.content, 'utf-8');
      return null;
    },
    sessionUpdate: async (params) => {
      console.log('Update:', JSON.stringify(params.update));
    },
  }),
  stream,
);

// 3. 初始化
const initResult = await connection.initialize({
  protocolVersion: PROTOCOL_VERSION,
  clientCapabilities: {
    fs: { readTextFile: true, writeTextFile: true },
    terminal: true,
  },
  clientInfo: { name: 'my-client', version: '1.0.0' },
});

// 4. 创建会话
const session = await connection.newSession({
  cwd: '/project/root',
  mcpServers: [],
});

// 5. 发送 prompt
const result = await connection.prompt({
  sessionId: session.sessionId,
  prompt: [{ type: 'text', text: 'Review the code in src/' }],
});

9. 常用 Agent 的 ACP 支持形态

核心规律：几乎所有 Agent CLI 的默认模式是交互式终端，需要特定子命令/标志才能进入 ACP stdio 模式。

9.1 分类总览

9.2 完整命令参考

9.3 常见错误

// ❌ 错误 — 启动交互终端，不是 ACP
{ "command": "gemini" }
{ "command": "cline" }

// ✅ 正确 — 进入 ACP stdio 模式
{ "command": "gemini", "args": ["--experimental-acp"] }
{ "command": "cline",  "args": ["--acp"] }

10. 调试与测试

10.1 手动测试 ACP Agent

通过管道直接向 Agent 发送 JSON-RPC 消息：

echo '{"jsonrpc":"2.0","id":0,"method":"initialize","params":{"protocolVersion":1,"clientCapabilities":{"fs":{"readTextFile":true,"writeTextFile":true},"terminal":true},"clientInfo":{"name":"test","version":"0.1.0"}}}' | gemini --experimental-acp

10.2 使用 acpx CLI 测试

acpx 是一个通用 ACP CLI 客户端：

npx acpx gemini --experimental-acp

10.3 在 Zed 中验证

在 ~/.config/zed/settings.json 中添加自定义 Agent：

{
  "agent_servers": {
    "My Agent": {
      "type": "custom",
      "command": "node",
      "args": ["./dist/my-agent.js"]
    }
  }
}

11. 协议 JSON Schema

完整的协议 JSON Schema 可从以下地址获取：

• https://agentclientprotocol.com/protocol/schema
• https://agentclientprotocol.com/api-reference/openapi.json

TypeScript SDK 中的类型定义直接从 Schema 生成，可作为实现参考。

12. 开发检查清单

实现新的 ACP Agent 或 Client 前，确认以下事项：