aws strands agents 单个agent和多个agent的调用方式的不同
·
🔍 单个Agent时使用A2ACardResolver
import asyncio
from a2a.client import A2ACardResolver, ClientConfig, ClientFactory
async def ask_question(question):
# 连接服务器
resolver = A2ACardResolver(base_url="http://localhost:9000")
agent_card = await resolver.get_agent_card()
# 发送问题,等结果
client = ClientFactory(ClientConfig(streaming=False)).create(agent_card)
response = await client.send_message(question)
return response
🔍 多个Agent时使用A2AClientToolProvider
💡 核心概念
A2ACardResolver就像一个"电话簿",它帮你找到并了解每个Agent能做什么。
🎯 两种主要场景
场景1:我有多个服务端Agent(提供不同服务)
比如你有:
- Agent A:计算器服务 (localhost:9000)
- Agent B:翻译服务 (localhost:9001)
- Agent C:天气服务 (localhost:9002)
方法1:让客户端Agent自动选择
from strands_tools.a2a_client import A2AClientToolProvider
# 告诉客户端所有可用的服务器地址
provider = A2AClientToolProvider(known_agent_urls=[
"http://localhost:9000", # 计算器
"http://localhost:9001", # 翻译
"http://localhost:9002" # 天气
])
# 客户端Agent会自动选择合适的服务
client_agent = Agent(tools=provider.tools)
client_agent("帮我算 5+3") # 自动选择计算器Agent
client_agent("翻译hello") # 自动选择翻译Agent
client_agent("今天天气怎么样") # 自动选择天气Agent
方法2:手动选择特定Agent
import asyncio
from a2a.client import A2ACardResolver
async def use_specific_agent(agent_url, question):
# 连接特定的Agent
resolver = A2ACardResolver(base_url=agent_url)
agent_card = await resolver.get_agent_card()
print(f"Agent名称: {agent_card.name}")
print(f"Agent描述: {agent_card.description}")
print(f"可用工具: {agent_card.tools}")
# 使用这个Agent
client = ClientFactory(ClientConfig()).create(agent_card)
response = await client.send_message(question)
return response
# 分别使用不同的Agent
asyncio.run(use_specific_agent("http://localhost:9000", "算5+3"))
asyncio.run(use_specific_agent("http://localhost:9001", "翻译hello"))
场景2:我想了解所有Agent的能力
async def discover_all_agents():
agent_urls = [
"http://localhost:9000",
"http://localhost:9001",
"http://localhost:9002"
]
for url in agent_urls:
try:
resolver = A2ACardResolver(base_url=url)
card = await resolver.get_agent_card()
print(f"🤖 发现Agent: {card.name}")
print(f"📝 描述: {card.description}")
print(f"🛠️ 工具: {[tool.name for tool in card.tools]}")
print("---")
except Exception as e:
print(f"❌ {url} 连接失败: {e}")
asyncio.run(discover_all_agents())
🚀 实际应用示例
智能路由客户端:
class SmartA2AClient:
def __init__(self):
self.agents = {
"calculator": "http://localhost:9000",
"translator": "http://localhost:9001",
"weather": "http://localhost:9002"
}
async def ask(self, question, agent_type=None):
if agent_type:
# 指定使用某个Agent
url = self.agents[agent_type]
else:
# 自动选择(这里简化了,实际可以用AI判断)
if any(word in question for word in ["算", "计算", "+"]):
url = self.agents["calculator"]
elif any(word in question for word in ["翻译", "translate"]):
url = self.agents["translator"]
else:
url = self.agents["weather"]
resolver = A2ACardResolver(base_url=url)
card = await resolver.get_agent_card()
client = ClientFactory(ClientConfig()).create(card)
return await client.send_message(question)
# 使用
client = SmartA2AClient()
asyncio.run(client.ask("5+3等于多少")) # 自动选择计算器
asyncio.run(client.ask("翻译hello", "translator")) # 指定翻译器
📋 关键要点
- A2ACardResolver每次只连接一个Agent
- 多个Agent = 多次调用A2ACardResolver
- 可以自动选择,也可以手动指定
- 先用get_agent_card()了解Agent能力,再决定是否使用
简单说就是:一个Resolver对应一个Agent,多个Agent就用多个Resolver!
更多推荐



所有评论(0)