SmartBizTrip/a2a_server/trip_server.py

# 【启动 S3】本文件是独立的 A2A 子助手进程：监听 5007，调用 MCP 工具服务 8003。
# 启动时先执行模块顶层配置、模型对象和代理卡片定义，再进入 __main__ 创建服务器。
# 类/函数定义本身不会执行查询；run_server 开始监听后，有 Task 到达才进入 handle_task。
# 主流程 7：接收任务 → 加载工具 → 模型选择工具和参数 → 执行器调用工具（主流程 8）。
# 主流程 9：工具结果交回模型生成回复 → 封装 Task 状态及 artifacts → 返回主助手。

"""
需求：实现基于A2A的差旅管家服务器，处理用户的商务用车、会务套餐、保险查询及预订请求

架构说明：
    本服务器是 SmartBizTrip 系统中的又一个子代理（Sub-Agent），负责处理所有差旅相关任务。
    它运行在独立的进程中（localhost:5007），通过 A2A（Agent2Agent）协议与主助手通信。

    与 ticket_server.py 和 weather_server.py 保持一致的架构模式：
    - 使用 LangChain Agent + MCP Tools 模式
    - LLM 从自然语言中提取参数（城市、日期、类型等）
    - 调用 MCP Server 的参数化工具（端口 8003）
    - MCP Server 返回结果后，LLM 将结果格式化为友好的中文回复

    工作流程：
    1. 主助手通过 A2A 协议向本服务器发送任务（Task）
    2. 本服务器收到任务后，提取用户的自然语言查询
    3. 使用 LangChain Agent 处理查询：
       a. LLM 分析用户输入，决定调用哪个 MCP 工具
       b. LangChain 自动调用 MCP Server 的工具（端口 8003）
       c. 工具返回结果后，LLM 将结果格式化为友好的中文回复
    4. 将结果返回给主助手

    MCP Server（端口 8003）提供的工具包括：
    - 商务用车查询工具、会务套餐查询工具、保险查询工具
    - 商务用车预订工具、会务套餐预订工具、保险购买工具

架构图（一次请求的调用关系）：

    主助手 → [A2A :5007] TripServer.handle_task
                              ↓
                    query_trip → 加载 MCP 工具定义
                              ↓
                    AgentExecutor → 模型选择工具和参数
                              ↓
                    [MCP :8003] 执行所选工具
                              ├─ 用车/保险查询 → MySQL
                              ├─ 会务套餐查询 → Embedding + Milvus
                              └─ 预订工具 → 模拟结果
                              ↓
                    工具结果 → 模型整理 → Task → 主助手

请求/响应流向：
    用户 → 主助手 → [A2A] → TripServer.handle_task()
                                  ↓
                          query_trip() → LangChain Agent (LLM)
                                  ↓
                          MCP Tools → MCP Server :8003
                                  ↓
                          MCP Server 返回工具结果
                                  ↓
                          LLM 格式化为中文回复
                                  ↓
                          handle_task() 设置 TaskStatus
                                  ↓
                          主助手 → 用户
"""

# ==================== 导入依赖 ====================
import asyncio  # 异步 IO 库

# create_tool_calling_agent: 创建基于工具调用的 Agent
# AgentExecutor: Agent 执行器，负责运行 Agent 循环
from datetime import datetime  # 时间处理

import pytz  # 时区库
from langchain.agents import AgentExecutor, create_tool_calling_agent
from langchain_core.prompts import ChatPromptTemplate  # 提示模板
from langchain_core.tools import StructuredTool  # 【新改动1】支持多个具名参数的工具
from langchain_mcp_adapters.tools import load_mcp_tools  # 从 MCP 会话加载工具

