文章目录

1、MCP技术体系介绍

1_MCP介绍

MCP（Model Context Protocol，模型上下文协议），旨在为大语言模型（LLM）提供统一的、 标准化的方式与外部数据源和工具之间进行通信。

从本质上来说，MCP 是一种技术协议，一种智能体 Agent 开发过程中共同约定的一种规范。

这就好比秦始皇的书同文、车同轨，在统一的规范下，大家的协作效率就能大幅提高，最终提升智能体 Agent 的开发效率。

img

传统AI集成的问题：这种为每个数据源构建独立连接的方式，可以被视为一个 M*N 问题。

问题：架构碎片化，难以扩展，限制了AI获取必要上下文信息的能力

MCP解决方案：提供统一且可靠的方式来访问所需数据，克服了以往集成方法的局限性。

截止目前，已上千种 MCP 工具诞生，在强悍的 MCP 生态加持下， 人人手搓 Manus 的时代即将到来。

特征 MCP TCP/IP、HTTPS
本质 协议（Protocol） 协议（Protocol）
作用 标准化 AI 模型与上下文来源/工具之间的数据交互方式 标准化设备之间的网络通信方式
目标 让不同模型应用可以用统一方式访问资源/工具 让不同设备、系统可以互通数据
好处 消除碎片化集成、形成生态闭环 解决设备互联、实现互联网基础

2_MCP推广时间线

2024年11月底，Anthropic 推出了 MCP，刚开始不温不火。目标就是能在 Agent 的开发过程中，让大模型更加便捷地调用外部工具。

2025年2月份， Cursor 正式宣布加入 MCP 功能支持， 一举将 MCP 推到了全体开发人员面前！

2025年3月27日，OpenAI 智能体支持 MCP。 OpenAI 联合创始人兼首席执行官 Sam Altman 也特意发文大赞 MCP，可见其对 Agent 的重要性。

3_Function calling到MCP

能调用外部工具，是大模型进化为智能体 Agent 的关键，如果不能使用外部工具，大模型就只能是个简单的聊天机器人，甚至连查询天气都做不到。

由于底层技术限制啊，大模型本身是无法和外部工具直接通信的，因此 Function calling 的思路，就是创建一个外部函数（function）作为中介，一边传递大模型的请求，另一边调用外部工具达成某一目的，最终让大模型能够间接的调用外部工具。

img

但唯一的问题就是，编写这个外部函数的工作量太大了，一个简单的外部函数往往就得上百行代码，而且，为了让大模型“认识”这些外部函数，还要额外为每个外部函数编写一个 JSON Schema 格式的功能说明，此外，我们还需要精心设计一个提示词模版，才能提高 Function calling 响应的准确率。

而 MCP 的目标，就是能在 Agent 开发过程中，让大模型更加便捷的调用外部工具，统一 Function calling 的运行规范。

首先是先统一名称，MCP 把大模型运行环境称作 MCP Client，也就是 MCP 客户端，同时，把外部函数运行环境称作 MCP Server，也就是 MCP 服务器，

img

接下来，统一 MCP 客户端和服务器的运行规范，并且要求 MCP 客户端和服务器之间，也统一按照某个既定的提示词模板进行通信。

这样的好处就是：可以避免 MCP 服务器的重复开发，也就是避免外部函数重复编写。

例如，像查询天气、网页爬取、查询本地 MySQL 数据库这种通用的需求，大家有一个人开发了一个服务器就好，开发完大家都能复制到自己的项目里来使用，不用每个人每次都单独写一套。

4_支持MCP的平台

Github查看：

其它平台：

5_MCP的架构

MCP 采用客户端-服务器架构，其中 MCP 主机与一个或多个 MCP 服务器建立连接。

MCP 主机通过为每个 MCP 服务器创建一个 MCP 客户端来实现这一点。

每个 MCP 客户端与其对应的 MCP 服务器保持一对一的专用连接。

MCP架构的主要参与者包括：

  • MCP Host：协调和管理一个或多个 MCP 客户端的 AI 应用程序
  • MCP 客户端：维护与 MCP 服务器的连接并从 MCP 服务器获取上下文以供 MCP 主机使用的组件
  • MCP 服务器：为 MCP 客户端提供上下文的程序
  • Resources：向 AI 应用程序提供上下文信息的数据源（例如文件内容、数据库记录、API 响应）

每个 Agent 可以有多个 MCP Client，MCP 客户端与 MCP 服务器之间是一对一关系。

MCP Host (AI Application)
One-to-one
connection
One-to-one
connection
One-to-one
connection
MCP Client 1
MCP Client 2
MCP Client 3
MCP Server 1
(e.g., Sentry)
MCP Server 2
(e.g., Filesystem)
MCP Server 3
(e.g., Database)

6_MCP通信方式

