CLI coding agents over ACP
Drive external CLI coding agents —
Claude Code,
Codex,
OpenCode — as first-class AG2
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 (
and its presets) — no changes to the
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 () and its terminal use mediated by AG2.
Not this skill: exposing an AG2 agent
to other systems — that is
(A2A protocol) or
(MCP server). For the HITL input plumbing that
relies on, see
.
Installation
The
extra pulls in the
SDK. Each CLI agent additionally needs its own ACP adapter on
:
| Agent | Adapter install | Auth |
|---|
| Claude Code | npm i -g @agentclientprotocol/claude-agent-acp
| in , or pointing at an existing Claude Code login |
| Codex | npm i -g @agentclientprotocol/codex-acp
| (takes precedence) or |
| OpenCode | CLI itself () | (or env / ) |
No global install? Override the launch command to use npx:
ClaudeCodeConfig(command=["npx", "-y", "@agentclientprotocol/claude-agent-acp"])
.
60-second recipe — ask a coding agent to do work
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
/
= 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.
Choosing an adapter
- — launches . Select the model via the adapter's env var.
- — launches . Model via the adapter's env var.
- — launches . Model in OpenCode's own config (:
"model": "provider/model"
The presets'
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.
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 | → final |
| thinking chunk | |
| tool call / tool result | / |
| plan | (entries with / / ) |
| mode change | () |
| available commands | () |
The ACP-specific events live in
; the rest are the standard events from
.
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.
decides the answer:
| Policy | Behavior |
|---|
| (default) | Route to the agent's / — a human decides |
| Approve automatically (headless orchestration) |
| Reject automatically |
python
agent = Agent(
"coder",
config=ClaudeCodeConfig(cwd="/repo", permission_policy="auto"), # fully autonomous
)
Pitfall: with no input route available (no
, no interactive context)
denies the request — a headless run with the default policy will quietly reject every sensitive action. For unattended runs set
explicitly.
Configuration reference
(and every preset) accepts:
| Field | Default | Purpose |
|---|
| preset per agent | Executable + args launching the agent in ACP mode |
| | Workspace root for the session |
| | Extra env vars, merged over a trimmed base env (, , , , , — not the full parent env); pass API keys here explicitly |
| | Response metadata only — see "Choosing an adapter" |
| | / / |
| | Root for mediated access (path-confined) |
| | Advertise the ACP terminal capability |
| | Extra workspace roots |
| | Subprocess spawn + handshake timeout (s) |
| | Per-prompt-turn timeout (s); on expiry the turn is cancelled and the reply body is whatever streamed so far |
| | Grace period (s) after a timed-out turn signals before the subprocess is hard-stopped |
File and terminal operations the agent requests are mediated by AG2: file access is confined to
, and commands run under AG2's control.
clones a config (sessions are not carried over).
Lifecycle
The ACP subprocess is spawned on the first turn and
reused across turns of the same run. Call
to tear down all live subprocesses started from a config (a finalizer terminates them as a safety net if you forget).
Testing — in-process, no subprocess, no API keys
ag2.acp.testing.fake_acp_config
wires an
to a scripted in-process agent: each
describes one prompt turn (the
s it emits and the stop reason). Your code exercises the full public
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())
blocks until cancelled — use it to exercise
handling.
Common pitfalls
- Missing extra — ; without it fails on the missing SDK.
- Exported API keys are not inherited — the subprocess env is a trimmed base set plus , so
export ANTHROPIC_API_KEY=...
env={"ANTHROPIC_API_KEY": ...}
- Deprecated adapter name — the old (
@zed-industries/claude-code-acp
@agentclientprotocol/claude-agent-acp
- in headless runs = deny — with no human input route, every permission request is rejected. Set for unattended orchestration.
- does nothing on the wire — select the model via / / instead.
- Adapter not on — errors usually mean the launch command wasn't found; install the adapter globally or use the command override.
- AG2 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.
Going deeper (source of truth)
- — + the three presets and their defaults.
- / / — the ACP Client, event bridging, subprocess lifecycle.
- — the exact ACP-update → AG2-event mapping.
- — how resolves permission requests.
- — , , .
- — , .
- ACP protocol: https://agentclientprotocol.com