一、什么是MCP ？

MCP（Model Context Protocol，模型上下文协议） 是由Anthropic公司（Claude模型的开发商）在2024年11月推出并开源的一种开放协议标准。它旨在标准化应用程序如何为大型语言模型（LLMs）提供上下文信息，可以形象地理解为"AI应用的USB-C接口"，为大模型连接各种数据源和工具提供了统一标准，使 LLM 能够连接各种数据源和外部工具，从而实现跨模型、跨平台无缝地访问和处理信息。

---

二、为什么需要MCP？

在MCP出现之前，AI应用开发基于 Function Calling 与外部工具交互能力显著，与此同时也面临一个巨大的痛点：N×M”的集成问题。

• N个AI模型：如GPT-4，Claude 3，文心一言，Llama等。
• M个外部工具/数据源：如第三方服务，GitHub，Slack，数据库，企业内部系统等。

为了让每个模型都能使用这些工具，开发者需要为每一个“模型-工具”组合都开发一套独立的连接器（Adapter）。这导致了：

• 开发成本极高：集成工作重复、繁琐，且无法复用。
• 能力扩展困难：模型本身被困在训练数据的“玻璃房”里，无法获取实时信息或执行操作。
• 生态割裂：不同模型平台的工具生态不互通，形成了新的数据孤岛。

MCP的诞生，就是为了将这个“N×M”的复杂问题，简化为“M+N”的简单模式：

• M个工具方：只需开发一次MCP服务器（Server），就能被所有支持MCP的AI模型使用。
• N个模型方：只需在AI应用（Host）中内置一个MCP客户端（Client），就能连接所有MCP服务器。

这种客户端-服务器解耦的架构，极大地降低了AI与外部世界连接的门槛，推动了AI生态的标准化和开放化。

---

三、MCP的核心优势

MCP 是一种标准化的通信协议，通过 Client、Host 和 Server 将大模型与外部交互抽象成了 “客户端 - 服务器” 架构，他实现了 AI 应用能力的动态扩展，降低了开发门槛和成本，其核心优势如下：

1. 动态发现 (Dynamic Discovery)：这是MCP的“杀手级”特性。LLM在运行时可以主动向服务器查询list_tools()，获取所有可用的工具列表及其功能描述。这意味着，即使是新上线的MCP服务器，模型也能立即“学会”如何使用它，无需任何预训练或代码修改。
2. 安全性 (Security)：MCP内置了严格的安全机制。
   • 权限控制：数据源的所有者始终掌握访问权，API密钥等敏感信息由MCP Server自己控制，无需暴露给AI模型。
   • 沙箱隔离：MCP Server可以在沙箱环境中运行，限制其对系统资源的访问。
   • 用户授权：高风险操作（如删除文件、转账）可以强制要求用户在客户端进行二次确认。
3. 标准化与互操作性 (Standardization & Interoperability)：MCP是供应商中立、模型无关的。为Claude开发的MCP服务器，理论上可以被GPT-4、DeepSeek等任何支持MCP的模型使用，打破了模型间的生态壁垒。
4. 灵活性与可扩展性 (Flexibility & Extensibility)：新工具的接入就像给手机安装App一样简单，只需开发一个MCP Server即可，无需改动底层AI模型或Host应用。这催生了一个围绕MCP的工具开发生态。

---

四、MCP技术架构

1. 核心组件

MCP遵循client-host-server架构，每个host可以运行多个client实例，该架构可以让任何应用通过清晰安全的方式集成人工智能能力，以下是他的三个核心组件：

• MCP主机（host）：使用MCP连接各种资源的AI应用程序，包括：AI助手（Claude Desktop）、开发环境（Cursor、Cline），以及专门的AI工具等。
• MCP客户端（client）：位于 MCP 主机内，帮助 LLM 与 MCP 服务器进行通信。它将 LLM 的请求转换为 MCP 可处理的格式，并将 MCP 的回复转换为 LLM 可理解的格式。它还会发现并调用可用的 MCP 服务器。
• MCP服务端（server）：MCP 服务器是为 LLM 提供上下文、数据和功能的外部服务。它通过连接数据库和 Web 服务等外部系统来帮助 LLM，将这些系统的响应转换为 LLM 可理解的格式，从而协助开发者提供多样化的功能。

2. 传输通信协议

MCP的传输通信协议建立在 JSON-RPC 2.0 之上，确保了消息格式的统一和规范。它支持两种主要的传输层模式：

• Streamable HTTP（推荐/默认）

  • 推出时间：2025年3月通过PR #206正式引入。
  • 状态：已取代HTTP+SSE成为默认的远程传输协议。
  • 核心优势：
    • 统一通信范式：同时支持流式和非流式传输。
    • 连接可恢复：支持断线重连和会话恢复。
    • 无状态设计：降低服务器压力，提高可扩展性。
    • 双向通信：客户端和服务端均可主动发起通信。