# A2AServer: A2A 服务器基类
# run_server: 启动服务器的函数
# AgentCard: 代理卡片
# AgentSkill: 代理技能
# TaskStatus: 任务状态对象
# TaskState: 任务状态枚举（COMPLETED / FAILED / INPUT_REQUIRED）
from langchain_openai import ChatOpenAI  # LangChain 的大模型接口
from python_a2a import (
    A2AServer,
    AgentCard,
    AgentSkill,
    MCPClient,
    TaskState,
    TaskStatus,
    run_server,
    to_langchain_tool,
)

from config import Config  # 项目配置
from create_logger import logger  # 日志模块

conf = Config()  # 全局配置实例

# ==================== 初始化大模型 ====================
# 创建一个 LLM 实例，供 Agent 使用
# 【启动 S3.1】创建模型客户端；模型推理在后面的 invoke/ainvoke 才发生。
llm = ChatOpenAI(
    model=conf.model_name,
    base_url=conf.base_url,
    api_key=conf.api_key,
    temperature=conf.temperature,
)

# ==================== MCP 客户端配置 ====================
# MCP Server 运行在 8003 端口，提供差旅相关的工具（查询 + 预订）
MCP_URL = "http://127.0.0.1:8003"


# ==================== 差旅查询函数 ====================
async def query_trip(conversation: str) -> dict:
    """
    通过 LangChain Agent + MCP Tools 执行差旅查询或预订

    与 ticket_server.py 的 handle_task 一样使用工具调用模式：
    1. 连接 MCP Server 加载所有可用的工具
    2. 创建一个 LangChain Agent，让它自主选择调用哪个工具
    3. AgentExecutor 负责执行 Agent 循环

    参数：
        conversation (str): 用户的查询内容，例如：
            "我想租一辆车，北京，明天"
            "有没有成都客户拜访的会务套餐"
            "买一份差旅保险"
            "帮我预订北京政企拜访3日套餐"

    返回值：
        dict: 查询结果，格式为：
            - {"status": "success", "message": "格式化后的查询结果"}
            - {"status": "error", "message": "错误信息"}
    """
    # 调用者：本文件的 handle_task。conversation 是整个任务文本，可能包含历史和当前需求。
    # 输出：字典 {"status": "success"/"error", "message": "回复或错误文本"}，由 handle_task 再转成 A2A 状态。
    # 这里的 success 表示执行器返回了结果，不等于工具一定查到了数据。
    try:
        # 【主流程 7.2】连接 MCP 读取工具定义；此处尚未执行具体业务查询。
        tools = to_langchain_tool(MCP_URL)
        # 【新改动1】保留 MCP 参数定义，将单参数 Tool 转为 StructuredTool，
        # 使取车城市、还车城市、日期等参数能按名称传给原工具函数。
        # 【主流程 7.3】把工具参数结构交给 StructuredTool，保留原 MCP 调用函数。
        # 模型输出参数名称和值，执行器按照这个结构传参。
        tools = [
            StructuredTool(
                # 工具对象的五个字段各有用途：name 用于匹配模型选中的工具，description 帮助模型理解用途；
                # func 保存真正发起 MCP 调用的函数；args_schema 规定参数名/类型；metadata 保留原工具信息。
                # 列表里的每一项转换成独立 StructuredTool，模型随后可在这些工具中选择。
                name=tool.name,
                description=tool.description,
                func=tool.func,
                args_schema=tool.metadata["parameters"],
                metadata=tool.metadata,
            )
            for tool in tools
        ]
        # 定义 Agent 的系统 Prompt
        # 提示模板分三部分：system 给规则和当前日期，human 放本次输入，placeholder 放执行过程消息。
        # {input}/{current_date} 由 invoke/ainvoke 的参数提供；{agent_scratchpad} 由 AgentExecutor 随工具调用更新。
        # 占位符不是立刻执行的代码，只是在实际请求模型前被替换成消息内容。
        prompt = ChatPromptTemplate.from_messages(
            [
                (
                    "system",
                    """你是一个差旅管家助手，能够调用工具来完成商务用车查询/预定、会务套餐查询/预订、保险查询/购买。
你需要仔细分析用户的问题，从问题中提取工具需要的参数，然后调用对应的工具。
如果用户提供的信息不足以提取到调用工具所有必要参数，则向用户追问，以获取该信息。不能自己编撰参数。预订工具仅用于模拟演示，回复必须明确说明模拟、未实际下单，不得声称已出票或生成保单。

查询到结果后，请用清晰的中文格式化输出，格式如下：
- 商务用车：取车城市 到 还车城市 日期: 车型XX，公司XX，价格XX元/天，余量XX
- 会务套餐：城市 套餐名称: 天数XX天，价格XX元，余位XX/总数XX，亮点XX，出发日期列表
- 保险：保险类型: 名称XX，价格XX元/份，保障范围XX，保险公司XX
预定成功后请直接告知用户预定结果。
当前日期是{current_date}。""",
                ),
                ("human", "{input}"),  # 用户输入会替换到这里
                ("placeholder", "{agent_scratchpad}"),  # Agent 思考过程的占位符
            ]
        )

        # 创建基于工具调用的 Agent
        # 【主流程 7.4】组装 Agent，随后创建执行器；此时尚未调用模型处理本次请求。
        # agent_scratchpad 承载本轮工具调用与工具返回消息，不是跨轮用户记忆。
        agent = create_tool_calling_agent(llm, tools, prompt)

        # 创建 Agent 执行器
        # 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}
        )

        # ainvoke 已经完成整轮工具循环，response["output"] 是模型整理后的文字。
        # 将它包装成普通字典返回给 handle_task，后者再决定 COMPLETED、INPUT_REQUIRED 或 FAILED。
        return {"status": "success", "message": response["output"]}

    except ExceptionGroup as eg:
        # MCP 工具内部使用 anyio.create_task_group()，失败时抛出 ExceptionGroup
        first_exc = eg.exceptions[0] if eg.exceptions else eg
        logger.error(f"差旅 MCP 查询出错：{first_exc}")
        return {"status": "error", "message": f"差旅 MCP 查询出错：{first_exc}"}
    except BaseException as e:
        # 兜底捕获（包括 ExceptionGroup 等 BaseException 子类）
        logger.error(f"差旅 MCP 查询出错：{str(e)}")
        return {"status": "error", "message": f"差旅 MCP 查询出错：{str(e)}"}


