# 在运行时部署 A2A 服务器 AgentCore
Amazon Bedrock AgentCore AgentCore Runtime 允许您在运行时中部署和运行 Agent-to-Agent (A2A) 服务器。 AgentCore 本指南将引导您创建、测试和部署您的第一台 A2A 服务器。
在本部分中，您将学习：
+ Amazon Bedrock 如何 AgentCore 支持 A2A
+ 如何创建具有代理功能的 A2A 服务器
+ 如何在本地测试服务器
+ 如何将服务器部署到 AWS
+ 如何调用已部署的服务器
+ 如何检索特工卡以供发现
有关 A2A 的更多信息，请参阅 [A2](runtime-a2a-protocol-contract.md) A 协议合约。
**Topics**
+ [Amazon Bedrock 如何 AgentCore 支持 A2A](#runtime-a2a-how-agentcore-supports)
+ [将 A2A 与运行时一起使用 AgentCore](#runtime-a2a-steps)
+ [附录](#runtime-a2a-appendix)
## Amazon Bedrock 如何 AgentCore 支持 A2A
Amazon Bedrock AgentCore 的 A2A 协议支持通过充当透明代理层来实现与 A2A 服务器的无缝集成。在 A2A 配置时，Amazon Bedrock AgentCore 希望容器`9000`在根路径 (`0.0.0.0:9000/`) 的端口上运行无状态、可流式传输的 HTTP 服务器，这与默认 A2A 服务器配置一致。
该服务提供企业级会话隔离，同时保持协议透明度——来自 [InvokeAgentRuntime](https://docs.aws.amazon.com/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 服务器安装在`/`（与 `/invocations` HTTP 相比，`/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 服务器部署到 Bedrock Runtime 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 协议和代理间通信概念
### 第 1 步：创建你的 A2A 项目
此示例使用 Strands Agents，但是 AgentCore CLI 也支持 A2A 项目 LangChain/LangGraph 和 Google ADK。
#### 为项目搭建脚手架
运行以下命令并在出现提示时选择 *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))
```
#### 了解代码
**Strands 特工**
使用特定工具和功能创建代理
**Strandsa2aExecutor**
封装 Strands 代理以提供 A2A 协议兼容性
**serve\_a2a**
启动 A Bedrock-compatible 2A 服务器的 Amazon Bedrock AgentCore SDK 帮助程序。它处理运行`/ping`状况端点、代理卡服务、`AGENTCORE_RUNTIME_URL`环境变量、Bedrock 标头传播，默认情况下在端口 9000 上运行。
**端口 9000**
A2A 服务器在运行时默认在端口 9000 上运行 AgentCore
要自定义此代理，请用您自己的`add_numbers`工具替换该工具并更新系统提示符。
### 第 2 步：在本地测试您的 A2A 服务器
在本地开发环境中运行并测试 A2A 服务器。
#### 启动你的 A2A 服务器
使用 CLI AgentCore 在本地启动 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 检查器[进行远程测试](https://github.com/a2aproject/a2a-inspector)中所述。
### 第 3 步：将 A2A 服务器部署到 Bedrock Runtime AgentCore
#### 设置 Cognito 用户池进行身份验证
在部署之前，请配置身份验证以安全访问已部署的服务器。有关 Cognito 的详细设置说明，请参阅设置 [Cognito 用户](runtime-mcp.md#runtime-mcp-appendix-a)池进行身份验证。这提供了安全访问已部署服务器所需的 OAuth 令牌。
#### 部署到 AWS
部署您的代理：
```
agentcore deploy
```
此命令将：
1. Package 你的代理代码和依赖关系
1. 将部署项目上传到 Amazon S3
1. 创建 Amazon Bedrock 运行 AgentCore 时
1. 将您的代理部署到 AWS
部署后，您将收到一个代理运行时 ARN，如下所示：
```
arn:aws:bedrock-agentcore:us-west-2:accountId:runtime/my_a2a_server-xyz123
```
### 第 4 步：获取代理卡
代理卡片是 JSON 元数据文档，描述了 A2A 服务器的身份、能力、技能、服务端点和身份验证要求。它们支持在 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 服务器并发送消息以测试其功能。
创建一个新文件`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 文档中的设置 [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 服务器必须在端口 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 服务器以 HTTP 200 状态代码的标准 JSON-RPC 错误响应形式返回错误。内部运行时错误会自动转换为 JSON-RPC 内部错误，以保持协议合规性。
该服务现在通过标准化 A2A-compliant 错误代码提供正确的 JSON-RPC 错误响应：
| JSON-RPC 错误码 | 运行时异常 | HTTP 错误代码 | JSON-RPC 错误消息 |
| --- | --- | --- | --- |
| N/A | `AccessDeniedException` | 403 | N/A |
| -32501 | `ResourceNotFoundException` | 404 | 未找到资源-请求的资源不存在 |
| -32502 | `ValidationException` | 400 | 验证错误-请求数据无效 |
| -32503 | `ThrottlingException` | 429 | 超出速率限制-请求太多 |
| -32503 | `ServiceQuotaExceededException` | 429 | 超出速率限制-请求太多 |
| -32504 | `ResourceConflictException` | 409 | 资源冲突-资源已存在 |
| -32505 | `RuntimeClientError` | 424 | 运行时客户端错误-请查看您的 CloudWatch 日志以获取更多信息。 |