Anthropic MCP发布通告：https://www.anthropic.com/news/model-context-protocol

MCP GitHub主页：https://github.com/modelcontextprotocol

MCP技术体系介绍

MCP

MCP（Model Context Protocol，模型上下文协议）是由Anthropic公司提出的一种开源协议，旨在让AI模型能够无缝连接外部工具和资源，就像一把“万能钥匙”，打通了AI与现实世界的交互通道。MCP发布于2024年11月，但起初并未引起太多关注，随着今年智能体（Agent）技术的爆发式发展，MCP逐渐成为开发者关注的焦点。今年2月，Cursor宣布支持MCP功能，进一步推动了这一协议的普及。

MCP的核心在于标准化（统一范式）。它通过统一的接口，简化了AI模型与外部工具的集成过程，消除了传统方法中需要为每个工具单独配置API的繁琐步骤。这种标准化的设计不仅降低了开发成本，还减少了错误的发生（调用外部工具时兼容性错误等bug），让AI模型能够更高效地与外部系统协作。

MCP的架构分为三层：主机（Host）、客户端（Client）和服务器（Server）。主机是发起任务请求的应用，客户端负责通信，而服务器则提供工具和服务。这种设计让开发者可以灵活地部署本地或远程的MCP服务器，形成一个开放但受控的智能体能力网络。

MCP的潜力不仅在于技术本身，还在于其生态的快速发展。目前，全球已有超过5000种MCP工具，覆盖数据处理、内容生成、系统控制等多个领域。这些工具的标准化和开放性，使得开发者可以轻松构建和测试智能体工作流，推动AI从“会说话”迈向“能干活”的新时代。

可以说，MCP正在成为AI领域的“HTTP协议”，推动大型语言模型（LLM）应用的标准化和去中心化。它不仅让AI模型摆脱了“缸中大脑”的孤岛状态，还赋予了它们“感知世界与行动执行”的真实能力。未来，随着MCP生态的进一步发展，人人都能轻松构建智能体的时代即将到来。

小结一下，MCP解决的最大痛点：降低Agent开发中调用外部工具（function calling）的技术门槛。

能调用外部工具，是大模型进化为智能体Agent的关键，如果不能使用外部工具，大模型就只能是个简单的聊天机器人，甚至连查询天气都做不到。由于底层技术限制，大模型本身是无法和外部工具直接通信的，因此Function calling的思路，就是创建一个外部函数（function）作为中介，一边传递大模型的请求，另一边调用外部工具，最终让大模型能够间接的调用外部工具，开展大模型的能力边界。

例如，当我们要查询当前天气时，让大模型调用外部工具的function calling的过程

MCP是AI领域的一场革命，它让大模型与外部工具的连接变得前所未有的简单和高效。想象一下，就像USB-C接口统一了各种设备的连接方式，MCP也正在成为AI的“通用接口”，让AI模型能够轻松地与外部世界互动。

为什么MCP如此重要？

在MCP出现之前，大模型与外部工具的集成就像一场噩梦。开发者需要为每个工具单独编写复杂的代码，还要为每个工具编写JSON Schema说明，甚至设计特定的提示词模板。这不仅工作量巨大，还容易出错。而MCP的出现，彻底改变了这一切。

MCP通过标准化的接口，让大模型（MCP客户端）和外部工具（MCP服务器）能够无缝协作。开发者只需编写一次工具，就可以在多个项目中复用，极大地提高了开发效率（类似理解为像python这种面向对象的变成，函数式编程使得一段代码在复用时的调用更加简洁高效，甚至一行代码就能搞定）。更重要的是，这种标准化还促进了全球开发者社区的协作，大家可以在GitHub上共享已经开发好的MCP服务器，从查询天气、网页爬取到数据分析、机器学习建模，应有尽有。

MCP如何运作？

MCP采用客户端-服务器架构，AI模型作为客户端，与提供特定功能的服务器进行通信。当AI需要访问外部数据或执行操作时，它会向相应的服务器发送请求，服务器处理后返回结果。这种设计不仅让AI模型能够专注于核心推理能力，还能通过外部工具扩展其功能。

例如，一个客服聊天机器人可以通过MCP实时查询订单状态，或者一个AI助手可以通过MCP分析数据库中的数据。这种灵活性和扩展性，让AI从“会说话的机器人”进化为“能干活的助手”。

MCP的未来：AI的“高速公路”

MCP不仅仅是一个技术协议，它更像是AI领域的“高速公路”，让AI模型和工具之间的数据流动更加顺畅。随着MCP生态的不断发展，开发者可以轻松构建和测试智能体工作流，推动AI从“孤岛”迈向“协作网络”的新时代。

想象一下，未来的AI助手不仅能回答问题，还能直接操作你的日程、管理你的邮件、分析你的数据，甚至帮你完成复杂的任务。这一切都得益于MCP的标准化和开放性。它不仅让AI更强大，也让开发者更高效，让每个人都能轻松构建智能体。

MCP社区的服务示例

MCP 协议如同大模型的 “万能插头”，本地大模型安装相应库支持该协议后，几行代码就能轻松接入海量外部工具，极大降低 Agent 开发门槛，让开发者专注创造性任务。它类似 HTTP 协议，旨在提升大模型 Agent 开发效率。Anthropic 为普及 MCP 协议，提供涵盖 Python、TS、Java 等多语言的 MCP 客户端和服务器开发 SDK，助力开发者快速搭建 MCP 服务器，吸引个人与团队参与生态建设。目前，MCP 协议在全球开发者社区掀起协作热潮，开发者可在 GitHub 分享或使用他人的 MCP 服务器，避免重复开发，加速生态繁荣，催生更多智能应用。

Function calling

Function calling 是一种让AI模型调用外部工具或功能的技术，就像给AI配了一个“万能助手”。当AI需要完成某些任务时，比如查询天气、分析数据或生成图表，它可以直接“打电话”给相应的工具，把任务交给工具处理，然后工具会把结果“回传”给AI。这样，AI不仅能回答问题，还能真正“动手”解决问题。

1.外部函数库准备

自定义函数（Step 1）：开发者预先定义好需要大模型调用的外部函数，明确函数功能；

