一、概念
MCP（Model Context Protocol，模型上下文协议），是由Anthropic公司在2024年11月推出的一种开放标准，旨在统一大语言模型（LLM）与外部数据源、工具之间的通信方式。

你可以把它理解为AI应用领域的“USB-C接口”或“万能插座”。在它出现之前，每让AI接入一个新工具（如数据库、API），都需要编写一套定制代码，非常低效。有了MCP，所有遵循此协议的AI模型和工具都能用同一种“语言”交流，解决了“信息孤岛”问题，让AI能够真正与真实世界交互，而不仅仅是聊天。

它解决了什么“原生”问题？
大语言模型（LLM）本身是一个“大脑”，但它有两个天生的局限：
信息滞后：它的知识只截止到训练时，无法获取实时数据（比如今天的天气）。
无法行动：它只能输出文字，无法直接操作外部系统（比如发邮件、查数据库）。

在没有MCP之前，要让AI做这两件事，开发者需要为每一个外部数据源或工具编写定制化的“胶水代码”。例如，连接A数据库写一套代码，连接B API又写另一套，效率低且难以复用。MCP的出现就是为了终结这种“重复造轮子”的局面。

二、MCP能做什么？（三大核心能力）
MCP定义了三种标准化的交互方式，让AI不仅能“说”，还能“做”：
Resources（资源）：让AI能够读取只读数据。比如读取你电脑上的本地文件、查询企业数据库或获取知识库内容。
Tools（工具）：让AI能够主动执行操作。比如帮你发一封邮件、运行一段代码、或者调用第三方API服务。
Prompts（提示模板）：让用户可以复用工作流。比如预设好的代码生成模板、报告撰写模板等，方便一键调用。

三、MCP是如何工作的？（核心架构）
MCP采用了一个经典的“客户端-主机-服务器”架构，你可以把它理解为一个三方协作的过程：
MCP Host（主机）：就是你直接使用的AI应用环境，比如ChatGPT、Claude或某个AI编程工具。它负责接收你的问题，并展示最终结果。
MCP Client（客户端）：它是主机内部的“翻译官”或“通信中间件”。它负责把你的需求翻译成标准化的指令，发送给外部工具。
MCP Server（服务器）：它是外部能力的提供者，也就是那个“适配器”。它封装了各种具体的工具（如发送邮件、查询数据库）或数据（如本地文件），并等待客户端的调用。

举个简单的例子：
当你在AI助手中问“今天杭州天气如何？”时：
主机（AI应用）理解了你的意图。
客户端（翻译官）将请求发送给专门提供天气服务的MCP Server。
天气MCP Server查询到数据后，将结果返回给客户端。
客户端把数据交还给主机，AI最终用自然语言回答你。

四、代码实现
整个流程分为三个文件，请你在电脑上创建一个新文件夹（比如 weather_mcp），然后跟着我一步步创建

第一步：在文件夹里新建一个文件，命名为 main.py，复制以下代码：

# 导入 os 模块：用于访问操作系统相关的功能（虽然这个文件里没直接用到，但预留着方便后续扩展）
import os

# 导入 json 模块：用于处理 JSON 数据的编码和解码
# 这里主要用来把 DeepSeek 返回的工具调用参数（字符串格式）转成 Python 字典
import json

# 导入 asyncio 模块：Python 的异步编程框架
# MCP 的客户端通信是异步的（非阻塞的），所以必须用 asyncio 来运行
import asyncio

# 从 openai 库中导入 OpenAI 类
# 注意：虽然叫 openai，但它兼容 DeepSeek 的 API 接口（DeepSeek 的 API 格式和 OpenAI 一致）
from openai import OpenAI

# 从 mcp 库中导入三个核心组件：
# ClientSession  —— 客户端会话对象，用来和 MCP 服务器收发消息
# StdioServerParameters  —— 配置本地 MCP 服务器的参数（命令、参数等）
from mcp import ClientSession, StdioServerParameters

