📚
ACP (Agent Communication Protocol) の4つの特長をQuickstartで体感する

nsatoh
2025/05/02に公開
 はじめに前回の記事では、「BeeAI」の全体像と、その中核を担うコンポーネント群（BeeAI Framework、BeeAI Platform、ACP）についてざっくり紹介しました。今回は、その中でも特にエージェント同士のやり取りを担う重要な仕組み、ACP(Agent Communication Protocol) に焦点を当てていきます。
ACPは、異なるAIエージェント同士が共通の言語で通信できるようにするための、オープンスタンダードなプロトコルです。たとえば、LangChain製のエージェントとBeeAI製のエージェントが、フレームワークを越えてタスクを協調できる──そんな未来を実現するための土台とも言えるでしょう。
このプロトコルは、Linux Foundationのプロジェクトとしてコミュニティ主導で開発されており、特定ベンダーのサービスにロックインされることなく、オープンかつ相互運用性のあるエージェントエコシステムを構築できるのが大きな魅力です。
この記事では、ACP公式ドキュメントの「Quickstart」セクションをベースに、実際にコードを動かしながら、ACPの4つの主要な特長を体感的に理解していきます。
!ACPは、2025年5月時点では アルファ版 として公開されており、今後仕様の変更や機能追加が行われる可能性があります。

 ACPの4つの特長ACPは、AIエージェント同士の通信を現実のユースケースに耐えうるものにするため、次の4つの特長を備えています。どれも「動くものを作る」現場で効いてくる設計です。

 1. 非同期優先・同期対応（Async-first, Sync Supported）ACPは非同期通信を基本としつつ、同期通信もサポートしています。
非同期優先：時間のかかる処理でも、レスポンスを待たずに次の処理を進められる
同期対応：単純なリクエスト・レスポンス型のやり取りも問題なく対応
用途に応じて通信スタイルを選べるため、エージェントの性質やタスク内容にあわせて柔軟に設計できます。

 2. RESTベースの通信（REST-based Communication）ACPはWebで広く使われているRESTの原則に則っています。
明快なエンドポイント設計と標準HTTPメソッド
Web開発者にとって馴染み深く、学習コストが小さい
ロードバランサーやセキュリティツールなど既存インフラとの相性も良好
JSON-RPCのような専用プロトコルよりも、実装と運用のハードルが低いのが強みです。

 3. SDKなしでも使える（No SDK Required）ACPは、curlやPostmanのような一般的なHTTPツールだけでも使えます。
特定言語やライブラリに依存しない
導入時のハードルが極めて低い
必要に応じて使える公式SDKも用意
これにより、PoCレベルから本番環境まで幅広い開発に対応できます。

 4. オフライン検出（Offline Discovery）ACPでは、起動していないエージェントも事前に検出できます。
マニフェスト機能により、エージェントの機能やインターフェースを事前公開
ゼロスケーリングとの親和性が高い設計
必要なときだけエージェントを起動できるため、コスト・パフォーマンスの最適化が可能
特に、クラウドや大規模な分散環境でのリソース最適化に有効です。
これら4つの特長が組み合わさることで、ACPは、実運用に適した、柔軟かつ扱いやすいエージェント間通信の基盤となっています。

次のセクションでは、Quickstartのコードを動かしながら、これらの特長がどのように現れるのかを具体的に見ていきましょう。
なお、今回取り上げるQuickstartのサンプルでは、Offline Discovery機能は含まれていないため、オンライン状態のエージェントを確認する方法のみ紹介します。マニフェストによるオフライン検出の仕組みについては、別の機会に掘り下げたいと思います。

 Quickstartをベースにエージェント環境を準備するここからは、ACPの公式ドキュメントにある「Quickstart」のサンプルコードをベースに、実際にエージェントを立ち上げて通信させてみます。

今回はいくつか手を加えて「より動きがわかりやすい」形に調整しています。

 動作環境Python 3.11以上（執筆時点のACP SDK前提）

acp-sdk v0.8.1（執筆時点の最新版）
SDKをインストールします：
pip install acp-sdk

 使用するサンプル構成今回使用するサンプルは以下の3ファイルです。Quickstartにあるコードをベースに、日本語対応や非同期処理の流れがわかりやすいように一部改変しています。