添加 Json Schema 表示（Step 2）：为每个自定义函数添加 Json Schema，用于描述函数的输入参数、输出格式等结构，使大模型能理解函数规范；

创建 Tool_List 列表（Step 3）：将所有函数的 Json Schema 整合到一个工具列表（Tool_List）中，为后续调用提供规范描述。

2.用户发起请求与模型判断

用户通过 send_messages 触发 client.chat.completions.create，传入 model（模型）、messages（对话消息）、tools（工具列表）。

模型判断是否需要调用外部函数：不需要调用：直接输出 assistant message，content 为模型返回的信息，tool_calls 设为 None。

需要调用：输出 assistant message，content 为 None，tool_calls 包含需调用的函数名和参数。

3.外部函数执行

将 user message 与含 tool_calls 的 assistant message 组合，传递给外部函数库：匹配函数（Step 1）：根据 tool_calls 中的函数名，找到对应的自定义函数。

传递参数（Step 2）：将 tool_calls 提供的参数传入该函数。

执行函数（Step 3）：运行函数并获取返回结果。

封装结果（Step 4）：将结果封装为 ToolMessage，格式为 {role: tool, name: 函数名, content: 执行结果, tool_call_id: id}。

4.生成最终答案

将封装好的 tool message 追加到原消息中，再次调用 client.chat.completions.create（传入更新后的 messages）。

模型基于外部函数的执行结果，输出 Final Answer，完成整个流程。

MCP客户端Client开发流程

uv工具

MCP开发要求借助uv进行虚拟环境创建和依赖管理。uv 是一个Python 依赖管理工具，类似于 pip 和 conda，但它更快、更高效，并且可以更好地管理 Python 虚拟环境和依赖项。它的核心目标是替代 pip、venv 和 pip-tools，提供更好的性能和更低的管理开销。

uv 的特点：

1. 速度更快：相比 pip，uv 采用 Rust 编写，性能更优
2. 支持 PEP 582：无需 virtualenv，可以直接使用 __pypackages__ 进行管理
3. 兼容 pip：支持 requirements.txt 和 pyproject.toml 依赖管理
4. 替代 venv：提供 uv venv 进行虚拟环境管理，比 venv 更轻量
5. 跨平台：支持 Windows、macOS 和 Linux

uv安装流程

1、使用 pip 安装

pip install uv

2、使用 curl 直接安装

curl -LsSf https://astral.sh/uv/install.sh | sh

uv的基本用法介绍

安装 uv 后，可以像 pip 一样使用它，它的语法更简洁，速度也更快。

示例

安装 Python 依赖

uv pip install numpy

与 pip install numpy 类似，但更快。

创建虚拟环境

uv venv mcpenv

等效于 python -m venv mcpenv，但更高效。

激活虚拟环境

source mcpenv/bin/activate  # Linux/macOS
mcpenv\Scripts\activate     # Windows

安装 requirements.txt

uv pip install -r requirements.txt

直接运行 Python 项目

如果项目中包含 pyproject.toml，可以直接运行：

uv run python script.py

这等效于：

pip install -r requirements.txt
python script.py

MCP是个让AI模型和外部工具协作的协议，就像给AI搭了个“工具箱”。但要管理这个工具箱里的东西，用传统的pip就像用手动拧螺丝，慢不说还容易出错。uv（虚拟环境管理工具）就厉害了，它就像个智能工具箱，能自动帮你搞定所有模块的安装和管理。

uv通过一个叫pyproject.toml的文件，把所有依赖都整得明明白白，就像给工具箱里的每个工具贴了标签，想找啥一目了然。而且uv的速度比pip快得多，安装东西就像开跑车，嗖一下就搞定了。对于MCP这种需要频繁折腾依赖的项目，uv简直是救星。

所以，MCP更推荐用uv来管理环境。接下来，咱们可以先搭个MCP客户端，就像先搭个积木架子，看看能不能跑起来。等客户端能正常工作了，再慢慢搭MCP服务器，就像给积木架子添砖加瓦。这样分阶段来，万一有问题也能一步步排查，不会一上来就被复杂性搞得晕头转向。

MCP极简客户端搭建流程

创建 MCP 客户端项目

# 创建项目目录
uv init mcp-client
cd mcp-client

建议在autodl上租用云算力进行测试部署

创建MCP客户端虚拟环境

# 创建虚拟环境
uv venv

# 激活虚拟环境
source .venv/bin/activate

这里需要注意的是，相比pip，uv会自动识别当前项目主目录并创建虚拟环境。

然后即可通过add方法在虚拟环境中安装相关的库。

# 安装 MCP SDK
uv add mcp

编写基础 MCP 客户端

然后在当前项目主目录中**创建 client.py **

并写入以下代码

import asyncio
from mcp import ClientSession
from contextlib import AsyncExitStack

class MCPClient:
    def __init__(self):
        """初始化 MCP 客户端"""
        self.session = None
        self.exit_stack = AsyncExitStack()

    async def connect_to_mock_server(self):
        """模拟 MCP 服务器的连接（暂不连接真实服务器）"""
        print(" MCP 客户端已初始化，但未连接到服务器")

    async def chat_loop(self):
        """运行交互式聊天循环"""
        print("\nMCP 客户端已启动！输入 'quit' 退出")

        while True:
            try:
                query = input("\nQuery: ").strip()
                if query.lower() == 'quit':
                    break
                print(f" [Mock Response] 你说的是：{query}")
            except Exception as e:
                print(f" 发生错误: {str(e)}")

    async def cleanup(self):
        """清理资源"""
        await self.exit_stack.aclose()

async def main():
    client = MCPClient()
    try:
        await client.connect_to_mock_server()
        await client.chat_loop()
    finally:
        await client.cleanup()

if __name__ == "__main__":
    asyncio.run(main())

这段代码能够初始化 MCP 客户端（但不连接服务器），并提供一个交互式 CLI，可以输入查询（但只返回模拟回复），通过输入 quit 退出程序。需要注意的是，此时客户端没有接入任何大模型，只会重复用户的输入。

代码解释：

