ag2-acp

通过 ACP 使用 CLI 编码 Agent

将外部 CLI 编码 Agent——Claude Code、Codex、OpenCode——作为一等 AG2 Agent 使用，基于 Agent Client Protocol（ACP）。AG2 扮演 ACP Client 角色；每个 CLI Agent 都作为 ACP Agent 子进程运行。Agent 的所有操作——消息输出、思考、工具调用、计划、权限提示——都会输出到 AG2 事件流中，因此你可以像观察、限制和编排其他 AG2 Agent 一样对其进行操作。

该集成只需一个配置类（ACPConfig 及其预设），无需更改 Agent API。

适用场景

• 你想要通过 Python 编排 CLI 编码 Agent：无头重构流水线、“manager” Agent 委派任务给 coder Agent、批量代码修改。
• 你想要在 AG2 事件流中实时观察编码 Agent 的工作——思考、工具调用、计划。
• 你想要通过权限策略或人工介入来限制 Agent 的敏感操作（文件写入、Shell 命令）。
• 你需要将 Agent 的文件访问限制在工作区根目录（fs_root）内，并通过 AG2 中介管理其终端使用。

此技能不适用于将 AG2 Agent 暴露给其他系统——这需要使用 ag2-a2a（A2A 协议）或 ag2-mcp（MCP 服务）。有关 permission_policy="ask" 所依赖的 HITL 输入处理，请参阅 ag2-hitl。

安装

pip install "ag2[acp]"

acp extra 会安装 agent-client-protocol SDK。每个 CLI Agent 还需要将自己的 ACP 适配器安装到 PATH 中：

Agent	适配器安装	身份验证
Claude Code	npm i -g @agentclientprotocol/claude-agent-acp（可执行文件 claude-agent-acp）	在 env 中设置 ANTHROPIC_API_KEY，或设置 CLAUDE_CONFIG_DIR 指向已有的 Claude Code 登录配置
Codex	npm i -g @agentclientprotocol/codex-acp（可执行文件 codex-acp）	CODEX_API_KEY（优先使用）或 OPENAI_API_KEY
OpenCode	opencode CLI 本身（opencode acp）	opencode auth login（或使用环境变量 / .env）

不想全局安装？可以覆盖启动命令，改用 npx：ClaudeCodeConfig(command=["npx", "-y", "@agentclientprotocol/claude-agent-acp"])。

公共 API（from ag2.acp import ...）：ACPConfig、ClaudeCodeConfig、CodexConfig、OpenCodeConfig。

60 秒快速上手——让编码 Agent 执行任务

import asyncio

from ag2 import Agent
from ag2.acp import ClaudeCodeConfig

async def main():
    config = ClaudeCodeConfig(cwd="/path/to/repo")  # workspace root
    agent = Agent("coder", config=config)
    try:
        reply = await agent.ask("Refactor the auth module and add tests")
        print(reply.body)
    finally:
        await config.aclose()  # tear down the CLI subprocess

asyncio.run(main())

每次 ask() / run() 调用都对应一个 ACP 提示词轮次：CLI Agent 会运行自己的内部工具循环（可能包含多次工具调用），AG2 会在每个步骤发生时实时输出。

选择适配器

• ClaudeCodeConfig()——启动 claude-agent-acp。通过适配器的 ANTHROPIC_MODEL 环境变量选择模型。
• CodexConfig()——启动 codex-acp。通过适配器的 MODEL_PROVIDER 环境变量指定模型。
• OpenCodeConfig()——启动 opencode acp。在 OpenCode 自身的配置中指定模型（opencode.json："model": "provider/model"）。

这些预设中的 model 字段仅用于响应元数据——不会发送给 Agent。请使用上述各适配器自身的机制（环境变量 / 配置文件）选择模型。

观察 Agent 的工作

在等待结果之前，先订阅运行事件流：

from ag2 import Agent
from ag2.acp import ClaudeCodeConfig
from ag2.acp.events import ACPPlan
from ag2.events import ModelMessageChunk, ModelReasoning
from ag2.events.tool_events import BuiltinToolCallEvent

agent = Agent("coder", config=ClaudeCodeConfig(cwd="/path/to/repo"))

def observe(event):
    if isinstance(event, ModelReasoning):
        print("thinking:", event.content)
    elif isinstance(event, ModelMessageChunk):
        print(event.content, end="")
    elif isinstance(event, BuiltinToolCallEvent):
        print(f"tool: {event.name}({event.arguments})")  # arguments = JSON string of the tool input
    elif isinstance(event, ACPPlan):
        for step in event.entries:
            print(f"  [{step.status}] {step.content}")

async with agent.run("Add a healthcheck endpoint") as run:
    run.stream.subscribe(observe)
    reply = await run.result()

ACP 会话更新与 AG2 事件的对应关系：

ACP 更新	AG2 事件
Agent 消息片段	ModelMessageChunk → 最终的 ModelResponse
思考片段	ModelReasoning
工具调用 / 工具结果	BuiltinToolCallEvent / BuiltinToolResultEvent
计划	ACPPlan（条目包含 .content / .status / .priority）
模式变更	ACPModeChange（.mode_id）
可用命令	ACPAvailableCommands（.commands）

ACP 专属事件位于 ag2.acp.events；其他事件均为 ag2.events 中的标准事件。

权限（human-in-the-loop）

当 Agent 想要执行敏感操作（写入文件、运行命令）时，会发送权限请求。permission_policy 决定如何处理：

