A2A学习（一）

# 定义Agent支持的两个技能，涵盖天气预报+空气质量报告
forecast_skill = AgentSkill(
    id='天气预告',
    name='天气预告',
    description='给出某地的天气预告',
    tags=['天气', '预告'],
    examples=['给我广州未来 7 天的天气预告'],
)
air_quality_skill = AgentSkill(
    id='空气质量报告',
    name='空气质量报告',
    description='给出某地当前时间的空气质量报告，不做预告',
    tags=['空气', '质量'],
    examples=['给我广州当前的空气质量报告'],
)

# 定义Agent卡片，告知外部该Agent的作用与技术指标
agent_card = AgentCard(
    name='天气 Agent',
    description='这是一个天气Agent，提供天气相关的查询功能',
    url=f'http://{host}:{port}',
    version='1.0.0',
    defaultInputModes=['text'],
    defaultOutputModes=['text'],
    capabilities=AgentCapabilities(streaming=True), # 声明支持流式返回
    skills=[forecast_skill, air_quality_skill],
)

async def connect_to_streamable_http_server(self, url: str, headers: Optional[dict] = None):
    """连接到高德 HTTP MCP 服务器"""
    # 1. 建立双向流通道
    self._streams_context = streamablehttp_client(url=url, headers=headers)
    read_stream, write_stream, _ = await self._streams_context.__aenter__()

    # 2. 建立 MCP 客户端会话
    self._session_context = ClientSession(read_stream, write_stream)
    self.session: ClientSession = await self._session_context.__aenter__()

    # 3. 初始化会话并动态拉取可用工具
    await self.session.initialize()
    response = await self.session.list_tools()

response = await self.session.list_tools()
available_tools = [
    {
        "type": "function",
        "function": {
            "name": tool.name,
            "description": tool.description,
            "parameters": tool.inputSchema, # 动态注入工具所需的参数描述
        }
    }
    for tool in response.tools
]

# 1. 第一次调用大模型，附带高德 MCP 工具箱
response = self.openai.chat.completions.create(
    model="deepseek-chat",
    messages=self.messages,
    tools=available_tools,
)

response_message = response.choices[0].message
tool_calls = response_message.tool_calls

# 2. 判断模型是否决定调用工具
if tool_calls:
    self.messages.append(response_message.model_dump())

    # 3. 执行工具调用
    for tool_call in tool_calls:
        tool_name = tool_call.function.name
        tool_args = json.loads(tool_call.function.arguments)

        # 4. 远程调用高德 MCP 服务器执行具体工具（如查询高德天气API）
        result = await self.session.call_tool(tool_name, tool_args)
        
        # 5. 将工具执行结果追加到对话历史中
        self.messages.append({
            "tool_call_id": tool_call.id,
            "role": "tool",
            "name": tool_name,
            "content": json.dumps(result.model_dump()),
        })

    # 6. 第二次调用大模型，让它结合高德返回的实时数据，生成最终的自然语言答复
    second_response = self.openai.chat.completions.create(
        model="deepseek-chat",
        messages=self.messages,
    )
    return second_response.choices[0].message.content

# agent_executor.py
class WeatherAgentExecutor(AgentExecutor):

    async def execute(self, context: RequestContext, event_queue: EventQueue) -> None:
        text = context.message.parts[0].root.text  # 用户提问
        agent = ReActAgent()
        
        try:
            # 连接高德 MCP 服务
            await agent.connect_to_streamable_http_server(
                url=f"https://mcp.amap.com/mcp?key={os.getenv('GAODE_API_KEY')}"
            )

            # 调用大模型执行 ReAct 获取响应
            response = await agent.process_query(text)

            # 将结果组装为 A2A 协议规定的事件格式，并推入异步事件队列
            await event_queue.enqueue_event(
                completed_task(
                    context.task_id,
                    context.context_id,
                    [new_artifact(parts=[Part(root=TextPart(text=response))], name="天气查询结果")],
                    [context.message],
                )
            )
        finally:
            await agent.cleanup()