# ==================== Agent Card（代理卡片） ====================
# 描述差旅管家代理的能力、技能等信息
# 【启动 S3.2】定义对外说明。skills 是能力介绍，实际可执行工具来自 MCP。
# 主助手当前按 Config.intent 路由；卡片声明的能力也不等于此方法已实现逐片输出。
# 卡片字段说明：name 是代理标识；url 是访问地址；description/skills 向调用方介绍能力。
# 这里描述技能不会自动生成对应 Python 函数，实际工具仍由 MCP 服务注册和提供。
# 主助手 Config.intent 中的名称、AgentNetwork 的注册名称和这里的 name 要相互对应。
agent_card = AgentCard(
    name="TripAssistant",  # 代理名称
    description="基于 LangChain 提供差旅管家服务的统一助手，支持商务用车、会务套餐、保险的查询与预订",  # 代理描述
    url="http://localhost:5007",  # 代理的访问地址
    version="1.0.0",  # 版本号
    capabilities={"streaming": True, "memory": True},  # 支持的能力
    skills=[
        # 技能1：商务用车查询
        AgentSkill(
            name="query car rental",
            description="查询商务用车信息，支持指定取车城市、还车城市、日期和车型",
            examples=[
                "商务用车 成都 成都 2026-05-01",
                "成都明天安排一辆商务SUV接送客户",
            ],
        ),
        # 技能2：会务套餐查询（语义搜索）
        AgentSkill(
            name="query meeting package",
            description="通过语义搜索查询会务套餐信息，支持自然语言描述需求（如成都办数字化转型峰会，需要接送和酒店）和可选的城市过滤",
            examples=[
                "成都办数字化转型峰会，需要接送和酒店",
                "适合销售团队的短途会务套餐",
                "客户接待晚宴方案",
                "北京政企拜访会务套餐",
            ],
        ),
        # 技能3：保险查询
        AgentSkill(
            name="query insurance",
            description="查询差旅保险产品，支持指定保险类型和日期",
            examples=["差旅保险 2025-08-01", "买一份综合型差旅保险"],
        ),
        # 技能4：商务用车预订
        AgentSkill(
            name="order car rental",
            description="根据日期、车型和数量预订商务用车服务",
            examples=["预订商务用车 2025-08-01 SUV 1辆"],
        ),
        # 技能5：会务套餐预订
        AgentSkill(
            name="order meeting package",
            description="根据日期、套餐名称和人数预订会务套餐",
            examples=["预订成都客户拜访2日套餐 2025-08-01 2人"],
        ),
        # 技能6：保险购买
        AgentSkill(
            name="order insurance",
            description="根据保险类型、日期和份数购买差旅保险",
            examples=["购买综合型保险 2025-08-01 1份"],
        ),
    ],
)


