SmartBizTrip/a2a_server/ticket_server.py

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

"""
需求：实现基于A2A的统一票务服务器，处理用户的高铁票、航班和会展门票查询及预订请求

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

    与 weather_server 一样，本模块使用参数化工具调用：
    - 模型选择工具并提取参数，MCP 服务内部负责 SQL 查询
    - ticket_server 使用 LangChain Agent + MCP Tools 模式（工具调用模式）
      LLM 直接决定调用哪个 MCP 工具（高铁票查询/航班查询/会展活动查询/预定），
      然后由 LangChain 的 AgentExecutor 自动完成工具调用和结果处理

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

    涉及的关键技术：
    - LangChain Agent: 一种让 LLM 自主选择和使用工具的框架
    - Tool Calling Agent: 一种 Agent 类型，LLM 以结构化格式调用工具
    - MCP Tools: MCP Server 提供的工具，本系统中包括高铁票查询、航班查询等
    - AgentExecutor: LangChain 的执行器，负责运行 Agent 循环（思考→调用工具→处理结果→继续）

    MCP Server（端口 8001）提供的工具包括：
    - 高铁票查询工具
    - 航班查询工具
    - 会展门票查询工具
    - 高铁票预定工具
    - 航班预定工具
    - 会展门票预定工具
"""

# 单次票务任务示例：主助手发送“北京到上海高铁”
# → handle_task → to_langchain_tool → StructuredTool → AgentExecutor.invoke
# → 模型选择 query_train 及城市/日期/座席参数 → MCP :8001 → MySQL
# → 工具返回 JSON → 模型整理文字 → task.artifacts → 主助手。
# 查询和预订工具按需要选择，不会把注册的六个工具依次全部执行。

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

# from mcp import ClientSession  # MCP 客户端会话
# from mcp.client.streamable_http import streamablehttp_client  # MCP HTTP 流式客户端

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

from langchain_openai import ChatOpenAI  # LangChain 的大模型接口
from langchain_core.tools import StructuredTool
from langchain_core.prompts import ChatPromptTemplate  # 提示模板
# from langchain_mcp_adapters.tools import load_mcp_tools  # 从 MCP 会话加载工具
from langchain.agents import create_tool_calling_agent, AgentExecutor
# create_tool_calling_agent: 创建基于工具调用的 Agent
# AgentExecutor: Agent 执行器，负责运行 Agent 循环

from datetime import datetime  # 时间处理
import pytz  # 时区库

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

# 1. 初始化大模型

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
)
# 2. 定义AgentCard，6个skill

# ==================== Agent Card（代理卡片） ====================
# 描述票务代理的能力、技能等信息
# 卡片用于说明和展示能力；当前主助手通过 Config.intent 的映射选择代理。
# TODO 这里的6个skill作用：是给主路由在选择子Agent的时候作为参考使用的，和MCP层的工具没有对应的关系。可以将6个skill合成2个，
# 1. 查询票务（高铁票、航班、会展活动） 2. 预定票（高铁票、航班、会展活动）
# 【启动 S3.2】定义对外说明。skills 是能力介绍，实际可执行工具来自 MCP。
# 主助手当前按 Config.intent 路由；卡片声明的能力也不等于此方法已实现逐片输出。
# 卡片字段说明：name 是代理标识；url 是访问地址；description/skills 向调用方介绍能力。
# 这里描述技能不会自动生成对应 Python 函数，实际工具仍由 MCP 服务注册和提供。
# 主助手 Config.intent 中的名称、AgentNetwork 的注册名称和这里的 name 要相互对应。
agent_card = AgentCard(
    name="TicketAssistant",  # 代理名称
    description="基于 LangChain 提供票务查询和预订服务的统一助手",  # 代理描述
    url="http://localhost:5006",  # 代理的访问地址
    version="2.0.0",  # 版本号
    capabilities={"streaming": True, "memory": True},  # 支持的能力
    skills=[
        # 技能1：高铁票查询
        AgentSkill(
            name="query train tickets",
            description="查询高铁票，支持指定出发城市、到达城市、日期和座位类型",
            examples=["高铁票 北京 上海 2025-07-31", "北京到广州的高铁 明天 二等座"]
        ),
        # 技能2：航班查询
        AgentSkill(
            name="query flight tickets",
            description="查询航班，支持指定出发城市、到达城市、日期和舱位类型",
            examples=["航班 北京 上海 2025-07-31", "北京到深圳的航班 后天 经济舱"]
        ),
        # 技能3：会展门票查询
        AgentSkill(
            name="query event tickets",
            description="查询会展门票，支持指定城市、活动名称、日期和票档类型",
            examples=["会展活动 成都企业数字化转型峰会 2026-05-01", "上海大型展会参展门票 2026-06-10"]
        ),
        # 技能4：高铁票预定
        AgentSkill(
            name="order train tickets",
            description="根据车次、座位类型和数量预定高铁票",
            examples=["预定G1次列车 2025-11-15 北京到上海 二等座 1张"]
        ),
        # 技能5：航班预定
        AgentSkill(
            name="order flight tickets",
            description="根据航班号、舱位类型和数量预定航班",
            examples=["预定MU5101 2025-12-11 上海到北京 公务舱 2张"]
        ),
        # 技能6：会展门票预定
        AgentSkill(
            name="order event tickets",
            description="根据活动名称、场地、日期、票档和数量预定会展门票",
            examples=["预定成都企业数字化转型峰会 2026-05-01 成都 VIP商务席 2张"]
        )
    ]
)