# 从 mcp.client.stdio 模块中导入 stdio_client 函数
# 它负责通过"标准输入/标准输出"（Stdio）建立客户端和服务器之间的通信管道
from mcp.client.stdio import stdio_client

# 初始化 DeepSeek 客户端（记得替换成你自己的 API Key）
client = OpenAI(
    api_key="sk-XXX",   #填写你自己的DeepSeek账号的API key
    base_url="https://api.deepseek.com/v1"
    # 【语法错误提示】这里缺少一个右括号 ")"，应该是 OpenAI(...) 后面再加一个 )
)

# 定义一个异步函数 main()，这是整个 AI 主机的核心入口
# 用 async 修饰是因为里面会进行异步网络请求（和 MCP 服务器通信）
async def main():
    # 1. 配置连接我们刚刚写的本地天气 MCP Server
    # StdioServerParameters 用来告诉客户端："怎么启动天气服务器"
    server_params = StdioServerParameters(
        command="py",           # 指定用 py 命令来运行（Windows 上 py 会调用默认 Python 版本）
        args=["weather_server.py"]  # 传入参数：要运行的文件名
    )

    # 2. 启动客户端并连接
    # stdio_client(server_params) 会：
    #   - 根据上面的配置，在后台启动一个 weather_server.py 的子进程
    #   - 创建一个双向通信管道（read 负责接收数据，write 负责发送数据）
    # async with 确保通信结束后自动关闭连接
    async with stdio_client(server_params) as (read, write):
        # ClientSession 封装了 MCP 协议的完整对话逻辑
        # 它基于 read/write 管道，自动处理 JSON-RPC 消息的发送和接收
        async with ClientSession(read, write) as session:
            # 发送初始化握手消息，告诉服务器"我准备好了"
            # 服务器收到后也会回一个初始化响应，双方确认协议版本等
            await session.initialize()
            
            # 获取天气工具的描述，准备发给 DeepSeek
            # list_tools() 会问 MCP 服务器："你有哪些工具可以给我用？"
            # 服务器会返回一份 JSON 描述，包含工具名、描述、参数格式等
            tools_response = await session.list_tools()
            
            # 把 MCP 服务器返回的工具描述，转成 OpenAI 兼容的格式
            # DeepSeek 的 chat.completions API 要求工具描述必须是特定的 JSON 结构
            tools = [
                {
                    "type": "function",            # 声明这是一个"函数调用"类型的工具
                    "function": {
                        "name": tool.name,                    # 工具名称，比如 "get_weather"
                        "description": tool.description,       # 工具说明，AI 靠它决定是否调用
                        "parameters": tool.inputSchema         # 参数 schema，告诉 AI 该传什么参数
                    }
                }
                for tool in tools_response.tools   # 遍历服务器返回的所有工具
            ]

            # 3. 模拟用户提问："今天杭州天气如何？"
            # messages 是和大模型对话的"消息列表"，大模型根据历史消息来理解上下文
            messages = [
                # system 消息：设定 AI 的角色和行为准则，优先级最高
                {"role": "system", "content": "你是一个贴心的天气助手，请根据工具提供的数据回答用户。"},
                # user 消息：用户的真实提问
                {"role": "user", "content": "今天杭州天气如何？"}
            ]

            # 4. 把问题连同工具列表发给 DeepSeek
            # 这一步是关键：把用户的问题 + 可用工具的描述一起发给大模型
            # 让大模型"看到"有哪些工具可以用，它自己决定要不要调用
            response = client.chat.completions.create(
                model="deepseek-chat",   # 指定使用 DeepSeek 的聊天模型
                messages=messages,        # 对话消息
                tools=tools,              # 可用工具列表（OpenAI 兼容格式）
                tool_choice="auto"        # 让模型自己决定要不要调用工具（不强制）
            )
            
            # 从响应中取出 AI 回复的消息对象
            msg = response.choices[0].message

            # 5. DeepSeek 决定调用工具，我们执行并返回结果
            # tool_calls 是一个列表，包含 DeepSeek 决定调用的工具及其参数
            # 如果为 None 或空列表，说明 DeepSeek 觉得不需要调用工具，直接回答了
            if msg.tool_calls:
                # 把 DeepSeek 的回复（包含工具调用指令）追加到消息历史中
                # 这样后续 DeepSeek 就能看到"我刚才决定调用什么工具了"
                messages.append(msg)
                
                # 遍历所有工具调用（一般只有 1 个，但 API 支持多个）
                for tool_call in msg.tool_calls:
                    # 调用天气 MCP Server 的对应工具
                    # tool_call.function.name  → 工具名，如 "get_weather"
                    # tool_call.function.arguments  → 参数字符串，如 '{"city": "杭州"}'
                    result = await session.call_tool(
                        tool_call.function.name, 
                        json.loads(tool_call.function.arguments)  # 把 JSON 字符串转成字典
                    )
                    # 把查到的天气数据交还给 DeepSeek
                    # role 设为 "tool"，表示这是工具执行的结果
                    # tool_call_id 关联到上面那条工具调用，让 DeepSeek 知道这是哪个工具的返回值
                    messages.append({
                        "role": "tool",
                        "tool_call_id": tool_call.id,
                        "content": result.content[0].text  # 提取工具返回的文本内容
                    })
                
                # 6. DeepSeek 拿到天气数据后，生成最终的自然语言回复
                # 把完整的对话历史（用户问题 + AI 调用工具 + 工具结果）再发给 DeepSeek
                # 这次不带 tools 参数了，因为 AI 已经拿到数据，直接用自然语言回答即可
                final_response = client.chat.completions.create(
                    model="deepseek-chat",
                    messages=messages
                )
                print("🤖 AI助手最终回复:", final_response.choices[0].message.content)

