MCP相关知识以及三种通信机制对比

MCP先导知识

假设在PC本地用Python写了一个程序，这个程序可以爬取某个UP主的粉丝数量，只要把LLM和这个爬虫程序结合起来，把程序返回结果传递给LLM，LLM就能知道某UP主的粉丝数，相当于LLM突破了预训练知识库的知识边界限制，能更好地回答用户提出的预训练知识库之外的知识。

如何让LLM运行用户本地程序呢?

程序一般在用户本地PC上，LLM部署在模型提供商那里，跟本地程序不在一个局域网上，LLM无法直调用程序。但没有什么是加一层中间层解决不了的。可以再写一个程序（中介程序），通过API的方式来访问LLM。中介程序会把用户提出的问题连同本地工具库的使用说明书一起发送给LLM，LLM分析用户提出的问题之后，决定是否需要使用工具，如果需要调用工具，就会向中介程序返回调用信息，说明要调用某个工具以及调用时传入的参数，中介程序收到后，按照AI的要求调用本地或远端工具，并把工具调用结果发送给LLM，LLM分析工具输出的内容后，整合最终答复返回给了中介程序，用户能通过中介程序看到大模型回复问题的结果。

中介程序可以是Claude Desktop、Cherry Studio、Cursor、Cline、Dify、Coze等，既有桌面客户端，也有IDE插件的方式，或者是Web方式。

可以把MCP协议类比为HTTP协议，MCP Host类比为各类浏览器，人主要与浏览器(相当于中介程序)打交道，MCP的原理与通过HTTP协议访问各种Server API相似。

工具使用说明书应该如何传递给大模型？

只要程序可交互，就一定有接口标准！！MCP只是制定了一个信息交换标准，信息交互标准由具体的MCP SDK实现，把最核心的Server端逻辑代码交给用户编写，其他的都交给MCP SDK内部实现

最开始的时候，工具使用说明书直接写在用户提示词里，用户可以自定义对接方式，把工具名称、功能、参数、返回值都描述清楚即可。约定好如果要调用某个工具，返回的内容应该是什么样的，根据LLM返回的内容可以自己写程序进行对接，然后就能调用本地程序或远端程序接口了，但是早期的时候AI还不够聪明，遵循指令的效果并不好，无法按照提前约好的提示词内容返回准确的信息（智商不够，无法理解用户提示词里的自定义规范），所以无法达到很好地使用工具的效果。

---

# 上述翻译
您可以通过函数调用让模型访问您自己的自定义代码。根据系统提示和消息，模型可能会决定调用这些函数——而不是（或除了）生成文本或音频。然后，您将执行函数代码，发回结果，模型会将它们合并到最终响应中。

后来，OpenAI在他们提供的ChatGPT模型里定义了规范，把工具传递的过程进行了标准化（Function calling），规范了工具使用说明书应该放在哪里、传递格式是什么、调用的返回格式是什么，把相关内容都约定好。实质上是独立使用JSON格式的交互信息来描述工具的相关信息，实现和用户输入提示词分离（不写在输入的提示词里），以便AI看到JSON格式的信息时能清楚各个工具的用法；

好好品鉴一下这段话

工具说明全放在提示词里是为了兼容市面上所有模型,其实openAI的function calling本质也是放在提示词里,只不过gpt的代码里给你封装了一层叫function calling的东西罢了,function calling也只是把指令按照约定的格式包装成提示词而已，只不过这个大模型可能针对这种格式做过调优，或者仅仅是系统提示词里有针对性的描述而已。一切大模型的调用最终都还是回到文本提示词。要记住,模型永远只能接受一段文本,是的,只有一段,无法并行读取,也无法分开,不管任何技术都是合并到一堆字符串里去,能不能完美理解出来全看LLM的智能程度,gpt刚出来的时候,我看到function calling就没去学了,因为这玩意只能GPT用,无法泛化到所有模型,现在的MCP封装了一层,弥补了function calling的短板,但如果你用的是ChatGPT api,那么MCP可以说是废的,多此一举,因为更消耗token,由于function calling是gpt源码里内置的,所以使用function calling的部分不计算token,更省钱,但现如今的技术肯定更看重泛化能力,谁都不想被一个LLM绑死,所以MCP是主流,但我认为这也不过是过渡性的产物罢了,海量的token消耗是个必须要解决的问题,如果MCP的源码在用GPT的时候,会自动转换成function calling来节省token,到后面LLM为了吸引用户,肯定会都内置function calling功能,这时MCP就变成了臃肿的老框架,因为它有很多适配的底层代码,这时候又会有一个新框架出现,主打效率、干净、整洁、无污染、低token,真香,计算机就是这样的,总是在封装,但封装过渡又无法很好向下兼容,所以又重开一个框架主打效率、简化,总是这么跳来跳去。