策略	行为
"ask"（默认）	转发给 Agent 的 hitl_hook / context.input，由人工决定
"auto"	自动批准（无头编排）
"deny"	自动拒绝

agent = Agent(
    "coder",
    config=ClaudeCodeConfig(cwd="/repo", permission_policy="auto"),  # fully autonomous
)

**注意：**如果没有可用的输入路由（没有 hitl_hook，也没有交互式上下文），"ask" 会拒绝请求——采用默认策略的无头运行会悄悄拒绝所有敏感操作。对于无人值守运行，请显式设置 permission_policy="auto"。

配置参考

ACPConfig（以及每个预设）接受以下字段：

字段	默认值	用途
command	每个 Agent 的预设值	以 ACP 模式启动 Agent 的可执行文件和参数
cwd	"."	会话的工作区根目录
env	None	额外环境变量，与精简后的基础环境变量（HOME、PATH、USER、SHELL、TERM、LOGNAME，不包含完整的父进程环境变量）合并；请在此处显式传入 API 密钥
model	None	仅用于响应元数据——参见“选择适配器”
permission_policy	"ask"	"ask" / "auto" / "deny"
fs_root	cwd	经中介处理的 fs/* 访问根目录（路径受限）
allow_terminal	True	声明 ACP 终端功能
additional_directories	[]	额外的工作区根目录
startup_timeout	30.0	子进程启动和握手超时时间（秒）
turn_timeout	None	每轮提示词的超时时间（秒）；超时后会取消该轮，对话回复正文为截至当时已流式传输的内容
cancel_timeout	5.0	超时轮次发出 session/cancel 后，强制停止子进程前的宽限时间（秒）

Agent 请求的文件和终端操作由 AG2 中介处理：文件访问仅限于 fs_root，命令在 AG2 的控制下运行。config.copy(**overrides) 会克隆配置（不会沿用会话）。

生命周期

ACP 子进程会在第一轮时启动，并在同一次运行的后续轮次中重复使用。调用 await config.aclose() 可拆除从该配置启动的所有活动子进程（如果忘记调用，终结器会作为安全保障终止它们）。

测试——进程内运行，无子进程，无 API 密钥

ag2.acp.testing.fake_acp_config 会将 ACPConfig 连接到一个脚本化的进程内 Agent：每个 ACPTurn 描述一轮提示词（它发出的 session/update 和停止原因）。你的代码会完整执行公开的 Agent.run 流程。

import asyncio

from acp import schema

from ag2 import Agent
from ag2.acp.testing import ACPTurn, fake_acp_config

def text(t):
    return schema.TextContentBlock(type="text", text=t)

async def main():
    config = fake_acp_config(
        ACPTurn(updates=[
            schema.AgentThoughtChunk(session_update="agent_thought_chunk", content=text("planning")),
            schema.AgentMessageChunk(session_update="agent_message_chunk", content=text("done")),
        ]),
        permission_policy="auto",  # overrides forward to ACPConfig
    )
    agent = Agent("coder", config=config)
    try:
        reply = await agent.ask("hello")
        assert reply.body == "done"
    finally:
        await config.aclose()

asyncio.run(main())

ACPTurn(hang=True) 会阻塞直到被取消，可用于测试 turn_timeout 处理逻辑。

常见注意事项

• 缺少 acp extra——运行 pip install "ag2[acp]"；否则，from ag2.acp import ... 会因缺少 acp SDK 而失败。
• 不会继承已导出的 API 密钥——子进程环境由精简的基础环境变量与 env= 合并而成，因此在 Shell 中执行 export ANTHROPIC_API_KEY=... 不会将其传给 Agent。请通过 env={"ANTHROPIC_API_KEY": ...} 传入。（CLAUDE_CONFIG_DIR 登录可用，因为 HOME 包含在基础环境变量中。）
• 已弃用的适配器名称——旧版 claude-code-acp（@zed-industries/claude-code-acp）已弃用；请使用 claude-agent-acp（@agentclientprotocol/claude-agent-acp），这是 ClaudeCodeConfig 启动的适配器。
• 无头运行中的 "ask" = 拒绝——没有人工输入路由时，每个权限请求都会被拒绝。对于无人值守编排，请设置 permission_policy="auto"。
• model= 不会影响实际请求——请改用 ANTHROPIC_MODEL / MODEL_PROVIDER / opencode.json 选择模型。
• 适配器不在 PATH 中——startup_timeout 错误通常意味着找不到启动命令；请全局安装适配器，或使用 npx -y 命令覆盖项。
• AG2 的 tools=[...] 尚未暴露给 CLI Agent——由 CLI 支持的 Agent 使用自己的内置工具；面向 AG2 所提供工具的 MCP 工具桥接仍在上游路线图中。

深入了解（事实来源）

• ag2/acp/config.py — ACPConfig 以及三个预设和它们的默认值。
• ag2/acp/client.py / bridge.py / session.py — ACP 客户端、事件桥接和子进程生命周期。
• ag2/acp/mappers.py — ACP 更新到 AG2 事件的精确映射。
• ag2/acp/permissions.py — permission_policy 如何处理权限请求。
• ag2/acp/events.py — ACPPlan、ACPModeChange、ACPAvailableCommands。
• ag2/acp/testing.py — fake_acp_config、ACPTurn。
• ACP 协议：https://agentclientprotocol.com