# 【启动 S3】本文件是独立的 A2A 子助手进程：监听 5005，调用 MCP 工具服务 8002。 # 启动时先执行模块顶层配置、模型对象和代理卡片定义，再进入 __main__ 创建服务器。 # 类/函数定义本身不会执行查询；run_server 开始监听后，有 Task 到达才进入 handle_task。 # 主流程 7：接收任务 → 加载工具 → 模型选择工具和参数 → 执行器调用工具（主流程 8）。 # 主流程 9：工具结果交回模型生成回复 → 封装 Task 状态及 artifacts → 返回主助手。 """ 需求：实现基于A2A的天气查询服务器，处理用户的天气查询请求并返回结果 架构说明： 本服务器是 SmartBizTrip 系统中的一个子代理（Sub-Agent），负责处理天气查询任务。 它运行在独立的进程中（localhost:5005），通过 A2A（Agent2Agent）协议与主助手通信。 工作流程（与票务 Agent 保持一致的架构模式）： 1. 主助手通过 A2A 协议向本服务器发送任务（Task） 2. 本服务器收到任务后，提取用户的自然语言查询 3. 使用 LangChain Agent（基于工具调用的 Agent）处理查询： a. LLM 分析用户输入，从自然语言中提取参数（城市、日期等） b. 调用 MCP Server 的参数化工具（query_weather） c. MCP Server 内部根据参数拼接 SQL 并执行查询 d. 工具返回结果后，LLM 将结果格式化为友好的中文回复 4. 将结果返回给主助手 与旧模式（Text-to-SQL）的区别： - 旧模式：LLM 直接生成 SQL 语句 → MCP 执行原始 SQL 问题：SQL 注入风险、LLM 生成的 SQL 可能带有代码块标记、容易出错 - 新模式：LLM 从自然语言中提取参数（city、date）→ MCP 内部拼 SQL 优势：参数化查询（安全）、SQL 由 MCP 统一管理（可控）、容错性好 涉及的关键技术： - LangChain Agent: 让 LLM 自主选择和使用工具的框架 - Tool Calling Agent: LLM 以结构化格式调用工具 - MCP Tools: MCP Server 提供的参数化工具（city、start_date、end_date） - AgentExecutor: 负责运行 Agent 循环（思考→调用工具→处理结果→继续） """ # ==================== 导入依赖 ==================== import asyncio # 异步 IO 库，用于在同步方法中调用异步 MCP 客户端 from python_a2a import A2AServer, run_server, AgentCard, AgentSkill, TaskStatus, TaskState # A2AServer: A2A 服务器基类，我们需要继承它来实现自己的天气查询服务器 # run_server: 启动 A2A 服务器的函数 # AgentCard: 代理卡片，描述本代理的能力、技能等信息 # AgentSkill: 代理技能，描述本代理能做什么 # TaskStatus: 任务状态，标记任务是完成、失败还是需要输入 # TaskState: 任务状态枚举（COMPLETED / FAILED / INPUT_REQUIRED） # to_langchain_tool: 从 MCP 服务器 URL 加载工具为 LangChain 工具 from langchain_openai import ChatOpenAI # LangChain 的大模型接口 from langchain_core.prompts import ChatPromptTemplate # LangChain 的提示模板 from langchain_core.tools import StructuredTool from python_a2a import to_langchain_tool # 从 MCP URL 加载工具 from langchain.agents import create_tool_calling_agent, AgentExecutor # create_tool_calling_agent: 创建基于工具调用的 Agent # AgentExecutor: Agent 执行器，负责运行 Agent 循环 from config import Config # 项目配置（模型地址、API Key等） from datetime import datetime # 时间处理，用于获取当前日期 import pytz # 时区库，用于转换到 Asia/Shanghai 时区 from create_logger import logger # 日志模块 conf = Config() # 初始化LLM # 【启动 S3.1】创建模型客户端；模型推理在后面的 invoke/ainvoke 才发生。 llm = ChatOpenAI( model=conf.model_name, base_url=conf.base_url, api_key=conf.api_key, temperature=conf.temperature ) # MCP 服务器运行在 8002 端口 MCP_URL = "http://127.0.0.1:8002" # ==================== Agent Card（代理卡片） ==================== # Agent Card 是 A2A 协议中描述一个代理能力的元数据 # 它告诉主助手：我是谁、我能做什么、我运行在哪里、我有哪些技能 # 主助手用卡片展示能力；当前业务路由依据 Config.intent 中的意图映射。 # 【启动 S3.2】定义对外说明。skills 是能力介绍，实际可执行工具来自 MCP。 # 主助手当前按 Config.intent 路由；卡片声明的能力也不等于此方法已实现逐片输出。 # 卡片字段说明：name 是代理标识；url 是访问地址；description/skills 向调用方介绍能力。 # 这里描述技能不会自动生成对应 Python 函数，实际工具仍由 MCP 服务注册和提供。 # 主助手 Config.intent 中的名称、AgentNetwork 的注册名称和这里的 name 要相互对应。 agent_card = AgentCard( name="WeatherQueryAssistant", # 代理名称 description="基于LangChain提供天气查询服务的助手", # 代理描述 url="http://localhost:5005", # 代理的访问地址 version="2.0.0", # 版本号（升级为2.0.0，表示架构变更） capabilities={"streaming": True, "memory": True}, # 支持的能力：流式输出、记忆 skills=[ # 技能列表：描述本代理具体能做什么 AgentSkill( name="execute weather query", # 技能名称 description="执行天气查询，返回天气数据库结果，支持自然语言输入，支持单天和范围查询", # 技能描述 examples=["北京 2025-07-30 天气", "上海未来5天", "今天天气如何"] # 使用示例 ) ] ) # ==================== 天气查询函数 ==================== async def query_weather(conversation: str) -> dict: """ 通过 LangChain Agent + MCP Tools 执行天气查询 与票务 Agent（ticket_server.py）使用完全相同的架构模式： 1. 连接 MCP Server 加载所有可用的工具 2. 创建一个 LangChain Agent，让它自主选择调用哪个工具 3. AgentExecutor 负责执行 Agent 循环： - LLM 分析用户输入，从自然语言中提取参数（城市、日期等） - 调用 MCP Server 的参数化工具（如 query_weather(city="北京", start_date="2025-07-30")） - MCP Server 内部根据参数拼接 SQL 并执行 - 工具返回结果后，LLM 将结果格式化为友好的中文回复 与旧模式（Text-to-SQL）的对比： 旧模式流程： 用户输入 "北京明天天气" → LLM 生成 SQL → MCP 执行原始 SQL 新模式流程： 用户输入 "北京明天天气" → LLM 提取 city="北京", date="明天" → MCP 工具接收参数 → MCP 内部拼 SQL → 执行 → LLM 格式化回复 参数： conversation (str): 用户的查询内容，例如： "北京明天天气怎么样？" "上海未来3天的天气" "北京2025-07-30的天气" 返回值： dict: 查询结果，格式为： - {"status": "success", "message": "格式化后的查询结果"} # 查询成功 - {"status": "error", "message": "错误信息"} # 查询失败 """ # 调用者：本文件的 handle_task。conversation 是整个任务文本，可能包含历史和当前需求。 # 输出：字典 {"status": "success"/"error", "message": "回复或错误文本"}，由 handle_task 再转成 A2A 状态。 # 这里的 success 表示执行器返回了结果，不等于工具一定查到了数据。 try: # 从 MCP Server 加载所有可用工具 # to_langchain_tool 会自动将 MCP Server 提供的工具转换为 LangChain 工具对象 # 当前只有一个工具：query_weather(city, start_date, end_date)，日期区间含首尾。 # 【主流程 7.2】连接 MCP 读取工具定义；此处尚未执行具体业务查询。 mcp_tools = to_langchain_tool(MCP_URL) # python-a2a 0.5.4 默认生成单输入 Tool，天气查询却需要城市和日期等多个参数。 # StructuredTool 按参数名称传值，例如 city="北京"、start_date="2026-09-26"。 # 【主流程 7.3】把工具参数结构交给 StructuredTool，保留原 MCP 调用函数。 # 模型输出参数名称和值，执行器按照这个结构传参。 tools = [] for mcp_tool in mcp_tools: weather_tool = StructuredTool( # 工具对象的五个字段各有用途：name 用于匹配模型选中的工具，description 帮助模型理解用途； # func 保存真正发起 MCP 调用的函数；args_schema 规定参数名/类型；metadata 保留原工具信息。 # 列表里的每一项转换成独立 StructuredTool，模型随后可在这些工具中选择。 name=mcp_tool.name, # 工具名称 description=mcp_tool.description, # 告诉模型工具能做什么 func=mcp_tool.func, # 保留原来调用 MCP 的函数 args_schema=mcp_tool.metadata['parameters'], # 城市、日期等参数的格式 metadata=mcp_tool.metadata, # 保留工具的附加信息 ) tools.append(weather_tool) # 定义 Agent 的系统 Prompt # 这个 Prompt 告诉 LLM：你是谁、你能做什么、如何提取参数、输出格式要求 # 提示模板分三部分：system 给规则和当前日期，human 放本次输入，placeholder 放执行过程消息。 # {input}/{current_date} 由 invoke/ainvoke 的参数提供；{agent_scratchpad} 由 AgentExecutor 随工具调用更新。 # 占位符不是立刻执行的代码，只是在实际请求模型前被替换成消息内容。 prompt = ChatPromptTemplate.from_messages([ ("system", """你是一个天气查询助手，能够调用工具来查询天气信息。 你需要仔细分析用户的问题，从问题中提取工具需要的参数（城市、日期等），然后调用对应的查询工具。 如果用户提供的信息不足以提取到调用工具的所有必要参数，则向用户追问，以获取该信息。不能自己编撰参数。 注意： - 用户可能使用相对时间，如"明天"、"后天"、"今天"、"未来3天"等，请根据当前日期转换为具体日期（YYYY-MM-DD格式）。 - 如果用户只给了城市没给日期，需要追问日期。 - 如果用户只给了日期没给城市，需要追问城市。 查询到结果后，请用清晰的中文格式化输出天气信息，包括城市、日期、天气状况、温度、湿度、风向、降水量等。 如果未查到数据，请回复"未找到相关天气数据，请确认或修改查询条件。" 当前日期是{current_date}。"""), ("human", "{input}"), # 用户输入会替换到这里 ("placeholder", "{agent_scratchpad}"), # Agent 思考过程的占位符 ]) # 创建基于工具调用的 Agent # create_tool_calling_agent 会创建一个能自主选择和使用工具的 Agent # LLM 以结构化格式（函数调用）来使用工具，而不是用自然语言描述 # 【主流程 7.4】组装 Agent，随后创建执行器；此时尚未调用模型处理本次请求。 # agent_scratchpad 承载本轮工具调用与工具返回消息，不是跨轮用户记忆。 agent = create_tool_calling_agent(llm, tools, prompt) # 创建 Agent 执行器 # AgentExecutor 负责运行 Agent 循环： # 1. 将用户输入发给 LLM # 2. LLM 决定调用哪个工具 + 提供参数 # 3. 执行工具调用 # 4. 将工具结果返回给 LLM # 5. LLM 生成最终回复 # agent 负责根据输入和已有工具结果产生下一步动作；executor 负责运行这个过程。 # 若模型请求工具，executor 按工具名找到对象并调用 func，再把返回值交回模型；直到得到最终回答。 # verbose=True 打印执行过程，便于观察调用轨迹，不代表业务结果会自动保存到记忆或数据库。 agent_executor = AgentExecutor(agent=agent, tools=tools, verbose=True) # 获取当前日期，注入到 Prompt 中（用户可能说"明天"、"后天"等相对时间） current_date = datetime.now(pytz.timezone('Asia/Shanghai')).strftime('%Y-%m-%d') # 执行 Agent，传入用户输入和当前日期 # 【主流程 7.5 → 8】执行器开始循环：问模型 → 执行所选工具 → 将结果交回模型。 # await 等待本次循环的最终结果；此处返回一个完整回复，不直接向主助手推送每个片段。 response = await agent_executor.ainvoke({ "input": conversation, # 用户输入的查询内容 "current_date": current_date # 当前日期，用于相对时间转换 }) # AgentExecutor 的 response["output"] 就是 LLM 生成的最终回复 # ainvoke 已经完成整轮工具循环，response["output"] 是模型整理后的文字。 # 将它包装成普通字典返回给 handle_task，后者再决定 COMPLETED、INPUT_REQUIRED 或 FAILED。 return {"status": "success", "message": response["output"]} except Exception as e: # MCP 连接失败、工具调用异常等情况的捕获 logger.error(f"天气 MCP 查询出错：{str(e)}") return {"status": "error", "message": f"天气 MCP 查询出错：{str(e)}"} # ==================== 天气查询服务器类 ==================== class WeatherQueryServer(A2AServer): """ 天气查询 A2A 服务器 这个类继承自 A2AServer，实现了天气查询的完整流程。 与旧模式（Text-to-SQL）不同，现在使用 LangChain Agent + MCP Tools 模式： - LLM 负责从自然语言中提取参数（城市、日期） - MCP Server 内部根据参数拼接 SQL 并执行（参数化查询，更安全） 任务状态说明： - COMPLETED: 任务成功完成，结果放在 task.artifacts 中 - FAILED: 任务失败，错误信息放在 task.status.message 中 - INPUT_REQUIRED: 需要用户补充输入，追问信息放在 task.status.message 中 """ def __init__(self): """ 初始化天气查询服务器 调用父类的 __init__ 注册 Agent Card。 """ # 【启动 S3.3】将卡片交给 A2AServer；父类提供通信框架，本类负责实现 handle_task。 # 输入：模块加载时创建的 agent_card。super() 将当前实例交给父类初始化。 # 本类继承了 A2A 通信框架，只覆盖业务回调 handle_task；此时还没有用户查询需要执行。 super().__init__(agent_card=agent_card) # 调用父类初始化，注册 Agent Card def handle_task(self, task): """ 处理来自 A2A 客户端的任务 —— 本服务器的核心方法 当主助手通过 A2A 协议发送任务过来时，这个方法会被调用。 它负责完成从"接收任务"到"返回结果"的完整流程。 完整流程： 1. 提取输入：从任务消息中获取用户的查询内容 2. 调用 MCP 查询：通过 query_weather 函数执行查询 - query_weather 内部使用 LangChain Agent + MCP Tools - LLM 自动从自然语言中提取参数（城市、日期） - MCP Server 根据参数拼接 SQL 并执行 - LLM 将结果格式化为友好的中文回复 3. 根据查询结果设置任务状态： - 成功：COMPLETED 状态，结果放在 task.artifacts - 需要输入：INPUT_REQUIRED 状态，追问信息放在 task.status.message - 失败：FAILED 状态，错误信息放在 task.status.message 参数： task: A2A 任务对象，包含： - task.message: 客户端发送的消息（包含用户输入） - task.artifacts: 用于存放任务结果（输出） - task.status: 任务状态（完成/失败/需要输入） 返回值： task: 处理后的任务对象（已设置状态和结果） """ # ========== 步骤1：提取输入 ========== # 从任务消息中获取内容（A2A 协议中，消息以字典格式存储） # 【主流程 7.1】A2A 框架收到任务后调用此方法；先从 task.message 取输入。 # 输入 task 来自主助手的 A2A 请求；它包含任务编号、message，处理后还会带上 status/artifacts。 # 本方法不是被 ChatService 直接当本地 Python 方法调用，而是由此进程的 A2A 框架接收请求后触发。 # 输出仍是 task 对象，只是在原对象上填写了结果状态和产物。 # 先取 message 内的 content；下一行确认它是字典，再取 text。 # 消息为空、content 类型不对或没有 text 时，本实现使用空字符串交给后续处理。 content = (task.message or {}).get("content", {}) # 提取对话内容，即客户端发起的任务中的用户输入 conversation = content.get("text", "") if isinstance(content, dict) else "" logger.info(f"对话历史及用户问题: {conversation}") try: # ========== 步骤2：调用 MCP 查询 ========== # query_weather 内部会： # 1. 连接 MCP Server 加载所有工具 # 2. 创建 LangChain Agent # 3. Agent 自动从自然语言中提取参数（城市、日期等） # 4. 调用 MCP 的参数化工具（query_weather(city="北京", start_date="2025-07-30")） # 5. MCP Server 内部拼接 SQL 并执行 # 6. LLM 将结果格式化为友好的中文回复 # 同步 handle_task 用 asyncio.run 驱动异步查询，等查询结束后才设置任务状态。 weather_result = asyncio.run(query_weather(conversation)) logger.info(f"MCP 查询返回: {weather_result}") # ========== 步骤3：根据结果设置任务状态 ========== # 这里按查询函数返回的 status 分支；先确认执行器是否正常返回，再从 message 检查是否为追问。 if weather_result.get("status") == "success": result_text = weather_result.get("message", "") # 检查是否是追问消息（LLM 发现信息不足时会追问） # 【主流程 9.1】用回复关键词判断是否需要补充信息；否则把文本放入 artifacts。 # COMPLETED 表示处理结束，仍需结合结果文本判断有无数据或是否仅为模拟。 # 任意一个关键词出现就视为需要补信息；只是当前示例的字符串规则，不是模型返回的专门状态字段。 if "请提供" in result_text or "请确认" in result_text: # 需要用户补充信息 task.status = TaskStatus( state=TaskState.INPUT_REQUIRED, message={"role": "agent", "content": {"text": result_text}} ) else: # 查询成功，将结果放入任务产物 # 成功文本放到 artifacts 的第一个 part；主助手按同一层级提取文字。 # 其他状态的解释文本放在 status.message，因此调用方需要同时处理这两种返回位置。 task.artifacts = [{"parts": [{"type": "text", "text": result_text}]}] task.status = TaskStatus(state=TaskState.COMPLETED) # 查询函数已捕获到错误时走这个分支；未知 status 走后面的 else，同样标为 FAILED。 elif weather_result.get("status") == "error": # MCP 查询出错：设置为失败状态 task.status = TaskStatus( state=TaskState.FAILED, message={"role": "agent", "content": {"text": weather_result.get("message", "查询失败，请重试。")}} ) else: # 未知的状态码，也视为失败 task.status = TaskStatus( state=TaskState.FAILED, message={"role": "agent", "content": {"text": "查询失败，请重试或提供更多细节。"}} ) # 【主流程 9.2】返回带有状态和结果的 Task，A2A 框架再把它传回主助手。 return task except Exception as e: # 捕获所有异常，确保即使出错也能返回有效的任务状态 logger.error(f"查询失败: {str(e)}") task.status = TaskStatus( state=TaskState.FAILED, message={"role": "agent", "content": {"text": f"查询失败: {str(e)} 请重试或提供更多细节。"}} ) return task if __name__ == "__main__": weather_server = WeatherQueryServer() print("\n=== 服务器信息 ===") print(f"名称: {weather_server.agent_card.name}") print(f"描述: {weather_server.agent_card.description}") print("\n技能:") # 启动时只打印卡片中声明的技能，帮助人工核对名称和配置；不会依次调用这些技能。 for skill in weather_server.agent_card.skills: print(f"- {skill.name}: {skill.description}") # 【启动 S3.4】开始监听 A2A 请求；后续 handle_task 按请求触发，不按函数定义顺序执行。 run_server(weather_server, host="127.0.0.1", port=5005)