A2A Protocol 1.0入門｜Pythonで最小エージェントサーバーを作る

ケーブルで接続した対等な2台の小型コンピューターを置いた共同作業机 A2A
カテゴリー
A2A
公開日
2026.09.07

はじめに

A2A（Agent2Agent Protocol）は、異なる実装のエージェントが能力を発見し、メッセージやタスクを交換するためのプロトコルです。MCPがツールやコンテキストの接続で使われるのに対し、A2Aは独立したエージェント同士の協調を主対象にします。

この記事では文字列を受け取り、同じ内容を返す最小サーバーの構成を学びます。LLM APIは使わないため、プロトコル部分だけを確認できます。

先に全体像をつかむ

この例では、クライアントが「こんにちは」と送ると、サーバーが「受信: こんにちは」と返します。AIらしい推論は行いません。まず通信部分だけを確認することで、問題がA2A接続にあるのか、LLMにあるのかを分けられます。

  • エージェント：依頼を受け、処理して結果を返すプログラム
  • Agent Card：そのエージェントが何をできるかを説明する公開プロフィール
  • JSON-RPC：JSON形式の要求と応答でメソッドを呼ぶ仕組み
  • ASGI／Uvicorn：PythonのWebアプリをHTTPサーバーとして動かすための仕組み

必要環境

空のフォルダーを作り、その中で次を実行します。Windows以外では仮想環境の有効化コマンドが異なります。

python -m venv .venv
.venv\Scripts\activate
pip install "a2a-sdk[http-server]==1.1.2" uvicorn

以下のコードをa2a_echo.pyとして保存します。最後にASGIアプリまで組み立てるため、そのままUvicornで起動できます。

Agent Card

Agent Cardは名前、説明、URL、対応能力、スキルなどをクライアントへ伝えるメタデータです。秘密情報を書かず、実際に提供できる能力だけを宣言します。

from a2a.types import AgentCapabilities, AgentCard, AgentInterface, AgentSkill

card = AgentCard(
    name="Echo Agent",
    description="受信テキストを返す検証用エージェント",
    supported_interfaces=[AgentInterface(
        url="http://localhost:9999/",
        protocol_binding="JSONRPC",
        protocol_version="1.0",
    )],
    version="1.0.0",
    default_input_modes=["text/plain"],
    default_output_modes=["text/plain"],
    capabilities=AgentCapabilities(streaming=False),
    skills=[AgentSkill(
        id="echo",
        name="Echo",
        description="テキストをそのまま返します",
        tags=["test"],
        examples=["こんにちは"],
    )],
)

SDKの型名やサーバー組み立てAPIは更新される可能性があります。1.1.2以外を使う場合は、公式quickstartの該当バージョンと照合してください。

実行部分を分離する

エージェント固有処理はAgentExecutorとして実装します。受信メッセージからテキストを取り出し、SDKのイベントキューへ結果を返します。

from a2a.server.agent_execution import AgentExecutor, RequestContext
from a2a.server.events import EventQueue
from a2a.helpers import new_text_message

class EchoExecutor(AgentExecutor):
    async def execute(self, context: RequestContext, event_queue: EventQueue):
        text = context.get_user_input()
        await event_queue.enqueue_event(
            new_text_message(f"受信: {text}")
        )

    async def cancel(self, context: RequestContext, event_queue: EventQueue):
        raise Exception("cancel is not supported")

本番では空入力、最大長、許可するPart種別を検査します。ユーザー入力をログへ無加工で残すと個人情報や秘密が混入するため、ログ方針も必要です。

サーバーとして公開する

Python SDKはrequest handler、task store、ASGI applicationを提供します。公式サンプルに合わせてDefaultRequestHandlerへexecutorとstoreを渡し、Agent Cardと共にアプリを構築します。起動はUvicornを利用します。

a2a_echo.pyの末尾へ次を追加します。

from starlette.applications import Starlette
from a2a.server.request_handlers import DefaultRequestHandler
from a2a.server.routes import create_agent_card_routes, create_jsonrpc_routes
from a2a.server.tasks import InMemoryTaskStore

