ag2-acp

Compare original and translation side by side

🇺🇸

Original

English

🇨🇳

Translation

Chinese

CLI coding agents over ACP

基于ACP的CLI编码代理

Drive external CLI coding agents — Claude Code, Codex, OpenCode — as first-class AG2

Agent

s, using the Agent Client Protocol (ACP). AG2 plays the ACP Client role; each CLI agent runs as an ACP Agent subprocess. Everything the agent does — message output, thinking, tool calls, plans, permission prompts — is externalized onto AG2's event stream, so you can observe, gate, and orchestrate it like any other AG2 agent.

The integration is just a config class (

ACPConfig

and its presets) — no changes to the

Agent

API.

通过Agent Client Protocol（ACP）将外部CLI编码代理——Claude Code、Codex、OpenCode——作为一等AG2

Agent

进行驱动。AG2扮演ACP Client角色；每个CLI代理以ACP Agent子进程的形式运行。代理的所有操作——消息输出、思考过程、工具调用、计划、权限提示——都会被外化到AG2的事件流中，因此你可以像管控其他AG2 Agent一样观察、管控和编排它。

该集成仅通过一个配置类（

ACPConfig

及其预设）实现——无需修改

Agent

API。

When to use

使用场景

You want to orchestrate CLI coding agents from Python: headless refactoring pipelines, "manager" agents delegating to coder agents, batch code-mod runs.
You want to observe a coding agent's work live — thoughts, tool calls, plans — on the AG2 event stream.
You want to gate the agent's sensitive actions (file writes, shell commands) with a permission policy or a human in the loop.
You need the agent's file access confined to a workspace root (
```
fs_root
```
) and its terminal use mediated by AG2.

Not this skill: exposing an AG2 agent to other systems — that is

ag2-a2a

(A2A protocol) or

ag2-mcp

(MCP server). For the HITL input plumbing that

permission_policy="ask"

relies on, see

ag2-hitl

你希望从Python中编排CLI编码代理：无头重构流水线、将任务委托给编码代理的「管理型」Agent、批量代码修改运行。
你希望实时观察编码代理的工作过程——思考内容、工具调用、计划——在AG2事件流中。
你希望通过权限策略或人机协同方式管控代理的敏感操作（文件写入、Shell命令）。
你需要将代理的文件访问限制在工作区根目录（
```
fs_root
```
），并由AG2管控其终端使用。

不适用场景：将AG2 Agent暴露给其他系统——这属于

ag2-a2a

（A2A协议）或

ag2-mcp

（MCP服务器）的功能。

permission_policy="ask"

依赖的人机协同输入流程，请查看

ag2-hitl

。

Installation

安装

bash

pip install "ag2[acp]"

The

acp

extra pulls in the

agent-client-protocol

SDK. Each CLI agent additionally needs its own ACP adapter on

PATH

Agent Adapter install Auth

Claude Code

Agent	Adapter install	Auth
Claude Code	`npm i -g @agentclientprotocol/claude-agent-acp` (bin `claude-agent-acp` )	`ANTHROPIC_API_KEY` in `env` , or `CLAUDE_CONFIG_DIR` pointing at an existing Claude Code login
Codex	`npm i -g @agentclientprotocol/codex-acp` (bin `codex-acp` )	`CODEX_API_KEY` (takes precedence) or `OPENAI_API_KEY`
OpenCode	`opencode` CLI itself ( `opencode acp` )	`opencode auth login` (or env / `.env` )

npm i -g @agentclientprotocol/claude-agent-acp

(bin

claude-agent-acp

)

ANTHROPIC_API_KEY

env

, or

CLAUDE_CONFIG_DIR

pointing at an existing Claude Code login

Codex

npm i -g @agentclientprotocol/codex-acp

(bin

codex-acp

)

CODEX_API_KEY

(takes precedence) or

OPENAI_API_KEY

OpenCode

opencode

CLI itself (

opencode acp

)

opencode auth login

(or env /

.env

)

No global install? Override the launch command to use npx:

ClaudeCodeConfig(command=["npx", "-y", "@agentclientprotocol/claude-agent-acp"])