• stdio（标准输入/输出）

  • 适用场景：本地进程间通信。
  • 状态：仍然是本地部署的首选方案。
  • 核心特点：
    • 通过stdin/stdout进行JSON-RPC消息交换。
    • 消息以换行符分隔，使用UTF-8编码。
    • 低延迟、高吞吐量的本地通信。
    • 适合开发环境和单机部署。

• Legacy Support（已废弃）

  • HTTP + SSE：已标记为废弃，但为兼容性暂时保留。
  • 状态：不推荐在新项目中使用，将在未来版本移除。

3. 核心原语

MCP协议定义了三大核心原语：工具（Tools）、资源（Resources）和提示（Prompts），以及最新扩展-能力原语（Capabilities）。这四者共同构成了AI智能体（Agent）感知和操作世界的完整能力闭环。

3.1. 工具 (Tools)

• 定义：工具是AI模型可以调用的可执行函数。它们代表了AI改变外部世界、执行具体操作的能力。
• 比喻：就像人类的双手，可以用来“做”事情，比如拿起杯子、发送邮件、执行代码。
• 核心特征：
  • 可执行性：工具是用来“调用”和“执行”的，而不是用来“读取”的。
  • 需要授权：出于安全考虑，工具的调用通常需要用户明确授权或在安全沙箱中运行。
  • 自描述：工具的名称、功能描述、参数结构都是标准化的，AI模型可以动态发现并理解如何使用它们。
• 工作流程：
  1. 发现：MCP客户端向服务器查询可用的工具列表（如 list_tools）。
  2. 决策：AI模型根据用户需求和工具描述，决定调用哪个工具，并生成包含参数的结构化请求（如 call_tool(name="get_weather", arguments={"city": "北京"})）。
  3. 执行：MCP客户端将请求发送给服务器，服务器执行相应的函数（如调用天气API）。
  4. 返回：服务器将执行结果（如“晴，25℃”）返回给客户端。
• 代码示例：

  {
    "name": "send_email",
    "description": "发送电子邮件给指定收件人",
    "parameters": {
      "type": "object",
      "properties": {
        "to": {"type": "string", "description": "收件人邮箱地址"},
        "subject": {"type": "string", "description": "邮件主题"},
        "body": {"type": "string", "description": "邮件正文内容"}
      },
      "required": ["to", "subject", "body"]
    },
    "returns": {
      "type": "object",
      "properties": {
        "success": {"type": "boolean"},
        "message_id": {"type": "string"}
      }
    }
  }
  

3.2. 资源 (Resources)

• 定义：资源是AI模型可以访问的只读数据。它们为AI提供了完成任务所需的背景信息、上下文和知识。
• 比喻：就像人类的眼睛和耳朵，用来“看”和“听”，获取信息，但不能直接修改它们。
• 核心特征：
  • 只读性：AI只能读取资源内容，不能修改或删除。
  • 多样性：资源可以是静态文件、动态数据流、结构化数据库记录或非结构化文本。
  • URI标识：每个资源都有一个唯一的URI（统一资源标识符），方便AI引用和访问。
• 资源分类：
  • 静态资源：如产品手册、API文档、历史知识库，内容稳定，适合缓存。
  • 动态资源：如实时股价、在线用户状态、系统监控指标，需要按需获取最新数据。
  • 结构化资源：如数据库表、CSV文件，支持条件筛选和字段投影。
  • 非结构化资源：如用户聊天记录、文档全文、图片描述，需要AI自行解析语义。
• 工作流程：
  1. 发现：AI模型通过 list_resources 获取可用的资源列表。
  2. 请求：模型生成资源获取请求，指定资源URI和可选的筛选条件（如 resource.get(resource_id="user_orders", filter={"status": "paid"})）。
  3. 读取：服务器读取数据（如查询数据库、读取文件）。
  4. 返回：服务器将数据内容、元信息（如更新时间）和状态码打包返回。
• 典型使用场景：
  • 读取本地文件系统中的代码文件。
  • 查询PostgreSQL数据库中的用户表。
  • 获取某个网站的API响应内容。
  • 读取当前会话的聊天历史。
• 代码示例：

  {
    "type": "file",
    "name": "user_documents",
    "description": "用户文档集合",
    "schema": {
      "properties": {
        "id": {"type": "string"},
        "name": {"type": "string"},
        "content": {"type": "string"},
        "created_at": {"type": "datetime"}
      }
    },
    "capabilities": {
      "read": true,
      "write": false,
      "search": true,
      "filter": ["name", "created_at"]
    }
  }
  

3.3. 提示 (Prompts)

• 定义：提示是服务器提供的、包含固定框架和动态变量的可复用指令模板。它不是由AI生成的，而是由开发者预定义的，用于标准化特定任务的交互流程。
• 比喻：就像一个填好固定格式的“表单”或“剧本”，AI只需要填入动态变量，就能快速完成一个标准化任务。
• 核心特征：
  • 模板化：包含固定文本和 {{variable}} 占位符，支持动态填充。
  • 可复用：一次定义，多次调用，避免为同类交互重复编写指令。
  • 用户控制：通常由用户或客户端主动触发，作为引导AI行为的“快捷方式”。
