SmartBizTrip/a2a_server/weather_server.py

# 【启动 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)