handler = DefaultRequestHandler(
    agent_executor=EchoExecutor(),
    task_store=InMemoryTaskStore(),
    agent_card=card,
)

app = Starlette(routes=[
    *create_agent_card_routes(card),
    *create_jsonrpc_routes(handler, "/"),
])

次のコマンドで起動します。

uvicorn a2a_echo:app --host 127.0.0.1 --port 9999

各部品の分担は次の通りです。

  • Agent Card：能力の説明
  • AgentExecutor：固有の処理
  • RequestHandler：A2A要求をexecutorへ橋渡し
  • TaskStore：タスク状態を保持
  • ASGI app：HTTPエンドポイントを公開

インメモリstoreは開発用です。複数プロセスや再起動をまたぐ本番環境では永続ストアへ置き換えます。

動作確認

  1. サーバーをlocalhostで起動する
  2. well-knownのAgent Cardを取得する
  3. SDKクライアントからテキストメッセージを送る
  4. 受信: ...を含む応答を確認する

異常系として、空メッセージ、未対応Part、巨大入力、不正なJSON-RPC、存在しないtask IDを試します。インターネットへ公開する前にTLS、認証、レート制限、入力上限を追加してください。

成功時には、Agent Card取得でエージェント名Echo Agentが確認でき、メッセージ送信後の応答に受信: こんにちはが含まれます。ブラウザでURLを開いただけではJSON-RPCのメッセージ送信テストにならないため、公式SDKクライアントまたは対応クライアントを使います。

Protocol 1.0のJSON-RPC要求にはA2A-Version: 1.0ヘッダーが必要です。公式Python SDKのクライアントはこのヘッダーを設定します。curlなどで生のHTTP要求を作る場合は自分で追加してください。ヘッダーがない要求は旧版の0.3として解釈され、1.0用サーバーではバージョン不一致になります。

SDKクライアントを用意する前にHTTP経路だけを確認する場合は、次をrequest.jsonとして保存します。messageIdは要求ごとに一意にします。

{
  "jsonrpc": "2.0",
  "id": "echo-1",
  "method": "SendMessage",
  "params": {
    "message": {
      "messageId": "c59b7156-b3d4-4d99-90ce-0d0eeb5e3570",
      "role": "ROLE_USER",
      "parts": [{ "text": "こんにちは" }]
    }
  }
}

サーバーを起動したまま、別のターミナルで送信します。

curl.exe -sS -X POST http://127.0.0.1:9999/ `
  -H "Content-Type: application/json" `
  -H "A2A-Version: 1.0" `
  --data-binary "@request.json"

応答のresult.message.parts内に{"text":"受信: こんにちは"}があれば正常です。不正JSONを送った場合はJSON-RPCエラーコード-32700、存在しないmethodでは-32601になることも確認します。

よくあるトラブル

  • ModuleNotFoundError：仮想環境を有効化したターミナルでpip installしたか確認する
  • Agent Cardを取得できない：UvicornのポートとCard内のURLが一致しているか確認する
  • メッセージがExecutorへ届かない：RequestHandlerへ同じExecutorを渡しているか確認する
  • 再起動するとTaskが消える：開発用のインメモリTaskStoreでは正常な挙動である
  • 外部PCから接続できない：まずlocalhostで確認し、公開設定、ファイアウォール、TLS、認証を別々に調べる

A2AとMCPをどう分けるか

A2A Agentの内部でMCP clientを使い、外部ツールを呼ぶ構成も可能です。A2AとMCPは競合する代替品ではなく、境界が異なります。ただし、A2A経由のユーザー権限をMCPツールへどう伝えるかは自動では決まりません。認証主体と委任範囲を明示します。

まとめ

A2Aサーバーは、発見用のAgent Card、固有処理のExecutor、プロトコル処理のHandler、状態保存のStoreに分けると理解しやすくなります。まずLLMなしのEchoで相互運用を確認してから、モデルや外部ツールを加えてください。

参考リンク

コメント

タイトルとURLをコピーしました