• 工作流程：
  1. 定义：开发者在MCP服务器上创建一个提示模板，如 “请分析{{dataset_name}}中{{date_range}}的销售数据，输出{{format}}格式的报告”。
  2. 调用：用户在AI应用中选择这个预设的提示模板（如“生成周报”）。
  3. 填充：应用或用户填入变量（如 dataset_name="Q1_sales", date_range="1月至3月"）。
  4. 执行：填充后的完整提示被发送给AI模型，模型根据这个高度结构化的指令生成内容。
• 核心提示类型：
  • 系统提示（System Prompts）：定义AI角色、能力边界和行为准则。
  • 用户提示（User Prompts）：用户输入的自然语言指令。
  • 上下文提示（Context Prompts）：基于当前上下文生成的动态提示。
  • 工具提示（Tool Prompts）：指导如何调用特定工具的专用提示。
  • 错误提示（Error Prompts）：处理错误和异常情况的标准响应模板。
• 代码示例：

  {
    "type": "system_prompt",
    "name": "data_analyst_role",
    "template": "你是一位专业的数据分析师，专注于{{domain}}领域。你的任务是分析用户提供的数据，并提供专业、准确的见解。请使用{{language}}语言回复。",
    "parameters": {
      "domain": {"type": "string", "default": "通用"},
      "language": {"type": "string", "default": "中文"}
    },
    "constraints": {
      "max_length": 500,
      "tone": "professional"
    }
  }
  

3.4. 能力原语（Capabilities）- 最新扩展

Capabilities原语 在2026年被正式纳入核心原语体系，作为对三大原语的补充和增强。

• 定义：Capabilities原语定义了MCP服务器自身的能力和特性，帮助客户端理解和适配服务器的具体功能。
• 技术特性：
  • 能力发现：自动发现服务器支持的功能和限制。
  • 兼容性检查：版本兼容性验证和功能降级支持。
  • 性能指标：提供服务器性能指标和负载信息。
  • 安全特性：支持的加密算法、认证机制等安全特性。
  • 扩展能力：自定义扩展功能的注册和发现机制。
• 代码示例：

  {
    "capabilities": {
      "protocol_version": "2.1",
      "max_concurrent_requests": 100,
      "supported_transports": ["streamable_http", "stdio"],
      "authentication_methods": ["oauth2", "api_key"],
      "rate_limits": {
        "requests_per_minute": 60,
        "tokens_per_minute": 100000
      },
      "extensions": ["real_time_updates", "multi_modal_support"]
    }
  }
  

---

五、MCP的完整工作流程

如上图，我们通过一个用户查询“北京今天天气怎么样？”来理解MCP的完整调用流程：

1. 用户请求发起：用户在AI Agent（MCP Host）中提问：“北京今天天气怎么样？”。
2. LLM智能决策：
   • AI Agent（MCP Client）将用户问题、以及已连接的MCP服务器信息（如一个名为weather-server的服务器及其工具get_weather）一同发送给LLM。
   • LLM分析后，判断需要调用外部工具，并生成一个结构化的工具调用请求，例如：call_tool(server="weather-server", tool="get_weather")。
3. 工具调用执行：MCP Client根据LLM的指令，通过STDIO或SSE协议向weather-server发起get_weather工具的调用。
4. 结果获取：weather-server执行查询天气的外部API调用操作，然后将结果（如{"weather": "sunny, 25℃"}）返回给MCP Client。
5. 内容整合与最终输出：
   • MCP Client将原始问题和工具返回的天气结果再次提交给LLM。
   • LLM对信息进行整合、优化，生成最终的自然语言回答：“今天北京的天气是：晴，25℃”
   • 最终结果由AI Agent呈现给用户。

---

六、MCP开发实战完整代码

实现一个本地开发基于flask简单的mcp案例，功能是为用户提供旅游景点推荐。实际项目中tools的管理和注册略有不同。

server端：

# server.py
"""
MCP 旅游推荐系统 Server 服务。

该模块作为核心后端服务，负责处理景点数据查询、
用户偏好分析，并调用大模型（Qwen）生成个性化的旅游推荐文案。
"""

import time
from flask import Flask, request, jsonify
from typing import List, Dict, Any
from openai import OpenAI
import os

# 这里我读的是配好的系统环境变量，你们可以替换成自己的模型相关信息
BASE_URL = os.getenv("DASHSCOPE_BASE_URL")
API_KEY = os.getenv("DASHSCOPE_API_KEY")
LLM_MODEL = os.getenv("DASHSCOPE_LLM_MODEL")

client = OpenAI(
    api_key=API_KEY,
    base_url=BASE_URL
)

app = Flask(__name__)
app.config['JSON_AS_ASCII'] = False  # 确保JSON响应使用UTF-8编码