MCP客户端基本代码结构

初始化 MCP 客户端

*提供一个命令行交互界面

模拟 MCP 服务器连接

支持用户输入查询并返回「模拟回复」

支持安全退出

首先是导入必要的库

import asyncio  # 让代码支持异步操作
from mcp import ClientSession  # MCP 客户端会话管理
from contextlib import AsyncExitStack  # 资源管理（确保客户端关闭时释放资源）

asyncio：Python 内置的异步编程库，让 MCP 可以非阻塞地执行任务（比如聊天、查询）。

mcp.ClientSession：用于管理 MCP 客户端会话（但目前我们先不连接 MCP 服务器）。

AsyncExitStack：自动管理资源，确保程序退出时正确关闭 MCP 连接。

然后创建 MCPClient 类

class MCPClient:
    def __init__(self):
        """初始化 MCP 客户端"""
        self.session = None  # 先不连接 MCP 服务器
        self.exit_stack = AsyncExitStack()  # 创建资源管理器

self.session = None：暂时不连接 MCP 服务器，后续可以修改来真正连接。

self.exit_stack = AsyncExitStack()：管理 MCP 客户端的资源，确保程序退出时可以正确释放资源。

模拟 MCP 服务器连接

async def connect_to_mock_server(self):
    """模拟 MCP 服务器的连接（暂不连接真实服务器）"""
    print(" MCP 客户端已初始化，但未连接到服务器")

这个函数不会真的连接 MCP 服务器，只是打印一条信息，表示客户端已经初始化。

async def：因为我们用的是 异步编程，所以需要用 async 关键字。

创建交互式聊天循环

async def chat_loop(self):
    """运行交互式聊天循环"""
    print("\nMCP 客户端已启动！输入 'quit' 退出")

    while True:  # 无限循环，直到用户输入 'quit'
        try:
            query = input("\nQuery: ").strip()  # 让用户输入问题
            if query.lower() == 'quit':  # 如果用户输入 quit，退出循环
                break
            print(f" [Mock Response] 你说的是：{query}")  # 返回模拟回复
        except Exception as e:  # 发生错误时捕获异常
            print(f" 发生错误: {str(e)}")

while True：无限循环，让用户可以不断输入查询。

query = input("\nQuery: ").strip()：获取用户输入的查询。

if query.lower() == 'quit'：如果用户输入 quit，退出循环。

print(f" [Mock Response] 你说的是：{query}")：模拟 MCP 服务器的响应，暂时只是回显用户输入的内容。

async def cleanup(self):
    """清理资源"""
    await self.exit_stack.aclose()  # 关闭资源管理器

aclose() 确保程序退出时正确关闭 MCP 连接（尽管目前没有真正的连接）。

并定义**main() 主函数**

async def main():
    client = MCPClient()  # 创建 MCP 客户端
    try:
        await client.connect_to_mock_server()  # 连接（模拟）服务器
        await client.chat_loop()  # 进入聊天循环
    finally:
        await client.cleanup()  # 确保退出时清理资源

client = MCPClient()：创建一个 MCP 客户端实例。

await client.connect_to_mock_server()：初始化 MCP 客户端（暂不连接服务器）。

await client.chat_loop()：启动交互式聊天。

finally: 确保 不管程序是否异常退出，都会正确释放资源。

运行代码脚本

if __name__ == "__main__":
    asyncio.run(main())

if __name__ == "__main__":：确保代码只能在 Python 直接运行时执行（而不是作为库导入时）

asyncio.run(main())：启动 main()，运行 MCP 客户端

MCP中一个基础的客户端代码结构总结如下：

运行 MCP 客户端

接下来尝试运行这个极简的MCP客户端：

uv run client.py

MCP客户端接入OpenAI、DeepSeek在线模型流程

接下来尝试在客户端中接入OpenAI和DeepSeek等在线模型进行对话。由于OpenAI和DeepSeek调用方法几乎完全一样，因此定义好一套的服务器client代码可以同时适用于GPT模型和DeepSeek模型。

新增依赖

为了支持调用OpenAI模型，以及在环境变量中读取API-KEY等信息，需要先安装如下依赖：

uv add mcp openai python-dotenv

创建.env文件

接下来创建.env文件，并写入OpenAI的API-Key，以及反向代理地址。借助反向代理，国内可以无门槛直连OpenAI官方服务器，并调用官方API。

写入如下内容

BASE_URL="反向代理地址"
MODEL=gpt-4o
OPENAI_API_KEY="OpenAI-API-Key"

而如果是使用DeepSeek模型，则需要在.env中写入如下内容：

BASE_URL=https://api.deepseek.com
MODEL=deepseek-chat      
OPENAI_API_KEY="DeepSeek API-Key"

import asyncio
import os
from openai import OpenAI
from dotenv import load_dotenv
from contextlib import AsyncExitStack

# 加载 .env 文件，确保 API Key 受到保护
load_dotenv()

class MCPClient:
    def __init__(self):
        
        """初始化 MCP 客户端"""
        self.exit_stack = AsyncExitStack()
        self.openai_api_key = os.getenv("OPENAI_API_KEY")  # 读取 OpenAI API Key
        self.base_url = os.getenv("BASE_URL")  # 读取 BASE YRL
        self.model = os.getenv("MODEL")  # 读取 model
        
        if not self.openai_api_key:
            raise ValueError(" 未找到 OpenAI API Key，请在 .env 文件中设置 OPENAI_API_KEY")
            
        self.client = OpenAI(api_key=self.openai_api_key, base_url=self.base_url) 
        

    async def process_query(self, query: str) -> str:
        """调用 OpenAI API 处理用户查询"""
        messages = [{"role": "system", "content": "你是一个智能助手，帮助用户回答问题。"},
                    {"role": "user", "content": query}]
        
        try:
            # 调用 OpenAI API
            response = await asyncio.get_event_loop().run_in_executor(
                None,
                lambda: self.client.chat.completions.create(
                    model=self.model,
                    messages=messages
                )
            )
            return response.choices[0].message.content
        except Exception as e:
            return f" 调用 OpenAI API 时出错: {str(e)}"

    async def chat_loop(self):
        """运行交互式聊天循环"""
        print(" MCP 客户端已启动！输入 'quit' 退出")

        while True:
            try:
                query = input("\n你: ").strip()
                if query.lower() == 'quit':
                    break
                
                response = await self.process_query(query)  # 发送用户输入到 OpenAI API
                print(f" OpenAI: {response}")

            except Exception as e:
                print(f" 发生错误: {str(e)}")

    async def cleanup(self):
        """清理资源"""
        await self.exit_stack.aclose()