agent_jp.py：受信メッセージに反応するエージェントのサンプル（日本語対応）
client_sync.py：同期的にメッセージを送信するクライアント
client_async.py：非同期にメッセージを送信するクライアント
agent_jp.pyACPエージェントの基本構造を示すシンプルなサンプルです。メッセージを受け取ると、日本語で順に処理状況を返してくれる「エコー・エージェント」になっています。
import asyncio
from collections.abc import AsyncGenerator

from acp_sdk.models import Message, MessagePart
from acp_sdk.server import Context, RunYield, RunYieldResume, Server

server = Server()

@server.agent()
async def japanese_echo(
    input: list[Message], context: Context
) -> AsyncGenerator[RunYield, RunYieldResume]:
    """日本語でメッセージを逐次的に処理するエージェント"""
    
    # 最初の挨拶を送信
    yield MessagePart(content="こんにちは！メッセージを受け取りました。順番に処理します。")
    
    for i, message in enumerate(input):
        # 2秒ごとに処理状況を送信
        await asyncio.sleep(2)
        yield MessagePart(content=f"メッセージ {i+1} を確認中です...")
        
        # メッセージ本体をエコーで返す
        await asyncio.sleep(2)
        for part in message.parts:
            yield MessagePart(content=f"メッセージ {i+1}: 「{part.content}」 を受信しました。")

    # 全ての処理が完了したことを通知
    yield MessagePart(content="全てのメッセージを処理しました！ありがとうございました。")

server.run()
client_sync.pyこのスクリプトでは、メッセージを送信したあと、run_sync() を使って同期的に処理し、エージェントからの出力をすべて受け取ってから表示します。
import asyncio

from acp_sdk.client import Client
from acp_sdk.models import Message, MessagePart


async def example() -> None:
    async with Client(base_url="http://localhost:8000") as client:
        run = await client.run_sync(
            agent="japanese_echo",
            input=[
                Message(parts=[MessagePart(content="こんにちは!!", content_type="text/plain")]),
                Message(parts=[MessagePart(content="今日の天気はどうですか？", content_type="text/plain")]),
                Message(parts=[MessagePart(content="明日の予定を教えてください。", content_type="text/plain")])
            ],
        )
        print(run.output)


if __name__ == "__main__":
    asyncio.run(example())
client_async.pyこちらは run_stream() を使って、エージェントからの応答をリアルタイムで1つずつ処理して逐次的に表示します。
import asyncio
from acp_sdk.client import Client
from acp_sdk.models import Message, MessagePart

async def example() -> None:
    async with Client(base_url="http://localhost:8000") as client:
        messages = [
            Message(parts=[MessagePart(content="こんにちは！", content_type="text/plain")]),
            Message(parts=[MessagePart(content="今日の天気はどうですか？", content_type="text/plain")]),
            Message(parts=[MessagePart(content="明日の予定を教えてください。", content_type="text/plain")])
        ]
        
        async for event in client.run_stream(
            agent="japanese_echo",
            input=messages
        ):
            if hasattr(event, 'message_part'):
                print(f"受信: {event.message_part.content}")
            else:
                print(f"イベント: {event}")

if __name__ == "__main__":
    asyncio.run(example())

 エージェントの実行エージェントを起動します：
$ python agent_jp.py 
INFO:     Started server process [21524]
INFO:     Waiting for application startup.
INFO:     Application startup complete.
INFO:     Uvicorn running on http://127.0.0.1:8000 (Press CTRL+C to quit)

 実際に試してみる
 特長1：非同期優先・同期対応ACPの大きな特徴のひとつが、「非同期通信が基本設計にありつつ、同期通信にも対応している」という点です。用意したクライアントコードを使えば、この特長を実際に体験できます。

 まずは同期クライアントからclient_sync.py を実行すると、メッセージをエージェントに送信し、すべての応答がまとまって返ってくるシンプルな挙動になります。