# function calling工具调用示例代码，要自己写一大堆工具schema，很麻烦

from openai import OpenAI

client = OpenAI()

# 要手动针对每一个工具做一大堆描述，工具越多，token消耗越多、context占用越多、需要手工写入的工具描述也越多
tools = [{
    "type": "function",
    "function": {
        "name": "get_weather",
        "description": "Get current temperature for a given location.",
        "parameters": {
            "type": "object",
            "properties": {
                "location": {
                    "type": "string",
                    "description": "City and country e.g. Bogotá, Colombia"
                }
            },
            "required": [
                "location"
            ],
            "additionalProperties": False
        },
        "strict": True
    }
}]

completion = client.chat.completions.create(
    model="gpt-4.1",
    messages=[{"role": "user", "content": "What is the weather like in Paris today?"}],
    tools=tools
)

print(completion.choices[0].message.tool_calls)

# LLM output——返回给本地的工具调用信息格式
[{
    "id": "call_12345xyz",
    "type": "function",
    "function": {
        "name": "get_weather",
        "arguments": "{\"location\":\"Paris, France\"}"
    }
}]

从HTTP到MCP

在Coze和Dify等低代码开发平台上，外部工具可以通过HTTP接口的方式暴露出来，只需要在平台上把外部工具的API schema配置好（API schema又是dify自定义的一种接口功能参数描述方法，可能会比function calling简单一点），Dify把API schema信息转换为标准的Function Calling工具说明书格式（套娃行为），从而实现让AI调用外部工具。在Dify上，每个外部工具都要编写专属的API schema，才能内置转换为LLMFunction Calling格式的说明书，就像使用function calling需要为每一个tool配置json格式的说明一样，底层实现的是一对一的转换。一个AI智能体可能有上百个外部工具，每一个都需要编写API schema超级麻烦，如果后续接口有变化还要手动进行更新维护，费时费力。

这种操作过程太过麻烦，最好是在写MCP Server的时候就顺手把上面的事情做了（实际应该是MCP SDK内部实现的），只需要在开发MCP Server时写清注释和参数类型即可（标准注释类型，既包含函数功能说明，也包含参数和返回值说明）。此时外部工具不必由开发人员手动将HTTP API暴露出来，而是套上了一层MCP壳，通过这个壳统一对外提供服务。Client和Server之间使用的通信协议就是MCP协议，容纳MCP Client的中介程序就是MCP Host。

之后，可以使用MCP SDK快速开发外部工具，通过装饰器@mcp.tool("func",description='xxx')的方式非常简洁地实现工具函数，无需再编写web server来暴露函数（SSE和Streamable HTTP方式会自动暴露，stdio方式无需暴露），也无需编写繁琐的API schema文件来描述工具函数，里面的转换细节都由MCP SDK封装好了。

编写MCP Server代码

Server的工具函数备注内容非常重要，传统的备注是给程序员看的，但是MCP服务的备注是给大模型看的，它通过注释信息理解接口的功能和使用方法。在下面的get_desktop_files()方法上增加@mcp.tool()装饰器，是MCP SDK定义的规范，用于MCP的客户端自动发现MCP Server。

然后就可以在MCP Host中配置MCP Server了，配置方法参见各种工具，如Cursor、Cherry Studio、VSCode等。

import os
from mcp.server.fastmcp import FastMCP
mcp = FastMCP()
@mcp.tool()
def get_desktop_files():
    """获取桌面上的文件列表"""    
    return os.listdir(os.path.expanduser("~/Desktop"))