async def main():
    client = MCPClient()
    try:
        await client.chat_loop()
    finally:
        await client.cleanup()

if __name__ == "__main__":
    asyncio.run(main())

运行client.py

然后即可输入如下命令开始运行对话客户端：

uv run client.py

clint.py代码解释

加载 OpenAI API Key

from dotenv import load_dotenv
import os

# 加载 .env 文件，确保 API Key 受到保护
load_dotenv()

self.openai_api_key = os.getenv("OPENAI_API_KEY")  # 读取 API Key
if not self.openai_api_key:
    raise ValueError(" 未找到 OpenAI API Key，请在 .env 文件中设置 OPENAI_API_KEY")

load_dotenv()：自动加载 .env 文件，避免在代码中直接暴露 API Key。

os.getenv("OPENAI_API_KEY")：从环境变量中读取 OPENAI_API_KEY。

raise ValueError(...)：如果 API Key 为空，则抛出错误，提醒用户必须配置 API Key。

创建 .env 文件

touch .env

在 .env 文件中添加 API Key

OPENAI_API_KEY=你的OpenAI API Key

发送用户输入到 OpenAI API

async def process_query(self, query: str) -> str:
    """调用 OpenAI API 处理用户查询"""
    messages = [
        {"role": "system", "content": "你是一个智能助手，帮助用户回答问题。"},
        {"role": "user", "content": query}
    ]

    try:
        # 调用 OpenAI GPT-4 API
        response = await asyncio.get_event_loop().run_in_executor(
            None,
            lambda: openai.ChatCompletion.create(
                model="gpt-4",
                messages=messages,
                max_tokens=1000,
                temperature=0.7
            )
        )
        return response["choices"][0]["message"]["content"].strip()
    except Exception as e:
        return f" 调用 OpenAI API 时出错: {str(e)}"

messages：创建对话上下文，让 OpenAI 知道如何回答问题：

system 角色：设定 AI 角色（如“你是一个智能助手”）。

user 角色：存储用户输入。

openai.ChatCompletion.create(...)

model="gpt-4"：使用 OpenAI 的 GPT-4 进行对话。

messages=messages：提供聊天记录，让 AI 生成回答。

max_tokens=1000：限制 AI 生成的最大字数。

temperature=0.7：控制 AI 回答的随机性（越高越随机）。

run_in_executor(...)：

因为 OpenAI API 是同步的，但我们用的是异步代码

这里用 asyncio.get_event_loop().run_in_executor(...) 将 OpenAI API 变成异步任务，防止程序卡顿。

交互式聊天

async def chat_loop(self):
    """运行交互式聊天循环"""
    print(" MCP 客户端已启动！输入 'quit' 退出")

    while True:
        try:
            query = input("\n你: ").strip()
            if query.lower() == 'quit':
                break

            response = await self.process_query(query)  # 发送用户输入到 OpenAI API
            print(f" OpenAI: {response}")

        except Exception as e:
            print(f" 发生错误: {str(e)}")

输入查询 query = input("\n你: ").strip()，支持多轮对话。

调用 process_query()，将用户输入发送到 OpenAI API 并获取回复。

显示 OpenAI 生成的回复：print(f" OpenAI: {response}")

用户输入 quit 退出。

需要注意的是，由于MCP的client SDK主要规定了client和server之间的通信方法，因此在没有创建server之前，一个单纯对话的client甚至不需要用到mcp功能。但本段代码的学习仍是有必要的，为了熟悉各类大模型本地调用对话流程。而后我们只需要围绕上述代码稍作修改，即可调用外部的server。

MCP对接外部工具服务器server原理

MCP服务器概念介绍

MCP协议中，Server可以提供三种类型的标准能力：Resources（资源）、Tools（工具）和Prompts（提示词）。这些能力分别对应不同的功能和应用场景。

1. Resources（资源）

定义：Resources 是一种数据提供能力，类似于文件数据读取或API响应返回的内容。

功能：它允许MCP Server向MCP Client提供数据资源，这些资源可以是文件、数据库查询结果、API响应等。

应用场景：

文件读取：例如，读取本地文件系统中的文件内容。

API响应：例如，从外部API获取数据，如天气预报、股票行情等。

数据库查询：例如，从MySQL或MongoDB中检索数据。

类比：Resources 就像是一个图书馆，MCP Client可以从中获取需要的书籍（数据）。

2. Tools（工具）

定义：Tools 是一种功能调用能力，允许MCP Server提供第三方服务或功能函数。

功能：它允许MCP Client调用预定义的函数或服务，从而扩展AI模型的能力。

应用场景：

第三方服务：例如，调用支付接口、发送邮件、发送短信等。

功能函数：例如，执行数学计算、数据处理、文件操作等。

系统控制：例如，执行命令行操作、控制硬件设备等。

类比：Tools 就像一个工具箱，MCP Client可以从中选择合适的工具来完成特定的任务。

3. Prompts（提示词）

定义：Prompts 是一种提示词模板能力，为用户预先定义好的完成特定任务的模板。

功能：它允许MCP Server向MCP Client提供预定义的提示词模板，帮助AI模型更好地理解和执行任务。

应用场景：

任务模板：例如，提供特定任务的提示词模板，如“生成一篇关于人工智能的文章”。

对话引导：例如，提供引导用户输入的提示词，如“请告诉我您的需求”。

任务优化：例如，提供优化任务执行的提示词，如“请用简洁的语言总结以下内容”。

类比：Prompts 就像是一本食谱，MCP Client可以按照这些模板来完成特定的任务。

Resources：提供数据资源，帮助MCP Client获取所需的数据。

