使用Bedrock模型
我们把上一节基于规则引擎的Agent升级为真正由 LLM 驱动的 Agent：
业务逻辑换成调用 Amazon Bedrock 的 converse API。我们会看到A2A 接口完全不变，变的只是 Executor 内部。这正是协议的价值：对外是标准 A2A，内部用什么模型、什么框架完全自由（不透明执行）。
架构
┌──────────────┐ A2A (JSON-RPC/SSE) ┌────────────────────────────┐
│ bedrock_ │ ────────────────────>│ bedrock_server.py │
│ client.py │ <════════════════════│ ┌──────────────────────┐ │
└──────────────┘ 状态/产物流式返回 │ │ BedrockAgentExecutor │ │
│ └──────────┬───────────┘ │
└─────────────┼──────────────┘
boto3 │ converse API
▼
┌────────────────────────────┐
│ Amazon Bedrock │
│ (amazon.nova-lite-v1:0) │
└────────────────────────────┘
环境准备
./venv/bin/pip install boto3 # 在 08 的 venv 基础上加装
# AWS 凭证走默认凭证链：env / ~/.aws/credentials / IAM Role 均可
export BEDROCK_REGION=us-east-1 # 可选，默认 us-east-1
export BEDROCK_MODEL_ID=amazon.nova-lite-v1:0 # 可选，可换 Claude 等
先用一行代码确认 Bedrock 可用（本环境实测返回 “Hi there, friend."）：
import boto3
c = boto3.client("bedrock-runtime", region_name="us-east-1")
r = c.converse(modelId="amazon.nova-lite-v1:0",
messages=[{"role": "user", "content": [{"text": "Say hi in 3 words"}]}],
inferenceConfig={"maxTokens": 30})
print(r["output"]["message"]["content"][0]["text"])
Server 代码
创建一个文件，名为bedrock_server.py：
"""实战二：Bedrock 驱动的 A2A Agent。
Agent 的"大脑"换成了 Amazon Bedrock（converse API），对外接口仍是标准 A2A。
任意具有 bedrock:InvokeModel 权限的 AWS 凭证（默认凭证链）都能运行。
运行: ./venv/bin/python bedrock_server.py (监听 http://localhost:10000)
环境变量: BEDROCK_REGION (默认 us-east-1)、BEDROCK_MODEL_ID (默认 amazon.nova-lite-v1:0)
"""
import asyncio
import os
import boto3
import uvicorn
from starlette.applications import Starlette
from a2a.server.agent_execution import AgentExecutor, RequestContext
from a2a.server.events import EventQueue
from a2a.server.request_handlers import DefaultRequestHandler
from a2a.server.routes import create_agent_card_routes, create_jsonrpc_routes
from a2a.server.tasks import InMemoryTaskStore, TaskUpdater
from a2a.types import (
AgentCapabilities,
AgentCard,
AgentInterface,
AgentSkill,
Part,
Task,
TaskState,
TaskStatus,
)
REGION = os.environ.get("BEDROCK_REGION", "us-east-1")
MODEL_ID = os.environ.get("BEDROCK_MODEL_ID", "amazon.nova-lite-v1:0")
SYSTEM_PROMPT = "你是一个乐于助人的中文助手，回答要简洁，不超过 3 句话。"
def call_bedrock(user_text: str) -> str:
"""调用 Bedrock converse API（同步 boto3，在线程池中执行）。"""
client = boto3.client("bedrock-runtime", region_name=REGION)
resp = client.converse(
modelId=MODEL_ID,
system=[{"text": SYSTEM_PROMPT}],
messages=[{"role": "user", "content": [{"text": user_text}]}],
inferenceConfig={"maxTokens": 300, "temperature": 0.5},
)
return resp["output"]["message"]["content"][0]["text"]
class BedrockAgentExecutor(AgentExecutor):
async def execute(
self, context: RequestContext, event_queue: EventQueue
) -> None:
if context.current_task is None:
await event_queue.enqueue_event(
Task(
id=context.task_id,
context_id=context.context_id,
status=TaskStatus(state=TaskState.TASK_STATE_SUBMITTED),
history=[context.message] if context.message else [],
)
)
updater = TaskUpdater(event_queue, context.task_id, context.context_id)
await updater.start_work()
user_text = context.get_user_input()
try:
# boto3 是同步库，放到线程池避免阻塞事件循环
answer = await asyncio.to_thread(call_bedrock, user_text)
await updater.add_artifact([Part(text=answer)], name="llm-answer")
await updater.complete()
except Exception as e: # 凭证/权限问题时任务进入 failed 终态
await updater.failed(
updater.new_agent_message([Part(text=f"Bedrock 调用失败: {e}")])
)
async def cancel(
self, context: RequestContext, event_queue: EventQueue
) -> None:
updater = TaskUpdater(event_queue, context.task_id, context.context_id)
await updater.cancel()
def build_agent_card() -> AgentCard:
return AgentCard(
name="Bedrock Chat Agent",
description=f"由 Amazon Bedrock（{MODEL_ID}）驱动的问答 Agent。",
version="1.0.0",
supported_interfaces=[
AgentInterface(
url="http://localhost:10000/",
protocol_binding="JSONRPC",
protocol_version="1.0",
)
],
capabilities=AgentCapabilities(streaming=True),
default_input_modes=["text/plain"],
default_output_modes=["text/plain"],
skills=[
AgentSkill(
id="chat",
name="智能问答",
description="基于 Bedrock 大模型回答任意问题",
tags=["llm", "chat", "bedrock"],
examples=["用一句话解释什么是 A2A 协议"],
)
],
)
def main() -> None:
card = build_agent_card()
handler = DefaultRequestHandler(
agent_executor=BedrockAgentExecutor(),
task_store=InMemoryTaskStore(),
agent_card=card,
)
app = Starlette(
routes=[
*create_agent_card_routes(card),
*create_jsonrpc_routes(handler, rpc_url="/"),
]
)
uvicorn.run(app, host="0.0.0.0", port=10000, log_level="warning")
if __name__ == "__main__":
main()
与上一节相比只有三处变化：
handle 逻辑→call_bedrock()：用 boto3 调 converse API；asyncio.to_thread：boto3 是同步库，必须丢进线程池，否则会阻塞 asyncio 事件循环、卡死其他请求；- 异常 →
updater.failed()：凭证/权限错误时任务进入failed终态，并附带错误说明消息——这是 A2A 标准的错误表达方式。
Client 代码
创建一个py文件，命名为bedrock_client.py：
"""实战二：调用 Bedrock A2A Agent 的客户端。
运行: ./venv/bin/python bedrock_client.py
"""
import asyncio
import uuid
import httpx
from a2a.client import A2ACardResolver, create_client
from a2a.types import Message, Part, Role, SendMessageRequest, TaskState
BASE_URL = "http://localhost:10000"
QUESTION = "用一句话解释：什么是 A2A（Agent2Agent）协议？"
async def main() -> None:
async with httpx.AsyncClient() as http:
card = await A2ACardResolver(http, BASE_URL).get_agent_card()
print(f"[发现] Agent: {card.name} — {card.description}")
client = await create_client(card)
request = SendMessageRequest(
message=Message(
message_id=str(uuid.uuid4()),
role=Role.ROLE_USER,
parts=[Part(text=QUESTION)],
)
)
print(f"\n[提问] {QUESTION}")
async for event in client.send_message(request):
if event.HasField("status_update"):
print(f"[状态] {TaskState.Name(event.status_update.status.state)}")
elif event.HasField("artifact_update"):
text = "".join(
p.text for p in event.artifact_update.artifact.parts
if p.HasField("text")
)
print(f"[回答] {text}")
await client.close()
if __name__ == "__main__":
asyncio.run(main())
运行结果
./venv/bin/python bedrock_server.py # 终端 1
./venv/bin/python bedrock_client.py # 终端 2