if __name__ == "__main__":
    mcp.run(transport='stdio')	# 选择stdio通信方式运行
    # mcp.run(transport='sse')	# 也可以选择sse通信方式，这种方式会默认启动一个Web Server

什么是MCP？

回顾计算机发展史，TCP/IP、HTTP等标准协议的建立曾催生互联网革命；在AI时代，MCP协议有望扮演类似的角色，成为智能应用互联互通的基石。

现在的LLM模型具有强大的推理和泛化能力，但LLM本身受限于预训练时使用的训练集数据，无法实时获取最新的知识内容，LLM本身也无法使用外部工具获取外部数据源。MCP的目的就是为了解决LLM因数据孤岛限制而无法充分发挥本身潜力的问题，让LLM获得连接外部数据的能力。有了MCP，LLM能进一步走向智能体。通过MCP服务，大模型不再局限于只扮演大脑的角色，而是拥有了四肢，拥有了行动能力和获取数据的能力。

MCP（Model Context Protocol,模型上下文协议），是由 Anthropic在 2024 年 11 月推出并开源的一项创新标准。它是一种开放标准协议，用于将 AI 模型连接到各种外部工具和数据源，它定义了一套规则，告诉 AI 如何调用工具，包括参数名、参数类型和参数描述等信息，还能响应 AI 的 Toolcall 并返回结果，让 AI 携带结果继续发起对话；

简单来说，MCP就像AI模型的”万能转接头”，让大模型能够以标准化的方式与外部数据源或工具进行交互，类似于TYPE-C能让不同设备能够通过相同的接口连接到PC。

MCP官方文档：Model Context Protocol

# 官方描述
Connect your AI applications to the world
MCP is an open protocol that standardizes how applications provide context to large language models (LLMs). Think of MCP like a USB-C port for AI applications. Just as USB-C provides a standardized way to connect your devices to various peripherals and accessories, MCP provides a standardized way to connect AI models to different data sources and tools. MCP enables you build agents and complex workflows on top of LLMs and connects your models with the world.

# Translate
MCP 是一个开放协议，它标准化了应用程序向大型语言模型 (LLM) 提供上下文的方式。MCP 就像 AI 应用程序的 USB-C 端口一样。正如 USB-C 提供了一种标准化的方式将您的设备连接到各种外围设备和配件一样，MCP 也提供了一种标准化的方式将 AI 模型连接到不同的数据源和工具。MCP 使您能够在 LLM 之上构建agents和复杂的工作流，并将您的模型与世界连接起来。

MCP架构与工作流程

核心架构

MCP遵循客户端-服务器架构（Client-Server，C—S架构），包含以下核心组件：

• MCP主机（MCP Hosts）

  • 发起API请求的LLM应用程序（Cursor、各种AI工具等）

  • 负责接收用户输入的提示词输入并与LLM进行交互

• MCP客户端（MCP Clients）

  • 位于MCP Hosts内部，与MCP Server保持1:1的连接

  • 充当LLM和MCP Server之间的桥梁

• MCP服务器（MCP Servers）

  • 自定义开发的各种工具、接口

  • 负责处理客户端发来的调用请求，根据传来的参数执行相应操作后返回执行结果

• 资源（Resources）

  • 本地资源：本地计算机中可供MCP server安全访问的资源（如文件、数据库）

  • 远程资源：MCP server可以连接的远程资源（如通过API访问服务）

工作流程

MCP的基本工作流程如下：

• 初始化连接

  • 客户端向服务器发送连接请求，建立通信通道

• 工具发现与选择

  • MCP客户端从MCP服务器获取可用的工具列表

  • 将用户的问题连同工具描述一起发送给LLM

  • LLM根据用户的问题决定是否需要使用工具以及使用哪些工具

• 工具调用与执行

  • 如果需要使用工具，MCP客户端会通过MCP服务器执行相应的工具调用

  • MCP服务器处理请求，执行相应的操作（如查询数据库、读取文件等）

• 结果处理与响应

  • 工具调用的结果会被发送回LLM

  • LLM基于所有信息生成自然语言响应

  • 最后将响应展示给用户

• 断开连接

  • 任务完成后，客户端可以主动关闭连接或等待服务器超时关闭