Tools：提供功能函数，帮助MCP Client执行特定的操作。

Prompts：提供提示词模板，帮助MCP Client更好地理解和完成任务。

MCP服务器通讯机制

MCP就像是给大模型和外部工具之间搭了一座桥，让它们能够顺畅地“聊天”。这座桥有两种通行方式：一种是本地的“面对面交流”（stdio），另一种是远程的“电话沟通”（HTTP+SSE）。不过，最近有开发者提议用“可流式传输的HTTP”来升级现有的“电话沟通”方式，让远程沟通更高效。

本地通信：stdio模式——“面对面交流”

想象一下，你和朋友在同一间房间里，可以直接面对面交流，不需要通过电话或者网络。这就是MCP的stdio模式。在这种模式下，MCP客户端会把服务器程序当作自己的“小跟班”，直接启动它作为一个子进程。它们之间的交流就像你和朋友直接说话一样，通过标准输入（stdin）和标准输出（stdout）进行。

工作流程：客户端通过stdin给服务器发消息，服务器通过stdout回复。这种方式简单直接，几乎没有延迟，就像你和朋友直接对话一样高效。

适用场景：适合客户端和服务器在同一台机器上运行的情况，比如本地开发或者一些对速度要求极高的应用。

远程通信：HTTP+SSE模式——“电话沟通”

如果客户端和服务器不在同一台机器上，就像你和朋友在不同的城市，那就需要通过“电话”来沟通了。MCP的HTTP+SSE模式就是这种“电话沟通”。

HTTP：传统的HTTP就像是打一次电话问一个问题，等对方回答后再挂电话。每次请求都是独立的，就像你每次打电话都要重新拨号。

SSE（服务器推送事件）：SSE就像是你和朋友保持通话状态，朋友可以随时给你推送信息，比如实时通知或者股票行情更新。这种方式适合服务器需要主动推送数据的场景。

不过，SSE有个小缺点，它只能单向通信，就像你只能听朋友说话，不能插嘴。而WebSocket则支持双向通信，但MCP目前主要用的是SSE。

为什么需要升级？—“可流式传输的HTTP”

虽然现有的HTTP+SSE模式已经很好用了，但开发者们觉得还可以更好。他们提出用“可流式传输的HTTP”来替代现有的方案，就像把普通的电话升级成视频通话，既能实时交流，又能更高效地处理数据。这种升级可以解决一些现有方案的限制，同时保留它的优点。

stdio模式：适合本地通信，就像面对面交流，简单高效，适合快速响应的场景。

HTTP+SSE模式：适合远程通信，就像电话沟通，适合需要实时更新的场景。

可流式传输的

HTTP PR：https://github.com/modelcontextprotocol/specification/pull/206

小结一下，MCP定义了Client与Server进行通讯的协议与消息格式，其支持两种类型通讯机制：标准输入输出通讯、基于SSE的HTTP通讯，分别对应着本地与远程通讯。Client与Server间使用JSON-RPC 2.0格式进行消息传输。

本地通讯：使用了stdio传输数据，具体流程Client启动Server程序作为子进程，其消息通讯是通过stdin/stdout进行的，消息格式为JSON-RPC 2.0。

远程通讯：Client与Server可以部署在任何地方，Client使用SSE与Server进行通讯，消息的格式为JSON-RPC 2.0，Server定义了/see与/messages接口用于推送与接收数据。

GraphRAG基于知识图谱的检索增强技术

GraphRAG 项目地址：

https://github.com/microsoft/graphrag/

GraphRAG

RAG（检索增强生成）是一种通过结合真实世界的信息来提升大型语言模型（LLM）输出质量的技术。它在帮助LLM推理私有数据集方面显示了很大的潜力，比如企业专有研究、业务文档或通信数据。然而，基线RAG（Baseline RAG）在某些情况下表现并不理想。

基线RAG的局限性：

难以将信息串联起来：当一个问题的答案需要通过多个不同的信息片段，并通过它们共享的属性来连接，进而提供新的综合见解时，基线RAG表现得很差。例如，在回答“如何通过现有数据推断出新结论”这种问题时，基线RAG无法很好地处理这些散布在不同文档中的相关信息，可能会遗漏一些关键联系点。

无法有效理解大型数据集或单一大文档的整体语义概念：当被要求在大量数据或复杂文档中进行总结、提炼和理解时，基线RAG往往表现不佳。例如，如果问题要求对整个文档或多篇文档的主题进行总结和理解，基线RAG的简单向量检索方法可能无法处理文档间的复杂关系，导致对全局语义的理解不完整。

为了应对这些挑战，微软研究院（Microsoft Research）提出了GraphRAG方法。GraphRAG使用LLM基于输入语料库构建知识图谱，这个图谱与社区总结和图谱机器学习输出结合，能够在查询时增强提示（prompt）。GraphRAG在回答以上两类问题时，展示了显著的改进，尤其是在复杂信息的推理能力和智能性上，超越了基线RAG之前应用于私有数据集的其他方法。

GraphRAG的优势：

更精确和上下文相关的回答：GraphRAG通过知识图谱提取信息，能够找到与问题直接相关的上下文，而不是随机的关键词。例如，当问到“AI在医疗中的应用”时，GraphRAG会解释不同类型的AI如何影响不同的医疗结构，提供更具体和相关的答案。

更好的语义理解：GraphRAG分析概念之间的连接，而不仅仅是文本匹配。它能够链接两个看似不相关的信息，因为它理解图谱结构，从而提供更符合用户需求的答案。

更高的准确性：知识图谱帮助GraphRAG专注于寻找有价值且上下文合适的数据，减少不相关或上下文错误的数据被包含的可能性。

更高效的数据检索：图谱结构使得GraphRAG能够快速导航和搜索网络化数据点，加速数据检索过程。

可扩展性和灵活性：GraphRAG能够随着数据量的增长无缝地整合新的数据点和关系，确保检索速度和准确性不受影响。

GraphRAG基本原理