MCP由两层组成：

  • 数据层：定义基于 JSON-RPC 的客户端-服务器通信协议，包括生命周期管理，以及核心原语，如工具、资源（文件或API响应内容等）、提示和通知。
  • 传输层：定义实现客户端和服务器之间数据交换的通信机制和渠道，包括传输特定的连接建立、消息框架和授权。

从概念上讲，数据层是内层，而传输层是外层。

MCP 支持的三种传输机制：

  • Stdio（本地操作，默认）：使用标准输入/输出流与同一台机器上的本地进程之间直接进行通信，提供最佳性能且无网络开销。
  • Server-Sent Events（SSE）：基于 HTTP 协议的流式传输机制，它允许服务器通过 HTTP 单向推送事件到客户端（即将被废弃）。
  • Streamable HTTP：基于 HTTP 协议的双向流式传输，可选用服务器发送事件来实现流式传输功能（HTTP 及 流式）。

stdio方式：

  • 简单、成本低：适用于客户端和服务器同机运行，无需外部服务器。
  • 通信速度快：无网络依赖，适合本地快速响应应用。
  • 可靠性高：本地环境运行，受网络干扰小，易调试。
  • 配置复杂：需提前安装命令行工具。
  • 升级不便：不支持热更新，更新需重启客户端。
  • 并发能力弱：单进程一对一通信，无法并行处理多个客户端请求（不支持分布式）。
  • 资源消耗高：本地运行大量服务时开销大，不适合复杂分布式场景。
  • 适用于本地开发、命令行工具、调试环境，或者模型和工具服务在同一进程内运行的情况。

SSE 方式（也是HTTP）：

  • 通过 HTTP GET 请求建立与服务器的连接，服务器以流式方式持续向客户端发送数据，客户端通过解析流数据来获取实时信息。
  • 这种方式即将被废弃，因为当 Client 和 Server 之间存在网络的中断时，无法感知到消息的丢失，发送失败的消息直接会被丢弃。
  • 适合实时推送和客户端/浏览器的单向通知，但无法满足双向复杂交互需求。
  • 既没有多通道并发同时也不够稳定，很难真正的适用于企业级应用场景。
  • 所以在新版本中被 Streamable HTTP 替代，两者之间其它部分差别不大。
  • 适用于需要服务器主动推送数据的场景，如实时聊天、天气预报、新闻更新等。

Streamable HTTP 方式：

  • 客户端通过 HTTP POST 向服务器发送请求，并可以接收流式响应（如 JSON-RPC 响应或 SSE 流）
  • 当请求数据较多或需要多次交互时，服务器可以通过长连接和分批推送的方式进行数据传输。
  • 配置简单：基本上配置就一个链接地址
  • 面向服务的：采用服务化设计，接口和功能模块独立，方便集成和复用，符合微服务架构理念。
  • 支持高并发：能够处理大量并发请求，保证在高访问量场景下仍有稳定响应。
  • OAuth2 权限控制：兼容 HTTP OAuth2 认证授权机制，支持细粒度的用户访问权限管理，保证数据安全。
  • 分布式云服务：支持部署在云端和分布式环境中，实现跨节点的负载均衡和高可用。
  • 热更新：服务端进行升级、客户端不必重启。
  • 适用于需要支持高并发、低延迟通信的分布式系统，尤其是跨服务或跨网络的应用。

MCP 传输方式优劣势对比：

特性 Stdio SSE Streamable HTTP
通信方向 双向（但仅限本地） 单向（服务器到客户端） 双向（适用于复杂交互）
使用场景 本地进程间通信 实时数据推送，浏览器支持 跨服务、分布式系统、大规模并发支持
支持并发连接数 低 中等 高（适合大规模并发）
适应性 局限于本地环境 支持浏览器，但单向通信 高灵活性，支持流式数据与请求批处理
实现难度 简单，适合本地调试 简单，但浏览器兼容性和长连接限制 复杂，需处理长连接和流管理
适合的业务类型 本地命令行工具，调试环境 实时推送，新闻、股票等实时更新 高并发、分布式系统，实时交互系统

注意：对于 Java 程序员来说 Spring AI 1.0.x 不支持 Streamable HTTP，目前仅有 SSE。

7_MCP工作流程

API 主要有两个

  • tools/list：列出 Server 支持的所有工具。
  • tools/call：Client 请求 Server 去执行某个工具， 并将结果返回。
User MCP Client MCP Server LLM 初始化阶段 启动客户端 连接服务器 确认连接 请求可用工具列表 返回工具列表和描述 查询处理阶段 输入查询 发送查询和可用工具信息 返回响应（文本或工具调用） 执行工具调用 返回工具执行结果 发送查询和工具结果 是文本响应 显示响应 alt [是工具调用] loop [工具调用循环] User MCP Client MCP Server LLM

