AgentCore ランタイムに A2A サーバーをデプロイする

Amazon Bedrock AgentCore AgentCore Runtime では、AgentCore Runtime で Agent-to-Agent (A2A) AgentCore サーバーをデプロイして実行できます。このガイドでは、最初の A2A サーバーの作成、テスト、デプロイについて説明します。

このセクションでは、以下を行います。

A2A の詳細については、「AA2Aプロトコル契約」を参照してください。

Amazon Bedrock AgentCore が A2A をサポートする方法

Amazon Bedrock AgentCore の A2A プロトコルサポートにより、透過的なプロキシレイヤーとして機能し、A2A サーバーとのシームレスな統合が可能になります。A2A 用に設定されている場合、Amazon Bedrock AgentCore は、デフォルトの A2A サーバー設定と一致するルートパス () 0.0.0.0:9000/ のポート9000でコンテナがステートレスでストリーミング可能な HTTP サーバーを実行することを期待します。 A2A

このサービスは、プロトコルの透明性を維持しながらエンタープライズグレードのセッション分離を提供します。InvokeAgentRuntime API からの JSON-RPC ペイロードは、変更なしで A2A コンテナに直接渡されます。このアーキテクチャは、 のエージェントカードによる組み込みエージェント検出/.well-known/agent-card.jsonや JSON-RPC 通信などの標準の A2A プロトコル機能を保持し、エンタープライズ認証 (SigV4/OAuth 2.0) とスケーラビリティを追加します。

他のプロトコルとの主な違いは、ポート (HTTP の場合は 9000 対 8080)、マウントパス ( /invocations /対 )、標準化されたエージェント検出メカニズムです。Amazon Bedrock AgentCore は、本番環境の A2A エージェントに最適なデプロイプラットフォームです。

他のプロトコルとの主な違い:

詳細については、「https://a2a-protocol.org/」を参照してください。

AgentCore ランタイムでの A2A の使用

このチュートリアルでは、A2A サーバーを作成、テスト、デプロイします。

前提条件

ステップ 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))

コードについて

このエージェントをカスタマイズするには、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 を使用してデプロイされたサーバーをテストすることもできます。

ステップ 3: A2A サーバーを Bedrock AgentCore ランタイムにデプロイする

認証用に Cognito ユーザープールを設定する

デプロイする前に、デプロイされたサーバーへの安全なアクセスのために認証を設定します。Cognito のセットアップ手順の詳細については、「認証用の Cognito ユーザープールのセットアップ」を参照してください。これにより、デプロイされたサーバーへの安全なアクセスに必要な OAuth トークンが提供されます。

にデプロイする AWS

エージェントをデプロイします。

agentcore deploy

このコマンドでは、次の操作を行います。

デプロイ後、エージェントランタイム ARN は次のようになります。

arn:aws:bedrock-agentcore:us-west-2:accountId:runtime/my_a2a_server-xyz123

ステップ 4: エージェントカードを取得する

エージェントカードは、A2A サーバーのアイデンティティ、機能、スキル、サービスエンドポイント、認証要件を説明する JSON メタデータドキュメントです。これにより、A2A エコシステムでエージェントの自動検出が有効になります。

環境変数をセットアップする

環境変数をセットアップする

エージェントカードを取得する

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/<ARN>/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"))

付録

認証用に Cognito ユーザープールを設定する

Cognito のセットアップ手順の詳細については、MCP ドキュメントの「認証用に Cognito ユーザープールを設定する」を参照してください。

A2A インスペクターによるリモートテスト

「https://github.com/a2aproject/a2a-inspector」を参照してください。

トラブルシューティング

A2A-specific一般的な問題

以下は、発生する可能性のある一般的な問題です。

例外処理

エラー処理の A2A 仕様: https://a2a-protocol.org/latest/specification/#81-standard-json-rpc-errors

A2A サーバーは、HTTP 200 ステータスコードを持つ標準の JSON-RPC エラーレスポンスとしてエラーを返します。内部ランタイムエラーは、プロトコルコンプライアンスを維持するために JSON-RPC 内部エラーに自動的に変換されます。

このサービスは、標準化された JSON-RPC エラーコードで適切な A2A-compliantのエラーレスポンスを提供するようになりました。