GraphRAG 是微软研究院开发的一种先进的增强检索生成（RAG）框架，旨在提升语言模型（LLM）在处理复杂数据时的性能。与传统的 RAG 方法依赖向量相似性检索不同，GraphRAG 利用 知识图谱 来显著增强语言模型的问答能力，特别是在处理私有数据集或大型、复杂数据集时表现尤为出色。

传统的 Baseline RAG 方法在某些情况下表现不佳，尤其是当查询需要在不同信息片段之间建立联系时，或是当需要对大规模数据集进行整体理解时。

GraphRAG 通过以下方式克服了这些问题：

更好的连接信息点：GraphRAG 能够处理那些需要从多个数据点合成新见解的任务。

更全面的理解能力：GraphRAG 更擅长对大型数据集进行全面理解，能够更好地处理复杂的抽象问题。

GraphRAG构建知识图谱的过程可以分为以下几个步骤：

数据预处理：包括数据清洗、实体识别等步骤，目的是将原始数据转换为适合构建知识图谱的格式。

实体关系识别：从清洗后的数据中提取实体和关系。可以使用命名实体识别（NER）技术，借助深度学习模型如LSTM或BERT等预训练模型来提高识别的准确率。

图谱构建：将提取出的实体和关系以图谱的形式进行可视化展示。每个实体作为图中的一个节点，节点之间的边表示实体之间的关系。

图谱存储：将构建好的知识图谱存储到图数据库中，如Neo4j，以实现高效查询和管理。

具体来说，GraphRAG会将输入的文档集合按一定策略拆分成多个chunks，然后解析每个chunk中的实体和关系。它借助大模型的few shot能力，通过特定的prompt模板来提取输入文档中的实体关系。这些prompt模板包括用于提取实体关系的prompt、用于总结实体关系描述的prompt以及用于提取实体属性的prompt。

在构建知识图谱的过程中，GraphRAG还会应用社区检测算法来识别相关节点的聚类，并生成社区摘要。这些摘要可以帮助用户更好地理解和分析知识图谱中的信息。

借助微软开源的GeaphRAG项目，可以快速做到以下事项：

基于图的检索：传统的 RAG 方法使用向量相似性进行检索，而 GraphRAG 引入了知识图谱来捕捉实体、关系及其他重要元数据，从而更有效地进行推理

层次聚类：GraphRAG 使用 Leiden 技术进行层次聚类，将实体及其关系进行组织，提供更丰富的上下文信息来处理复杂的查询

多模式查询：支持多种查询模式：

全局搜索：通过利用社区总结来进行全局性推理

局部搜索：通过扩展相关实体的邻居和关联概念来进行具体实体的推理

DRIFT 搜索：结合局部搜索和社区信息，提供更准确和相关的答案

图机器学习：集成了图机器学习技术，提升查询响应质量，并提供来自结构化和非结构化数据的深度洞察

Prompt调优：提供调优工具，帮助根据特定数据和需求调整查询提示，从而提高结果质量

GraphRAG运行流程

索引（Indexing）过程

文本单元切分：将输入文本分割成 TextUnits，每个 TextUnit 是一个可分析的单元，用于提取关键信息。

实体和关系提取：使用 LLM 从 TextUnits 中提取实体、关系和关键声明。

图构建：构建知识图谱，使用 Leiden 算法进行实体的层次聚类。每个实体用节点表示，节点的大小和颜色分别代表实体的度数和所属社区。

社区总结：从下到上生成每个社区及其成员的总结，帮助全局理解数据集。

查询（Query）过程

索引完成后，用户可以通过不同的搜索模式进行查询：

全局搜索：当我们想了解整个语料库或数据集的整体概况时，GraphRAG 可以利用 社区总结 来快速推理和获取信息。这种方式适用于大范围问题，如某个主题的总体理解。

局部搜索：如果问题关注于某个特定的实体，GraphRAG 会向该实体的 邻居（即相关实体）扩展搜索，以获得更详细和精准的答案。

DRIFT 搜索：这是对局部搜索的增强，除了获取邻居和相关概念，还引入了 社区信息 的上下文，从而提供更深入的推理和连接。

Prompt 调优

为了获得最佳性能，GraphRAG 强烈建议进行 Prompt 调优，确保模型可以根据你的特定数据和查询需求进行优化，从而提供更准确和相关的答案。

GraphRAG计算流程示例

流程解析

1.查询预处理

用户输入自然语言查询后，系统对其进行解析，识别查询中的实体、关系及意图。例如，查询 “某公司的产品有哪些合作伙伴”，需提取 “公司”“产品”“合作伙伴” 等关键元素，将自然语言转化为图数据库可理解的结构化形式，为后续检索做准备。

2.图数据库检索

图数据库存储着由实体（节点）和关系（边）构成的知识图谱。根据预处理后的查询，在图数据库中检索与之相关的子图。例如，以 “公司” 节点为起点，沿 “产品”“合作伙伴” 等关系边扩展，提取包含相关节点和边的子图，精准捕捉实体间的关联结构。

3.子图处理

利用图神经网络（GNN）等技术对检索到的子图进行处理。GNN 通过聚合节点邻域信息，捕捉节点间的复杂关系，生成节点的向量表示，进而压缩或转换子图信息，提取关键内容，为后续生成上下文做准备。

4.上下文生成

将处理后的子图信息转换为文本形式，整合成适合大语言模型（LLM）输入的上下文。例如，把节点和关系转化为自然语言句子（如 “某公司的产品与某合作伙伴存在合作关系”），清晰呈现实体间联系。

5.LLM 生成回答

将原始查询与生成的上下文输入 LLM，LLM 基于这些信息生成自然语言回答。例如，综合查询意图与上下文关系，输出 “某公司的产品与 X、Y 等合作伙伴有合作”。

原理阐释

关系建模优势：传统 RAG 从文本块中检索信息，对实体间关系的处理不够直观。而 GraphRAG 借助图数据库的图结构，显式建模实体与关系，能更精准地捕捉和利用复杂关系信息。

提升回答质量：通过图数据库检索与子图处理，GraphRAG 为 LLM 提供更具针对性和结构性的上下文，使生成的回答在涉及关系推理（如供应链分析、知识关联查询等场景）时更准确、相关，弥补了传统 RAG 在关系型知识处理上的不足。

