本文属于机器翻译版本。若本译文内容与英语原文存在差异，则一律以英文原文为准。
# 在 AgentCore 运行时部署 A2A 服务器
Amazon Bedrock AgentCore Runtime 允许您在运行时部署和运行 Agent-to-Agent (A2A) 服务器。 AgentCore 本指南引导您创建、测试和部署您的第一台 A2A 服务器。
在本部分中，您将学习：
+ 亚马逊 Bedrock 如何 AgentCore 支持 A2A
+ 如何创建具有代理功能的 A2A 服务器
+ 如何在本地测试服务器
+ 如何将服务器部署到 Amazon
+ 如何调用已部署的服务器
+ 如何检索代理卡以供发现
有关 A2A 的更多信息，请参阅 [ A2A 协议合约。](runtime-a2a-protocol-contract.md)
**Topics**
+ [亚马逊 Bedrock 如何 AgentCore 支持 A2A](#runtime-a2a-how-agentcore-supports)
+ [在运行时中使用 A2A AgentCore](#runtime-a2a-steps)
+ [附录](#runtime-a2a-appendix)
## 亚马逊 Bedrock 如何 AgentCore 支持 A2A
亚马逊 Bedrock AgentCore 的 A2A 协议支持通过充当透明代理层实现与 A2A 服务器的无缝集成。配置为 A2A 时，Amazon Bedrock AgentCore 期望容器`9000`在根路径 (`0.0.0.0:9000/`) 的端口上运行无状态、可流式传输的 HTTP 服务器，这与默认 A2A 服务器配置一致。
该服务提供企业级会话隔离，同时保持协议透明度——来自 [ InvokeAgentRuntime ](https://docs.amazonaws.cn/bedrock-agentcore/latest/APIReference/API_InvokeAgentRuntime.html) 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 代理的理想部署平台。
与其他协议的主要区别：
**端口**：
A2A 服务器在端口 9000 上运行（而 HTTP 为 8080，MCP 为 8000）
**路径**
A2A 服务器安装在`/`（与 HTTP 相比，`/invocations`对`/mcp`于 MCP）
**代理卡 **
A2A 通过代理卡提供内置代理发现功能，网址为 `/.well-known/agent-card.json`
**协议**
代理与代理 JSON-RPC 之间通信的用途
**身份验证**
支持 SigV4 和 OAuth 2.0 身份验证方案
有关更多信息，请参阅 [https://a2a-protocol.org/](https://a2a-protocol.org/)。
## 在运行时中使用 A2A AgentCore
在本教程中，您将创建、测试和部署 A2A 服务器。
**Topics**
+ [先决条件](#runtime-a2a-prerequisites)
+ [第 1 步：创建您的 A2A 项目](#runtime-a2a-create-server)
+ [第 2 步：在本地测试您的 A2A 服务器](#runtime-a2a-test-locally)
+ [第 3 步：将 A2A 服务器部署到 Bedro AgentCore ck Runtime](#runtime-a2a-deploy)
+ [第 4 步：获取代理卡](#runtime-a2a-step-4)
+ [第 5 步：调用已部署的 A2A 服务器](#runtime-a2a-step-5)
### 先决条件
+ 已安装 Python 3.10 或更高版本，对 Python 有基本的了解
+ Node.js 已安装 20 或更高版本（ AgentCore CLI 必需）
+ 安装了 AgentCore CLI：`npm install -g @aws/agentcore`
+ 配置了适当权限和本地凭据的 Amazon 账户
+ 理解 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))
```
#### 理解代码
**Strands 代理人 **
使用特定工具和功能创建代理
**Strandsa2aExecutor **
封装 Strands 代理以提供 A2A 协议兼容性
**serve\_a2a **
启动 Bedrock-compatible A2A 服务器的亚马逊 Bedrock AgentCore SDK 助手。它处理运行`/ping`状况端点、代理卡服务、`AGENTCORE_RUNTIME_URL`环境变量、基岩标头传播，并默认在端口 9000 上运行。
**端口 9000 **
在 Runtime 中，A2A 服务器默认在端口 9000 上运行 AgentCore
要自定义此代理，请使用您自己的`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 检查器进行[远程测试中所述。](https://github.com/a2aproject/a2a-inspector)
### 第 3 步：将 A2A 服务器部署到 Bedro AgentCore ck Runtime
#### 设置 Cognito 用户池进行身份验证
在部署之前，配置身份验证以安全访问已部署的服务器。有关 Cognito 的详细设置说明，请参阅[设置 Cognito 用户池以进行身份验证。](runtime-mcp.md#runtime-mcp-appendix-a)这提供了安全访问已部署服务器所需的 OAuth 令牌。
#### 部署到 Amazon
部署您的代理：
```
agentcore deploy
```
此命令将：
1. 打包您的代理代码和依赖关系
1. 将部署项目上传到 Amazon S3
1. 创建亚马逊 Bedrock 运行 AgentCore 时
1. 将您的代理部署到 Amazon
部署后，您将收到一个代理运行时 ARN，如下所示：
```
arn:aws:bedrock-agentcore:us-west-2:accountId:runtime/my_a2a_server-xyz123
```
### 第 4 步：获取代理卡
代理卡是 JSON 元数据文档，用于描述 A2A 服务器的身份、功能、技能、服务端点和身份验证要求。它们支持在 A2A 生态系统中自动发现代理。
#### 设置环境变量
设置环境变量
1. 将持有者令牌作为环境变量导出。有关不记名令牌设置，请参阅不[记名令牌设置。](https://docs.amazonaws.cn/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 服务器并发送消息来测试功能。
创建一个新文件`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"))
```
## 附录
**Topics**
+ [设置 Cognito 用户池进行身份验证](#runtime-a2a-setup-cognito-appendix)
+ [使用 A2A 检查器进行远程测试](#runtime-a2a-remote-testing)
+ [问题排查](#runtime-a2a-troubleshooting)
### 设置 Cognito 用户池进行身份验证
有关详细的 Cognito 设置说明，请参阅 MCP 文档[中的设置 ](https://docs.amazonaws.cn/bedrock-agentcore/latest/devguide/runtime-mcp.html#set-up-cognito-user-pool-for-authentication) Cognito 用户池以进行身份验证。
### 使用 A2A 检查器进行远程测试
请参阅[https://github.com/a2aproject/a2a-inspector](https://github.com/a2aproject/a2a-inspector)。
### 问题排查
**常见 A2A-specific 问题 **
以下是您可能会遇到的常见问题：
港口冲突
在运行环境中，A2A 服务器必须在端口 9000 上运行 AgentCore
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 服务器将大多数错误作为标准 JSON-RPC 错误响应返回。该服务以原生 HTTP 错误的形式返回身份验证和授权失败（例如`AccessDeniedException`），并使用自己的状态码，如下表所示。该服务自动将内部运行时错误转换为内部错误，以 JSON-RPC 保持协议合规性。
该服务使用标准化 A2A-compliant 错误代码提供 JSON-RPC 错误响应：
| JSON-RPC 错误码 | 运行时异常 | HTTP 错误代码 | JSON-RPC 错误消息 |
| --- | --- | --- | --- |
| 不适用 | `AccessDeniedException` | 403 | 访问被拒绝（以标准 HTTP 错误的形式返回，而不是 JSON-RPC 错误） |
| -32051 | `ResourceNotFoundException` | 404 | 未找到资源-请求的资源不存在 |
| -32052 | `ValidationException` | 400 | 验证错误-请求数据无效 |
| -32053 | `ThrottlingException` | 429 | 超出速率限制 — 请求过多 |
| -32053 | `ServiceQuotaExceededException` | 429 | 超出速率限制 — 请求过多 |
| -32054 | `ConflictException` | 409 | 资源冲突-资源已经存在 |
| -32054 | `RetryableConflictException` | 409 | 会话操作正在进行中，请重试 |
| -32055 | `RuntimeClientError` | 424 | 运行时客户端错误-查看您的 CloudWatch 日志以获取更多信息。 |
| -32603 | `Any other exception` | 500 | 内部错误-处理请求时出现意外错误 |