MCP技术实现和应用实战

【超详细】MCP技术细节和应用实战

MCP(Model Context Protocol,模型上下文协议),是一个开放标准,它让AI能够直接连接到外部工具,数据库和服务。

在MCP发布之前,开发者需要为每个AI接入应用编写特定的代码,过程繁琐,MCP出现就如同给AI打造一个USB-C接口,提供了标准化接入方式。

MCP

MCP 核心组件详解

MCP采用客户端-服务器架构,核心组件有:

  • MCP Host (宿主应用): 发起请求的llm应用程序,如claude desktop,cursor等或者其他ai工具
  • MCP Client(客户端):内置在宿主应用Host中,负责与MCP 服务器通信,而且每个客户端只会直接对接一个MCP服务器
  • MCP Server (服务器): 为MCP client提供对特定外部服务的访问
  • 传输层: 客户端和服务器之间的消息传递方式

Pasted image 20260419210607.png

MCP Client

在我自学MCP的时候,很大阻碍就是搞不明白MCP的Client和Server究竟有啥区别。后来我终于搞明白了:

  • MCP Client主要由Host应用内置自己实现,比如Claude Code,就是Anthropic自己做的,Cursor等其他的软件,也会有自己的实现
  • 作为开发者,我们大部分只需要关心写好server,选择传输层stdio还是SSE

Client的大概功能

  • MCP Client首先从 MCP Server 获取可用的工具列表 将用户查询联通工具描述一起发送给llm
  • LLM决定是否需要工具
  • 如果需要工具,MCP client会通过MCP server执行相应的工具调用,并将工具调用结果返回给llm
  • LLM 基于所有信息生成自然语言响应

MCP server

MCP sever 可以提供3种类型的功能:

  • 资源 --- 数据访问
  • 工具 --- 是ai可以调用来执行操作的函数
  • 提示 --- 可以复用的指令,用于引导ai如何使用工具和资源

三者如何协同:prompt结构化意图 -- tool 执行操作 -- 资源提供或者捕获数据

整体调用链路:

用户:提出xxxx需求 (涉及到MCP工具提供的服务)
↓
Claude(Host) 决定要调用MCP工具
↓
MCP Client 把调用请求序列化,通过stdin/http等传输协议发送给server进程
↓
MCP Server 我们开发的代码收到,执行服务,通过协议返回给Client 
↓
Claude 拿到数据,组织成自然语言回答 

通信机制

MCP协议支持一下通信机制:

传输方式 工作原理 适用场景
stdio Server 作为子进程运行,通过 stdin/stdout 通信 本地开发
HTTP + SSE 客户端发送 POST 请求,服务器通过 Server-Sent Events 流式响应 网络部署
Streamable HTTP 增强的 HTTP,支持流式传输 实时应用

STDIO和SSE机制都是用JSON-RPC2.0 格式进行消息传输,确保通信的标准化和可扩展性

对比Functional Calling

Function Calling MCP
本质 LLM 输出结构化的函数调用 完整的交互协议
范围 单次函数调用 发现 + 调用 + 响应
可移植性 供应商专属(OpenAI、Anthropic) 通用标准
生态 按应用隔离 所有工具共享 Server
复杂度 简单 更全面

启动与发现

以Claude Code为例,我们写好的Server,将需要配置一个Json文件,这个JSON需要放在约定的位置,Claude 启动的时候就能找到。

  • Mac: ~/Library/Application Support/Claude/claude_desktop_config.json
  • Windows: %APPDATA%\Claude\claude_desktop_config.json