# 模拟景点数据库
SCENIC_SPOTS = {
    "北京": [
        {"name": "故宫", "description": "明清皇家宫殿，世界文化遗产", "rating": 4.8, "type": "历史"},
        {"name": "长城", "description": "世界七大奇迹之一，中国古代军事防御工程",
            "rating": 4.9, "type": "历史"},
        {"name": "颐和园", "description": "中国现存规模最大、保存最完整的皇家园林",
            "rating": 4.7, "type": "园林"},
    ],
    "上海": [
        {"name": "外滩", "description": "上海标志性景观，万国建筑博览群",
            "rating": 4.6, "type": "城市景观"},
        {"name": "东方明珠", "description": "上海地标性建筑，广播电视塔",
            "rating": 4.5, "type": "现代建筑"},
        {"name": "豫园", "description": "明代江南古典园林，历史悠久", "rating": 4.4, "type": "园林"},
    ],
    "广州": [
        {"name": "广州塔", "description": "世界第四高塔，广州新地标",
            "rating": 4.7, "type": "现代建筑"},
        {"name": "沙面岛", "description": "欧陆风情历史建筑群，拍照胜地", "rating": 4.6, "type": "历史"},
        {"name": "长隆旅游度假区", "description": "大型主题乐园，适合家庭游玩",
            "rating": 4.8, "type": "娱乐"},
    ],
    "成都": [
        {"name": "宽窄巷子", "description": "成都历史文化街区，体验老成都生活",
            "rating": 4.5, "type": "文化"},
        {"name": "青城山", "description": "道教名山，自然风光秀丽", "rating": 4.7, "type": "自然"},
        {"name": "大熊猫基地", "description": "世界最大的大熊猫保护研究机构",
            "rating": 4.9, "type": "动物"},
    ]
}


def get_scenic_spots(city: str) -> List[Dict[str, Any]]:
    """获取指定城市的景点列表"""
    return SCENIC_SPOTS.get(city, [])


def filter_spots_by_type(spots: List[Dict[str, Any]], spot_type: str) -> List[Dict[str, Any]]:
    """按类型筛选景点"""
    if not spot_type:
        return spots
    return [spot for spot in spots if spot_type.lower() in spot["type"].lower()]


def generate_recommendation_with_qwen3(city: str, spots: List[Dict[str, Any]], preferences: str) -> str:
    """使用Qwen3生成推荐文案"""
    if not spots:
        return f"抱歉，在{city}暂时没有找到符合您偏好的旅游景点。建议您可以考虑其他城市或调整偏好条件。"

    # 构建景点信息字符串
    spots_info = "\n".join([
        f"- {spot['name']} ({spot['type']}): {spot['description']} (评分: {spot['rating']})"
        for spot in spots[:5]  # 只取前5个
    ])

    prompt = f"""你是一位专业的旅游顾问，请根据以下信息为用户生成个性化的旅游景点推荐：

    城市：{city}
    用户偏好：{preferences if preferences else '无特定偏好'}

    推荐景点：
    {spots_info}

    请用温馨、专业的语气回复，突出景点的特色和适合人群，控制在200字以内。"""

    try:

        messages = [
            {"role": "system", "content": "你是一位专业的旅游顾问。"},
            {"role": "user", "content": prompt}
        ]

        response = client.chat.completions.create(
            model=LLM_MODEL,
            messages=messages,
            # 取值范围： [0, 2),temperature越高,生成的文本更多样,反之,生成的文本更确定。
            temperature=0.7,
            response_format={"type": "text"}  # 返回消息的格式
        )

        # 返回模型生成的文本
        msg = response.choices[0].message.content
        return msg
    except Exception as e:
        return f"生成推荐时发生异常: {str(e)}"


@app.route('/mcp/invoke', methods=['POST'])
def mcp_invoke():
    """MCP核心接口：处理用户请求"""
    try:
        data = request.get_json()
        if not data:
            return jsonify({"error": "无效的JSON请求"}), 400

        context_id = data.get("context_id", f"ctx_{int(time.time())}")
        user_input = data.get("user_input", "")

        if not user_input:
            return jsonify({
                "context_id": context_id,
                "response": "请输入您想要了解的旅游景点信息，例如'北京旅游推荐'或'上海有什么好玩的地方'。",
                "status": "success"
            }), 200

        # 简单解析用户输入，提取城市和偏好
        city = ""
        preferences = ""

        # 简单的关键词提取
        for c in ["北京", "上海", "广州", "成都", "杭州", "西安", "深圳"]:
            if c in user_input:
                city = c
                break

        # 提取偏好关键词
        preference_keywords = ["历史", "文化", "自然", "美食", "购物", "拍照", "家庭", "情侣"]
        for keyword in preference_keywords:
            if keyword in user_input:
                preferences += f"{keyword} "

        preferences = preferences.strip()

        if not city:
            return jsonify({
                "context_id": context_id,
                "response": "请告诉我您想了解哪个城市的旅游景点，例如'北京旅游推荐'或'上海有什么好玩的地方'。",
                "status": "success"
            }), 200

        # 调用工具获取景点数据
        spots = get_scenic_spots(city)

        # 根据偏好筛选
        if preferences:
            filtered_spots = filter_spots_by_type(spots, preferences)
            if not filtered_spots and spots:
                filtered_spots = spots  # 如果没有匹配的，使用所有景点
        else:
            filtered_spots = spots

        # 使用Qwen3生成推荐
        recommendation = generate_recommendation_with_qwen3(
            city, filtered_spots, preferences)

        # 构建工具结果
        tool_results = [{
            "tool_name": "get_scenic_spots",
            "city": city,
            "preferences": preferences,
            "spots_found": len(filtered_spots),
            "spots": filtered_spots
        }]

        response_data = {
            "context_id": context_id,
            "response": recommendation,
            "tool_results": tool_results,
            "status": "success"
        }

        return jsonify(response_data), 200

    except Exception as e:
        return jsonify({
            "error": f"服务器错误: {str(e)}",
            "status": "error"
        }), 500


