"""Minimal MCP Streamable HTTP adapter for Baixinghui customer-service tools.""" from __future__ import annotations import json import os import time import uuid from typing import Any from fastapi import APIRouter, Request from fastapi.responses import JSONResponse, Response, StreamingResponse from app.api.travel_assistant import _customer_service_query_response router = APIRouter() MCP_PROTOCOL_VERSION = "2025-06-18" SUPPORTED_MCP_PROTOCOL_VERSIONS = {"2024-11-05", "2025-03-26", MCP_PROTOCOL_VERSION} DEFAULT_GRAPH_NAME = "baixinghui_travel_agency" def _tool_definitions() -> list[dict[str, Any]]: return [ { "name": "ask_customer_service", "description": ( "向百姓惠旅行社知识图谱智能客服提问。适用于线路价格、行程天数、景区、酒店、费用、车型、" "线路推荐和复杂多跳问答。只读工具，不承诺最终价格、余位、房型或景区政策。" ), "inputSchema": { "type": "object", "properties": { "question": {"type": "string", "description": "用户自然语言问题"}, "session_id": {"type": "string", "description": "外部会话 ID，可选"}, "request_id": {"type": "string", "description": "外部请求 ID，可选"}, "customer_context": { "type": "object", "description": "客户上下文，例如人数、预算、出发日期、偏好，可选", }, }, "required": ["question"], }, }, { "name": "query_route_price", "description": "查询指定旅行线路的参考价格、成人/儿童价格区间和不可直接承诺事项。只读工具。", "inputSchema": { "type": "object", "properties": { "route_name": {"type": "string", "description": "线路名称或关键词，例如 黄小西三日游"}, "traveler_type": { "type": "string", "description": "游客类型，可选，例如 成人、儿童、老人", }, "session_id": {"type": "string", "description": "外部会话 ID，可选"}, }, "required": ["route_name"], }, }, { "name": "list_travel_routes", "description": "查询百姓惠旅行线路清单，可按天数、景区或关键词筛选。只读工具。", "inputSchema": { "type": "object", "properties": { "keyword": {"type": "string", "description": "线路或目的地关键词，可选"}, "duration_days": {"type": "integer", "description": "期望天数，可选"}, "limit": {"type": "integer", "description": "返回条数上限，可选，默认 20"}, "session_id": {"type": "string", "description": "外部会话 ID，可选"}, }, }, }, { "name": "query_route_detail", "description": "查询某条线路的天数、价格、景区、附近酒店和费用边界。只读工具。", "inputSchema": { "type": "object", "properties": { "route_name": {"type": "string", "description": "线路名称或关键词"}, "detail_focus": { "type": "string", "description": "关注点，可选，例如 价格、景区、酒店、费用、车型", }, "session_id": {"type": "string", "description": "外部会话 ID，可选"}, }, "required": ["route_name"], }, }, { "name": "compare_routes", "description": "对比两条旅行线路的适配度、景点数量、轻松程度或客服推荐理由。只读工具。", "inputSchema": { "type": "object", "properties": { "route_a": {"type": "string", "description": "第一条线路名称或关键词"}, "route_b": {"type": "string", "description": "第二条线路名称或关键词"}, "compare_focus": { "type": "string", "description": "对比关注点，例如 老人小孩、不太累、景点更多、价格", }, "session_id": {"type": "string", "description": "外部会话 ID，可选"}, }, "required": ["route_a", "route_b"], }, }, ] def _json_rpc_result(message_id: Any, result: dict[str, Any]) -> dict[str, Any]: return {"jsonrpc": "2.0", "id": message_id, "result": result} def _json_rpc_error(message_id: Any, code: int, message: str, data: Any = None) -> dict[str, Any]: error: dict[str, Any] = {"code": code, "message": message} if data is not None: error["data"] = data return {"jsonrpc": "2.0", "id": message_id, "error": error} def _mcp_auth_enabled() -> bool: return bool(os.getenv("BXH_MCP_TOKEN", "").strip()) def _is_authorized(request: Request) -> bool: expected = os.getenv("BXH_MCP_TOKEN", "").strip() if not expected: return True auth = request.headers.get("authorization", "").strip() api_key = request.headers.get("x-mcp-api-key", "").strip() bearer = auth[7:].strip() if auth.lower().startswith("bearer ") else auth return expected in {bearer, api_key} def _normalize_args(params: dict[str, Any]) -> tuple[str, dict[str, Any]]: name = str(params.get("name") or "").strip() args = params.get("arguments") if args is None: args = {} if not isinstance(args, dict): raise ValueError("arguments must be an object") return name, args def _question_for_tool(name: str, args: dict[str, Any]) -> str: if name == "ask_customer_service": return str(args.get("question") or "").strip() if name == "query_route_price": route = str(args.get("route_name") or "").strip() traveler = str(args.get("traveler_type") or "").strip() suffix = f"，{traveler}怎么收费" if traveler else "" return f"{route}多少钱{suffix}？".strip() if name == "list_travel_routes": keyword = str(args.get("keyword") or "").strip() days = args.get("duration_days") pieces = ["旅行车线路有哪些"] if keyword: pieces.append(f"和{keyword}相关") if days: pieces.append(f"{days}天") return "，".join(pieces) + "？" if name == "query_route_detail": route = str(args.get("route_name") or "").strip() focus = str(args.get("detail_focus") or "").strip() if focus: return f"{route}的{focus}是什么？" return f"{route}多少钱，可以玩几天，期间可以去哪些景区，附近可以入住什么酒店？" if name == "compare_routes": route_a = str(args.get("route_a") or "").strip() route_b = str(args.get("route_b") or "").strip() focus = str(args.get("compare_focus") or "哪个更适合客户").strip() return f"{route_a}和{route_b}{focus}，为什么？" raise KeyError(name) def _mcp_content_text(payload: dict[str, Any]) -> str: reply = str(payload.get("customer_reply") or payload.get("answer") or "").strip() if reply: return reply return "当前知识图谱没有返回可直接回复的话术，请补充线路、人数、天数或出行日期后再试。" def _mcp_structured_data(payload: dict[str, Any]) -> dict[str, Any]: knowledge = payload.get("knowledge") if isinstance(payload.get("knowledge"), dict) else {} return { "status": payload.get("status") or "ok", "service": payload.get("service") or "baixinghui_customer_service", "request_id": payload.get("request_id"), "trace_id": payload.get("trace_id"), "session_id": payload.get("session_id"), "graph_name": payload.get("graph_name"), "question": payload.get("question"), "customer_reply": payload.get("customer_reply"), "answer": payload.get("answer"), "confidence": payload.get("confidence"), "follow_up_questions": payload.get("follow_up_questions") or [], "risk_notes": payload.get("risk_notes") or [], "knowledge": { "plans": knowledge.get("plans") or [], "evidence": knowledge.get("evidence") or [], }, "routing": payload.get("routing") or {}, "trace": payload.get("trace") or {}, } def _negotiate_protocol_version(params: dict[str, Any]) -> str: requested = str(params.get("protocolVersion") or "").strip() if requested in SUPPORTED_MCP_PROTOCOL_VERSIONS: return requested return MCP_PROTOCOL_VERSION async def _call_customer_service_tool(name: str, args: dict[str, Any]) -> dict[str, Any]: question = _question_for_tool(name, args) if not question: raise ValueError("question or route fields are required") body = { "request_id": args.get("request_id") or f"mcp-{uuid.uuid4().hex[:16]}", "session_id": args.get("session_id") or "", "channel": args.get("channel") or "mcp", "graph_name": args.get("graph_name") or DEFAULT_GRAPH_NAME, "question": question, "customer_context": args.get("customer_context") if isinstance(args.get("customer_context"), dict) else {}, "limit": int(args.get("limit") or 80), } started = time.perf_counter() payload = await _customer_service_query_response(body) structured = _mcp_structured_data(payload) structured["tool_name"] = name structured["tool_latency_ms"] = max(1, round((time.perf_counter() - started) * 1000)) return { "content": [{"type": "text", "text": _mcp_content_text(payload)}], "structuredContent": structured, "isError": False, } async def _handle_request(message: dict[str, Any]) -> dict[str, Any] | None: message_id = message.get("id") method = str(message.get("method") or "") params = message.get("params") if isinstance(message.get("params"), dict) else {} if not method: return _json_rpc_error(message_id, -32600, "Invalid Request") # JSON-RPC notifications have no id and do not require a response. if "id" not in message: return None if method == "initialize": return _json_rpc_result( message_id, { "protocolVersion": _negotiate_protocol_version(params), "capabilities": {"tools": {"listChanged": False}}, "serverInfo": { "name": "baixinghui-customer-service-mcp", "version": "0.1.0", }, "instructions": "使用百姓惠旅行社知识图谱工具回答线路、价格、景区、酒店和费用问题。", }, ) if method == "ping": return _json_rpc_result(message_id, {}) if method == "tools/list": return _json_rpc_result(message_id, {"tools": _tool_definitions()}) if method == "tools/call": try: name, args = _normalize_args(params) tool_names = {tool["name"] for tool in _tool_definitions()} if name not in tool_names: return _json_rpc_error(message_id, -32602, f"Unknown tool: {name}") result = await _call_customer_service_tool(name, args) return _json_rpc_result(message_id, result) except ValueError as exc: return _json_rpc_error(message_id, -32602, str(exc)) except Exception as exc: # noqa: BLE001 return _json_rpc_error(message_id, -32603, f"Tool execution failed: {str(exc)[:300]}") if method in {"resources/list", "prompts/list"}: key = "resources" if method == "resources/list" else "prompts" return _json_rpc_result(message_id, {key: []}) return _json_rpc_error(message_id, -32601, f"Method not found: {method}") @router.get("/mcp") async def mcp_get(request: Request): if not _is_authorized(request): return JSONResponse({"detail": "Unauthorized"}, status_code=401) accept = request.headers.get("accept", "") if "text/event-stream" in accept: async def stream(): yield ": baixinghui customer-service MCP endpoint is ready\n\n" return StreamingResponse(stream(), media_type="text/event-stream") return { "name": "baixinghui-customer-service-mcp", "protocolVersion": MCP_PROTOCOL_VERSION, "auth": "bearer" if _mcp_auth_enabled() else "disabled", "tools": [tool["name"] for tool in _tool_definitions()], } @router.post("/mcp") async def mcp_post(request: Request): if not _is_authorized(request): return JSONResponse( _json_rpc_error(None, -32001, "Unauthorized"), status_code=401, media_type="application/json", ) try: payload = await request.json() except Exception: return JSONResponse( _json_rpc_error(None, -32700, "Parse error"), status_code=400, media_type="application/json", ) if isinstance(payload, list): responses = [] for message in payload: if not isinstance(message, dict): responses.append(_json_rpc_error(None, -32600, "Invalid Request")) continue response = await _handle_request(message) if response is not None: responses.append(response) if not responses: return Response(status_code=202) return JSONResponse(responses, media_type="application/json") if not isinstance(payload, dict): return JSONResponse( _json_rpc_error(None, -32600, "Invalid Request"), status_code=400, media_type="application/json", ) response = await _handle_request(payload) if response is None: return Response(status_code=202) return JSONResponse(response, media_type="application/json")