┌─────────────────────────────────────────────────────────────┐
│  Claude Code 启动流程                                         │
└─────────────────────────────────────────────────────────────┘
                          │
                          ▼
        ┌─────────────────────────────────┐
        │  1. 读取配置文件                  │
        │  %APPDATA%\Claude\               │
        │  claude_desktop_config.json      │
        └─────────────────────────────────┘
                          │
                          ▼
        ┌─────────────────────────────────┐
        │  2. 解析 MCPServers 配置         │
        │  {                              │
        │    "mpc001": {             │
        │      "command": "python.exe",   │
        │      "args": ["-m", "src.server"]│
        │    }                            │
        │  }                              │
        └─────────────────────────────────┘
                          │
                          ▼
        ┌─────────────────────────────────┐
        │  3. 启动MCP服务器进程             │
        │  执行: python -m src.server      │
        │  在指定目录(cwd)下运行            │
        └─────────────────────────────────┘
                          │
                          ▼
        ┌─────────────────────────────────┐
        │  4. 建立MCP协议连接              │
        │  - 标准输入/输出通信             │
        │  - JSON-RPC消息格式              │
        │  - 服务器初始化握手              │
        └─────────────────────────────────┘
                          │
                          ▼
        ┌─────────────────────────────────┐
        │  5. 工具发现                     │
        │  - 服务器返回工具列表            │
        │  - Claude Code注册工具           │
        │  - 工具可用于后续调用            │
        └─────────────────────────────────┘

官方的MCP SDK

Anthropic 官方提供了ts,py和C#的sdk,以便开发者构建MCP客户端和服务器。

python SDK

安装方式

pip  install "MCP[cli]" #支持MCP规范,stdio,sse,streamable http等传输方式

#或者 

pip install MCP 

# 这两个区别在于,MCP只安装核心sdk,无法使用MCP命令,"MCP[cli]"安装MCP 核心sdk+cli工具依赖,可以获得完整的开发和调试体验

官方推荐用uv来管理py项目

FastMCP

FastMCP是一个用于构建MCP的框架,在官方的基础上提供了更加简单的方式来创建MCP服务器,使AI能够访问本地工具和资源。

核心功能:

  • 工具: 类似api的post端点
  • 资源:
  • 提示模板: 定义可重用的交互模式,支持结构化的消息序列
  • 图片处理: 内置图片数据处理,自动处理格式转换,支持工具和资源中使用。

对比官方SDK

官方提供了两层sdk,低级别的协议层和高级别的应用层

低级别api(协议层),可以完全控制通信细节

from MCP.server import Server, NotificationOptions
from MCP.server.models import InitializationOptions
import MCP.server.stdio
import MCP.types as types

# 创建服务器实例
server = Server("example-server")

# 注册工具处理器
@server.list_tools()
async def handle_list_tools() -> list[types.Tool]:
    return [
        types.Tool(
            name="add",
            description="Add two numbers",
            inputSchema={
                "type": "object",
                "properties": {
                    "a": {"type": "number"},
                    "b": {"type": "number"}
                },
                "required": ["a", "b"]
            }
        )
    ]

@server.call_tool()
async def handle_call_tool(
    name: str, 
    arguments: dict | None
) -> list[types.TextContent | types.ImageContent | types.EmbeddedResource]:
    if name == "add":
        result = arguments["a"] + arguments["b"]
        return [types.TextContent(type="text", text=str(result))]
    raise ValueError(f"Unknown tool: {name}")

# 运行服务器
async def main():
    async with MCP.server.stdio.stdio_server() as (read_stream, write_stream):
        await server.run(
            read_stream,
            write_stream,
            InitializationOptions(
                server_name="example-server",
                server_version="0.1.0",
                capabilities=server.get_capabilities(
                    notification_options=NotificationOptions(),
                    experimental_capabilities={},
                ),
            ),
        )

高级别API(server类) MCP.server.Server提供了更加方便的注册方式,但依然需要手动处理类型定义

from MCP.server import Server
from MCP.server.stdio import stdio_server
import MCP.types as types

server = Server("my-server")

@server.list_tools()
async def list_tools():
    return [
        types.Tool(
            name="calculate",
            description="Run a calculation",
            inputSchema={
                "type": "object",
                "properties": {
                    "operation": {"type": "string"},
                    "x": {"type": "number"},
                    "y": {"type": "number"}
                }
            }
        )
    ]

@server.call_tool()
async def call_tool(name: str, arguments: dict):
    if name == "calculate":
        op = arguments["operation"]
        x, y = arguments["x"], arguments["y"]
        
        if op == "add":
            result = x + y
        elif op == "multiply":
            result = x * y
        else:
            raise ValueError(f"Unknown operation: {op}")
            
        return types.CallToolResult(
            content=[types.TextContent(type="text", text=str(result))]
        )

