Amazon Web Services 文档中描述的 Amazon Web Services 服务或功能可能因区域而异。要查看适用于中国区域的差异，请参阅 中国的 Amazon Web Services 服务入门 (PDF)。

本文属于机器翻译版本。若本译文内容与英语原文存在差异，则一律以英文原文为准。

在 AgentCore 运行时部署 A2A 服务器

Amazon Bedrock AgentCore Runtime 允许您在运行时部署和运行 Agent-to-Agent (A2A) 服务器。 AgentCore 本指南引导您创建、测试和部署您的第一台 A2A 服务器。

在本部分中，您将学习：

有关 A2A 的更多信息，请参阅 A2A 协议合约。

亚马逊 Bedrock 如何 AgentCore 支持 A2A

亚马逊 Bedrock AgentCore 的 A2A 协议支持通过充当透明代理层实现与 A2A 服务器的无缝集成。配置为 A2A 时，Amazon Bedrock AgentCore 期望容器9000在根路径 (0.0.0.0:9000/) 的端口上运行无状态、可流式传输的 HTTP 服务器，这与默认 A2A 服务器配置一致。

该服务提供企业级会话隔离，同时保持协议透明度——来自 InvokeAgentRuntime API的 JSON-RPC 有效载荷无需修改即可直接传递到A2A容器。该架构保留了标准的 A2A 协议功能，例如通过代理卡进行内置代理发现/.well-known/agent-card.json和 JSON-RPC 通信，同时增加了企业身份验证 (SigV4/OAuth 2.0) 和可扩展性。

与其他协议的主要区别在于端口（HTTP 为 9000 与 8080）、挂载路径（/vs/invocations）和标准化代理发现机制，这使得 Amazon Bedrock AgentCore 成为生产环境中 A2A 代理的理想部署平台。

与其他协议的主要区别：

有关更多信息，请参阅 https://a2a-protocol.org/。

在运行时中使用 A2A AgentCore

在本教程中，您将创建、测试和部署 A2A 服务器。

先决条件

第 1 步：创建您的 A2A 项目

此示例使用了 Strands 代理，但是 AgentCore CLI 也支持带有 LangChain/LangGraph Google ADK 的 A2A 项目。

为项目搭建脚手架

运行以下命令：

agentcore create \
  --project-name A2AProject \
  --name A2AAgent \
  --language Python \
  --framework Strands \
  --model-provider Bedrock \
  --memory none \
  --protocol A2A
cd A2AProject

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 检查器测试已部署的服务器，如使用 A2A 检查器进行远程测试中所述。

第 3 步：将 A2A 服务器部署到 Bedro AgentCore ck Runtime

设置 Cognito 用户池进行身份验证

在部署之前，配置身份验证以安全访问已部署的服务器。有关 Cognito 的详细设置说明，请参阅设置 Cognito 用户池以进行身份验证。这提供了安全访问已部署服务器所需的 OAuth 令牌。

部署到 Amazon

部署您的代理：

agentcore deploy

此命令将：

部署后，您将收到一个代理运行时 ARN，如下所示：

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

第 4 步：获取代理卡

代理卡是 JSON 元数据文档，用于描述 A2A 服务器的身份、功能、技能、服务端点和身份验证要求。它们支持在 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 服务器将大多数错误作为标准 JSON-RPC 错误响应返回。该服务以原生 HTTP 错误的形式返回身份验证和授权失败（例如AccessDeniedException），并使用自己的状态码，如下表所示。该服务自动将内部运行时错误转换为内部错误，以 JSON-RPC 保持协议合规性。

该服务使用标准化 A2A-compliant 错误代码提供 JSON-RPC 错误响应：