简而言之：Client 首先与 Server 握手并获取工具清单，Host 把这些清单连同用法注入系统提示词；LLM 根据提示输出结构化调用（通常是 JSON）；Client 解析调用并向 Server 发起调用请求，收到结果后回传 Host，Host 再携带最新上下文进行下一轮模型推理。

MCP应用场景

功能分类	功能项	说明
本地文件操作	文件读写	允许 AI 模型读取和创建本地文件
	文档分析	分析本地文档并提取信息
	代码管理	读取、分析和修改代码文件
数据库交互	数据查询	从本地或远程数据库获取信息
	数据分析	分析数据库中的数据并生成报告
	数据管理	更新、插入或删除数据库记录
API集成	网络服务调用	调用各种网络 API 获取信息
	第三方服务集成	与第三方服务（如天气、股票、新闻等）集成
	企业系统对接	与企业内部系统（如 CRM、ERP 等）对接
开发工具增强	IDE集成	增强代码编辑器和 IDE 的功能
	调试辅助	帮助开发者调试代码
个人助理功能	日程管理	访问和管理日历
	邮件处理	读取和撰写邮件
	信息整理	整理和归类个人信息

JSON-RPC 2.0 协议

MCP 中 Client 与 Server 使用 JSON-RPC 2.0 作为通信消息格式。JSON-RPC 是轻量级通信协议 一种轻量级的通信协议，其核心目标是屏蔽网络细节，使远程调用如同本地调用般简单，并可基于多种底层网络协议（如 TCP/HTTP）实现。

JSON-RPC 的设计原则是简单性和互操作性，旨在为不同编程语言之间提供一种方便的方法来执行远程过程调用。通过定义一组简单的规则，任何能够解析 JSON 的语言或平台都可以实现 JSON-RPC 客户端或服务器。

JSON-RPC 2.0中只定义了两种消息，就是请求和响应，通知是一种特殊的请求，特点是不包含id字段。

请求（Request）

请求由MCP客户端或MCP服务器之一发给对方，调用对方的一个方法，并预期获得一个响应。

{
  "jsonrpc": "2.0",
  "method": "方法名",
  "params": {"参数名": "值"} | ["值 1", "值 2"],  // 对象或数组
  "id": "唯一ID"                             // 可选（通知请求可省略）
}

响应（Response）

{
  "jsonrpc": "2.0",
  "result": "返回值",
  "id": "对应请求 ID"
}

MCP 与 Function Calling 的关系

Function Calling是个非常不错的设计思路。但该技术有很大的不足——编写外部函数的工作量太大，一个简单的外部函数往往就得上百行代码（注释+参数说明+实现逻辑），而且为了让大模型认识这些函数，还需要额外为每个外部函数编写一个JSON Schema格式的功能说明(tools调用时放在tool列表里)。此外，还要为函数设计一个提示词模板description, 才能提高Function Calling响应的准确率。这种开发模式使得开发者要为每个系统单独编写适配接口，处理不同的认证方式、数据格式和调用规范，即使同样的功能，不同的人也会有N种开发版本，造成开发过程费时费力且不利于后期维护。简单来说，Function Calling其实就是用JSON重新描述了一个函数（接口）+注释

MCP技术借助大模型Function Calling的基础能力，凭借高效的开发规范，被广为关注。MCP统一规范采用分布式架构，分为Client客户端和Server服务端两部分，客户端用户基于大模型编写(可以是.exe之类的可运行带GUI的程序，也可以是封装了LLM的py代码)，服务端暴露扩展大模型功能函数或接口。 MCP最突出的优势在于它的标准化程度：一次开发，多人复用。用户只需要编写一套客户端代码，就可以接入符合MCP开发规范的所有服务端（不管是自己编写的，还是利用别人编写的MCP Server）。这种统一的开发模式就像秦始皇的“书同文、车同轨”，有了统一的规范，只要本地运行的环境支持MCP协议，仅需几行代码就可以接入海量的外部MCP工具，提升智能体AI Agent的开发效率。Anthropic提供了一整套MCP客户端、服务端SDK等各种开发工具，并且支持Python， TypeScript和Java等多种语言，借助SDK, 仅需几行代码就可以快速开发一个MCP服务器。