MCP鉴权

MCP服务器认证主要分两种形式,API密钥 和 OAuth2.1 ,各有使用场景。

API密钥:适合简单场景的快速方案

很多MCP服务器在刚起步的时候会使用API密钥认证,这种方式简单直接,本地部署,快速制作原型很方便,但是在生产环境风险很大:

  • 权限无限制:拿到密钥就等于拿到服务器所有功能,没法精确限制操作范围
  • 轮换麻烦:密钥一旦泄露,所有用该密钥的地方都要替换,容易导致集成中断,服务停止。
  • 无过期和追溯性:API密钥通常永久有效,没法绑定到具体用户或者设备,审计或者事故响应很被动。

OAuth2.1:生产环境的标准方案

OAuth 2.0 是一个授权框架(Authorization Framework),它允许第三方应用在用户同意的前提下,有限制地访问用户在另一个服务上的资源,而无需获取用户的密码

举个例子,去商场停车,停车卡只能在停车场内使用,你需要第三方待取车服务帮你取车。 但是你不想将停车卡交给第三方服务。

OAuth的核心思想就是:你授权停车场服务,我允许这个代取服务在2小时内帮我取车,代取服务拿到一张临时通行证,在2小时内有效,过期就作废。

应用到MCP场景就是:

  • (用户) = 车主
  • 停车厂 = 金蝶 MCP Server(资源服务器,存放着你的业务数据)
  • 代取服务 = ChatBox 应用(第三方客户端)
  • 临时通行证 = Access Token(访问令牌)

对于大多数生产环境,OAuth2.1是MCP认证的推荐标准,它用带有范围,有过期时间的令牌替代静态密钥,令牌由可信的授权服务器颁发,优势很明显。

  • 权限精确:每个客户端或用户只分配了特定范围的权限
  • 可撤销,容易管理: 令牌会自动过期
  • 可审计:每个令牌都绑定唯一的客户端身份,操作轨迹可查。

认证流程

  1. 用户打开MCP客户端,尝试访问需要对接的MCP服务器
  2. 客户端打开浏览器,将用户重定向到OAuth服务器
  3. 用户完成登录,授权客户端访问自己账户
  4. OAuth服务器向客户端返回一个授权码
  5. 客户端用这个授权码,换取访问令牌 Access Token
  6. 客户端携带访问令牌,向MCP服务器发起请求

用PKCE

在上面的认证流程的第4步和第5步之间,有一个重大的安全风险,就是授权码通常是通过浏览器HTTP重定向来传递的。 过程可能被恶意程序截取,在移动设备中尤其令人担忧,因为其他应用程序可能注册相同的重定向URI并截获授权码。

这时候就需要PKCE (授权码交换证明密钥,RFC 7636)

PKCE 工作原理(简化版)

  1. 客户端提前生成一个随机字符串 code_verifier(例如 E9Melhoa2OwvFrEMTJguCHaoeK1t8URWbuGJSstw-cM
  2. 客户端计算其哈希值 code_challenge = SHA256(code_verifier) 发送给授权服务器。
  3. 授权服务器存储这个 code_challenge 并与生成的授权码绑定。
  4. 换取令牌时,客户端除了提供授权码,还必须提供原始的 code_verifier
  5. 授权服务器验证SHA256(code_verifier) 是否等于之前存储的 code_challenge

具体实现

为了实现OAuth2.1认证,协议通常要求MCP服务器提供以下API,用来和MCP客户端完成OAuth2.1认证过程:

/.well-known/oauth-authorization-server : OAuth 服务器元数据
/.well-known/oauth-protected-resource : 受保护资源元数据的列表
/authorize : 授权端点,用于授权请求
/token : 令牌端点,用于令牌交换和刷新
/register :  客户端注册端点,用于动态客户端注册

参考资料: https://grapecity.csdn.net/6911831d82fbe0098caa1f77.html#devmenu8