@app.route('/health', methods=['GET'])
def health_check():
    """健康检查接口"""
    return jsonify({
        "status": "healthy",
        "timestamp": time.time(),
        "service": "mcp_travel_server"
    }), 200


if __name__ == "__main__":
    print("Starting MCP Travel Server on http://localhost:8000")
    print("Available cities: 北京, 上海, 广州, 成都")
    print("Example request: POST /mcp/invoke with JSON body")
    app.run(host="0.0.0.0", port=8000, debug=True, threaded=True)

host端：

# host.py
"""
MCP 旅游推荐系统 Host 服务。

该模块作为 Flask 后端服务，负责接收客户端请求，
协调与 MCP Server 的交互，并管理用户会话状态。
"""

import time
import requests
from flask import Flask, request, jsonify
import uuid

app = Flask(__name__)
app.config['JSON_AS_ASCII'] = False  # 确保JSON响应使用UTF-8编码

# 配置
SERVER_URL = "http://localhost:8000/mcp/invoke"
HOST_PORT = "8001"

# 会话管理 - 使用内存字典存储会话
sessions = {}


def create_context_id(session_id: str) -> str:
    """创建上下文ID"""
    return f"ctx_{session_id}_{int(time.time())}"


@app.route('/travel/recommend', methods=['POST'])
def get_recommendation():
    """主机端核心接口：协调client和server"""
    try:
        data = request.get_json()
        if not data:
            return jsonify({"error": "无效的JSON请求"}), 400

        user_input = data.get("user_input", "")
        session_id = data.get("session_id")

        if not user_input:
            return jsonify({
                "error": "请输入旅游需求",
                "session_id": session_id or str(uuid.uuid4())
            }), 400

        # 获取或创建会话
        if not session_id:
            session_id = str(uuid.uuid4())
            sessions[session_id] = {
                "created_at": time.time(),
                "requests": []
            }
        elif session_id not in sessions:
            sessions[session_id] = {
                "created_at": time.time(),
                "requests": []
            }

        # 创建上下文ID
        context_id = create_context_id(session_id)

        # 构建MCP请求
        mcp_request = {
            "context_id": context_id,
            "user_input": user_input,
            "tools": [{
                "tool_name": "get_scenic_spots",
                "parameters": {"city": user_input}
            }]
        }

        # 调用Server
        try:
            server_response = requests.post(
                SERVER_URL,
                json=mcp_request,
                timeout=300
            )

            if server_response.status_code != 200:
                return jsonify({
                    "error": f"Server returned status {server_response.status_code}: {server_response.text}",
                    "session_id": session_id
                }), 502

            result = server_response.json()

        except requests.exceptions.RequestException as e:
            return jsonify({
                "error": f"Server communication error: {str(e)}",
                "session_id": session_id
            }), 502

        # 保存请求历史
        sessions[session_id]["requests"].append({
            "timestamp": time.time(),
            "user_input": user_input,
            "context_id": context_id,
            "server_response": result
        })

        # 构建主机响应
        response_data = {
            "session_id": session_id,
            "response": result.get("response", "无推荐结果"),
            "context_data": {
                "context_id": context_id,
                "tool_results": result.get("tool_results", []),
                "server_status": result.get("status", "unknown")
            },
            "timestamp": time.time()
        }

        return jsonify(response_data), 200

    except Exception as e:
        return jsonify({
            "error": f"Host processing error: {str(e)}",
            "session_id": session_id if 'session_id' in locals() else None
        }), 500


@app.route('/session/<session_id>', methods=['GET'])
def get_session(session_id):
    """获取会话详情"""
    if session_id not in sessions:
        return jsonify({"error": "Session not found"}), 404

    session_data = sessions[session_id]
    return jsonify({
        "session_id": session_id,
        "created_at": session_data["created_at"],
        "request_count": len(session_data["requests"]),
        "last_request": session_data["requests"][-1] if session_data["requests"] else None
    }), 200


@app.route('/health', methods=['GET'])
def health_check():
    """健康检查"""
    return jsonify({
        "status": "healthy",
        "host_port": HOST_PORT,
        "server_url": SERVER_URL,
        "active_sessions": len(sessions),
        "timestamp": time.time(),
        "service": "mcp_travel_host"
    }), 200