例如，使用Function Calling技术开发天气查询功能要编写description, json schema前后大约150行代码，使用MCP SDK开发只需要编写一个请求函数get_weather并给出函数注释，借助@mcp.tool()装饰器就可以快速添加天气查询功能。

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

# 初始化MCP服务器
mcp = FastMCP("WeatherServer")

@mcp.tool()
async def get_weather(city: str):
    """
    输入指定城市的名称，返回当前天气情况
    :param city: 城市名称
    :return: json格式的天气信息
    """
    url="https://api.seniverse.com/v3/weather/now.json"
    params={
        "key": "注册的心知天气私钥",
        "location": city,
        "language": "zh-Hans",
        "unit": "c"
    }
    response = requests.get(url, params=params)
    temperature = response.json()['results'][0]['now']
    return json.dumps(temperature)

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

不同之处

MCP和Function Calling两种技术目标都是为了增强大模型的外部交互能力，但Function Calling技术采用集中式架构，函数定义，模型调用和执行逻辑是紧密耦合在单一应用中的。设计简单直接，适合小型项目但难以拓展，一遇到复杂项目代码就变得很臃肿。

MCP则采用了分布式架构，明确分离了MCP Server和MCP Client，很像现在开发项目的前后端分离模式，这种解耦使各组件能够独立演化，更适合企业级应用。

还有一点很重要，Function Calling通常采用同步执行模型，模型发起请求后等待响应，这在处理耗时操作时一定会导致性能瓶颈，比如当一个用户查询天气时，另一个用户只能排队等他查询完。MCP原生就支持异步操作，允许长时间任务在后台执行，通过回调机制通知用户结果。这种模式更符合真实业务场景，使AI能并行处理多个请求，显著提升吞吐量。

client.py -> 大模型客户端

该客户端代码最大的优势就是完全没有绑定具体函数调用和大模型的逻辑，也就是说该客户端代码无论面对哪种类型的MCP Server都可以成功执行，这样在以后开发中可以保持MCP Client的代码不变，专注于快速开发或者访问不同的MCP Server，可以便捷的使用大模型开发不同能力的AI Agent智能体，这就是MCP相比于Function Calling最大的技术优势！分布式的架构，解耦的代码等于更便捷快速的开发速度。

许多桌面级大模型连接工具如cherry studio、vscode都集成了MCP，其实就是在现有工具的代码上加上一个client.py的调用模块，再自定义一下读取用户设置的MCP server，就能实现支持MCP的功能了，没有很复杂，MCP只是完整客户端支持的一个小模块，还有很多其他的模块，例如通过API连接远程LLM等。

对比sse客户端代码和stdio客户端代码的区别，可以发现只是在连接服务函数层面有变化（connect_to_server (self, server_script_path)或者 connect_to_sse_server(self, server_url))。

• LLM ↔ Host：OpenAI/DeepSeek 的 SDK 自动把 function/tool calling 的 JSON 打包/解析。
• Host ↔ MCP Server：mcp 库里的 ClientSession 自动把调用转换成 JSON-RPC 2.0 请求并通过 stdio 传输；返回值也自动解析。

所以看不到显式的 JSON 拼接是正常的， 使用的代码只是在做“结构之间的映射”，不是在手写协议。

import asyncio
import json
from typing import Optional
from contextlib import AsyncExitStack
from openai import OpenAI

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

class MCPClient:
    def __init__(self):
        """初始化MCP客户端"""
        self.exit_stack = AsyncExitStack()
        self.opanai_api_key = "替换你的api_key" # 调用模型的api_key
        self.base_url = "https://api.deepseek.com" # 调用模型url, 这里以deepseek作演示
        self.model = "deepseek-chat" # 调用deepseek-v3模型
        self.client = OpenAI(api_key=self.opanai_api_key, base_url=self.base_url)
        self.session: Optional[ClientSession] = None # Optional提醒用户该属性是可选的，可能为None
        self.exit_stack = AsyncExitStack() # 用来存储和清楚对话中上下文的，提高异步资源利用率

    async def connect_to_server(self, server_script_path):	# stdio的server，因为是本地文件，需要传入本地代码的路径
        """连接到MCP服务器并列出MCP服务器的可用工具函数"""
        server_params = StdioServerParameters(
        command="python",
            args=[server_script_path],
            env=None
        ) # 设置启动服务器的参数, 这里是要用python执行server.py文件

        # 启动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() # 与服务器建立stdio连接

        # 列出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 Server可用的MCP工具"""
        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]

        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)
                print(f"\nDeepSeek-V3-0324: {response}")
            except Exception as e:
                print(f"发生错误: {str(e)}")

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