MCP+GraphRAG搭建检索增强智能体

根据GraphRAG API的调用方法，创建一个基于GraphRAG的MCP智能体服务器，并尝试在本地client对其进行调用。

创建 MCP 客户端项目

# 创建项目目录
uv init mcp-graphrag
cd mcp-graphrag

创建MCP客户端虚拟环境

# 创建虚拟环境
uv venv

# 激活虚拟环境
source .venv/bin/activate

这里需要注意的是，相比pip，uv会自动识别当前项目主目录并创建虚拟环境。

然后即可通过add方法在虚拟环境中安装相关的库。

# 安装 MCP SDK
uv add mcp graphrag pathlib pandas

创建GraphRAG并构建索引（Index）

创建项目目录并进行初始化

mkdir -p ./graphrag/input
graphrag init --root ./graphrag

修改配置文件

打开.env文件，填写DeepSeek API-KEY或OpenAI API-Key；

打开setting.yaml文件，填写模型名称和代理地址

上传文本数据

index过程

graphrag index --root ./graphrag

创建GraphRAG服务器Server

当前创建的GraphRAG Server只负责进行对某一个完成Index的知识库进行Query，更加复杂的如文件管理、实时增加检索、多文件库检索等。

在当前项目中创建一个名为rag_server.py的server，

写入代码：

from pathlib import Path
from pprint import pprint

import pandas as pd

import graphrag.api as api
from graphrag.config.load_config import load_config
from graphrag.index.typing.pipeline_run_result import PipelineRunResult

from typing import Any
from mcp.server.fastmcp import FastMCP

# 初始化 MCP 服务器
mcp = FastMCP("rag_ML")
USER_AGENT = "rag_ML-app/1.0"

@mcp.tool()
async def rag_ML(query: str) -> str:
    """
    用于查询机器学习决策树相关信息。
    :param query: 用户提出的具体问题
    :return: 最终获得的答案
    """
    PROJECT_DIRECTORY = "/root/autodl-tmp/MCP/mcp-graphrag/graphrag"
    graphrag_config = load_config(Path(PROJECT_DIRECTORY))
    
    # 加载实体
    entities = pd.read_parquet(f"{PROJECT_DIRECTORY}/output/entities.parquet")
    # 加载社区
    communities = pd.read_parquet(f"{PROJECT_DIRECTORY}/output/communities.parquet")
    # 加载社区报告
    community_reports = pd.read_parquet(
        f"{PROJECT_DIRECTORY}/output/community_reports.parquet"
    )
    # 进行全局搜索
    response, context = await api.global_search(
        config=graphrag_config,
        entities=entities,
        communities=communities,
        community_reports=community_reports,
        community_level=2,
        dynamic_community_selection=False,
        response_type="Multiple Paragraphs",
        query=query,
    )
    
    return response

if __name__ == "__main__":
    # 以标准 I/O 方式运行 MCP 服务器
    mcp.run(transport='stdio')

代码解释：

导入必要的模块和库：

Path 和 pprint：用于路径操作和美化打印输出。

pandas：用于数据处理，特别是读取 Parquet 格式的数据文件。

graphrag.api 和相关配置模块：用于加载配置和调用 GraphRAG 的 API。

FastMCP：MCP 协议的快速实现，用于创建 MCP 服务器。

初始化 MCP 服务器：

mcp = FastMCP("rag_ML")：创建一个名为 rag_ML 的 MCP 服务器实例。

USER_AGENT = "rag_ML-app/1.0"：定义用户代理字符串，可能用于标识客户端应用程序的版本信息。

定义工具函数 rag_ML：

使用装饰器 @mcp.tool() 将函数注册为 MCP 工具，使其可被客户端调用。

函数为异步函数，接受一个字符串类型的 query 参数，表示用户的查询。

函数内部执行以下操作：

加载 GraphRAG 配置：

PROJECT_DIRECTORY：定义项目目录路径。

graphrag_config = load_config(Path(PROJECT_DIRECTORY))：加载 GraphRAG 的配置文件。

加载数据文件：

使用 pandas 的 read_parquet 方法分别加载实体、社区和社区报告的 Parquet 文件。

调用

api.global_search

方法进行全局搜索：

传入配置、实体、社区和社区报告等参数。

设置 community_level=2 和 dynamic_community_selection=False，用于控制社区层级和是否动态选择社区。

设置 response_type="Multiple Paragraphs"，指定响应类型为多段落文本。

返回搜索结果 response。

运行 MCP 服务器：

在主程序中，调用 mcp.run(transport='stdio') 以标准输入输出（stdio）的方式运行 MCP 服务器，使其能够接收和响应客户端的请求。

创建GraphRAG服务器client

接下来继续创建客户端，在项目主目录下创建一个名为client.py的客户端

写入代码：

import asyncio
import os
import json
from typing import Optional
from contextlib import AsyncExitStack

from openai import OpenAI  
from dotenv import load_dotenv

from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client

# 加载 .env 文件，确保 API Key 受到保护
load_dotenv()

