ag2-acp

ACP を介した CLI コーディングエージェント

外部の CLI コーディングエージェント — Claude Code、Codex、OpenCode — を、Agent Client Protocol（ACP）を使って、AG2 のファーストクラスの Agent として操作します。AG2 は ACP の Client の役割を担い、各 CLI エージェントは ACP の Agent サブプロセスとして実行されます。エージェントのあらゆる動作 — メッセージ出力、思考、ツール呼び出し、プラン、権限の確認 — は AG2 のイベントストリームに公開されるため、他の AG2 エージェントと同じように、動作を監視し、制御し、オーケストレーションできます。

この連携は設定クラス（ACPConfig とそのプリセット）だけで実現でき、Agent API を変更する必要はありません。

使用する場面

• Python から CLI コーディングエージェントをオーケストレーションしたい場合：ヘッドレスのリファクタリングパイプライン、「マネージャー」エージェントからコーダーエージェントへの委任、一括コード変更の実行など。
• AG2 のイベントストリームで、コーディングエージェントの作業 — 思考、ツール呼び出し、プラン — をリアルタイムに監視したい場合。
• 権限ポリシーまたは人間の介在によって、エージェントの重要な操作（ファイル書き込み、シェルコマンド）を制御したい場合。
• エージェントのファイルアクセスをワークスペースのルート（fs_root）内に制限し、AG2 を介してターミナルの使用を制御する必要がある場合。

このスキルの対象外：AG2 エージェントを他のシステムに公開する場合は、ag2-a2a（A2A プロトコル）または ag2-mcp（MCP サーバー）を使用します。permission_policy="ask" が使用する HITL 入力の接続方法については、ag2-hitl を参照してください。

インストール

pip install "ag2[acp]"

acp エクストラをインストールすると、agent-client-protocol SDK が追加されます。また、各 CLI エージェントには、それぞれの ACP アダプターを PATH 上に用意する必要があります。

エージェント	アダプターのインストール	認証
Claude Code	npm i -g @agentclientprotocol/claude-agent-acp（実行ファイル claude-agent-acp）	env に ANTHROPIC_API_KEY を設定するか、既存の Claude Code ログインを指す CLAUDE_CONFIG_DIR を設定
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 秒で始める — コーディングエージェントに作業を依頼

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

1 回の ask()／run() が、ACP のプロンプトターン 1 回に相当します。CLI エージェントは独自の内部ツールループ（複数回のツール呼び出しを含む場合があります）を実行し、AG2 はそのすべてのステップを発生と同時にストリーム配信します。

アダプターの選択

• ClaudeCodeConfig() — claude-agent-acp を起動します。モデルはアダプターの ANTHROPIC_MODEL 環境変数で選択します。
• CodexConfig() — codex-acp を起動します。モデルはアダプターの MODEL_PROVIDER 環境変数で指定します。
• OpenCodeConfig() — opencode acp を起動します。モデルは OpenCode 独自の設定（opencode.json の "model": "provider/model"）で指定します。

プリセットの model フィールドは応答メタデータ専用で、エージェントには送信されません。上記のとおり、各アダプター独自の方法（環境変数／設定ファイル）でモデルを選択してください。

エージェントの作業を監視する

結果を待つ前に、実行のストリームを購読します。

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 イベント
エージェントのメッセージチャンク	ModelMessageChunk → 最終的な ModelResponse
思考チャンク	ModelReasoning
ツール呼び出し／ツールの結果	BuiltinToolCallEvent／BuiltinToolResultEvent
プラン	ACPPlan（.content／.status／.priority を持つエントリー）
モード変更	ACPModeChange（.mode_id）
使用可能なコマンド	ACPAvailableCommands（.commands）

ACP 固有のイベントは ag2.acp.events にあり、それ以外は ag2.events の標準イベントです。

権限（人間参加型）

エージェントが機密性の高い操作（ファイルの書き込み、コマンドの実行）を行おうとすると、権限リクエストを送信します。permission_policy によって応答が決まります。

ポリシー	動作
"ask"（デフォルト）	エージェントの hitl_hook / context.input にルーティングし、人間が判断します
"auto"	自動的に承認します（ヘッドレスオーケストレーション）
"deny"	自動的に拒否します

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

注意点： "ask" に利用可能な入力ルート（hitl_hook や対話型コンテキスト）がない場合、リクエストは拒否されます。デフォルトのポリシーを使用したヘッドレス実行では、機密性の高い操作がすべて通知なく拒否されます。無人実行では、permission_policy="auto" を明示的に設定してください。

設定リファレンス

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) は設定を複製します（セッションは引き継がれません）。

ライフサイクル

ACP サブプロセスは最初のターンで起動され、同じ実行の以降のターンでも再利用されます。設定から起動した実行中のサブプロセスをすべて終了するには、await config.aclose() を呼び出してください（呼び出しを忘れた場合の安全策として、ファイナライザーがサブプロセスを終了します）。

テスト — インプロセスで実行、サブプロセス不要、API キー不要

ag2.acp.testing.fake_acp_config は、ACPConfig をスクリプト化されたインプロセスエージェントに接続します。各 ACPTurn には、1 回のプロンプトターン（エージェントが送信する 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 エクストラが不足している — pip install "ag2[acp]" を実行してください。これがないと、acp SDK が見つからず from ag2.acp import ... が失敗します。
• エクスポートした API キーは引き継がれない — サブプロセスの環境変数は、絞り込まれた基本環境変数に env= を加えたものです。そのため、シェルで export ANTHROPIC_API_KEY=... を実行しても、エージェントには渡されません。env={"ANTHROPIC_API_KEY": ...} で渡してください。（HOME は基本環境変数に含まれるため、CLAUDE_CONFIG_DIR のログインは機能します。）
• アダプター名が非推奨 — 以前の claude-code-acp（@zed-industries/claude-code-acp）は非推奨です。ClaudeCodeConfig が起動する claude-agent-acp（@agentclientprotocol/claude-agent-acp）を使用してください。
• ヘッドレス実行での "ask" = 拒否 — 人間からの入力ルートがない場合、権限リクエストはすべて拒否されます。無人オーケストレーションでは permission_policy="auto" を設定してください。
• model= は通信内容に影響しません — 代わりに ANTHROPIC_MODEL / MODEL_PROVIDER / opencode.json でモデルを選択してください。
• アダプターが PATH にない — startup_timeout エラーは通常、起動コマンドが見つからないことを示します。アダプターをグローバルにインストールするか、npx -y コマンドで上書きしてください。
• AG2 の tools=[...] は、まだ CLI エージェントに公開されていません — CLI ベースのエージェントは、組み込みのツールを使用します。AG2 が提供するツール向けの MCP ツールブリッジは、上流のロードマップ項目です。

さらに詳しく（信頼できる情報源）

• ag2/acp/config.py — ACPConfig、3 つのプリセットとそのデフォルト値。
• 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