大模型应用技术之MCP面试题(一)
MCP(Model Context Protocol)是一个开放标准协议,旨在规范AI应用与外部数据源、工具和服务之间的交互。它采用客户端-服务器架构,支持多种传输方式(stdio/HTTP/WebSocket/SSE),提供标准化的资源访问和工具调用机制。MCP解决了AI应用集成不同数据源时的重复开发问题,实现了"一次开发,多处使用"的效果。其核心组件包括客户端(发起请求的A
1. 什么是 MCP(Model Context Protocol)?它解决了什么问题?
官方回答
**MCP(Model Context Protocol)**是一个开放标准协议,用于标准化 AI 应用与外部数据源、工具和服务之间的交互方式。它定义了客户端(AI 应用)和服务器(数据源/工具提供者)之间的通信规范,使得 AI 模型能够以统一的方式访问各种外部资源。
解决的问题:① 标准化问题:不同 AI 应用需要为每个数据源单独开发集成,MCP 提供统一接口;② 上下文管理:统一管理模型所需的上下文信息;③ 可扩展性:通过标准协议轻松添加新的数据源和工具;④ 互操作性:不同 AI 应用可以使用相同的 MCP 服务器。
教学回答
MCP = “AI 应用的’USB 接口’”
问题:
没有 MCP:
- AI应用A要连数据库 → 自己写代码
- AI应用B要连数据库 → 又写一遍
- AI应用C要连API → 再写一遍
→ 每个应用都要重复开发
有了 MCP:
MCP服务器(数据库):
- 提供标准接口
- AI应用A、B、C都能用
→ 一次开发,到处使用
形象比喻:
- 没有 MCP = 每个电器都要定制插头
- 有了 MCP = 所有电器都用 USB-C,通用!
例子:
【MCP服务器】文件系统服务器
功能:读取文件、写入文件
【AI应用】
- Claude Desktop → 通过MCP连接
- Cursor → 通过MCP连接
- 其他AI工具 → 都能用
→ 一个MCP服务器,服务所有AI应用
2. MCP 的架构是什么样的?核心组件有哪些?
官方回答
MCP 架构采用客户端-服务器模式:① MCP 客户端(Client):AI 应用(如 Claude Desktop、Cursor),发起请求;② MCP 服务器(Server):提供数据源、工具或服务的后端系统;③ 传输层(Transport):支持 stdio、HTTP、WebSocket 等传输方式;④ 协议层(Protocol):定义消息格式、请求/响应结构;⑤ 资源(Resources):服务器提供的可访问数据;⑥ 工具(Tools):服务器提供的可执行操作;⑦ 提示词(Prompts):服务器提供的提示词模板。
通信流程:客户端初始化连接 → 交换能力信息 → 客户端请求资源/工具 → 服务器响应 → 持续交互。
教学回答
MCP架构 = “客户点餐,厨房做菜”
核心组件:
| 组件 | 像什么 | 作用 |
|---|---|---|
| MCP客户端 | 顾客 | AI应用,发起请求 |
| MCP服务器 | 厨房 | 提供数据和工具 |
| 传输层 | 传菜通道 | 怎么传递消息 |
| 资源 | 菜单上的菜 | 可访问的数据 |
| 工具 | 厨房的操作 | 可执行的功能 |
| 提示词 | 菜谱模板 | 可用的提示词 |
架构图:
AI应用(客户端)
↓ 请求
MCP协议
↓
MCP服务器
├─ 资源:文件、数据库
├─ 工具:搜索、计算
└─ 提示词:模板
↓ 响应
AI应用(客户端)
通信流程:
【第1步:连接】
客户端:"你好,我想连接"
服务器:"好的,我支持文件读取、搜索工具"
【第2步:请求】
客户端:"给我文件A的内容"
服务器:"这是文件A的内容..."
【第3步:调用工具】
客户端:"执行搜索工具,关键词:AI"
服务器:"搜索结果:..."
【持续交互】
→ 可以一直这样来回通信
3. MCP 客户端和 MCP 服务器有什么区别?它们如何协作?
官方回答
MCP 客户端是发起请求的一方,通常是 AI 应用(如 Claude Desktop、Cursor),负责:发起连接、请求资源、调用工具、使用提示词、处理响应。
MCP 服务器是提供服务的一方,提供:资源访问(文件、数据库)、工具执行(API调用、计算)、提示词模板。
协作方式:客户端通过标准协议连接服务器,交换能力信息,然后客户端可以请求资源、调用工具、获取提示词,服务器响应并提供服务。
教学回答
客户端 vs 服务器 = “点餐的 vs 做菜的”
| 项目 | MCP客户端 | MCP服务器 |
|---|---|---|
| 角色 | 点餐的顾客 | 做菜的厨房 |
| 主动/被动 | 主动发起请求 | 被动响应请求 |
| 例子 | Claude Desktop、Cursor | 文件服务器、数据库服务器 |
| 做什么 | “给我文件”、“执行工具” | “这是文件”、“工具执行完成” |
协作流程:
【客户端】Claude Desktop
【服务器】文件系统MCP服务器
【第1步:连接】
客户端:"我想连接文件服务器"
服务器:"好的,我支持读取、写入文件"
【第2步:请求资源】
客户端:"给我 /home/user/doc.txt 的内容"
服务器:"文件内容:..."
【第3步:调用工具】
客户端:"执行写入工具,文件:test.txt,内容:Hello"
服务器:"写入成功"
【第4步:获取提示词】
客户端:"给我'代码审查'提示词"
服务器:"提示词模板:..."
关系:
- 客户端 = 需要服务的AI应用
- 服务器 = 提供服务的后端系统
- 协议 = 它们之间的"语言"
4. MCP 支持哪些传输方式(Transport)?各有什么特点?
官方回答
MCP 支持的传输方式:① stdio(标准输入输出):通过进程的标准输入输出通信,适合本地进程间通信;② HTTP:基于 HTTP 协议的请求-响应模式,适合远程服务;③ WebSocket:全双工通信,支持实时双向数据传输;④ SSE(Server-Sent Events):服务器向客户端推送事件流。
特点对比:stdio 简单但仅限本地;HTTP 通用但需要请求-响应;WebSocket 实时但复杂度高;SSE 适合服务器推送场景。
教学回答
传输方式 = “怎么传递消息”
| 方式 | 像什么 | 特点 | 适用场景 |
|---|---|---|---|
| stdio | 面对面说话 | 简单、本地 | 本地进程通信 |
| HTTP | 打电话 | 通用、远程 | 远程服务调用 |
| WebSocket | 视频通话 | 实时、双向 | 实时数据流 |
| SSE | 广播 | 服务器推送 | 事件通知 |
stdio 例子:
【本地通信】
AI应用 ←→ MCP服务器(同一台机器)
→ 通过标准输入输出
→ 简单快速
HTTP 例子:
【远程通信】
AI应用(本地) → HTTP请求 → MCP服务器(远程)
← HTTP响应 ←
→ 通用,适合网络环境
WebSocket 例子:
【实时通信】
AI应用 ←→ WebSocket ←→ MCP服务器
→ 双向实时通信
→ 适合流式数据
选择建议:
- 本地开发 → stdio(简单)
- 远程服务 → HTTP(通用)
- 实时数据 → WebSocket(实时)
5. MCP 中的资源(Resources)是什么?如何使用?
官方回答
**资源(Resources)**是 MCP 服务器提供的可访问数据对象,如文件、数据库记录、API 数据等。资源具有:URI(唯一标识)、名称、描述、MIME 类型。
使用流程:① 客户端列出可用资源;② 客户端请求特定资源;③ 服务器返回资源内容;④ 客户端使用资源内容。
资源类型:文件资源、数据库资源、API 资源、计算资源等。
教学回答
资源 = “服务器提供的’数据仓库’”
资源例子:
【文件系统MCP服务器】
资源列表:
- file:///home/user/doc1.txt
- file:///home/user/doc2.md
- file:///home/user/data.json
【数据库MCP服务器】
资源列表:
- db://users/123
- db://products/456
使用流程:
【第1步:列出资源】
客户端:"有哪些资源?"
服务器:
- 资源1:file:///doc1.txt
- 资源2:file:///doc2.md
【第2步:请求资源】
客户端:"给我 file:///doc1.txt"
服务器:"文件内容:这是文档1..."
【第3步:使用资源】
客户端拿到内容 → 传给AI模型 → 生成回答
资源特点:
- 有URI → 唯一标识,像网址
- 有类型 → 文本、JSON、图片等
- 可访问 → 客户端可以读取
实际应用:
AI应用需要文件内容
→ 通过MCP请求资源
→ 获取文件内容
→ 作为上下文传给AI模型
6. MCP 中的工具(Tools)是什么?如何调用工具?
官方回答
**工具(Tools)**是 MCP 服务器提供的可执行操作,如搜索、计算、API 调用等。工具包含:名称、描述、输入参数(schema)、输出格式。
调用流程:① 客户端列出可用工具;② 客户端调用工具(传入参数);③ 服务器执行工具;④ 服务器返回结果;⑤ 客户端处理结果。
工具类型:查询工具、操作工具、计算工具、转换工具等。
教学回答
工具 = “服务器提供的’功能按钮’”
工具例子:
【搜索MCP服务器】
工具列表:
- search_web:搜索网页
- search_code:搜索代码
- search_docs:搜索文档
【文件MCP服务器】
工具列表:
- read_file:读取文件
- write_file:写入文件
- list_files:列出文件
调用流程:
【第1步:列出工具】
客户端:"有哪些工具?"
服务器:
- search_web:搜索网页
- read_file:读取文件
【第2步:调用工具】
客户端:"调用 search_web,参数:{query: 'AI'}"
【第3步:执行】
服务器执行搜索 → 返回结果
【第4步:返回结果】
服务器:"搜索结果:..."
【第5步:使用结果】
客户端拿到结果 → 传给AI模型
工具参数:
工具:search_web
参数:
- query(必填):搜索关键词
- num_results(可选):结果数量
调用:
{
"tool": "search_web",
"arguments": {
"query": "MCP协议",
"num_results": 10
}
}
工具 vs 资源:
- 资源 = 静态数据(文件内容)
- 工具 = 动态操作(搜索、计算)
7. MCP 中的提示词(Prompts)是什么?如何使用?
官方回答
**提示词(Prompts)**是 MCP 服务器提供的提示词模板,客户端可以获取并使用。提示词包含:名称、描述、参数(可选的变量)、模板内容。
使用流程:① 客户端列出可用提示词;② 客户端请求提示词(传入参数);③ 服务器返回填充后的提示词;④ 客户端使用提示词。
应用场景:标准化提示词模板、动态生成提示词、提示词版本管理。
教学回答
提示词 = “服务器提供的’提示词模板库’”
提示词例子:
【代码审查提示词】
名称:code_review
参数:
- code:要审查的代码
- language:编程语言
模板:
"请审查以下{language}代码:
{code}
要求:检查bug、性能问题、代码风格"
使用流程:
【第1步:列出提示词】
客户端:"有哪些提示词?"
服务器:
- code_review:代码审查
- summarize:总结文档
【第2步:请求提示词】
客户端:"给我 code_review 提示词,参数:{code: 'def hello(): print(\"hi\")', language: 'Python'}"
【第3步:返回模板】
服务器:"请审查以下Python代码:
def hello(): print(\"hi\")
要求:检查bug、性能问题、代码风格"
【第4步:使用】
客户端拿到提示词 → 传给AI模型 → 生成审查结果
提示词参数化:
【模板】
"分析{language}代码,关注{focus}方面"
【填充】
参数:{language: 'Python', focus: '性能'}
结果:"分析Python代码,关注性能方面"
好处:
- 标准化 → 提示词统一管理
- 可复用 → 一次定义,多次使用
- 可更新 → 服务器更新,所有客户端受益
8. MCP 与 RAG(检索增强生成)有什么关系?如何结合使用?
官方回答
MCP 与 RAG 的关系:MCP 可以作为 RAG 系统的数据源提供者,通过 MCP 服务器提供文档检索、向量搜索等功能。结合方式:① MCP 服务器实现 RAG 检索功能;② 客户端通过 MCP 调用 RAG 工具;③ RAG 结果作为上下文注入 AI 模型。
优势:标准化 RAG 接口、可插拔的 RAG 实现、多客户端共享 RAG 服务。
教学回答
MCP + RAG = “标准化的知识检索系统”
传统RAG:
每个AI应用自己实现RAG:
- 应用A:自己写检索代码
- 应用B:又写一遍
- 应用C:再写一遍
→ 重复开发
MCP + RAG:
MCP服务器(RAG服务):
- 提供标准检索接口
- 所有AI应用都能用
应用A、B、C → 都通过MCP调用RAG
→ 一次开发,到处使用
结合方式:
【MCP服务器:RAG服务器】
工具:
- search_documents:检索文档
- get_context:获取上下文
【AI应用】
用户问:"AI发展历史"
【流程】
1. AI应用调用MCP工具:search_documents("AI发展历史")
2. MCP服务器检索向量数据库
3. 返回相关文档片段
4. AI应用将结果作为上下文
5. AI模型基于上下文生成回答
架构:
AI应用
↓ MCP调用
RAG MCP服务器
├─ 向量数据库
├─ 文档存储
└─ 检索算法
↓ 返回结果
AI应用 → 注入上下文 → AI模型
好处:
- 标准化 → 统一的RAG接口
- 可复用 → 多个应用共享RAG服务
- 易扩展 → 添加新的数据源很容易
9. MCP 与 Agent 有什么关系?如何结合使用?
官方回答
MCP 与 Agent 的关系:MCP 为 Agent 提供标准化的工具和数据源访问方式。Agent 可以通过 MCP 调用外部工具、访问资源、获取提示词,实现更强大的能力。
结合方式:① Agent 作为 MCP 客户端;② Agent 通过 MCP 调用工具;③ MCP 服务器提供 Agent 所需的能力;④ Agent 使用 MCP 资源作为上下文。
教学回答
MCP + Agent = “Agent的’工具库’”
Agent需要工具:
Agent要完成任务:
- 需要搜索 → 调用搜索工具
- 需要文件 → 读取文件工具
- 需要计算 → 计算工具
→ 这些工具可以通过MCP提供!
结合架构:
【Agent】
任务:写一份报告
【通过MCP调用工具】
1. 调用搜索工具(MCP服务器1)
→ 搜索资料
2. 调用文件工具(MCP服务器2)
→ 读取模板
3. 调用数据库工具(MCP服务器3)
→ 获取数据
【Agent处理】
综合所有结果 → 生成报告
实际例子:
【Agent任务】
用户:"帮我分析市场趋势"
【Agent规划】
1. 获取历史数据(MCP:数据库服务器)
2. 搜索相关报告(MCP:搜索服务器)
3. 分析数据(Agent自己)
4. 生成报告
【执行】
Agent → MCP调用 → 获取数据
Agent → 分析 → 生成报告
好处:
- 标准化工具 → Agent用统一方式调用工具
- 可扩展 → 添加新工具很容易
- 可复用 → 多个Agent共享工具
10. 如何创建一个 MCP 服务器?基本步骤是什么?
官方回答
创建 MCP 服务器的步骤:① 选择实现方式:使用 MCP SDK(如 TypeScript、Python);② 定义能力:确定提供哪些资源、工具、提示词;③ 实现服务器逻辑:处理客户端请求;④ 配置传输方式:选择 stdio、HTTP 等;⑤ 测试服务器:使用 MCP 客户端测试;⑥ 部署服务器:部署到目标环境。
技术栈:TypeScript/JavaScript(@modelcontextprotocol/sdk)、Python(mcp)、其他语言(根据SDK支持)。
教学回答
创建MCP服务器 = “开一家’工具店’”
步骤:
1. 确定卖什么(定义能力)
我要提供:
- 资源:文件读取
- 工具:搜索、计算
- 提示词:代码审查模板
2. 选工具(选择SDK)
【TypeScript】
npm install @modelcontextprotocol/sdk
【Python】
pip install mcp
3. 写代码(实现服务器)
// TypeScript例子
import { Server } from "@modelcontextprotocol/sdk/server/index.js";
const server = new Server({
name: "my-mcp-server",
version: "1.0.0",
});
// 定义工具
server.setRequestHandler(ListToolsRequestSchema, async () => ({
tools: [
{
name: "search",
description: "搜索功能",
inputSchema: {
type: "object",
properties: {
query: { type: "string" }
}
}
}
]
}));
// 处理工具调用
server.setRequestHandler(CallToolRequestSchema, async (request) => {
if (request.params.name === "search") {
// 执行搜索逻辑
return { content: [{ type: "text", text: "搜索结果..." }] };
}
});
4. 配置传输(选择方式)
【stdio方式】
server.connect(stdioTransport());
【HTTP方式】
server.listen(3000);
5. 测试
使用MCP客户端连接 → 测试工具调用 → 确认正常
6. 部署
打包 → 部署到服务器 → 配置客户端连接
完整例子:文件服务器
// 提供文件读取工具
server.setRequestHandler(CallToolRequestSchema, async (request) => {
if (request.params.name === "read_file") {
const filePath = request.params.arguments?.path;
const content = await fs.readFile(filePath, 'utf-8');
return { content: [{ type: "text", text: content }] };
}
});
11. 如何在 AI 应用中集成 MCP 客户端?以 Claude Desktop 为例
官方回答
集成 MCP 客户端:① 安装 MCP 服务器:确保 MCP 服务器已部署;② 配置客户端:在 AI 应用的配置文件中添加 MCP 服务器配置;③ 指定传输方式:配置 stdio、HTTP 等;④ 提供服务器信息:命令、参数、环境变量;⑤ 重启应用:使配置生效;⑥ 验证连接:确认客户端能连接服务器。
Claude Desktop 配置:在配置文件中添加 MCP 服务器配置,指定命令、参数等。
教学回答
集成MCP = “给AI应用装’插件’”
Claude Desktop例子:
1. 找到配置文件
macOS: ~/Library/Application Support/Claude/claude_desktop_config.json
Windows: %APPDATA%\Claude\claude_desktop_config.json
2. 添加MCP服务器配置
{
"mcpServers": {
"filesystem": {
"command": "node",
"args": ["/path/to/filesystem-server.js"],
"env": {}
},
"database": {
"command": "python",
"args": ["/path/to/database-server.py"],
"env": {
"DB_URL": "postgresql://localhost/mydb"
}
}
}
}
3. 重启应用
关闭Claude Desktop → 重新打开
→ 自动连接MCP服务器
4. 验证
在Claude Desktop中:
- 能看到MCP服务器提供的工具
- 能调用工具
- 能访问资源
配置说明:
- command:启动服务器的命令
- args:命令参数
- env:环境变量
其他AI应用:
- Cursor:类似配置方式
- 自定义应用:使用MCP SDK集成
12. MCP 协议的消息格式是什么样的?请求和响应如何定义?
官方回答
MCP 协议消息格式:基于 JSON-RPC 2.0,包含:方法名(method)、参数(params)、ID(请求标识)、结果(result)或错误(error)。
请求格式:{"jsonrpc": "2.0", "method": "...", "params": {...}, "id": 1}
响应格式:{"jsonrpc": "2.0", "result": {...}, "id": 1} 或 {"jsonrpc": "2.0", "error": {...}, "id": 1}
标准方法:initialize、tools/list、tools/call、resources/list、resources/read、prompts/list、prompts/get 等。
教学回答
消息格式 = “MCP的’语言规则’”
请求格式:
{
"jsonrpc": "2.0",
"method": "tools/call",
"params": {
"name": "search",
"arguments": {
"query": "MCP"
}
},
"id": 1
}
响应格式(成功):
{
"jsonrpc": "2.0",
"result": {
"content": [
{
"type": "text",
"text": "搜索结果..."
}
]
},
"id": 1
}
响应格式(错误):
{
"jsonrpc": "2.0",
"error": {
"code": -32602,
"message": "Invalid params",
"data": "Missing required parameter: query"
},
"id": 1
}
标准方法:
| 方法 | 作用 | 例子 |
|---|---|---|
| initialize | 初始化连接 | 客户端连接服务器 |
| tools/list | 列出工具 | 获取可用工具列表 |
| tools/call | 调用工具 | 执行工具 |
| resources/list | 列出资源 | 获取资源列表 |
| resources/read | 读取资源 | 获取资源内容 |
| prompts/list | 列出提示词 | 获取提示词列表 |
| prompts/get | 获取提示词 | 获取具体提示词 |
通信流程:
客户端 → 发送请求(JSON-RPC格式)
服务器 → 处理请求
服务器 → 返回响应(JSON-RPC格式)
客户端 → 处理响应
13. MCP 如何处理错误和异常?错误码有哪些?
官方回答
MCP 错误处理:遵循 JSON-RPC 2.0 错误规范,包含:错误码(code)、错误消息(message)、错误数据(data,可选)。
标准错误码:-32700(解析错误)、-32600(无效请求)、-32601(方法不存在)、-32602(无效参数)、-32603(内部错误)、-32000 到 -32099(服务器错误,自定义)。
错误响应格式:{"jsonrpc": "2.0", "error": {"code": -32602, "message": "...", "data": {...}}, "id": 1}
教学回答
错误处理 = “MCP的’错误报告’”
错误响应格式:
{
"jsonrpc": "2.0",
"error": {
"code": -32602,
"message": "Invalid params",
"data": {
"param": "query",
"reason": "Missing required parameter"
}
},
"id": 1
}
标准错误码:
| 错误码 | 含义 | 例子 |
|---|---|---|
| -32700 | 解析错误 | JSON格式不对 |
| -32600 | 无效请求 | 请求格式错误 |
| -32601 | 方法不存在 | 调用了不存在的方法 |
| -32602 | 无效参数 | 参数类型不对 |
| -32603 | 内部错误 | 服务器内部出错 |
| -32000 | 自定义错误 | 业务逻辑错误 |
错误处理例子:
1. 参数错误
请求:{"method": "tools/call", "params": {"name": "search"}}
错误:缺少arguments参数
响应:{"error": {"code": -32602, "message": "Invalid params"}}
2. 方法不存在
请求:{"method": "tools/nonexistent"}
错误:方法不存在
响应:{"error": {"code": -32601, "message": "Method not found"}}
3. 服务器错误
请求:正常请求
错误:数据库连接失败
响应:{"error": {"code": -32603, "message": "Internal error"}}
客户端处理:
收到错误响应 → 检查错误码 → 根据错误码处理
- 参数错误 → 修正参数重试
- 服务器错误 → 记录日志,提示用户
14. MCP 支持哪些认证和授权机制?
官方回答
MCP 认证授权:① 传输层安全:使用 TLS/SSL 加密通信;② API 密钥:在环境变量或配置中提供 API 密钥;③ OAuth:支持 OAuth 2.0 流程;④ Token 认证:使用 JWT 或其他 token;⑤ 自定义认证:在服务器端实现自定义认证逻辑。
实现方式:在 MCP 服务器初始化时验证凭证,在请求处理中检查权限,返回适当的错误响应。
教学回答
认证授权 = “MCP的’门禁系统’”
认证方式:
1. API密钥
【配置】
环境变量:MCP_API_KEY=secret123
【服务器验证】
if (request.headers['api-key'] !== process.env.MCP_API_KEY) {
return error("Unauthorized");
}
2. Token认证
【客户端】
请求头:Authorization: Bearer <token>
【服务器验证】
验证token有效性 → 允许/拒绝
3. OAuth
【流程】
1. 客户端请求授权
2. 服务器返回授权URL
3. 用户授权
4. 服务器返回token
5. 客户端使用token访问
4. TLS加密
【传输加密】
使用HTTPS → 数据加密传输
→ 防止窃听
实际例子:
API密钥方式:
{
"mcpServers": {
"database": {
"command": "python",
"args": ["server.py"],
"env": {
"API_KEY": "secret123"
}
}
}
}
服务器端验证:
import os
API_KEY = os.getenv("API_KEY")
def authenticate(request):
if request.headers.get("api-key") != API_KEY:
raise UnauthorizedError()
权限控制:
【不同用户不同权限】
用户A → 只能读取文件
用户B → 可以读写文件
【服务器检查】
if (user.role === "readonly" && action === "write") {
return error("Permission denied");
}
15. MCP 如何处理流式响应(Streaming)?适用场景有哪些?
官方回答
MCP 流式响应:支持服务器向客户端推送数据流,适用于:长时间运行的任务、实时数据更新、大文件传输、进度报告。
实现方式:使用 WebSocket 或 SSE(Server-Sent Events)传输,服务器发送多个部分响应,客户端逐步接收和处理。
消息格式:使用进度通知(progress notifications)或流式内容块(streaming content chunks)。
教学回答
流式响应 = “边做边传,不等全部做完”
传统方式(不好):
任务:处理大文件(10分钟)
→ 等10分钟 → 一次性返回结果
→ 用户等得着急
流式方式(好):
任务:处理大文件
→ 处理10% → 通知进度
→ 处理20% → 通知进度
→ ...
→ 用户能看到进度,不着急
适用场景:
1. 长时间任务
任务:分析1000个文件
→ 流式返回:已完成100个、200个...
→ 用户知道进度
2. 实时数据
任务:监控系统状态
→ 流式推送:状态更新1、状态更新2...
→ 实时显示
3. 大文件传输
文件:100MB
→ 流式传输:先传10MB、再传10MB...
→ 不用等全部下载完
实现方式:
WebSocket:
【连接】
客户端 ←→ WebSocket ←→ 服务器
【推送】
服务器 → 发送进度消息
客户端 → 实时接收
SSE(Server-Sent Events):
【连接】
客户端 → HTTP请求
服务器 → SSE流式响应
【推送】
服务器持续发送事件
客户端持续接收
消息格式:
{
"jsonrpc": "2.0",
"method": "progress",
"params": {
"progress": 50,
"message": "处理中..."
}
}
16. 如何实现 MCP 服务器的资源缓存?有什么好处?
官方回答
MCP 资源缓存:在服务器端缓存资源内容,减少重复计算和数据库查询。实现方式:内存缓存(Redis、Memcached)、TTL(Time-To-Live)机制、缓存失效策略、缓存键设计。
好处:提升响应速度、降低后端负载、减少成本、改善用户体验。
教学回答
资源缓存 = “把常用的东西放’抽屉’里,随时拿”
没有缓存:
客户端请求资源A → 服务器查数据库(2秒)→ 返回
客户端又请求资源A → 服务器又查数据库(2秒)→ 返回
→ 每次都慢
有缓存:
第1次:客户端请求资源A → 查数据库(2秒)→ 存入缓存 → 返回
第2次:客户端请求资源A → 从缓存取(0.01秒)→ 返回
→ 快200倍!
实现方式:
1. 内存缓存
cache = {}
def get_resource(uri):
if uri in cache:
return cache[uri] # 从缓存取
else:
data = fetch_from_database(uri) # 查数据库
cache[uri] = data # 存入缓存
return data
2. TTL机制
cache = {}
TTL = 3600 # 1小时过期
def get_resource(uri):
if uri in cache:
entry = cache[uri]
if time.time() - entry['timestamp'] < TTL:
return entry['data'] # 缓存有效
# 缓存过期,重新获取
data = fetch_from_database(uri)
cache[uri] = {'data': data, 'timestamp': time.time()}
return data
3. Redis缓存
import redis
r = redis.Redis()
def get_resource(uri):
cached = r.get(uri)
if cached:
return cached # 从Redis取
else:
data = fetch_from_database(uri)
r.setex(uri, 3600, data) # 存入Redis,1小时过期
return data
好处:
- 速度快 → 缓存比数据库快100倍
- 省资源 → 减少数据库查询
- 省钱 → 减少API调用成本
17. MCP 服务器如何实现负载均衡和高可用?
官方回答
负载均衡和高可用:① 多实例部署:部署多个 MCP 服务器实例;② 负载均衡器:使用负载均衡器(如 Nginx、HAProxy)分发请求;③ 健康检查:定期检查服务器健康状态;④ 故障转移:主服务器故障时切换到备用服务器;⑤ 数据同步:多个实例共享状态或使用共享存储。
架构:客户端 → 负载均衡器 → 多个MCP服务器实例 → 共享数据存储。
教学回答
负载均衡 = “多个’服务员’,谁闲谁干活”
单服务器(不好):
客户端 → MCP服务器(1个)
→ 请求多了就卡
→ 服务器挂了就全完
负载均衡(好):
客户端 → 负载均衡器 → MCP服务器1
→ MCP服务器2
→ MCP服务器3
→ 请求分散,不卡
→ 一个挂了,其他的还能用
实现方式:
1. 负载均衡器
【Nginx配置】
upstream mcp_servers {
server mcp1:3000;
server mcp2:3000;
server mcp3:3000;
}
server {
location / {
proxy_pass http://mcp_servers;
}
}
2. 健康检查
【定期检查】
每10秒检查一次:
- 服务器1:健康 ✅
- 服务器2:健康 ✅
- 服务器3:故障 ❌
【移除故障服务器】
负载均衡器 → 不再转发请求到服务器3
3. 故障转移
【主服务器】
主服务器:mcp1
备用服务器:mcp2, mcp3
【主服务器故障】
检测到故障 → 切换到备用服务器
→ 用户无感知
4. 数据同步
【共享存储】
所有服务器 → 连接同一个数据库
→ 数据一致
【或】
服务器1更新 → 同步到服务器2、3
→ 数据一致
好处:
- 高性能 → 多个服务器处理请求
- 高可用 → 一个挂了不影响
- 可扩展 → 随时加服务器
18. MCP 协议如何支持版本控制?如何处理兼容性?
官方回答
MCP 版本控制:① 协议版本:MCP 协议本身有版本号;② 服务器版本:服务器声明支持的协议版本;③ 能力协商:客户端和服务器协商共同支持的版本;④ 向后兼容:新版本保持对旧版本的兼容;⑤ 弃用机制:逐步弃用旧功能。
兼容性处理:版本检测、功能检测、降级策略、迁移指南。
教学回答
版本控制 = “MCP的’升级系统’”
问题:
客户端:支持MCP v1.0
服务器:支持MCP v2.0
→ 能通信吗?
解决方案:
1. 版本协商
【初始化时】
客户端:"我支持v1.0和v1.1"
服务器:"我支持v1.1和v2.0"
→ 协商:使用v1.1(共同支持)
2. 能力检测
【检查功能】
客户端:"支持流式响应吗?"
服务器:"支持"
→ 使用流式响应
服务器:"不支持"
→ 使用普通响应
3. 向后兼容
【v1.0功能】
工具调用:tools/call
【v2.0新增】
批量工具调用:tools/call_batch
【兼容性】
v2.0服务器 → 仍支持v1.0的tools/call
→ 旧客户端仍能用
版本声明:
{
"jsonrpc": "2.0",
"method": "initialize",
"params": {
"protocolVersion": "2024-11-05",
"capabilities": {
"tools": {},
"resources": {}
},
"clientInfo": {
"name": "claude-desktop",
"version": "1.0.0"
}
}
}
处理不兼容:
【检测到不兼容】
客户端版本太旧 → 提示升级
服务器版本太旧 → 提示升级服务器
【降级策略】
新功能不可用 → 使用旧功能
→ 保证基本功能可用
19. 如何测试 MCP 服务器?有哪些测试方法?
官方回答
MCP 服务器测试:① 单元测试:测试单个功能(工具调用、资源读取);② 集成测试:测试完整流程;③ 端到端测试:使用真实客户端测试;④ 性能测试:测试响应时间、吞吐量;⑤ 压力测试:测试高并发场景;⑥ Mock客户端:创建模拟客户端进行测试。
测试工具:MCP SDK 提供的测试工具、自定义测试脚本、CI/CD 集成。
教学回答
测试MCP服务器 = “检查’工具店’是否正常营业”
测试方法:
1. 单元测试
# 测试工具调用
def test_search_tool():
result = call_tool("search", {"query": "test"})
assert result["content"][0]["text"] == "搜索结果"
# 测试资源读取
def test_read_resource():
content = read_resource("file:///test.txt")
assert content == "文件内容"
2. 集成测试
# 测试完整流程
def test_complete_flow():
# 1. 列出工具
tools = list_tools()
assert "search" in [t["name"] for t in tools]
# 2. 调用工具
result = call_tool("search", {"query": "test"})
assert result is not None
# 3. 读取资源
resource = read_resource("file:///test.txt")
assert resource is not None
3. Mock客户端
# 模拟客户端请求
class MockClient:
def send_request(self, method, params):
# 模拟客户端发送请求
return server.handle_request(method, params)
# 测试
client = MockClient()
result = client.send_request("tools/call", {
"name": "search",
"arguments": {"query": "test"}
})
assert result["result"] is not None
4. 性能测试
import time
def test_performance():
start = time.time()
for i in range(100):
call_tool("search", {"query": f"test{i}"})
end = time.time()
avg_time = (end - start) / 100
assert avg_time < 0.1 # 平均响应时间<100ms
5. 压力测试
import concurrent.futures
def test_concurrent():
with concurrent.futures.ThreadPoolExecutor(max_workers=10) as executor:
futures = [
executor.submit(call_tool, "search", {"query": f"test{i}"})
for i in range(100)
]
results = [f.result() for f in futures]
assert all(r is not None for r in results)
测试清单:
- ✅ 工具调用正常
- ✅ 资源读取正常
- ✅ 错误处理正确
- ✅ 性能达标
- ✅ 并发安全
20. MCP 在实际项目中的应用场景有哪些?
官方回答
MCP 应用场景:① 文件系统集成:AI 应用访问本地或远程文件;② 数据库集成:AI 应用查询数据库;③ API 集成:AI 应用调用外部 API;④ 代码仓库集成:AI 应用访问 Git 仓库;⑤ 知识库集成:AI 应用访问文档库、Wiki;⑥ 工具集成:AI 应用使用各种开发工具;⑦ 监控集成:AI 应用访问监控数据。
优势:标准化接口、可复用、易扩展、统一管理。
教学回答
应用场景 = “MCP能干什么活”
1. 文件系统集成
【场景】
AI应用需要读取代码文件
【MCP服务器】
提供文件读取工具:
- read_file:读取文件
- list_files:列出文件
【使用】
AI应用 → MCP调用 → 读取文件 → 分析代码
2. 数据库集成
【场景】
AI应用需要查询用户数据
【MCP服务器】
提供数据库查询工具:
- query:执行SQL查询
【使用】
AI应用 → MCP调用 → 查询数据库 → 分析数据
3. API集成
【场景】
AI应用需要调用外部API
【MCP服务器】
封装API调用:
- call_api:调用外部API
【使用】
AI应用 → MCP调用 → 调用API → 获取数据
4. 代码仓库集成
【场景】
AI应用需要分析Git仓库
【MCP服务器】
提供Git工具:
- git_log:查看提交历史
- git_diff:查看差异
【使用】
AI应用 → MCP调用 → 分析代码变更
5. 知识库集成
【场景】
AI应用需要检索文档
【MCP服务器】
提供文档检索:
- search_docs:搜索文档
- get_doc:获取文档内容
【使用】
AI应用 → MCP调用 → 检索文档 → 作为上下文
6. 开发工具集成
【场景】
AI应用需要使用开发工具
【MCP服务器】
封装工具:
- run_tests:运行测试
- build_project:构建项目
【使用】
AI应用 → MCP调用 → 执行开发任务
实际项目例子:
代码审查助手:
MCP服务器:
- Git服务器:获取代码
- 测试服务器:运行测试
- 文档服务器:检索文档
AI应用:
- 读取代码
- 运行测试
- 检索相关文档
- 生成审查报告
21. MCP 与 LangChain、LlamaIndex 等框架如何集成?
官方回答
MCP 与框架集成:MCP 可以作为这些框架的工具提供者。集成方式:① MCP 作为工具:框架通过 MCP 客户端调用 MCP 服务器提供的工具;② 框架作为 MCP 服务器:框架功能通过 MCP 暴露;③ 双向集成:框架既使用 MCP 工具,也提供 MCP 服务。
LangChain 集成:使用 LangChain 的 MCP 工具包装器,将 MCP 工具集成到 LangChain Agent。
LlamaIndex 集成:MCP 服务器提供数据源,LlamaIndex 通过 MCP 访问数据。
教学回答
框架集成 = “让MCP和现有工具’握手’”
LangChain集成:
from langchain.agents import AgentExecutor
from langchain.tools import MCPTool
# 创建MCP工具
mcp_tool = MCPTool(
server_url="http://localhost:3000",
tool_name="search"
)
# 集成到Agent
agent = AgentExecutor(
tools=[mcp_tool, ...],
llm=llm
)
# 使用
result = agent.run("搜索AI相关内容")
LlamaIndex集成:
from llama_index import VectorStoreIndex
from llama_index.tools import MCPToolSpec
# 创建MCP工具规范
mcp_tools = MCPToolSpec(
server_url="http://localhost:3000"
)
# 集成到索引
index = VectorStoreIndex.from_tools(
mcp_tools.get_tools()
)
# 使用
query_engine = index.as_query_engine()
response = query_engine.query("查询文档")
双向集成:
【框架作为MCP客户端】
LangChain Agent → MCP调用 → 外部MCP服务器
→ 使用外部工具
【框架作为MCP服务器】
LangChain功能 → MCP暴露 → 其他应用使用
→ 提供框架能力
实际例子:
LangChain + MCP文件服务器:
# LangChain使用MCP文件工具
agent = create_agent(
tools=[MCPFileTool()],
llm=llm
)
# Agent可以读取文件
agent.run("读取config.json并分析")
22. MCP 服务器的性能优化有哪些策略?
官方回答
MCP 服务器性能优化:① 异步处理:使用异步I/O处理请求;② 连接池:复用数据库、API连接;③ 缓存策略:缓存常用资源;④ 批量处理:合并多个请求;⑤ 资源限制:限制并发请求数;⑥ 压缩传输:压缩响应数据;⑦ CDN加速:静态资源使用CDN;⑧ 数据库优化:索引、查询优化。
教学回答
性能优化 = “让MCP服务器跑得更快”
优化策略:
1. 异步处理
# 同步(慢)
def handle_request(request):
result = slow_operation() # 阻塞
return result
# 异步(快)
async def handle_request(request):
result = await slow_operation() # 不阻塞
return result
2. 连接池
# 没有连接池(慢)
def query_db():
conn = create_connection() # 每次创建
result = conn.query(...)
conn.close()
return result
# 有连接池(快)
pool = create_connection_pool()
def query_db():
conn = pool.get_connection() # 复用连接
result = conn.query(...)
pool.return_connection(conn)
return result
3. 缓存
cache = {}
def get_resource(uri):
if uri in cache:
return cache[uri] # 从缓存取,快
data = fetch_from_db(uri)
cache[uri] = data
return data
4. 批量处理
# 单个处理(慢)
for request in requests:
process(request) # 100次,每次10ms = 1000ms
# 批量处理(快)
batch_process(requests) # 100次,批量10ms = 10ms
5. 压缩传输
import gzip
def compress_response(data):
compressed = gzip.compress(data.encode())
return compressed # 体积小,传输快
性能对比:
【优化前】
响应时间:500ms
吞吐量:100请求/秒
【优化后】
响应时间:50ms(快10倍)
吞吐量:1000请求/秒(快10倍)
23. MCP 的安全最佳实践有哪些?
官方回答
MCP 安全最佳实践:① 传输加密:使用 TLS/SSL;② 身份验证:API密钥、Token、OAuth;③ 权限控制:基于角色的访问控制(RBAC);④ 输入验证:验证和清理用户输入;⑤ 输出过滤:过滤敏感信息;⑥ 日志审计:记录所有操作;⑦ 速率限制:防止滥用;⑧ 安全配置:最小权限原则。
教学回答
安全实践 = “保护MCP服务器不被攻击”
安全措施:
1. 传输加密
【使用HTTPS】
HTTP → 明文传输 ❌
HTTPS → 加密传输 ✅
【配置TLS】
server.listen(443, {
key: fs.readFileSync('key.pem'),
cert: fs.readFileSync('cert.pem')
});
2. 身份验证
# API密钥验证
def authenticate(request):
api_key = request.headers.get('api-key')
if api_key != os.getenv('VALID_API_KEY'):
raise UnauthorizedError()
3. 权限控制
# 基于角色的访问控制
def check_permission(user, resource, action):
if user.role == 'admin':
return True
if user.role == 'user' and action == 'read':
return True
return False
4. 输入验证
# 验证输入
def validate_input(params):
if 'path' in params:
# 防止路径遍历攻击
if '..' in params['path']:
raise InvalidInputError()
return sanitize(params)
5. 输出过滤
# 过滤敏感信息
def filter_output(data):
# 移除密码、密钥等
filtered = remove_sensitive_fields(data)
return filtered
6. 速率限制
from flask_limiter import Limiter
limiter = Limiter(
app,
key_func=get_remote_address,
default_limits=["100 per hour"]
)
安全检查清单:
- ✅ 使用HTTPS
- ✅ 身份验证
- ✅ 权限控制
- ✅ 输入验证
- ✅ 输出过滤
- ✅ 日志审计
- ✅ 速率限制
24. 如何监控和调试 MCP 服务器?有哪些工具和方法?
官方回答
MCP 监控调试:① 日志记录:详细记录请求、响应、错误;② 指标监控:响应时间、错误率、吞吐量;③ 追踪:请求追踪、调用链分析;④ 健康检查:定期检查服务器状态;⑤ 调试工具:MCP Inspector、自定义调试工具;⑥ 性能分析:性能分析工具。
工具:Prometheus(指标)、Grafana(可视化)、ELK(日志)、Jaeger(追踪)。
教学回答
监控调试 = “看清楚MCP服务器在干嘛”
监控方法:
1. 日志记录
import logging
logger = logging.getLogger('mcp_server')
def handle_request(request):
logger.info(f"收到请求: {request.method}")
try:
result = process_request(request)
logger.info(f"请求成功: {result}")
return result
except Exception as e:
logger.error(f"请求失败: {e}", exc_info=True)
raise
2. 指标监控
from prometheus_client import Counter, Histogram
request_count = Counter('mcp_requests_total', 'Total requests')
request_duration = Histogram('mcp_request_duration_seconds', 'Request duration')
def handle_request(request):
request_count.inc()
with request_duration.time():
return process_request(request)
3. 健康检查
@app.route('/health')
def health_check():
# 检查数据库连接
if not check_database():
return {"status": "unhealthy"}, 503
# 检查其他依赖
return {"status": "healthy"}, 200
4. 请求追踪
import uuid
def handle_request(request):
trace_id = str(uuid.uuid4())
logger.info(f"Trace ID: {trace_id}")
# 在响应中包含trace_id
return {
"result": process_request(request),
"trace_id": trace_id
}
监控面板:
【Grafana面板】
- 请求数:1000/分钟
- 平均响应时间:50ms
- 错误率:0.1%
- 活跃连接数:50
【告警】
- 错误率>1% → 发邮件
- 响应时间>1秒 → 发短信
调试工具:
MCP Inspector:
可视化工具:
- 查看请求/响应
- 查看工具调用
- 查看资源访问
- 性能分析
25. MCP 的未来发展趋势是什么?有哪些改进方向?
官方回答
MCP 发展趋势:① 标准化:更多AI应用采用MCP标准;② 生态扩展:更多MCP服务器和工具;③ 性能优化:更好的性能和可扩展性;④ 安全性增强:更强的安全机制;⑤ 易用性提升:更好的开发工具和文档;⑥ 云原生:支持容器化、Kubernetes部署;⑦ 多模态支持:支持图像、音频等多模态数据;⑧ 实时协作:支持多客户端协作。
教学回答
发展趋势 = “MCP会变成什么样”
发展方向:
1. 更广泛采用
现在:Claude Desktop、Cursor支持
未来:更多AI应用支持
→ 成为行业标准
2. 生态扩展
现在:基础MCP服务器
未来:丰富的MCP服务器生态
- 数据库服务器
- API服务器
- 工具服务器
→ 像npm包一样丰富
3. 性能提升
现在:基础性能
未来:
- 更快的响应
- 更好的并发
- 更低的延迟
→ 支持更大规模应用
4. 安全性增强
现在:基础安全
未来:
- 更强的加密
- 更细的权限控制
- 更好的审计
→ 企业级安全
5. 易用性提升
现在:需要手动配置
未来:
- 可视化配置工具
- 自动发现服务器
- 一键部署
→ 像安装App一样简单
6. 云原生
现在:本地部署
未来:
- 容器化支持
- Kubernetes集成
- 自动扩缩容
→ 云原生架构
7. 多模态支持
现在:主要是文本
未来:
- 图像资源
- 音频资源
- 视频资源
→ 支持多模态AI应用
8. 实时协作
现在:单客户端
未来:
- 多客户端协作
- 实时同步
- 共享上下文
→ 团队协作场景
更多推荐



所有评论(0)