$ python client_sync.py 
[Message(parts=[MessagePart(name=None, content_type='text/plain', content='こんにちは！メッセージを受け
取りました。順番に処理します。', content_encoding='plain', content_url=None), MessagePart(name=None, content_type='text/plain', content='メッセージ 1 を確認中です...', content_encoding='plain', content_url=None), MessagePart(name=None, content_type='text/plain', content='メッセージ 1: 「こんにちは!!」 を受信し ました。', content_encoding='plain', content_url=None), MessagePart(name=None, content_type='text/plain', content='メッセージ 2 を確認中です...', content_encoding='plain', content_url=None), MessagePart(name=None, content_type='text/plain', content='メッセージ 2: 「今日の天気はどうですか？」 を受信しました。', content_encoding='plain', content_url=None), MessagePart(name=None, content_type='text/plain', content='メッセージ 3 を確認中です...', content_encoding='plain', content_url=None), MessagePart(name=None, content_type='text/plain', content='メッセージ 3: 「明日の予定を教えてください。」 を受信しました。', content_encoding='plain', content_url=None), MessagePart(name=None, content_type='text/plain', content='全ての メッセージを処理しました！ありがとうございました。', content_encoding='plain', content_url=None)], created_at=datetime.datetime(2025, 5, 2, 8, 45, 50, 409807, tzinfo=TzInfo(UTC)), completed_at=datetime.datetime(2025, 5, 2, 8, 46, 2, 499402, tzinfo=TzInfo(UTC)))]
このように、非同期な内部処理が行われていても、外から見れば「1回のリクエストに対して、1回の応答が返ってくる」同期的な使い方ができます。既存のAPI設計に慣れている開発者にとって、馴染みやすいスタイルです。

 次に非同期ストリームクライアント一方で、client_async.py を使うと、エージェントからの応答が部分的に逐次送られてくる様子が観察できます。
$ python client_async.py 
イベント: type='run.created' run=Run(run_id=UUID('c79e049a-4b40-41ee-8c5a-9c3c14999c64'), agent_name='japanese_echo', session_id=UUID('e2ec7baf-a49f-4867-b085-fafff4898dc6'), status=<RunStatus.CREATED: 'created'>, await_request=None, output=[], error=None, created_at=datetime.datetime(2025, 5, 2, 8, 48, 48, 919130, tzinfo=TzInfo(UTC)), finished_at=None)
イベント: type='run.in-progress' run=Run(run_id=UUID('c79e049a-4b40-41ee-8c5a-9c3c14999c64'), agent_name='japanese_echo', session_id=UUID('e2ec7baf-a49f-4867-b085-fafff4898dc6'), status=<RunStatus.IN_PROGRESS: 'in-progress'>, await_request=None, output=[], error=None, created_at=datetime.datetime(2025, 5, 2, 8, 48, 48, 919130, tzinfo=TzInfo(UTC)), finished_at=None)
イベント: type='message.created' message=Message(parts=[], created_at=datetime.datetime(2025, 5, 2, 8, 48, 48, 923126, tzinfo=TzInfo(UTC)), completed_at=None)
イベント: type='message.part' part=MessagePart(name=None, content_type='text/plain', content='こんにちは！メッセージを受け取りました。順番に処理します。', content_encoding='plain', content_url=None)
イベント: type='message.part' part=MessagePart(name=None, content_type='text/plain', content='メッセージ 1 を確認中です...', content_encoding='plain', content_url=None)
イベント: type='message.part' part=MessagePart(name=None, content_type='text/plain', content='メッセージ 1: 「こんにちは！」 を受信しました。', content_encoding='plain', content_url=None)
イベント: type='message.part' part=MessagePart(name=None, content_type='text/plain', content='メッセージ 2 を確認中です...', content_encoding='plain', content_url=None)
イベント: type='message.part' part=MessagePart(name=None, content_type='text/plain', content='メッセージ 2: 「今日の天気はどうですか？」 を受信しました。', content_encoding='plain', content_url=None)
イベント: type='message.part' part=MessagePart(name=None, content_type='text/plain', content='メッセージ 3 を確認中です...', content_encoding='plain', content_url=None)
イベント: type='message.part' part=MessagePart(name=None, content_type='text/plain', content='メッセージ 3: 「明日の予定を教えてください。」 を受信しました。', content_encoding='plain', content_url=None)     
イベント: type='message.part' part=MessagePart(name=None, content_type='text/plain', content='全てのメッセージを処理しました！ありがとうございました。', content_encoding='plain', content_url=None)
イベント: type='message.completed' message=Message(parts=[MessagePart(name=None, content_type='text/plain', content='こんにちは！メッセージを受け取りました。順番に処理します。', content_encoding='plain', content_url=None), MessagePart(name=None, content_type='text/plain', content='メッセージ 1 を確認中です...', content_encoding='plain', content_url=None), MessagePart(name=None, content_type='text/plain', content='メッセージ 1: 「こんにちは！」 を受信しました。', content_encoding='plain', content_url=None), MessagePart(name=None, content_type='text/plain', content='メッセージ 2 を確認中です...', content_encoding='plain', content_url=None), MessagePart(name=None, content_type='text/plain', content='メッセージ 2: 「今日の天気はどうですか？」 を受信しました。', content_encoding='plain', content_url=None), MessagePart(name=None, content_type='text/plain', content='メッセージ 3 を確認中です...', content_encoding='plain', content_url=None), MessagePart(name=None, content_type='text/plain', content='メッセージ 3: 「明日の予定を教え てください。」 を受信しました。', content_encoding='plain', content_url=None), MessagePart(name=None, content_type='text/plain', content='全てのメッセージを処理しました！ありがとうございました。', content_encoding='plain', content_url=None)], created_at=datetime.datetime(2025, 5, 2, 8, 48, 48, 923126, tzinfo=TzInfo(UTC)), completed_at=datetime.datetime(2025, 5, 2, 8, 49, 0, 979161, tzinfo=TzInfo(UTC)))
イベント: type='run.completed' run=Run(run_id=UUID('c79e049a-4b40-41ee-8c5a-9c3c14999c64'), agent_name='japanese_echo', session_id=UUID('e2ec7baf-a49f-4867-b085-fafff4898dc6'), status=<RunStatus.COMPLETED: 'completed'>, await_request=None, output=[Message(parts=[MessagePart(name=None, content_type='text/plain', content='こんにちは！メッセージを受け取りました。順番に処理します。', content_encoding='plain', content_url=None), MessagePart(name=None, content_type='text/plain', content='メッセージ 1 を確認中です...', content_encoding='plain', content_url=None), MessagePart(name=None, content_type='text/plain', content='メッセージ 1: 「こんにちは！」 を受信しました。', content_encoding='plain', content_url=None), MessagePart(name=None, content_type='text/plain', content='メッセージ 2 を確認中です...', content_encoding='plain', content_url=None), MessagePart(name=None, content_type='text/plain', content='メッセージ 2: 「今日の天 気はどうですか？」 を受信しました。', content_encoding='plain', content_url=None), MessagePart(name=None, content_type='text/plain', content='メッセージ 3 を確認中です...', content_encoding='plain', content_url=None), MessagePart(name=None, content_type='text/plain', content='メッセージ 3: 「明日の予定を教えて ください。」 を受信しました。', content_encoding='plain', content_url=None), MessagePart(name=None, content_type='text/plain', content='全てのメッセージを処理しました！ありがとうございました。', content_encoding='plain', content_url=None)], created_at=datetime.datetime(2025, 5, 2, 8, 48, 48, 923126, tzinfo=TzInfo(UTC)), completed_at=datetime.datetime(2025, 5, 2, 8, 49, 0, 979161, tzinfo=TzInfo(UTC)))], error=None, created_at=datetime.datetime(2025, 5, 2, 8, 48, 48, 919130, tzinfo=TzInfo(UTC)), finished_at=datetime.datetime(2025, 5, 2, 8, 49, 0, 980151, tzinfo=TzInfo(UTC)))
このように、リアルタイムに進行状況を伝えられる非同期処理は、マルチステップの推論やチャットインターフェース、あるいは処理時間が長いAIエージェントにとって極めて重要です。

 なぜ重要なのか？この「同期でも非同期でも動かせる」柔軟な設計は、以下のような使い分けが可能です。
Webアプリやモバイルアプリなど、即時応答が求められるUIでは同期
バックグラウンド処理やストリーミングが必要な複雑なワークフローでは非同期
つまり、用途や設計方針に応じて、最適な通信スタイルを選べるのがACPの強みです。

 特長2 & 3：RESTベースの通信 ＆ SDKなしでも利用可能ACPは、Web標準のRESTアーキテクチャに基づいて設計されており、さらに専用SDKがなくても利用可能という柔軟性があります。このふたつの特長の組み合わせにより、学習コストを抑えながら、多様な開発環境での導入が可能です。
ここではcurlを使って、SDKを使わずにACPエージェントにアクセスしてみましょう。

 curlでエージェントにアクセスしてみるまずは、以下のような内容の request.json を用意します。
request.json
{
    "agent_name": "japanese_echo",
    "input": [
      { "parts": [ { "content": "こんにちは！", "content_type": "text/plain" } ] },
      { "parts": [ { "content": "今日の天気はどうですか？", "content_type": "text/plain" } ] },
      { "parts": [ { "content": "明日の予定を教えてください。", "content_type": "text/plain" } ] }
    ]
  }
そして以下のコマンドを実行します。
$ curl -X POST http://localhost:8000/runs \
  -H "Content-Type: application/json; charset=utf-8" \
  -d @request.json
{"run_id":"0246262a-b26d-49eb-b8fa-d25a9a68de58","agent_name":"japanese_echo","session_id":"30e0a1f5-1fc0-4c9d-812a-f835bd020bdf","status":"completed","await_request":null,"output":[{"parts":[{"name":null,"content_type":"text/plain","content":"こんにちは！メッセージを受け取りました。順番に処理します。","content_encoding":"plain","content_url":null},{"name":null,"content_type":"text/plain","content":"メッセ ージ 1 を確認中です...","content_encoding":"plain","content_url":null},{"name":null,"content_type":"text/plain","content":"メッセージ 1: 「こんにちは！」 を受信しました。","content_encoding":"plain","content_url":null},{"name":null,"content_type":"text/plain","content":"メッセージ 2 を確認中です...","content_encoding":"plain","content_url":null},{"name":null,"content_type":"text/plain","content":"メッセージ 2: 「今日の天気はどうですか？」 を受信しました。","content_encoding":"plain","content_url":null},{"name":null,"content_type":"text/plain","content":"メッセージ 3 を確認中です...","content_encoding":"plain","content_url":null},{"name":null,"content_type":"text/plain","content":"メッセージ 3: 「明日の予定を教えてください。」 を受信しました。","content_encoding":"plain","content_url":null},{"name":null,"content_type":"text/plain","content":"全てのメッセージを処理しました！ありがとうございました。","content_encoding":"plain","content_url":null}],"created_at":"2025-05-01T06:13:50.006470Z","completed_at":"2025-05-01T06:13:53.071320Z"}],"error":null,"created_at":"2025-05-01T06:13:50.006470Z","finished_at":"2025-05-01T06:13:53.071320Z"}

 RESTベースのメリット（特長2）この例からわかるように、ACPはHTTPのPOSTメソッドと明確なエンドポイント設計（ここでは /runs）を使い、一般的なRESTful APIと同じ感覚で使えます。これは以下のような利点があります。
特別なプロトコル知識が不要（例：gRPCやJSON-RPCなど）
Webインフラ（ロードバランサー、APIゲートウェイ、モニタリングツールなど）との高い互換性
エンドポイントが明確でドキュメント化しやすい

 SDKなしでも動くという柔軟さ（特長3）上記の例のように、curlやPostmanといったHTTPツールだけで、エージェントの実行・応答の取得が可能です。
開発初期のプロトタイピングが素早くできる
言語やフレームワークに縛られない
SDKを使う前の動作確認にも便利

 特長4：エージェント検出機能（Offline Discovery）ーのかわりに稼働中のエージェント確認ACPには、AIエージェントを事前に把握できる「検出」機能が備わっています。これは、本来、非稼働状態のエージェントであってもマニフェストファイルを通じてその仕様を公開し、ゼロスケーリングやダイナミックな起動に対応するための仕組みです。
今回のQuickStart環境では、この完全なオフライン検出（manifestベース）の機能は実装されていませんが、稼働中のエージェント情報を確認する仕組みは利用可能です。
以下のコマンドで、現在起動中のエージェント一覧を確認できます。
$ curl http://localhost:8000/agents
{"agents":[{"name":"japanese_echo","description":"日本語でメッセージを逐次的に処理するエージェント","metadata":{"annotations":null,"documentation":null,"license":null,"programming_language":null,"natural_languages":null,"framework":null,"capabilities":null,"domains":null,"tags":null,"created_at":null,"updated_at":null,"author":null,"contributors":null,"links":null,"dependencies":null,"recommended_models":null}}]}
このように、エージェントの名前や説明文、付随するメタデータなどを外部から取得できることで、自動化されたサービス連携やインフラリソースの動的管理への応用が期待されます。

 さいごに今回は、ACP（Agent Communication Protocol）の基本的な特長と、簡単なサンプルを通じた使い方を確認しました。
特に以下の点が理解できました。
非同期処理と同期処理の両対応により、用途に応じた柔軟な通信設計が可能であること
RESTベースの通信により、一般的なWebの知識で扱いやすく、SDKを使わずにcurlでも操作可能であること
稼働中のエージェント検出が可能で、簡単にインターフェース情報を取得できること
これにより、ACPが現実的な開発環境や本番運用を意識した設計になっていることがよく分かりました。
一方で、今回扱えなかった部分もいくつかあり、今後の調査・検証ポイントとして以下のようなテーマが挙げられます。
オフラインディスカバリーの仕組み：マニフェストを使って、エージェントが稼働していなくても機能情報を取得する仕組みの実態
エージェントの本格的な公開手法：どのようにACP準拠のエージェントをクラウドや本番環境にデプロイ・運用するのか
セキュリティやアクセス制御の設計：エージェント間の認証・認可の扱いや、外部からのアクセス制限の方法
これらの観点を掘り下げていくことで、ACPをより実用的に活用する道筋が見えてくるはずです。次のステップとして、公式ドキュメントやコミュニティ情報を参照しながら、これらの点を具体的に試していきたいと思います。
DXC LabPublication
AI / Data Science / Robotics のエンジニア集団として情報を公開していきます。日本IBMシステムズ・エンジニアリングのLab組織が有志で活動をしています。
ACP (Agent Communication Protocol) の4つの特長をQuickstartで体感する

はじめに

ACPの4つの特長

1. 非同期優先・同期対応（Async-first, Sync Supported）

2. RESTベースの通信（REST-based Communication）

3. SDKなしでも使える（No SDK Required）

4. オフライン検出（Offline Discovery）

Quickstartをベースにエージェント環境を準備する

動作環境

使用するサンプル構成

エージェントの実行

実際に試してみる

特長1：非同期優先・同期対応

まずは同期クライアントから

次に非同期ストリームクライアント

なぜ重要なのか？

特長2 & 3：RESTベースの通信＆ SDKなしでも利用可能

curlでエージェントにアクセスしてみる

RESTベースのメリット（特長2）

SDKなしでも動くという柔軟さ（特長3）

特長4：エージェント検出機能（Offline Discovery）ーのかわりに稼働中のエージェント確認

さいごに

Discussion

はじめに

ACPの4つの特長

1. 非同期優先・同期対応（Async-first, Sync Supported）

2. RESTベースの通信（REST-based Communication）

3. SDKなしでも使える（No SDK Required）

4. オフライン検出（Offline Discovery）

Quickstartをベースにエージェント環境を準備する

動作環境

使用するサンプル構成

エージェントの実行

実際に試してみる

特長1：非同期優先・同期対応

まずは同期クライアントから

次に非同期ストリームクライアント

なぜ重要なのか？

特長2 & 3：RESTベースの通信 ＆ SDKなしでも利用可能

curlでエージェントにアクセスしてみる

RESTベースのメリット（特長2）

SDKなしでも動くという柔軟さ（特長3）

特長4：エージェント検出機能（Offline Discovery）ーのかわりに稼働中のエージェント確認

さいごに

Discussion

特長2 & 3：RESTベースの通信＆ SDKなしでも利用可能
