在 AgentCore 執行期中部署 A2A 伺服器

Amazon Bedrock AgentCore執行期可讓您在 AgentCore 執行期中部署和執行 Agent-to-Agent (A2A) 伺服器。本指南會逐步引導您建立、測試和部署第一個 A2A 伺服器。

在本區段，您會學習：

如需 A2A 的詳細資訊，請參閱 A2A 通訊協定合約。

Amazon Bedrock AgentCore 如何支援 A2A

Amazon Bedrock AgentCore 的 A2A 通訊協定支援可做為透明代理層，與 A2A 伺服器無縫整合。為 A2A 設定時，Amazon Bedrock AgentCore 預期容器在根路徑 (0.0.0.0:9000/) 9000的連接埠上執行無狀態、可串流的 HTTP 伺服器，這符合預設的 A2A 伺服器組態。

此服務提供企業級工作階段隔離，同時維持通訊協定透明度 - 來自 InvokeAgentRuntime API 的 JSON-RPC 承載會直接傳遞至 A2A 容器，無需修改。此架構會保留標準 A2A 通訊協定功能，例如透過 的代理程式卡/.well-known/agent-card.json和 JSON-RPC 通訊的內建代理程式探索，同時新增企業身分驗證 (SigV4/OAuth 2.0) 和可擴展性。

與其他通訊協定的主要區別在於連接埠 (9000 vs 8080 for HTTP)、掛載路徑 ( / vs /invocations ) 和標準化代理程式探索機制，讓 Amazon Bedrock AgentCore 成為生產環境中 A2A 代理程式的理想部署平台。

與其他通訊協定的主要差異：

如需詳細資訊，請參閱https://a2a-protocol.org/。

搭配 AgentCore 執行期使用 A2A

在本教學課程中，您會建立、測試和部署 A2A 伺服器。

先決條件

步驟 1：建立您的 A2A 專案

此範例使用 Strands Agents，但 AgentCore CLI 也支援具有 LangChain/LangGraph 和 Google ADK 的 A2A 專案。

Scaffold 專案

執行下列命令，並在出現提示時選取字串做為架構：

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

這會在 Web 瀏覽器中開啟 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 檢查器的遠端測試中所述。

步驟 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 伺服器，並傳送訊息以測試功能。

建立新的檔案my_a2a_client_remote.py以叫用您部署的 A2A 伺服器：

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錯誤回應：