class MCPClient:
    def __init__(self):
        """初始化 MCP 客户端"""
        self.exit_stack = AsyncExitStack()
        self.openai_api_key = os.getenv("OPENAI_API_KEY")  # 读取 OpenAI API Key
        self.base_url = os.getenv("BASE_URL")  # 读取 BASE YRL
        self.model = os.getenv("MODEL")  # 读取 model
        if not self.openai_api_key:
            raise ValueError(" 未找到 OpenAI API Key，请在 .env 文件中设置 OPENAI_API_KEY")
        self.client = OpenAI(api_key=self.openai_api_key, base_url=self.base_url) # 创建OpenAI client
        self.session: Optional[ClientSession] = None   

    async def transform_json(self, json2_data):
        """
        将Claude Function calling参数格式转换为OpenAI Function calling参数格式，多余字段会被直接删除。
        
        :param json2_data: 一个可被解释为列表的 Python 对象（或已解析的 JSON 数据）
        :return: 转换后的新列表
        """
        result = []
        
        for item in json2_data:
            # 确保有 "type" 和 "function" 两个关键字段
            if not isinstance(item, dict) or "type" not in item or "function" not in item:
                continue
        
            old_func = item["function"]
        
            # 确保 function 下有我们需要的关键子字段
            if not isinstance(old_func, dict) or "name" not in old_func or "description" not in old_func:
                continue
        
            # 处理新 function 字段
            new_func = {
                "name": old_func["name"],
                "description": old_func["description"],
                "parameters": {}
            }
        
            # 读取 input_schema 并转成 parameters
            if "input_schema" in old_func and isinstance(old_func["input_schema"], dict):
                old_schema = old_func["input_schema"]
                
                # 新的 parameters 保留 type, properties, required 这三个字段
                new_func["parameters"]["type"] = old_schema.get("type", "object")
                new_func["parameters"]["properties"] = old_schema.get("properties", {})
                new_func["parameters"]["required"] = old_schema.get("required", [])
            
            new_item = {
                "type": item["type"],
                "function": new_func
            }
        
            result.append(new_item)
    
        return result

    async def connect_to_server(self, server_script_path: str):
        """连接到 MCP 服务器并列出可用工具"""
        is_python = server_script_path.endswith('.py')
        is_js = server_script_path.endswith('.js')
        if not (is_python or is_js):
            raise ValueError("服务器脚本必须是 .py 或 .js 文件")

        command = "python" if is_python else "node"
        server_params = StdioServerParameters(
            command=command,
            args=[server_script_path],
            env=None
        )

        # 启动 MCP 服务器并建立通信
        stdio_transport = await self.exit_stack.enter_async_context(stdio_client(server_params))
        self.stdio, self.write = stdio_transport
        self.session = await self.exit_stack.enter_async_context(ClientSession(self.stdio, self.write))

        await self.session.initialize()

        # 列出 MCP 服务器上的工具
        response = await self.session.list_tools()
        tools = response.tools
        print("\n已连接到服务器，支持以下工具:", [tool.name for tool in tools])     
        
    async def process_query(self, query: str) -> str:
        """
        使用大模型处理查询并调用可用的 MCP 工具 (Function Calling)
        """
        messages = [{"role": "user", "content": query}]
        
        response = await self.session.list_tools()
        
        available_tools = [{
            "type": "function",
            "function": {
                "name": tool.name,
                "description": tool.description,
                "input_schema": tool.inputSchema
            }
        } for tool in response.tools]
        # print(available_tools)

        # 进行参数格式转化
        available_tools = await self.transform_json(available_tools)
        
        response = self.client.chat.completions.create(
            model=self.model,            
            messages=messages,
            tools=available_tools     
        )
        
        # 处理返回的内容
        content = response.choices[0]
        if content.finish_reason == "tool_calls":
            # 如何是需要使用工具，就解析工具
            tool_call = content.message.tool_calls[0]
            tool_name = tool_call.function.name
            tool_args = json.loads(tool_call.function.arguments)
            
            # 执行工具
            result = await self.session.call_tool(tool_name, tool_args)
            print(f"\n\n[Calling tool {tool_name} with args {tool_args}]\n\n")
            
            # 将模型返回的调用哪个工具数据和工具执行完成后的数据都存入messages中
            messages.append(content.message.model_dump())
            messages.append({
                "role": "tool",
                "content": result.content[0].text,
                "tool_call_id": tool_call.id,
            })
            
            # 将上面的结果再返回给大模型用于生产最终的结果
            response = self.client.chat.completions.create(
                model=self.model,
                messages=messages,
            )
            return response.choices[0].message.content
            
        return content.message.content
    
    async def chat_loop(self):
        """运行交互式聊天循环"""
        print(" MCP 客户端已启动！输入 'quit' 退出")

        while True:
            try:
                query = input("\n你: ").strip()
                if query.lower() == 'quit':
                    break
                
                response = await self.process_query(query)  # 发送用户输入到 OpenAI API
                print(f" OpenAI: {response}")

            except Exception as e:
                print(f" 发生错误: {str(e)}")

    async def cleanup(self):
        """清理资源"""
        await self.exit_stack.aclose()

async def main():
    if len(sys.argv) < 2:
        print("Usage: python client.py <path_to_server_script>")
        sys.exit(1)

    client = MCPClient()
    try:
        await client.connect_to_server(sys.argv[1])
        await client.chat_loop()
    finally:
        await client.cleanup()

if __name__ == "__main__":
    import sys
    asyncio.run(main())

这段代码实现了一个 MCP 客户端，用于连接 MCP 服务器，并利用 OpenAI 的 API 进行 Function Calling（函数调用）。该客户端能够与 MCP 服务器交互，列出可用工具，并根据用户输入选择适当的工具调用。

初始化

AsyncExitStack() 处理多个异步上下文（如 MCP 连接）。

读取 .env配置：

OPENAI_API_KEY

BASE_URL（可选，用于自定义 API 代理）

MODEL（指定 OpenAI 使用的模型）

self.client = OpenAI(...) 创建 OpenAI API 客户端。

转换 API 格式 (transform_json)

OpenAI 和 Claude API 的 Function Calling 格式不同。

该函数将 Claude 的 input_schema 转换为 OpenAI 兼容格式。

连接 MCP 服务器

连接 MCP 服务器，支持 Python 或 JavaScript 服务器脚本。

stdio_client(server_params) 通过 stdio 进行通信。

await self.session.list_tools() 列出 MCP 服务器上可用的工具。

处理用户查询 (process_query)

获取 MCP 服务器上可用的工具 (list_tools)。

让 OpenAI 选择是否需要调用 MCP 服务器上的工具 (tool_calls)。

若需要工具调用：

解析 tool_calls

call_tool(tool_name, tool_args) 调用 MCP 服务器上的工具

再次向 OpenAI 提交新信息，获取最终答案

交互式聊天 (chat_loop)

允许用户输入查询，自动选择 MCP 工具或直接回答。

输入 quit 退出聊天。

然后创建配置文件.env：

并手动输入

BASE_URL=
MODEL=
OPENAI_API_KEY=

随后在命令行输入，来运行mcp客服端和项目的启动代码

uv run client.py rag_server.py

运行项目