async def main():
    if len(sys.argv) < 2:
        print("使用方法是： python client.py server.py")
        sys.exit(1)

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

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

server.py

from mcp.server.fastmcp import FastMCP

# 创建MCP 服务
mcp = FastMCP('Demo')

@mcp.tool()
def add(a:int, b:int) ->int:
  """"
  计算两个整数的和并返回
  """
  return a+b

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

MCP Host与LLM之间的通信

LLM模型永远只能接受一段文本,是的,只有一段,无法并行读取,也无法分开,不管任何技术都是合并到一堆字符串里去,能不能完美理解出来全看LLM的智能程度

大模型（如 GPT-4/5）在返回推理结果时，只会输出结构化的 JSON，这就是 function calling / tool calling 的产物。本身不会直接发 HTTP 或 RPC 请求去调用 MCP Server，它只是根据上下文生成或提示词生成“工具调用”的 JSON 结构化输出。

• 在 OpenAI API 里，就是 function_call / tool_call 的 JSON。
• 在 Anthropic、Claude、Gemini 里也有类似机制。

MCP Host 的作用相当于一个翻译器。LLM 把 JSON 格式化结果交给 Host，Host 内部的 MCP Client 模块会把它转换成 MCP 规定的 JSON-RPC 2.0 请求，这才是 MCP Server 能识别的标准调用。可以这么理解：

• function calling JSON 是 LLM 与 MCP Host 的接口格式；
• JSON-RPC 则是 MCP Client 与 MCP Server 的接口格式。

JSON-RPC 2.0 请求会通过一个 传输层（Transport） 发给 MCP Server，传输层可以是：

• stdio（进程间标准输入输出，最常见，延迟低）
• SSE（跨网络时）
• Streamable HTTP

``` LLM输出
{
  "tool_call": {
    "name": "query_database",
    "arguments": {
      "sql": "SELECT * FROM users LIMIT 10"
    }
  }
}


```
JSON-RPC 2.0 格式
{
  "jsonrpc": "2.0",
  "id": 42,
  "method": "tools/call",
  "params": {
    "name": "query_database",
    "arguments": {
      "sql": "SELECT * FROM users LIMIT 10"
    }
  }
}

MCP Client与MCP Server的三种通信机制对比

STDIO

STDIO（标准输入输出）是一种用于本地通信的传输方式。其核心思想是利用操作系统提供的管道机制实现父子进程间的数据交换，客户端通过标准输入发送请求，服务器通过标准输出返回响应。在这种模式下，客户端(MCP-Client)通过启动MCP Server可执行文件作为子进程， 双方通过约定的标准输入和标准输出进行数据交换。

这种方式适用于MCP-Client和MCP-Server都在同一台计算机上的场景，例如本地自行编写服务端代码或将别人编写的服务端代码pull到本地执行。设计极其简单直接，完全依赖操作系统内核提供的进程管理能力。无需配置网络环境，数据仅在本地流转，避免远程泄露风险，直接通过内存或管道传输，确保了高效、低延迟的通信性能；但无法同时处理多客户端请求，扩展性差，无法支持分布式或远程资源访问。

SSE(已停止，官网更新为Streamable HTTP)

SSE（Server-Sent Events，服务器发送事件） 是一种基于 HTTP 协议的单向通信机制，由服务器主动向浏览器单向推送数据; SSE是解决“HTTP请求只能由客户端主动向服务器发消息，不能由服务器主动向客户端发消息”问题的一种方案；就像一条单向的传送带，只能由服务器向客户端发送数据，客户端无法通过这条传送带将物品送回服务器，适合需要服务器持续向客户端更新的场景。

