🔍 单个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"))  # 指定翻译器

📋 关键要点

  1. A2ACardResolver每次只连接一个Agent
  2. 多个Agent = 多次调用A2ACardResolver
  3. 可以自动选择,也可以手动指定
  4. 先用get_agent_card()了解Agent能力,再决定是否使用

简单说就是:一个Resolver对应一个Agent,多个Agent就用多个Resolver!

Logo

码道开发者社区,聚焦华为云码道 CodeArts 代码智能体,沉淀 Agent、Skill、鸿蒙开发实战内容,供开发者查阅资料、交流技术、分享工程实践

更多推荐