# ==================== 差旅查询服务器类 ====================
class TripServer(A2AServer):
    """
    差旅管家 A2A 服务器

    这个类继承自 A2AServer，负责处理所有差旅相关的任务。
    与票务子助手一样使用工具调用；本类的具体流程如下：
    1. 接收来自主助手的任务（Task）
    2. 从任务中提取用户的查询内容
    3. 调用 query_trip 函数执行查询
    4. 将查询结果设置到任务状态中，返回给主助手
    """

    def __init__(self):
        """
        初始化差旅查询服务器
        """
        # 【启动 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 客户端的任务 —— 本服务器的核心方法

        完整流程：
        1. 提取输入：从任务消息中获取用户的查询内容
        2. 调用 MCP 查询：通过 query_trip 函数执行查询
        3. 根据查询结果设置任务状态：
           - 成功：COMPLETED 状态，结果放在 task.artifacts
           - 需要输入：INPUT_REQUIRED 状态，追问信息放在 task.status.message
           - 失败：FAILED 状态，错误信息放在 task.status.message

        参数：
            task: A2A 任务对象

        返回值：
            task: 处理后的任务对象（已设置状态和结果）
        """
        # ========== 步骤1：提取输入 ==========
        # 【主流程 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 查询 ==========
            # 同步 handle_task 用 asyncio.run 驱动异步查询，等查询结束后才设置任务状态。
            trip_result = asyncio.run(query_trip(conversation))
            logger.info(f"MCP 查询返回: {trip_result}")

            # ========== 步骤3：根据结果设置任务状态 ==========
            # 这里按查询函数返回的 status 分支；先确认执行器是否正常返回，再从 message 检查是否为追问。
            if trip_result.get("status") == "success":
                result_text = trip_result.get("message", "")

                # 检查是否是追问消息
                # 【主流程 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 trip_result.get("status") == "error":
                task.status = TaskStatus(
                    state=TaskState.FAILED,
                    message={
                        "role": "agent",
                        "content": {
                            "text": trip_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__":
    # 创建差旅查询服务器实例
    trip_server = TripServer()

    # 打印服务器信息
    print("\n=== 服务器信息 ===")
    print(f"名称: {trip_server.agent_card.name}")
    print(f"描述: {trip_server.agent_card.description}")
    print("\n技能:")
    # 启动时只打印卡片中声明的技能，帮助人工核对名称和配置；不会依次调用这些技能。
    for skill in trip_server.agent_card.skills:
        print(f"- {skill.name}: {skill.description}")

    # 启动 A2A 服务器，监听 5007 端口
    # 【启动 S3.4】开始监听 A2A 请求；后续 handle_task 按请求触发，不按函数定义顺序执行。
    run_server(trip_server, host="127.0.0.1", port=5007)