2、MCP工具接入

目前网络上 MCP 服务 Stdio 接入方式大多数为 TypeScript 和 Python 的方式，分别对应着 uvx 和 npx 两种指令。

1_准备工作

提前下载并配置 Python 或 Node.js。

第1种：若已配置 Python 环境，可使用以下命令安装：

pip install uv

第2种：在 Windows 下可以通过 PowerShell 运行命令来安装：

powershell -ExecutionPolicy ByPass -c "irm https://astral.sh/uv/install.ps1 | iex”

npx 为 npm 自带的包管理工具，无需额外安装。

2_Cursor中使用MCP

Cursor 中国区官网：https://www.cursor.com/cn 。

在 Cursor 中添加 MCP Server 有两种配置方式：

  • 全局设置：通过 Cursor Settings -> MCP -> Add new global MCP server 添加全局可用的 MCP 服务。
  • 项目级别：在项目目录的 .cursor 目录中新建 mcp.json 文 件进行配置，仅对特定项目生效。

全局设置步骤如下：

1、点击右上角的齿轮按键，弹出设置栏。2、点击 Tool & Integrations 然后通过写入JSON的形式导入MCP Server。

img

项目级别设置如下：

1、在项目根目录创建 .cursor 文件夹（如果不存在）。2、在该文件夹中创建 mcp.json 文件。

img

在“支持的MCP平台”中选择一个服务添加到 mcp.json 或全局配置中（以百度地图为例）

{
  "mcpServers": {
    "baidu-map": {
      "command": "cmd",
      "args": [
        "/c",
        "npx",
        "-y",
        "@baidumap/mcp-server-baidu-map"
      ],
      "env": {
        "BAIDU_MAP_API_KEY": "xxx"
      }
    }
  }
}

然后在设置中启用，如果出现绿色则配置成功

img

测试验证

img

3_Cline中使用MCP

在 VSCoed 中打开 Cline 点击 MCP Servers 中的设置

img

打开 MCP 配置文件 cline_mcp_settings.json 后添加如下配置：

{
  "mcpServers": {
    "baidu-map": {
      "command": "cmd",
      "args": [
        "/c",
        "npx",
        "-y",
        "@baidumap/mcp-server-baidu-map"
      ],
      "env": {
        "BAIDU_MAP_API_KEY": "xxx"
      }
    }
  }
}

配置成功后即可看到如下信息：

img

3、手动开发MCP

创建 mcp 项目：

uv init mcp-project
cd mcp-project
# 添加依赖项
uv add mcp
# 创建环境文件
touch .env

需要在 .env 中配置模型的基本信息。

1_Stdio方式

编写 server-stdio.py 服务端代码，以加法计算为例

# serve.py
from mcp.server.fastmcp import FastMCP
# create an mcp server

mcp = FastMCP("Demo")

# Add an addition tool
@mcp.tool()  # 类似于http 的post请求，调用后会产生一个副作用
def add(a: int, b: int)-> int:
    """add two numbers"""
    return a * b # 错误的逻辑方便进行验证

# Add a dynamic greeting resource

@mcp.resource("greeting://{name}")# 类似于http 的get请求
def get_greeting(name: str)-> str:
    """Get a personalized greeting"""
    return f"Hello, {name}!"


if __name__ == "__main__":
    mcp.run(transport='stdio')