要注意SSE协议和WebSocket协议的区别，SSE协议是单向的，客户端和服务端建立连接后，只能由服务端向客户端进行消息推送。而WebSocket协议是全双工的，客户端和服务端建立连接后，可以双向通信。

从架构层面看，SSE构建在标准的HTTP协议之上，利用其长连接特性实现服务器到客户端的单向实时数据流，这种设计使其天然适合分布式系统和网络化部署场景， 适用于客户端和服务端位于不同物理位置的场景，尤其是对于分布式或远程部署的场景。SSE 对网络波动的适应性不强，一旦网络不太稳定，数据传输就容易卡顿甚至中断。

MCP Client和MCP Server 在HTTP SSE方式下通过两个主要通道进行通信：

• HTTP请求/响应：客户端通过标准HTTP请求发送消息到服务端（一般是POST，由client先发送一个post请求到服务器）
• 服务器推送事件（SSE）：通过专门的/sse端点向客户端推送消息

SSE的不足

虽然SSE这种设计方式简单直观，但它存在如下关键问题：

1. 不支持断线重连/恢复：SSE连接断开所有会话状态丢失，客户端必须重新建立连接并初始化整个会话。
2. 服务器需维护长连接：服务器必须为每个客户端维护一个长时间的SSE连接，大量并发用户会导致资源消耗剧增，当服务器重启或扩容时所有连接会中断影响用户体验和系统可靠性。
3. 服务器消息只能通过SSE传递：即使是简单的请求-响应交互、服务器也必须通过SSE通道返回信息，这就需要服务器一直保持SSE连接，造成不必要的复杂性和开销。
4. 基础设施兼容性限制：目前很多Web基础设施如CDN、负载均衡器、API网关等对长时间SSE连接支持性不够，企业防火墙有可能强制关闭超时SSE连接，造成连接不可用。

Streamable HTTP

在2025年3月26日，MCP官方Github仓库出现了采用“可流式传输的 HTTP”来替代现有的 HTTP+SSE 方案的Streamable HTTP提议（issue）。该提议详细说明了Streamable HTTP MCP 服务器与客户端之间的通信流程，以及外部工具调用信息同步格式与流程；

2025年5月9日，MCP 迎来重磅升级——Streamable HTTP正式发布，取代了HTTP SSE, 成为AI模型通信的新标准！

在Streamable HTTP传输中，服务器作为独立进程运行，可以处理多个客户端连接。传输使用 HTTP POST 和 GET 请求。服务器可以选择使用服务器发送事件(SSE) 来传输多条服务器消息。这不仅允许基本的 MCP 服务器，还允许功能更丰富的服务器支持流式传输以及服务器到客户端的通知和请求。

什么是Streamable HTTP？

“Streamable HTTP” 是一种基于标准 HTTP 协议的 流式传输机制，允许服务器在保持连接不断开的前提下，持续/分批地发送数据给客户端。这种模式适用于需要实时数据推送或大数据逐步传输的场景。

Streamable HTTP 是为了解决传统 SSE（Server-Sent Events）传输方式的局限性而提出的。SSE 需要长连接，且仅支持单向通信（服务器到客户端），导致其在某些场景下不够灵活。Streamable HTTP 则通过改进传输机制，支持更灵活的双向通信和无状态服务器运行。

Streamable HTTP 的核心特性

• 支持流式传输：Streamable HTTP 允许数据以流的形式分块传输，而不是一次性传输完整数据。这种方式可以减少客户端和服务器的内存占用，并提高实时性。
• 无状态服务器：服务器可以选择完全无状态运行，不再需要维持长期连接。这使得服务器资源利用率更高，更适合高并发场景。
• 兼容性与易用性：Streamable HTTP 基于标准的 HTTP 协议，兼容现有的 HTTP 基础设施，包括 CDN、API 网关和负载均衡等。

解决了SSE哪些问题？

1. Streamable HTTP如何解决SSE不支持断线重连的问题？

​ 答：Streamable HTTP在每次通信时会记录id编号对应请求与响应，将这里请求与响应存储可断线重连进行恢复。

2. Streamable HTTP如何解决SSE服务器需要维持长连接的问题？

​ 答：在需要发送响应过程中会保持连接，但一旦流式响应结束，服务器随后便会关闭流。