# Python 的标准入口判断
# 只有直接运行这个文件时（python main.py），才会执行下面的代码
# 如果被其他文件 import 导入，则不会自动运行
if __name__ == "__main__":
    # 用 asyncio.run() 启动异步事件循环，运行 main() 函数
    asyncio.run(main())

第二步：在文件夹里新建一个文件，命名为 weather_server.py，复制以下代码：

# 导入 Python 的系统模块（sys），用于访问与 Python 解释器紧密相关的变量和函数
import sys

# 导入 builtins 模块，它包含了 Python 的所有内置函数（比如我们常用的 print）
import builtins

# 【核心修复】：将所有 print 输出重定向到 stderr，保持 stdout 只传输 MCP 协议数据
# 解释：MCP 客户端和服务器之间是通过“标准输出(stdout)”来传递 JSON 数据的。
# 如果你的服务器代码里不小心写了一个 print("调试信息")，
# 这个文本也会混在 stdout 里发给客户端，客户端收到后会因为格式错误而崩溃。
# 所以，我们把内置的 print 函数替换掉，强制让它把内容打印到“标准错误(stderr)”里。
# 这样既不影响调试，又保证了 stdout 的绝对纯净。
_orig_print = builtins.print  # 先把原本的 print 函数保存起来，防止丢失
builtins.print = lambda *a, **k: _orig_print(*a, file=sys.stderr, **k) 
# 用 lambda 匿名函数重新定义 print：无论传入什么参数，都调用原本的 print，
# 但强制指定输出目标为 sys.stderr（标准错误输出流）

# 导入 asyncio 模块，用于支持异步编程（MCP 服务器底层是异步运行的）
import asyncio

# 从 fastmcp 库中导入 FastMCP 类。
# FastMCP 是一个高级封装库，能帮我们省去大量复杂的底层配置，轻松创建 MCP 服务器
from fastmcp import FastMCP

# 初始化一个名为 "Weather-Server" 的 MCP 服务器实例
# 这个名字会作为标识符，当客户端连接时，能知道连的是哪个服务
mcp = FastMCP("Weather-Server")