编写 client-stdio.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("DASHSCOPE_API_KEY")  # 读取 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("！！！！！未找到 API Key，请在 .env 文件中设置 DASHSCOPE_API_KEY")
        self.client = OpenAI(api_key=self.openai_api_key, base_url=self.base_url)  # 创建OpenAI client
        self.session: Optional[ClientSession] = None
        self.exit_stack = AsyncExitStack()

    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)

        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("\n MCP 客户端已启动！输入 'quit' 退出")

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

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

            except Exception as e:
                print(f"\n 发生错误: {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())

代码说明

代码部分 作用
MCPClient.__init__() 初始化 MCP 客户端
connect_to_mock_server() 模拟 MCP 服务器连接
chat_loop() 提供交互式聊天界面
cleanup() 释放资源
main() 启动客户端
asyncio.run(main()) 运行程序

运行代码：

uv run client-stdio.py server-stdio.py

测试结果：

img

2_SSE方式

创建 server-sse.py，仅将传输方式修改为 sse，启动服务端：

mcp.run(transport='sse')

创建 client-sse.py，仅修改 connect_to_server 方法，使用 sse 方式建立连接：

    async def connect_to_server(self, url: str):

        # 启动 MCP 服务器并建立通信
        sse_transport = await self.exit_stack.enter_async_context(sse_client(url))
        self.read, self.write = sse_transport
        self.session = await self.exit_stack.enter_async_context(ClientSession(self.read,self.write))

        await self.session.initialize()

        # 列出 MCP 服务器上的工具
        response = await self.session.list_tools()
        tools = response.tools
        print("\n已连接到服务器，支持以下工具:", [tool.name for tool in tools])

运行代码：

uv run client_sse.py http://127.0.0.1:8000/sse

测试结果

img

3_Streamable HTTP方式

创建 server-http.py，将传输方式修改为 streamable-http，启动服务端：

mcp.run(transport='streamable-http')

创建 client-http.py，仅修改 connect_to_server 方法，将传输方式变为 streamable-http：

    async def connect_to_server(self, url: str):
        # 启动 MCP 服务器并建立通信
        sse_transport = await self.exit_stack.enter_async_context(streamablehttp_client(url))
        # 解构,这里三个参数
        self.read, self.write, _ = sse_transport
        self.session = await self.exit_stack.enter_async_context(ClientSession(self.read, self.write))
        await self.session.initialize()
        # 列出 MCP 服务器上的工具
        response = await self.session.list_tools()
        tools = response.tools
        print("\n已连接到服务器，支持以下工具:", [tool.name for tool in tools])

运行代码：

uv run client_sse.py http://127.0.0.1:8000/mcp

测试结果

img

4_MCP Inspector功能介绍

在实际开发 MCP 服务器的过程中，Anthropic 提供了一个非常便捷的 debug 工具：Inspector。

借助 Inspector，我们能够非常快捷的调用各类 server，并测试其功能。

首先，安装 nodejs：

curl -fsSL https://deb.nodesource.com/setup_20.x | sudo bash -
sudo apt install -y nodejs

运行 Inspector：

npx -y @modelcontextprotocol/inspector uv run server.py

然后即可在本地浏览器查看当前工具运行情况：

img

4、总结与补充

Streamable HTTP 请求响应顺序

Client Server 初始化阶段 POST /mcp initialize (协商协议版本 & 能力) InitializeResponse protocolVersion & capabilities POST /mcp notifications/initialized (客户端已就绪) 202 Accepted POST /mcp tools/list (工具清单请求) 返回工具列表 result.tools[] + nextCursor 工具调用阶段 POST /mcp tools/call params.name=add params.arguments.a=123 返回工具执行结果 result.content[] 推送 SSE 消息 (执行进度/中间结果) loop [连接保持打开时] SSE 事件: 最终执行结果 result.content[] alt [单次HTTP响应] [服务器开启SSE流] 客户端通知/响应 POST /mcp ... notification/response ... 202 Accepted 服务器请求 GET (保持长连接获取SSE) Mcp-Session-Id: 1868a90c... 推送 SSE 消息 (来自服务器的请求/数据) loop [连接保持打开时] Client Server

HTTP 进程生命周期可以分为三个阶段：初始化、工作、连接关闭。

HTTP流式传输服务器与客户端之间的通信流程，以及外部工具信息同步格式与流程：

LLM Client Server Discovery tools/list List of tools Tool Selection Select tool to use Invocation tools/call Tool result Process result Updates tools/list_changed tools/list Updated tools LLM Client Server

消息的传递格式都是 JSON-RPC 形式的，这是一种用 JSON 编写的、结构化的远程调用协议：

类型 字段 说明
请求 jsonrpc 固定为 "2.0"
id 请求编号，用于对应请求与响应
method 要调用的方法名（比如 "tools/call"）
params 方法参数（可以是对象或数组）
响应 jsonrpc 也要写 "2.0"
id 与请求的 ID 一致
result 成功返回值（只需 result）
error 如果出错则返回 error 对象
# 人工智能 # 大数据 # 数据库 # python
Logo
武汉城市开发者社区

为武汉地区的开发者提供学习、交流和合作的平台。社区聚集了众多技术爱好者和专业人士，涵盖了多个领域，包括人工智能、大数据、云计算、区块链等。社区定期举办技术分享、培训和活动，为开发者提供更多的学习和交流机会。

更多推荐

cover

吴亚洲：企业AI落地的关键——从幻觉走向可靠的工业级 AI标准

avatar 武汉城市开发者社区
cover

高通骁龙8精英版第五代发布，游戏与AI性能大幅升级

avatar 武汉城市开发者社区

介绍 τ-bench：一个评估语言智能体在真实场景中与人、工具、规则交互能力的新基准

近年来，基于大语言模型（LLM）的智能体（agent）在自动化任务处理方面展现出巨大潜力。然而，现有的评测基准大多聚焦于单轮指令执行或纯工具调用，很少涉及以及这些在实际应用中至关重要的能力。为了填补这一空白，来自 Sierra 的研究团队提出了（Tool-Agent-User Interaction Benchmark），一个专注于评估语言智能体在中与用户交互、使用工具并遵循领域规则能力的全新基准

avatar 武汉城市开发者社区