@app.route('/sessions/cleanup', methods=['POST'])
def cleanup_sessions():
    """清理过期会话"""
    current_time = time.time()
    expired_sessions = []

    for session_id, session_data in list(sessions.items()):
        # 清理1小时未活动的会话
        if current_time - session_data["created_at"] > 3600:
            expired_sessions.append(session_id)
            del sessions[session_id]

    return jsonify({
        "message": f"清理了 {len(expired_sessions)} 个过期会话",
        "expired_sessions": expired_sessions,
        "remaining_sessions": len(sessions)
    }), 200


if __name__ == "__main__":
    print(f"Starting MCP Travel Host on http://localhost:{HOST_PORT}")
    print(f"Connected to server: {SERVER_URL}")
    app.run(host="0.0.0.0", port=HOST_PORT, debug=True, threaded=True)

client端：

# client.py
"""
MCP 旅游推荐系统客户端。

该模块提供了一个命令行界面，用于与后端旅游推荐服务交互。
支持获取景点推荐、查看会话信息等功能。
"""

import requests
import json
import time
from typing import Dict, Any
import os

# 配置
HOST_URL = "http://localhost:8001/travel/recommend"
# CLIENT_SESSION_ID = None


class TravelClient:
    """
    旅游推荐客户端，用于与后端服务交互获取推荐结果和会话信息。

    参数：
    - host_url: 后端服务URL
    """

    def __init__(self, host_url: str):
        self.host_url = host_url
        self.session_id = None

    def get_recommendation(self, user_input: str) -> Dict[str, Any]:
        """获取旅游推荐"""
        payload = {
            "user_input": user_input,
            "session_id": self.session_id
        }

        try:
            response = requests.post(
                self.host_url,
                json=payload,
                timeout=300
            )

            if response.status_code != 200:
                return {
                    "error": f"请求失败: HTTP {response.status_code} - {response.text}",
                    "session_id": self.session_id
                }

            return response.json()

        except requests.exceptions.RequestException as e:
            return {
                "error": f"网络请求失败: {str(e)}",
                "session_id": self.session_id
            }

    def show_session_info(self) -> Dict[str, Any]:
        """显示会话信息"""
        if not self.session_id:
            return {"error": "没有活跃的会话"}

        try:
            response = requests.get(
                f"http://localhost:8001/session/{self.session_id}",
                timeout=300
            )

            if response.status_code != 200:
                return {
                    "error": f"获取会话信息失败: HTTP {response.status_code} - {response.text}"
                }

            return response.json()
        except requests.exceptions.RequestException as e:
            return {"error": f"获取会话信息失败: {str(e)}"}


def print_welcome():
    """显示欢迎信息"""
    print("""
    MCP旅游景点推荐系统 (Flask版)
    
    支持城市：北京、上海、广州、成都
    支持偏好：历史、文化、自然、美食、购物、拍照、家庭、情侣
    
    示例输入：
    - "北京旅游推荐"
    - "上海有什么历史景点"
    - "广州适合家庭游玩的地方"
    - "成都美食景点推荐"
    
    命令：
    /session - 查看当前会话信息
    /exit    - 退出程序
    
    请输入您的需求：
    """)


def format_response(response_data: Dict[str, Any]) -> str:
    """格式化响应输出"""
    if "error" in response_data:
        return f"错误: {response_data['error']}"

    result = f"推荐结果：\n\n{response_data['response']}\n"

    if response_data.get("context_data", {}).get("tool_results"):
        tool_results = response_data["context_data"]["tool_results"][0]
        if tool_results.get("spots"):
            result += "\n找到的相关景点：\n"
            for i, spot in enumerate(tool_results["spots"][:3], 1):
                result += f"  {i}. {spot['name']} - {spot['description']}\n"
                result += f"     类型: {spot['type']} | 评分: {spot['rating']}\n"

    result += f"\n会话ID: {response_data['session_id']}"
    return result


def main():
    """主程序"""

    # global CLIENT_SESSION_ID

    client = TravelClient(HOST_URL)
    print_welcome()

    while True:
        try:
            user_input = input("\n> ").strip()

            if not user_input:
                continue

            # 处理命令
            if user_input.startswith("/"):
                if user_input.lower() == "/exit":
                    print("感谢使用MCP旅游推荐系统！")
                    break
                elif user_input.lower() == "/session":
                    session_info = client.show_session_info()
                    print("\n会话信息：")
                    print(json.dumps(session_info, indent=2, ensure_ascii=False))
                    continue
                else:
                    print(f"未知命令: {user_input}")
                    print("可用命令: /session, /exit")
                    continue

            # 获取推荐
            print("\n正在为您生成推荐，请稍候...")
            start_time = time.time()

            response = client.get_recommendation(user_input)

            elapsed_time = time.time() - start_time
            print(f"\n响应时间: {elapsed_time:.2f}秒")

            # 显示结果
            formatted_response = format_response(response)
            print("\n" + formatted_response)

        except KeyboardInterrupt:
            print("\n感谢使用MCP旅游推荐系统！")
            break
        except Exception as e:
            print(f"发生错误: {str(e)}")
            import traceback
            print(traceback.format_exc())