3. Streamable HTTP如何解决SSE服务器消息只能通过SSE传输的问题？

​ Streamable HTTP服务器可灵活选择是返回普通HTTP响应还是升级为SSE流，对于简单请求直接使用普通HTTP响应，对于内容复杂等需要流式传输的请求场景自动升级为SSE。

4. Streamable HTTP如何解决SSE服务器基础设施兼容性限制？

​ 答：Streamable HTTP各基础设施的兼容性很完备。

三种方式对比

特性	Stdio	SSE	Streamable HTTP
通信方式	本地进程管道	HTTP 长连接 + SSE	标准 HTTP + 动态流式升级
适用场景	本地隐私数据处理	实时远程通知	云原生、分布式系统
网络依赖	无	必需	必需
多客户端支持	否	是	是
协议演进	长期支持	已废弃	官方推荐替代方案

当前环境下MCP的不足之处

安全问题

• 认证机制混乱：缺乏统一认证标准：部分服务器采用OAuth 2.0 + MFA，部分甚至无API密钥保护。权限过大可能会损害数据安全，如误删本地文件等。
• 本地执行风险：通过 stdio 模式安装的第三方工具可能携带恶意代码（如伪装成PDF解析工具上传用户 .bash_history 文件）。

执行延迟问题

• 协议开销导致高延迟：MCP在处理高并发请求时，协议本身的复杂性（如多层通信架构、数据序列化）会引入显著延迟。在高负载场景下，其整体吞吐量甚至可能低于传统函数调用。尤其在边缘计算或分布式系统中，网络通信延迟会进一步放大这一问题。

• 上下文管理效率低下：工具描述、返回的JSON数据等占用大量上下文窗口，挤压模型推理空间。实验表明，当工具数量从5个增至20个时，模型指令遵循准确率从78%暴跌至34%，形成“上下文污染”的恶性循环。

大模型输出错误缺陷

• 错误传播与“幻觉”放大：当工具返回部分错误数据时，大模型的“脑补”能力会合理化错误，导致更危险的输出。例如：医疗AI将仪器“-1”错误码解读为“检测值偏低”，生成错误诊断建议等。
• 工具选择与参数解析错误：模型常混淆功能相似的工具（如混淆 search_flights 与 query_timetable）。
• 复杂逻辑处理失败：（如时区转换错误），在结构化任务（如航班预订）中成功率不足20%。

补充

MCP协议并没有规定MCP Host如何与 LLM 之间进行通信，只是规定了MCP Client与MCP Server之间的通信方式，至于和AI大模型之间的通信方式，常见的有两种方式：

• 可以直接用最原始的方式将工具说明书直接写到提示词里（简单粗暴），这种方法的好处是理论上来说任何指令遵循能力好大模型都可以使用MCP技术，但是可能会导致token爆炸，尤其是工具库包含很多工具时，可能会一次性传好几万token给大模型，如果LLM的API是按token计费方式则成本高昂；
• 也可以用Function Calling的方式传递将工具列表传递给AI大模型，但是对大模型有要求，必须支持Function Calling技术的大模型才行，本质上应该是模型底层实现了一种基于Function Calling的提示词技术，但是使用Function Calling是不计算在用户输入的token中的，应该还是离不开提示词技术。

成熟一点的MCP与LLM连接方案本质上还是通过调用模型的Function Calling功能实现的，只是将MCP Server代码端的注释、参数、返回值等json_schema通过MCP 的SDK进行转换，再填充到Function Calling的待补充参数之中，可从下面的示例代码中略窥一二。

async def process_query(self, query:str)->str:
        """使用大模型处理查询并调用MCP Server可用的MCP工具"""
        messages = [{"role":"user", "content":query}]
        response = await self.session.list_tools()

        available_tools = [{
            "type": "function",	# 还是离不开Function Calling
            "function": {
                "name": tool.name,
                "description": tool.description,
                "input_schema": tool.inputSchema
            }
        } for tool in response.tools]

        response = self.client.chat.completions.create(	
            model=self.model,
            messages=messages,
            tools=available_tools	# 把MCP对齐后的Function Calling传到工具列表中，还是用到Function Calling
        )