简单代理核心示例
在Amazon Bedrock AgentCore上部署代理和MCP服务器的简单示例。
目录
simple-agent/ - HTTP agent on AgentCore Runtime (tested, working)
simple-a2a-agent/ - A2A protocol agent on AgentCore Runtime (tested, working)
simple-mcp/ - MCP server on AgentCore Gateway (tested, working)| 文件夹 | 协议 | 身份验证 | 它的作用 |
|---|---|---|---|
| 简单代理/ | HTTP | IAM(SigV4)或Cognito(JWT) | 使用BedrockAgentCoreApp的计算器代理 |
| simple-a2a-agent/ | a2a | IAM(SigV4) | 与a2a服务器相同的计算器代理,代理卡发现 |
| simple-mcp/ | mcp | Cognito JWT | Geolocation API(ipwho.is)作为mcp工具公开 |
先决条件
- Python 3.11+
- 配置了访问Amazon Bedrock AgentCore的AWS凭据
uv已安装包管理器
设置
uv sync用法
代理名称必须以字母开头,并且只能包含字母、数字和下划线(没有连字符)。
简单代理/(HTTP代理)
支持两种身份验证模式:IAM(SigV4,默认)和Cognito(JWT承载令牌)。
cd simple-agent
# Deploy with IAM auth (default)
uv run python deploy_agent.py --agent-name my_simple_agent --region us-east-1
# Deploy with Cognito auth (auto-creates User Pool, App Client, test user)
uv run python deploy_agent.py --agent-name my_simple_agent --region us-east-1 --auth cognito
# Invoke with IAM auth
uv run python deploy_agent.py --agent-name my_simple_agent --region us-east-1 --invoke-only --prompt "What is 42 * 17?"
# Invoke with Cognito auth (auto-refreshes bearer token)
uv run python deploy_agent.py --agent-name my_simple_agent --region us-east-1 --auth cognito --invoke-only --prompt "What is 42 * 17?"
# Setup Cognito only (saves config to .cognito_config.json)
uv run python deploy_agent.py --agent-name my_simple_agent --region us-east-1 --setup-cognito
# Check status
uv run python deploy_agent.py --agent-name my_simple_agent --region us-east-1 --status-only
# Refresh Cognito token only (standalone script, saves to .token)
./get_token.sh
# Delete agent and Cognito resources
uv run python deploy_agent.py --agent-name my_simple_agent --region us-east-1 --deletesimple-a2a-agent/(a2a代理)
相同的计算器代理,但使用A2A协议。这使得代理卡发现和代理到代理的通信成为可能。使用单独的 client.py 用于调用(A2A代理需要A2A格式的消息,而不是启动器工具包的invoke方法)。
这 client.py 从自动读取代理ARN .bedrock_agentcore.yaml (在部署后创建),因此不需要硬编码。
cd simple-a2a-agent
# Full deployment (configure with A2A protocol, launch, wait for READY)
uv run python deploy_a2a_agent.py --agent-name my_a2a_agent --region us-east-1
# Invoke the agent via A2A client (reads ARN from .bedrock_agentcore.yaml)
uv run python client.py --prompt "What is 100 / 4 + 25?"
# Get the agent card (pretty printed, saved to agent_card.json)
uv run python client.py --agent-card-only
# Override agent ARN if needed
uv run python client.py --agent-arn --prompt "What is 2 + 2?"
# Check status
uv run python deploy_a2a_agent.py --agent-name my_a2a_agent --region us-east-1 --status-only
# Delete the agent
uv run python deploy_a2a_agent.py --agent-name my_a2a_agent --region us-east-1 --delete获取A2A代理卡
由于这是一个A2A代理,代理卡可以直接从运行时端点访问 /.well-known/agent.json。端点需要SigV4身份验证(纯 curl 将不起作用),因此使用 client.py 它自动处理身份验证:
cd simple-a2a-agent
# Fetch the agent card (handles SigV4 auth, saves to agent_card.json)
uv run python client.py --agent-card-only代理卡URL从代理ARN导出,如下所示:
Runtime URL: https://bedrock-agentcore.{region}.amazonaws.com/runtimes/{url-encoded-arn}/invocations/
Agent card URL: https://bedrock-agentcore.{region}.amazonaws.com/runtimes/{url-encoded-arn}/invocations/.well-known/agent.json要打印已部署代理的完整URL,请执行以下操作:
python3 -c "
import yaml
from urllib.parse import quote
cfg = yaml.safe_load(open('.bedrock_agentcore.yaml'))
agent = cfg['agents'][cfg['default_agent']]
arn = agent['bedrock_agentcore']['agent_arn']
region = agent['aws']['region']
escaped = quote(arn, safe='')
runtime = f'https://bedrock-agentcore.{region}.amazonaws.com/runtimes/{escaped}/invocations/'
print(f'Agent endpoint: {runtime}')
print(f'Agent card URL: {runtime}.well-known/agent.json')
"简单mcp/(网关上的mcp服务器)
将ipwho.is(免费HTTPS地理位置API)部署为具有Cognito JWT授权的AgentCore网关上的MCP服务器。
注意:与代理名称不同,网关名称使用连字符(不是下划线)。
cd simple-mcp
# Full deployment (IAM role, Cognito, Gateway, S3 upload, target, invoke)
uv run python deploy_mcp_server.py --gateway-name geo-mcp --region us-east-1
# Invoke only (requires prior deployment, reads saved deployment info)
uv run python deploy_mcp_server.py --gateway-name geo-mcp --region us-east-1 --invoke-only
# Custom prompt
uv run python deploy_mcp_server.py --gateway-name geo-mcp --region us-east-1 --invoke-only \
--prompt "Where is IP address 1.1.1.1 located?"
# Delete and redeploy from scratch
uv run python deploy_mcp_server.py --gateway-name geo-mcp --region us-east-1 --delete
uv run python deploy_mcp_server.py --gateway-name geo-mcp --region us-east-1
# Delete all resources (gateway, credential provider, Cognito, IAM role)
uv run python deploy_mcp_server.py --gateway-name geo-mcp --region us-east-1 --delete
# Refresh Cognito token only (standalone script)
./get_token.sh生成的文件(全部为gitignored)
部署后或 --invoke-only,生成以下文件 simple-mcp/:
| 文件 | 目的 |
|---|---|
.deployment_info.json | 网关URL、ID、Cognito池/客户端ID。被...使用 --invoke-only 和 get_token.sh. |
.roo.json | 已准备好使用Roo Code MCP配置。将其内容复制到Roo Code的MCP设置中,以连接到网关。 |
.token | 生干邑持有者代币。代币将在约1小时后过期,请重新运行 --invoke-only 或 ./get_token.sh 刷新。 |
get_token.sh | 用于刷新Cognito令牌的独立脚本。更新 .token. |
使用curl测试MCP服务器
MCP协议使用HTTP上的JSON-RPC。在运行这些命令之前,您需要一个有效的令牌。
cd simple-mcp
# Refresh token first (tokens expire after ~1 hour)
./get_token.sh
# Set variables
GW_URL=$(python3 -c "import json;print(json.load(open('.deployment_info.json'))['gateway_url'])")
TOKEN=$(cat .token)
# Initialize MCP session and save session ID
curl -s -X POST "$GW_URL" \
-H "Authorization: Bearer $TOKEN" \
-H "Content-Type: application/json" \
-d '{"jsonrpc":"2.0","id":1,"method":"initialize","params":{"protocolVersion":"2025-03-26","capabilities":{},"clientInfo":{"name":"curl","version":"1.0"}}}' \
-D /tmp/mcp_headers > /dev/null
SESSION_ID=$(grep -i mcp-session-id /tmp/mcp_headers | tr -d '\r' | awk '{print $2}')
# List available tools
curl -s -X POST "$GW_URL" \
-H "Authorization: Bearer $TOKEN" \
-H "Content-Type: application/json" \
-H "Mcp-Session-Id: $SESSION_ID" \
-d '{"jsonrpc":"2.0","id":2,"method":"tools/list","params":{}}' | python3 -m json.tool
# Call a tool (geolocation lookup for 8.8.8.8)
curl -s -X POST "$GW_URL" \
-H "Authorization: Bearer $TOKEN" \
-H "Content-Type: application/json" \
-H "Mcp-Session-Id: $SESSION_ID" \
-d '{"jsonrpc":"2.0","id":3,"method":"tools/call","params":{"name":"geolocation___getGeolocationByIp","arguments":{"ipAddress":"8.8.8.8"}}}' | python3 -m json.tool建筑
HTTP代理(简单代理/)
User -> deploy_agent.py -> AgentCore Runtime -> BedrockAgentCoreApp -> Strands Agent + CalculatorA2A代理(简单-A2A-Agent/)
client.py (SigV4) -> AgentCore Runtime -> A2AServer (FastAPI:9000) -> Strands Agent + Calculator
|
/.well-known/agent-card.jsonMCP服务器(简单MCP/)
User -> deploy_mcp_server.py -> Cognito (JWT) -> AgentCore Gateway (MCP) -> ipwho.is