if __name__ == "__main__":
    # 检查环境变量
    if not os.getenv("DASHSCOPE_API_KEY"):
        print("警告: 未设置DASHSCOPE_API_KEY环境变量")
        print("请在运行前设置: export DASHSCOPE_API_KEY='your_api_key'")
        print("或者在.env文件中添加: DASHSCOPE_API_KEY=your_api_key")
        confirm = input("是否继续运行(使用示例模式)? (y/n): ")
        if confirm.lower() != 'y':
            exit(0)

    # 检查服务是否可用
    try:
        test_response = requests.get(
            "http://localhost:8000/health", timeout=10)
        if test_response.status_code != 200:
            print(f"警告: MCP Server服务未正常运行 (状态码: {test_response.status_code})")
    except requests.exceptions.ConnectionError:
        print("警告: 无法连接到MCP Server (http://localhost:8000)")

    try:
        test_response = requests.get(
            "http://localhost:8001/health", timeout=10)
        if test_response.status_code != 200:
            print(f"警告: MCP Host服务未正常运行 (状态码: {test_response.status_code})")
    except requests.exceptions.ConnectionError:
        print("警告: 无法连接到MCP Host (http://localhost:8001)")
        print("请先启动server.py和host.py服务")
        exit(1)

    # 启动客户端
    main()

tools本地模拟知识库或联网查询：

# tools.py
"""
MCP 旅游推荐系统工具模块。

该模块定义了旅游相关的核心工具函数，包括获取景点信息、
根据偏好筛选景点以及获取天气信息等。
"""

from typing import Dict, List, Any
import time


class TravelTools:
    """旅游相关的工具集合"""

    @staticmethod
    def get_scenic_spots(city: str) -> Dict[str, Any]:
        """
        获取城市景点信息

        Args:
            city: 城市名称

        Returns:
            包含景点信息的字典
        """
        # 实际项目中这里会调用真实API或数据库
        # 这里使用模拟数据
        mock_data = {
            "北京": [
                {"name": "故宫", "description": "明清皇家宫殿，世界文化遗产",
                    "rating": 4.8, "type": "历史"},
                {"name": "长城", "description": "世界七大奇迹之一，中国古代军事防御工程",
                    "rating": 4.9, "type": "历史"},
                {"name": "颐和园", "description": "中国现存规模最大、保存最完整的皇家园林",
                    "rating": 4.7, "type": "园林"},
            ],
            "上海": [
                {"name": "外滩", "description": "上海标志性景观，万国建筑博览群",
                    "rating": 4.6, "type": "城市景观"},
                {"name": "东方明珠", "description": "上海地标性建筑，广播电视塔",
                    "rating": 4.5, "type": "现代建筑"},
                {"name": "豫园", "description": "明代江南古典园林，历史悠久",
                    "rating": 4.4, "type": "园林"},
            ],
            "广州": [
                {"name": "广州塔", "description": "世界第四高塔，广州新地标",
                    "rating": 4.7, "type": "现代建筑"},
                {"name": "沙面岛", "description": "欧陆风情历史建筑群，拍照胜地",
                    "rating": 4.6, "type": "历史"},
                {"name": "长隆旅游度假区", "description": "大型主题乐园，适合家庭游玩",
                    "rating": 4.8, "type": "娱乐"},
            ],
            "成都": [
                {"name": "宽窄巷子", "description": "成都历史文化街区，体验老成都生活",
                    "rating": 4.5, "type": "文化"},
                {"name": "青城山", "description": "道教名山，自然风光秀丽",
                    "rating": 4.7, "type": "自然"},
                {"name": "大熊猫基地", "description": "世界最大的大熊猫保护研究机构",
                    "rating": 4.9, "type": "动物"},
            ]
        }

        return {
            "city": city,
            "spots": mock_data.get(city, []),
            "total": len(mock_data.get(city, [])),
            "timestamp": time.time()
        }

    @staticmethod
    def filter_spots_by_preference(spots: List[Dict[str, Any]], preference: str) -> List[Dict[str, Any]]:
        """
        根据偏好筛选景点

        Args:
            spots: 景点列表
            preference: 偏好类型 (历史、自然、美食等)

        Returns:
            筛选后的景点列表
        """
        if not preference or not spots:
            return spots

        preference_lower = preference.lower()
        filtered = []

        for spot in spots:
            if (preference_lower in spot["type"].lower() or
                    preference_lower in spot["description"].lower()):
                filtered.append(spot)

        return filtered if filtered else spots[:3]  # 如果没有匹配，返回前3个

    @staticmethod
    def get_weather_info(city: str) -> Dict[str, Any]:
        """
        获取天气信息 (示例工具)

        Args:
            city: 城市名称

        Returns:
            天气信息
        """
        # 实际项目中会调用天气API
        return {
            "city": city,
            "temperature": 25,
            "condition": "晴朗",
            "recommendation": "天气适宜出游"
        }