Public API (

from ag2.acp import ...

ACPConfig

ClaudeCodeConfig

CodexConfig

OpenCodeConfig

bash

pip install "ag2[acp]"

acp

扩展依赖会引入

agent-client-protocol

SDK。每个CLI代理还需要在

PATH

中配置其对应的ACP适配器：

代理	适配器安装方式	认证方式
Claude Code	`npm i -g @agentclientprotocol/claude-agent-acp` （可执行文件 `claude-agent-acp` ）	环境变量 `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-second recipe — ask a coding agent to do work

60秒快速入门——让编码代理执行任务

python

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())

One

ask()

run()

= one ACP prompt turn: the CLI agent runs its own internal tool loop (possibly many tool calls) and AG2 streams every step as it happens.

python

import asyncio

from ag2 import Agent
from ag2.acp import ClaudeCodeConfig

async def main():
    config = ClaudeCodeConfig(cwd="/path/to/repo")  # 工作区根目录
    agent = Agent("coder", config=config)
    try:
        reply = await agent.ask("重构认证模块并添加测试用例")
        print(reply.body)
    finally:
        await config.aclose()  # 终止CLI子进程

asyncio.run(main())

一次

ask()

run()

调用对应一次ACP 对话轮次：CLI代理会运行自身的内部工具循环（可能包含多次工具调用），AG2会实时流式传输每一步操作。

Choosing an adapter

选择适配器

```
ClaudeCodeConfig()
```
— launches
```
claude-agent-acp
```
. Select the model via the adapter's
```
ANTHROPIC_MODEL
```
env var.
```
CodexConfig()
```
— launches
```
codex-acp
```
. Model via the adapter's
```
MODEL_PROVIDER
```
env var.

OpenCodeConfig()

— launches

opencode acp

. Model in OpenCode's own config (

opencode.json

"model": "provider/model"

The presets'
model
field is response metadata only — it is not sent to the agent. Pick the model through each adapter's own mechanism (env var / config file) as above.

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

预设中的
model
字段仅作为响应元数据——不会发送给代理。请通过上述适配器自身的机制（环境变量/配置文件）选择模型。

Observing the agent's work

观察代理工作过程

Subscribe to the run's stream before awaiting the result:

python

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()

How ACP session updates map onto AG2 events:

ACP update	AG2 event
agent message chunk	`ModelMessageChunk` → final `ModelResponse`
thinking chunk	`ModelReasoning`
tool call / tool result	`BuiltinToolCallEvent` / `BuiltinToolResultEvent`
plan	`ACPPlan` (entries with `.content` / `.status` / `.priority` )
mode change	`ACPModeChange` ( `.mode_id` )
available commands	`ACPAvailableCommands` ( `.commands` )

The ACP-specific events live in

ag2.acp.events

; the rest are the standard events from

ag2.events

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

python

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("思考中:", event.content)
    elif isinstance(event, ModelMessageChunk):
        print(event.content, end="")
    elif isinstance(event, BuiltinToolCallEvent):
        print(f"工具调用: {event.name}({event.arguments})")  # arguments = 工具输入的JSON字符串
    elif isinstance(event, ACPPlan):
        for step in event.entries:
            print(f"  [{step.status}] {step.content}")

async with agent.run("添加健康检查端点") as run:
    run.stream.subscribe(observe)
    reply = await run.result()

ACP会话更新与AG2事件的映射关系：

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

ACP专属事件位于

ag2.acp.events

；其余为

ag2.events

中的标准事件。

Permissions (human-in-the-loop)

权限管控（人机协同）

When the agent wants to perform a sensitive action (write a file, run a command), it sends a permission request.

permission_policy

decides the answer:

Policy	Behavior
`"ask"` (default)	Route to the agent's `hitl_hook` / `context.input` — a human decides
`"auto"`	Approve automatically (headless orchestration)
`"deny"`	Reject automatically

python

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

Pitfall:
"ask"
with no input route available (no
hitl_hook
, no interactive context) denies the request — a headless run with the default policy will quietly reject every sensitive action. For unattended runs set
permission_policy="auto"
explicitly.

当代理想要执行敏感操作（写入文件、运行命令）时，会发送权限请求。

permission_policy

决定处理方式：

策略	行为
`"ask"` （默认）	路由到代理的 `hitl_hook` / `context.input` ——由人工决定
`"auto"`	自动批准（无头编排场景）
`"deny"`	自动拒绝

python

agent = Agent(
    "coder",
    config=ClaudeCodeConfig(cwd="/repo", permission_policy="auto"),  # 完全自主运行
)

注意事项：在没有可用输入路由的情况下使用
"ask"
（无
hitl_hook
、无交互上下文）会拒绝请求——默认策略下的无头运行会静默拒绝所有敏感操作。无人值守运行请显式设置
permission_policy="auto"
。

Configuration reference

配置参考

ACPConfig

(and every preset) accepts:

Field	Default	Purpose
`command`	preset per agent	Executable + args launching the agent in ACP mode
`cwd`	`"."`	Workspace root for the session
`env`	`None`	Extra env vars, merged over a trimmed base env ( `HOME` , `PATH` , `USER` , `SHELL` , `TERM` , `LOGNAME` — not the full parent env); pass API keys here explicitly
`model`	`None`	Response metadata only — see "Choosing an adapter"
`permission_policy`	`"ask"`	`"ask"` / `"auto"` / `"deny"`
`fs_root`	`cwd`	Root for mediated `fs/*` access (path-confined)
`allow_terminal`	`True`	Advertise the ACP terminal capability
`additional_directories`	`[]`	Extra workspace roots
`startup_timeout`	`30.0`	Subprocess spawn + handshake timeout (s)
`turn_timeout`	`None`	Per-prompt-turn timeout (s); on expiry the turn is cancelled and the reply body is whatever streamed so far
`cancel_timeout`	`5.0`	Grace period (s) after a timed-out turn signals `session/cancel` before the subprocess is hard-stopped

File and terminal operations the agent requests are mediated by AG2: file access is confined to

fs_root

, and commands run under AG2's control.

config.copy(**overrides)

clones a config (sessions are not carried over).

ACPConfig

（及其所有预设）支持以下配置项：

字段	默认值	用途
`command`	各代理预设值	启动代理进入ACP模式的可执行文件+参数
`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` 信号后的宽限期（秒），之后强制终止子进程

代理请求的文件和终端操作由AG2管控：文件访问被限制在

fs_root

，命令在AG2控制下运行。

config.copy(**overrides)

可克隆配置（不会保留会话状态）。

Lifecycle

生命周期

The ACP subprocess is spawned on the first turn and reused across turns of the same run. Call

await config.aclose()

to tear down all live subprocesses started from a config (a finalizer terminates them as a safety net if you forget).

ACP子进程在第一次轮次时启动，并在同一运行的多轮对话中复用。调用

await config.aclose()

可终止该配置启动的所有活跃子进程（如果忘记调用，终结器会作为安全网自动终止它们）。

Testing — in-process, no subprocess, no API keys

测试——进程内、无子进程、无需API密钥

ag2.acp.testing.fake_acp_config

wires an

ACPConfig

to a scripted in-process agent: each

ACPTurn

describes one prompt turn (the

session/update

s it emits and the stop reason). Your code exercises the full public

Agent.run

path.

python

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)

blocks until cancelled — use it to exercise

turn_timeout

handling.

ag2.acp.testing.fake_acp_config

将

ACPConfig

连接到一个脚本化的进程内代理：每个

ACPTurn

描述一轮对话（包含其发送的

session/update

和终止原因）。你的代码可以完整测试

Agent.run

的公开路径。

python

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",  # 覆盖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

的处理逻辑。

Common pitfalls

常见陷阱

Missing
acp
extra —

pip install "ag2[acp]"

; without it

from ag2.acp import ...

fails on the missing

acp

SDK.

Exported API keys are not inherited — the subprocess env is a trimmed base set plus
```
env=
```
, so
```
export ANTHROPIC_API_KEY=...
```
in your shell does not reach the agent. Pass it via
```
env={"ANTHROPIC_API_KEY": ...}
```
. (
```
CLAUDE_CONFIG_DIR
```
logins work because
```
HOME
```
is in the base set.)

Deprecated adapter name — the old

claude-code-acp

(

@zed-industries/claude-code-acp

) is deprecated; use

claude-agent-acp

(

@agentclientprotocol/claude-agent-acp

), which is what

ClaudeCodeConfig

launches.

"ask"
in headless runs = deny — with no human input route, every permission request is rejected. Set
```
permission_policy="auto"
```
for unattended orchestration.
model=
does nothing on the wire — select the model via
```
ANTHROPIC_MODEL
```
/
```
MODEL_PROVIDER
```
/
```
opencode.json
```
instead.
Adapter not on
PATH
—
```
startup_timeout
```
errors usually mean the launch command wasn't found; install the adapter globally or use the
```
npx -y
```
command override.
AG2
tools=[...]
are not exposed to the CLI agent yet — CLI-backed agents use their own built-in tools; the MCP tool bridge for AG2-provided tools is an upstream roadmap item.

缺少
acp
扩展依赖——请执行
```
pip install "ag2[acp]"
```
；否则
```
from ag2.acp import ...
```
会因缺少
```
acp
```
SDK而失败。
导出的API密钥未被继承——子进程环境是精简的基础环境加上
```
env=
```
配置的变量，因此Shell中
```
export ANTHROPIC_API_KEY=...
```
不会传递给代理。请通过
```
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代理——基于CLI的代理使用自身内置工具；AG2提供工具的MCP工具桥是上游路线图中的功能。

Going deeper (source of truth)

深入了解（权威来源）

```
ag2/acp/config.py
```
—
```
ACPConfig
```
+ the three presets and their defaults.
```
ag2/acp/client.py
```
/
```
bridge.py
```
/
```
session.py
```
— the ACP Client, event bridging, subprocess lifecycle.
```
ag2/acp/mappers.py
```
— the exact ACP-update → AG2-event mapping.
```
ag2/acp/permissions.py
```
— how
```
permission_policy
```
resolves permission requests.

ag2/acp/events.py

—

ACPPlan

ACPModeChange

ACPAvailableCommands

ag2/acp/testing.py

—

fake_acp_config

ACPTurn

ACP protocol: https://agentclientprotocol.com

```
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