# 使用 @mcp.tool() 装饰器将下面的函数注册为一个“工具”
# 加上这个装饰器后，FastMCP 会自动提取函数的名字、参数类型和注释，
# 生成一份标准的 JSON 格式工具描述，发给 AI 大模型看
@mcp.tool()
async def get_weather(city: str) -> str:
    """
    获取指定城市的当前天气情况。
    参数: city - 城市名称，例如：杭州、北京
    """
    # 【注意】：这里的三引号注释（Docstring）非常重要！
    # AI 大模型就是靠阅读这段文字来理解“这个工具是干嘛的”以及“参数该怎么填”的。
    # 写得越清晰，AI 调用的准确率就越高。

    # 这里我们模拟天气数据，不需要真实联网
    # 创建一个字典，键是城市名，值是天气描述
    fake_weather_data = {
        "杭州": "晴天，气温 28℃，微风，非常适合散步。",
        "北京": "多云，气温 25℃，空气质量良。",
        "上海": "小雨，气温 22℃，出门记得带伞。"
    }
    
    # 使用字典的 get 方法安全地获取数据：
    # 如果传入了字典里有的城市，就返回对应的天气；
    # 如果传入了没有的城市，就返回一句友好的抱歉提示，避免程序报错崩溃
    return fake_weather_data.get(city, f"抱歉，暂时无法获取 {city} 的天气信息。")

# Python 的标准入口判断：
# 只有当这个文件被直接运行（比如 python weather_server.py）时，才会执行下面的代码。
# 如果是被其他文件 import 导入，则不会执行，防止意外启动服务器。
if __name__ == "__main__":
    # 调用 mcp.run() 启动服务器。
    # 此时服务器会进入“监听状态”，等待客户端（你的 AI 主机）通过 stdin 发送指令
    mcp.run()

第三步：运行测试
打开终端（命令行），进入你的 weather_mcp 文件夹。
确保安装了依赖（如果之前装过可以跳过）：
pip install openai mcp
运行主程序：
python main.py

执行过程分析：
为了让你更直观地理解这两个文件是如何配合工作的，我们可以把整个过程想象成“老板（main.py）雇佣了一个专职员工（weather_server.py）去查资料，然后老板亲自写总结”。
以下是 main.py 和 weather_server.py 从启动到结束的详细执行过程：

1. 启动阶段：建立连接
   main.py 启动：当你运行 python main.py 时，它首先初始化了 DeepSeek 客户端。接着，它通过 stdio_client 在后台启动了一个新的子进程，也就是运行 python weather_server.py。
   weather_server.py 启动：它运行到 mcp.run() 时，并没有像普通程序那样执行完就退出，而是进入了“静默监听状态”。它不打印任何内容，只是安静地等待 main.py 通过标准输入（stdin）发送指令。
   双方握手：main.py 发送初始化请求，weather_server.py 收到后回复确认。双方确认通信协议无误，成功建立连接。
2. 准备阶段：获取“员工技能清单”
   main.py 询问：main.py 向 weather_server.py 发送 list_tools 请求，意思是：“你都能干什么？”
   weather_server.py 汇报：它检查自己身上带 @mcp.tool() 装饰器的函数，把 get_weather 的名称、作用（Docstring）、需要的参数（city: str）打包成一份标准的 JSON 格式，返回给 main.py。
   main.py 翻译：main.py 收到这份 JSON 后，把它转换成 DeepSeek 能看懂的格式，准备发给 AI。
3. 决策阶段：AI 大脑思考
   main.py 提问：main.py 把用户的提问（“今天杭州天气如何？”）连同刚才拿到的“工具清单”一起发送给 DeepSeek。
   DeepSeek 决策：DeepSeek 看到问题后，发现需要查天气，并且工具清单里刚好有 get_weather 这个工具。于是它不直接回答，而是返回一条特殊的指令：“我需要调用 get_weather 工具，参数是 city=“杭州””。
4. 执行阶段：员工干活
   main.py 派发任务：main.py 拿到 DeepSeek 的指令，立刻通过 MCP 协议向 weather_server.py 发送调用请求。
   weather_server.py 查数据：weather_server.py 收到请求，执行 get_weather(“杭州”) 函数。它去字典里查到了“晴天，气温 28℃…”这段模拟数据，并将其返回给 main.py。