运行步骤：

• 1.启动Server端服务:

  • 在vscode中新建一个终端执行命令：python server.py
  • 访问 http://localhost:8000/health 验证服务

• 2.启动Host端服务:

  • 在vscode中新建一个终端执行命令：python host.py
  • 访问 http://localhost:8001/health 验证服务

• 3.启动Client端进行交互:

  • 在vscode中新建一个终端执行命令：python client.py
  • 客户端启动成功，根据客户端提示交互

---

七、MCP与Prompt 、Function Call、RAG差异

MCP (模型上下文协议)，RAG（检索生成增强），Function Call (函数调用)，Prompt (提示词) 是AI应用开发中几种不同层面、不同定位的技术手段。简单来说，这四者是构建现代AI智能体（Agent）的四大支柱，它们分别解决了AI的如何思考、如何获取知识、如何行动以及如何标准化连接的问题。

我们可以用一个生动的比喻来理解：将AI智能体想象成一位超级助理。

• LLM (大模型)： 是助理的大脑，负责核心的思考、推理与决策。
• Prompt：是你给助理的清晰指令，告诉他要做什么、怎么做。
• RAG：是助理的动态知识库，让他能查阅最新的资料，避免凭空捏造。
• Function Calling：是助理的手和脚，让他能实际操作工具、执行任务。
• MCP：则是连接大脑与手脚的标准化神经系统和USB接口，确保所有工具都能即插即用，高效协作。

差异对比总结

特性维度	Prompt (提示词)	RAG (检索增强生成)	Function Calling (函数调用)	MCP (模型上下文协议)
核心角色	指令与控制	外部知识库	行动与执行	标准化连接协议
解决痛点	LLM输出模糊、不可控	知识滞后、幻觉、无法访问私有数据	LLM无法与现实世界交互	工具集成碎片化、N×M集成难题
作用对象	LLM的推理与生成过程	LLM的输入上下文	LLM的输出行为	整个AI应用与外部世界的连接层
工作方式	通过自然语言设定角色、规则和示例	检索外部数据 → 注入Prompt → LLM生成	LLM生成结构化调用请求 → 外部程序执行 → 结果返回LLM	定义标准化的Server/Client/Transport层，实现即插即用
通俗比喻	给助理的工作备忘录	助理的参考书和数据库	助理的手和脚	连接大脑与手脚的USB接口和神经系统
关系	基础：所有技术都依赖Prompt来引导	补充：为Prompt提供动态知识	扩展：让Prompt的指令能被物理执行	使能者：为RAG和Function Calling提供标准化实现框架

---

八、MCP的生态现状

截至目前，MCP已迅速从一个概念演变为行业事实标准，应用场景极其广泛：

• 软件开发：在Cursor、Zed、VS Code等IDE中，MCP让AI助手可以直接读取代码库、执行终端命令、查询数据库架构、甚至操作DevOps工具，成为真正的“结对编程”伙伴。
• 企业级智能助理：Block（Square）等公司利用MCP让内部AI助手安全地访问CRM、知识库和专有文档，处理订单查询、客户服务等业务。
• 数据分析与科研：AI代理通过MCP连接专业的数据分析软件（如Tableau）、科学仪器接口，实现从数据查询到可视化报告生成的全流程自动化。
• 办公自动化：AI助理可以跨平台连接日历、邮件、网盘和笔记软件，根据用户指令自动安排会议、整理文档、预订行程。
• 物联网(IoT)控制：智能家居设备可以被封装成MCP服务器，让AI模型成为整个家庭的统一语音控制中心。

生态现状：

• 巨头支持：除了发起者Anthropic，Google、OpenAI、微软等大厂已纷纷宣布在其产品（如ChatGPT桌面版、Agents SDK、Windows 11）中原生支持MCP。
• 丰富的服务器：社区已涌现出海量的开源MCP服务器，覆盖了Google Drive、Slack、GitHub、PostgreSQL、Stripe、AWS S3等常见服务，并在持续快速增长。
• 开发工具成熟：Anthropic及社区提供了Python、TypeScript、Java、C#等多种语言的MCP SDK，大大降低了开发者的接入门槛。

---

九、MCP未来

MCP协议的出现，标志着大模型技术正从能说会道的聊天机器人向能做事的智能体演进。它不是要取代大模型，而是为其装上“手”和“眼”，让它能够安全、标准、高效地连接和操作整个数字世界。

如果说大模型是AI时代的“操作系统内核”，那么MCP就是这个操作系统的“USB-C接口”和“神经系统”。它通过统一的连接标准，正在构建一个开放、繁荣的AI工具生态，是通往通用人工智能（AGI）的关键基础设施。对于任何希望构建下一代AI应用的开发者和企业来说，理解并拥抱MCP已不再是“可选项”，而是“必选项”。