MCP_URL = 'http://127.0.0.1:8001'


# 3. 定义类，继承A2AServer
class TicketServer(A2AServer):
    # 4. 实现初始化
    def __init__(self):
        # 【启动 S3.3】将卡片交给 A2AServer；父类提供通信框架，本类负责实现 handle_task。
        # 输入：模块加载时创建的 agent_card。super() 将当前实例交给父类初始化。
        # 本类继承了 A2A 通信框架，只覆盖业务回调 handle_task；此时还没有用户查询需要执行。
        super().__init__(agent_card=agent_card)

    # 5. 实现handle_task
    # 接收用户的请求，使用langchain-agent自动去调用对应的工具集
    def handle_task(self, task):
        # 5.1 从task里面获取message
        # 【主流程 7.1】A2A 框架收到任务后调用此方法；先从 task.message 取输入。
        # 输入 task 来自主助手的 A2A 请求；它包含任务编号、message，处理后还会带上 status/artifacts。
        # 本方法不是被 ChatService 直接当本地 Python 方法调用，而是由此进程的 A2A 框架接收请求后触发。
        # 输出仍是 task 对象，只是在原对象上填写了结果状态和产物。
        # (task.message or {}) 在消息为空时退回空字典，再用 get 避免直接按键访问报 KeyError。
        # 与天气、差旅助手一致，取出 content 中的文本交给模型。
        content = (task.message or {}).get("content", {})
        query = content.get("text", "") if isinstance(content, dict) else ""
        # 5.2 获取MCP Server所有的工具集（tools） 
        # 【主流程 7.2】先请求 MCP 工具定义，取得工具名、说明、参数结构和调用包装器。
        # 加载定义并不等于已查票；只有执行器调用选中的工具时才执行查询。
        tools = to_langchain_tool(MCP_URL)  ## 向票务工具服务获取可用工具，把它们包装成LangChain执行器可以使用的工具对象
        # python-a2a 0.5.4 返回单参数 Tool；保留 MCP JSON Schema，以支持查票的多个命名参数。
        # 【主流程 7.3】保留多个具名参数的结构，例如 departure_city、arrival_city、date。
        # func 仍是原 MCP 调用函数，StructuredTool 负责让参数结构与执行器匹配。
        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]
        # 5.3 构建提示词模板(prompt)
        # 提示模板分三部分：system 给规则和当前日期，human 放本次输入，placeholder 放执行过程消息。
        # {input}/{current_date} 由 invoke/ainvoke 的参数提供；{agent_scratchpad} 由 AgentExecutor 随工具调用更新。
        # 占位符不是立刻执行的代码，只是在实际请求模型前被替换成消息内容。
        prompt = ChatPromptTemplate.from_messages([
            ("system",
             "你是一个票务预定助手，能够调用工具来完成高铁票、航班或会展票的预定。你需要仔细分析工具需要的参数，然后从用户提供的信息中提取信息。如果用户提供的信息不足以提取到调用工具所有必要参数，则向用户追问，以获取该信息。不能自己编撰参数。预订工具仅用于模拟演示，回复必须明确说明模拟、未实际下单，不得声称已出票或生成保单。当前日期{current_date}"),
            ("human", "{input}"),
            ("placeholder", "{agent_scratchpad}"),
        ])
        # 5.4 构建Agent(tools + prompt + llm)
        # 【主流程 7.4】组装模型、工具和提示词；agent_scratchpad 保存工具调用/返回消息。
        # Agent 决定下一步，AgentExecutor 驱动循环并实际调用工具。
        agent = create_tool_calling_agent(llm, tools, prompt) ## agent 组织提示词和模型调用，决定下一步动作
        # agent 负责根据输入和已有工具结果产生下一步动作；executor 负责运行这个过程。
        # 若模型请求工具，executor 按工具名找到对象并调用 func，再把返回值交回模型；直到得到最终回答。
        # verbose=True 打印执行过程，便于观察调用轨迹，不代表业务结果会自动保存到记忆或数据库。
        agent_executor = AgentExecutor(agent=agent, tools=tools, verbose=True)  ## agent_executor负责跑执行过程，把模型选择的工具真正调用起来，再把结果交给模型
        # 5.5 执行请求Agent的invoke  这里，真正开始处理请求
        try:
            # 【主流程 7.5 → 8】从这里开始运行工具调用循环；必要时可调用多次工具。
            result = agent_executor.invoke({'input': query,
                                            'current_date':datetime.strftime(datetime.now(), '%Y-%m-%d')
                                            })
            # invoke 返回字典，output 键里才是 Agent 最终回复文本；它不是原始 SQL 行列表。
            # 若缺少信息，模型可能直接追问并结束，不会保证每次任务都调用一个工具。
            output = result['output']
            # 5.6 根据返回结果封装Task.Status和artifacts
            # ① 缺少参数
            # 【主流程 9.1】当前用回复中的关键词区分追问，否则标为 COMPLETED。
            # 这是文字判断，不是业务成功校验；预订工具返回的仍然是模拟结果。
            # 这个 if 检查的是文本中是否出现“缺少”两个字，不能识别所有可能的追问表达。
            # 命中就把追问放进 status.message；未命中才进入 else 填写正常产物。
            if "缺少" in output:
                task.status = TaskStatus(
                    state=TaskState.INPUT_REQUIRED,
                    message={"role": "agent", "content": {"text": output}}
                )
            else:
                # ② 运行成功
                task.status = TaskStatus(
                    state=TaskState.COMPLETED,
                    message={"role": "agent", "content": {"text": "运行成功"}}
                )
                # 产物结构是列表→parts 列表→文本项；主助手从 artifacts[0]["parts"][0]["text"] 读取 output。
                # status 中的“运行成功”只说明此回调走完，真正给用户的文本在 artifacts。
                task.artifacts = [{"parts": [{"type": "text", "text": output}]}]
            # 【主流程 9.2】把同一个 Task 连同状态或结果交回 A2A 框架，再由框架发送给主助手。
            return task
        # 这个 try 从执行器 invoke 才开始；前面的工具加载和 Agent 构造异常不在此 try 的捕获范围。
        # 被捕获的执行异常写入 FAILED 状态，让主助手能读取错误文本。
        except Exception as e:
            # ③ 运行失败
            task.status = TaskStatus(TaskState.FAILED,

                                     message={"role": "agent", "content": {"text": f"agent执行失败：{e}"}}
                                     )
            return task
if __name__ == "__main__":
    # 创建票务查询服务器实例
    ticket_server = TicketServer()

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

    # 启动 A2A 服务器，监听 5006 端口
    # 主助手会通过 http://localhost:5006 连接本服务器
    # 【启动 S3.4】开始监听 A2A 请求；后续 handle_task 按请求触发，不按函数定义顺序执行。
    run_server(ticket_server, host="127.0.0.1", port=5006)