# AgentCore ランタイムに A2A サーバーをデプロイする
Amazon Bedrock AgentCore AgentCore Runtime では、AgentCore Runtime で Agent-to-Agent (A2A) AgentCore サーバーをデプロイして実行できます。このガイドでは、最初の A2A サーバーの作成、テスト、デプロイについて説明します。
このセクションでは、以下を行います。
+ Amazon Bedrock AgentCore が A2A をサポートする方法
+ エージェント機能を使用して A2A サーバーを作成する方法
+ サーバーをローカルでテストする方法
+ サーバーを にデプロイする方法 AWS
+ デプロイされたサーバーを呼び出す方法
+ 検出用にエージェントカードを取得する方法
A2A の詳細については、「A[A2Aプロトコル契約](runtime-a2a-protocol-contract.md)」を参照してください。
**Topics**
+ [Amazon Bedrock AgentCore が A2A をサポートする方法](#runtime-a2a-how-agentcore-supports)
+ [AgentCore ランタイムでの A2A の使用](#runtime-a2a-steps)
+ [付録](#runtime-a2a-appendix)
## Amazon Bedrock AgentCore が A2A をサポートする方法
Amazon Bedrock AgentCore の A2A プロトコルサポートにより、透過的なプロキシレイヤーとして機能し、A2A サーバーとのシームレスな統合が可能になります。A2A 用に設定されている場合、Amazon Bedrock AgentCore は、デフォルトの A2A サーバー設定と一致するルートパス () `0.0.0.0:9000/` のポート`9000`でコンテナがステートレスでストリーミング可能な HTTP サーバーを実行することを期待します。 A2A
このサービスは、プロトコルの透明性を維持しながらエンタープライズグレードのセッション分離を提供します。[InvokeAgentRuntime](https://docs.aws.amazon.com/bedrock-agentcore/latest/APIReference/API_InvokeAgentRuntime.html) API からの JSON-RPC ペイロードは、変更なしで A2A コンテナに直接渡されます。このアーキテクチャは、 のエージェントカードによる組み込みエージェント検出`/.well-known/agent-card.json`や JSON-RPC 通信などの標準の A2A プロトコル機能を保持し、エンタープライズ認証 (SigV4/OAuth 2.0) とスケーラビリティを追加します。
他のプロトコルとの主な違いは、ポート (HTTP の場合は 9000 対 8080)、マウントパス ( `/invocations` `/`対 )、標準化されたエージェント検出メカニズムです。Amazon Bedrock AgentCore は、本番環境の A2A エージェントに最適なデプロイプラットフォームです。
他のプロトコルとの主な違い:
**[ポート]**
A2A サーバーはポート 9000 (HTTP の場合は 8080、MCP の場合は 8000) で実行されます。
**[Path]** (パス)
A2A サーバーは にマウントされます `/` (HTTP `/invocations`の場合は 、MCP `/mcp`の場合は )
**エージェントカード**
A2A は、 のエージェントカードを通じて組み込みエージェント検出を提供します。 `/.well-known/agent-card.json`
**[プロトコル]**
agent-to-agent通信に JSON-RPC を使用する
**認証**
SigV4 認証スキームと OAuth 2.0 認証スキームの両方をサポート
詳細については、「[https://a2a-protocol.org/](https://a2a-protocol.org/)」を参照してください。
## AgentCore ランタイムでの A2A の使用
このチュートリアルでは、A2A サーバーを作成、テスト、デプロイします。
**Topics**
+ [前提条件](#runtime-a2a-prerequisites)
+ [ステップ 1: A2A プロジェクトを作成する](#runtime-a2a-create-server)
+ [ステップ 2: A2A サーバーをローカルでテストする](#runtime-a2a-test-locally)
+ [ステップ 3: A2A サーバーを Bedrock AgentCore ランタイムにデプロイする](#runtime-a2a-deploy)
+ [ステップ 4: エージェントカードを取得する](#runtime-a2a-step-4)
+ [ステップ 5: デプロイされた A2A サーバーを呼び出す](#runtime-a2a-step-5)
### 前提条件
+ Python 3.10 以降がインストールされ、Python の基本的な理解
+ Node.js 18 以上がインストールされている (AgentCore CLI に必要)
+ AgentCore CLI がインストールされました。 `npm install -g @aws/agentcore`
+ 適切なアクセス許可とローカル認証情報が設定されている AWS アカウント
+ A2A プロトコルとagent-to-agent通信の概念を理解する
### ステップ 1: A2A プロジェクトを作成する
この例では Strands エージェントを使用していますが、AgentCore CLI は LangChain/LangGraph と Google ADK を使用した A2A プロジェクトもサポートしています。
#### プロジェクトの足場
次のコマンドを実行し、プロンプトが表示されたらフレームワークとして *Strands* を選択します。
```
agentcore create --protocol A2A
```
CLI は、必要なすべての依存関係と設定を含む完全なプロジェクトをスキャフォールドします。生成された `main.py`には A2A サーバーが含まれています。
```
from strands import Agent, tool
from strands.multiagent.a2a.executor import StrandsA2AExecutor
from bedrock_agentcore.runtime import serve_a2a
from model.load import load_model
@tool
def add_numbers(a: int, b: int) -> int:
"""Return the sum of two numbers."""
return a + b
tools = [add_numbers]
agent = Agent(
model=load_model(),
system_prompt="You are a helpful assistant. Use tools when appropriate.",
tools=tools,
)
if __name__ == "__main__":
serve_a2a(StrandsA2AExecutor(agent))
```
#### コードについて
**ストランドエージェント**
特定のツールと機能を持つエージェントを作成します。
**StrandsA2AExecutor**
Strands エージェントをラップして A2A プロトコルの互換性を提供します
**serve\_a2a**
Bedrock 互換 A2A サーバーを起動する Amazon Bedrock AgentCore SDK ヘルパー。`/ping` ヘルスエンドポイント、エージェントカードサービス、`AGENTCORE_RUNTIME_URL`環境変数、Bedrock ヘッダーの伝播を処理し、デフォルトでポート 9000 で実行されます。
**ポート 9000**
エージェントAgentCoreランタイムでデフォルトでポート 9000 で実行される A2A サーバー
このエージェントをカスタマイズするには、`add_numbers`ツールを独自のツールに置き換え、システムプロンプトを更新します。
### ステップ 2: A2A サーバーをローカルでテストする
ローカル開発環境で A2A サーバーを実行してテストします。
#### A2A サーバーを起動する
AgentCore CLI を使用して A2A サーバーをローカルで起動します。
```
agentcore dev
```
これにより、ウェブブラウザで AgentCore エージェントインスペクターが開きます。代わりにターミナルベースの TUI を使用するには、 を使用します`agentcore dev --no-browser`。
または、サーバーを直接実行することもできます。
```
python main.py
```
サーバーがポート で実行されていることを示す出力が表示されます`9000`。
#### エージェントを呼び出す
```
curl -X POST http://localhost:9000/ \
-H "Content-Type: application/json" \
-d '{
"jsonrpc": "2.0",
"id": "req-001",
"method": "message/send",
"params": {
"message": {
"role": "user",
"parts": [
{
"kind": "text",
"text": "what is 101 * 11?"
}
],
"messageId": "12345678-1234-1234-1234-123456789012"
}
}
}' | jq .
```
#### エージェントカードの取得をテストする
エージェントカードエンドポイントはローカルでテストできます。
```
curl http://localhost:9000/.well-known/agent-card.json | jq.
```
A2A Inspector [を使用したリモートテスト」で説明されているように、A2A Inspector を使用してデプロイされたサーバーをテスト](https://github.com/a2aproject/a2a-inspector)することもできます。
### ステップ 3: A2A サーバーを Bedrock AgentCore ランタイムにデプロイする
#### 認証用に Cognito ユーザープールを設定する
デプロイする前に、デプロイされたサーバーへの安全なアクセスのために認証を設定します。Cognito のセットアップ手順の詳細については、[「認証用の Cognito ユーザープールのセットアップ](runtime-mcp.md#runtime-mcp-appendix-a)」を参照してください。これにより、デプロイされたサーバーへの安全なアクセスに必要な OAuth トークンが提供されます。
#### にデプロイする AWS
エージェントをデプロイします。
```
agentcore deploy
```
このコマンドでは、次の操作を行います。
1. エージェントコードと依存関係をパッケージ化する
1. デプロイアーティファクトを Amazon S3 にアップロードする
1. Amazon Bedrock AgentCore ランタイムを作成する
1. エージェントを にデプロイする AWS
デプロイ後、エージェントランタイム ARN は次のようになります。
```
arn:aws:bedrock-agentcore:us-west-2:accountId:runtime/my_a2a_server-xyz123
```
### ステップ 4: エージェントカードを取得する
エージェントカードは、A2A サーバーのアイデンティティ、機能、スキル、サービスエンドポイント、認証要件を説明する JSON メタデータドキュメントです。これにより、A2A エコシステムでエージェントの自動検出が有効になります。
#### 環境変数をセットアップする
環境変数をセットアップする
1. ベアラートークンを環境変数としてエクスポートします。ベアラートークンの設定については、[「ベアラートークンの設定](https://docs.aws.amazon.com/bedrock-agentcore/latest/devguide/runtime-mcp.html#runtime-mcp-appendix)」を参照してください。
```
export BEARER_TOKEN=""
```
1. エージェント ARN をエクスポートします。
```
export AGENT_ARN="arn:aws:bedrock-agentcore:us-west-2:accountId:runtime/my_a2a_server-xyz123"
```
#### エージェントカードを取得する
```
import os
import json
import requests
from uuid import uuid4
from urllib.parse import quote
def fetch_agent_card():
# Get environment variables
agent_arn = os.environ.get('AGENT_ARN')
bearer_token = os.environ.get('BEARER_TOKEN')
if not agent_arn:
print("Error: AGENT_ARN environment variable not set")
return
if not bearer_token:
print("Error: BEARER_TOKEN environment variable not set")
return
# URL encode the agent ARN
escaped_agent_arn = quote(agent_arn, safe='')
# Construct the URL
url = f"https://bedrock-agentcore.us-west-2.amazonaws.com/runtimes/{escaped_agent_arn}/invocations/.well-known/agent-card.json"
# Generate a unique session ID
session_id = str(uuid4())
print(f"Generated session ID: {session_id}")
# Set headers
headers = {
'Accept': '*/*',
'Authorization': f'Bearer {bearer_token}',
'X-Amzn-Bedrock-AgentCore-Runtime-Session-Id': session_id
}
try:
# Make the request
response = requests.get(url, headers=headers)
response.raise_for_status()
# Parse and pretty print JSON
agent_card = response.json()
print(json.dumps(agent_card, indent=2))
return agent_card
except requests.exceptions.RequestException as e:
print(f"Error fetching agent card: {e}")
return None
if __name__ == "__main__":
fetch_agent_card()
```
エージェントカードから URL を取得したら、環境変数`AGENTCORE_RUNTIME_URL`としてエクスポートします。
```
export AGENTCORE_RUNTIME_URL="https://bedrock-agentcore.us-west-2.amazonaws.com/runtimes//invocations/"
```
### ステップ 5: デプロイされた A2A サーバーを呼び出す
クライアントコードを作成して、デプロイされた Amazon Bedrock AgentCore A2A サーバーを呼び出し、機能テスト用のメッセージを送信します。
デプロイされた A2A サーバー`my_a2a_client_remote.py`を呼び出す新しいファイルを作成します。
```
import asyncio
import logging
import os
from uuid import uuid4
import httpx
from a2a.client import A2ACardResolver, ClientConfig, ClientFactory
from a2a.types import Message, Part, Role, TextPart
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
DEFAULT_TIMEOUT = 300 # set request timeout to 5 minutes
def create_message(*, role: Role = Role.user, text: str) -> Message:
return Message(
kind="message",
role=role,
parts=[Part(TextPart(kind="text", text=text))],
message_id=uuid4().hex,
)
async def send_sync_message(message: str):
# Get runtime URL from environment variable
runtime_url = os.environ.get('AGENTCORE_RUNTIME_URL')
# Generate a unique session ID
session_id = str(uuid4())
print(f"Generated session ID: {session_id}")
# Add authentication headers for Amazon Bedrock AgentCore
headers = {"Authorization": f"Bearer {os.environ.get('BEARER_TOKEN')}",
'X-Amzn-Bedrock-AgentCore-Runtime-Session-Id': session_id}
async with httpx.AsyncClient(timeout=DEFAULT_TIMEOUT, headers=headers) as httpx_client:
# Get agent card from the runtime URL
resolver = A2ACardResolver(httpx_client=httpx_client, base_url=runtime_url)
agent_card = await resolver.get_agent_card()
# Agent card contains the correct URL (same as runtime_url in this case)
# No manual override needed - this is the path-based mounting pattern
# Create client using factory
config = ClientConfig(
httpx_client=httpx_client,
streaming=False, # Use non-streaming mode for sync response
)
factory = ClientFactory(config)
client = factory.create(agent_card)
# Create and send message
msg = create_message(text=message)
# With streaming=False, this will yield exactly one result
async for event in client.send_message(msg):
if isinstance(event, Message):
logger.info(event.model_dump_json(exclude_none=True, indent=2))
return event
elif isinstance(event, tuple) and len(event) == 2:
# (Task, UpdateEvent) tuple
task, update_event = event
logger.info(f"Task: {task.model_dump_json(exclude_none=True, indent=2)}")
if update_event:
logger.info(f"Update: {update_event.model_dump_json(exclude_none=True, indent=2)}")
return task
else:
# Fallback for other response types
logger.info(f"Response: {str(event)}")
return event
# Usage - Uses AGENTCORE_RUNTIME_URL environment variable
asyncio.run(send_sync_message("what is 101 * 11"))
```
## 付録
**Topics**
+ [認証用に Cognito ユーザープールを設定する](#runtime-a2a-setup-cognito-appendix)
+ [A2A インスペクターによるリモートテスト](#runtime-a2a-remote-testing)
+ [トラブルシューティング](#runtime-a2a-troubleshooting)
### 認証用に Cognito ユーザープールを設定する
Cognito のセットアップ手順の詳細については、MCP ドキュメントの[「認証用に Cognito ユーザープール](https://docs.aws.amazon.com/bedrock-agentcore/latest/devguide/runtime-mcp.html#set-up-cognito-user-pool-for-authentication)を設定する」を参照してください。
### A2A インスペクターによるリモートテスト
「[https://github.com/a2aproject/a2a-inspector](https://github.com/a2aproject/a2a-inspector)」を参照してください。
### トラブルシューティング
**A2A-specific一般的な問題**
以下は、発生する可能性のある一般的な問題です。
ポートの競合
A2A サーバーは AgentCore ランタイム環境のポート 9000 で実行する必要があります
JSON-RPC エラー
クライアントが適切にフォーマットされた JSON-RPC 2.0 メッセージを送信していることを確認します。
認可方法の不一致
リクエストで、エージェントが設定されたのと同じ認証方法 (OAuth または SigV4) を使用していることを確認します。
**例外処理**
エラー処理の A2A 仕様: [https://a2a-protocol.org/latest/specification/#81-standard-json-rpc-errors](https://a2a-protocol.org/latest/specification/#81-standard-json-rpc-errors)
A2A サーバーは、HTTP 200 ステータスコードを持つ標準の JSON-RPC エラーレスポンスとしてエラーを返します。内部ランタイムエラーは、プロトコルコンプライアンスを維持するために JSON-RPC 内部エラーに自動的に変換されます。
このサービスは、標準化された JSON-RPC エラーコードで適切な A2A-compliantのエラーレスポンスを提供するようになりました。
| JSON-RPC エラーコード | ランタイム例外 | HTTP エラーコード | JSON-RPC エラーメッセージ |
| --- | --- | --- | --- |
| 該当なし | `AccessDeniedException` | 403 | 該当なし |
| -32501 | `ResourceNotFoundException` | 404 | リソースが見つかりません — リクエストされたリソースは存在しません |
| -32502 | `ValidationException` | 400 | 検証エラー – 無効なリクエストデータ |
| -32503 | `ThrottlingException` | 429 | レート制限の超過 – リクエストが多すぎます |
| -32503 | `ServiceQuotaExceededException` | 429 | レート制限の超過 – リクエストが多すぎます |
| -32504 | `ResourceConflictException` | 409 | リソースの競合 — リソースが既に存在します |
| -32505 | `RuntimeClientError` | 424 | ランタイムクライアントエラー – 詳細については、CloudWatch ログを確認してください。 |