大模型应用技术之 MCP 扩展功能
1. MCP 扩展功能概览
1.1 什么是 MCP 扩展功能?
MCP 扩展功能是 MCP 服务器提供给 AI 客户端的标准化能力接口,让 AI 应用能够:
- 访问外部数据:通过 Resources 获取文件、数据库、API 数据
- 执行操作:通过 Tools 调用搜索、计算、API 等功能
- 使用模板:通过 Prompts 获取预定义的提示词模板
- 测试连接:通过 Ping 验证服务器可用性
- AI 推理:通过 Sampling 让服务器请求 AI 进行推理
- 管理路径:通过 Roots 定义资源根目录
1.2 功能对比表
| 功能 | 类型 | 作用 | 使用频率 |
|---|---|---|---|
| Resources | 数据访问 | 提供静态或动态数据资源 | ⭐⭐⭐⭐⭐ |
| Prompts | 模板库 | 提供预定义的提示词模板 | ⭐⭐⭐⭐ |
| Tools | 功能调用 | 执行搜索、计算、API 调用等操作 | ⭐⭐⭐⭐⭐ |
| Ping | 连接测试 | 验证服务器连接和响应速度 | ⭐⭐⭐ |
| Sampling | AI 推理 | 服务器请求 AI 进行推理 | ⭐⭐⭐ |
| Roots | 路径管理 | 定义资源根目录和访问范围 | ⭐⭐⭐⭐ |
2. Resources(资源)详解
2.1 什么是 Resources?
**Resources(资源)**是 MCP 服务器提供的可访问数据,类似于"数据仓库"。每个资源都有唯一的 URI(统一资源标识符),客户端可以通过 URI 请求获取资源内容。
2.2 Resources 的核心特点
- 唯一标识:每个资源都有 URI,如
file:///path/to/file.txt - 类型多样:可以是文件、数据库记录、API 响应、配置数据等
- 可访问性:客户端可以列出和读取资源内容
- 动态更新:资源内容可以实时更新(如数据库查询结果)
2.3 Resources 的使用流程
【第1步:列出资源】
客户端 → 服务器:"有哪些资源?"
服务器 → 客户端:
- file:///documents/readme.md
- file:///documents/config.json
- db://users/123
【第2步:请求资源】
客户端 → 服务器:"给我 file:///documents/readme.md"
服务器 → 客户端:"# 项目说明\n这是一个示例项目..."
【第3步:使用资源】
客户端拿到内容 → 传给AI模型 → 生成回答
2.4 Resources 的实际应用场景
场景1:文件系统资源服务器
作用:提供本地文件系统的访问能力
使用场景:
- 代码审查:AI 需要读取项目文件进行代码分析
- 文档生成:AI 读取多个文档文件,生成综合报告
- 配置管理:AI 读取配置文件,提供配置建议
示例:
// 资源列表
{
"resources": [
{
"uri": "file:///project/src/main.py",
"name": "主程序文件",
"description": "项目主入口文件",
"mimeType": "text/x-python"
},
{
"uri": "file:///project/README.md",
"name": "项目说明",
"description": "项目README文档",
"mimeType": "text/markdown"
}
]
}
实际应用:
用户:"帮我审查这个项目的代码质量"
AI → MCP:列出所有 .py 文件资源
AI → MCP:读取每个文件内容
AI → 分析代码 → 生成审查报告
场景2:数据库资源服务器
作用:提供数据库记录的访问能力
使用场景:
- 数据分析:AI 读取数据库记录进行数据分析
- 报表生成:AI 读取数据生成业务报表
- 数据查询:AI 根据用户问题查询相关数据
示例:
// 资源列表
{
"resources": [
{
"uri": "db://users/123",
"name": "用户信息",
"description": "用户ID为123的详细信息",
"mimeType": "application/json"
},
{
"uri": "db://orders/recent",
"name": "最近订单",
"description": "最近30天的订单数据",
"mimeType": "application/json"
}
]
}
实际应用:
用户:"最近一周的销售情况如何?"
AI → MCP:读取 db://orders/recent 资源
AI → 分析数据 → 生成销售报告
场景3:API 资源服务器
作用:提供外部 API 数据的访问能力
使用场景:
- 实时数据:获取天气、股票、新闻等实时数据
- 第三方服务:集成 GitHub、Slack、Jira 等服务数据
- 数据聚合:聚合多个 API 的数据源
示例:
// 资源列表
{
"resources": [
{
"uri": "api://weather/beijing",
"name": "北京天气",
"description": "北京市当前天气信息",
"mimeType": "application/json"
},
{
"uri": "api://github/repos/owner/repo",
"name": "GitHub仓库信息",
"description": "指定仓库的详细信息",
"mimeType": "application/json"
}
]
}
实际应用:
用户:"今天北京适合出门吗?"
AI → MCP:读取 api://weather/beijing 资源
AI → 分析天气数据 → 给出建议
2.5 Resources 的优势
- 统一接口:不同数据源使用统一的访问方式
- 按需加载:只请求需要的资源,节省带宽
- 类型安全:通过 MIME 类型确保数据格式正确
- 可扩展:轻松添加新的资源类型
3. Prompts(提示词)详解
3.1 什么是 Prompts?
**Prompts(提示词)**是 MCP 服务器提供的预定义提示词模板,类似于"提示词模板库"。客户端可以获取这些模板,填充参数后直接使用,避免重复编写相似的提示词。
3.2 Prompts 的核心特点
- 模板化:使用变量占位符,支持参数化
- 标准化:提供经过验证的提示词模板
- 可复用:一次定义,多处使用
- 版本管理:支持提示词版本控制
3.3 Prompts 的使用流程
【第1步:列出提示词】
客户端 → 服务器:"有哪些提示词模板?"
服务器 → 客户端:
- code_review:代码审查模板
- data_analysis:数据分析模板
- bug_fix:Bug修复模板
【第2步:获取提示词】
客户端 → 服务器:"给我 code_review 模板,参数:{code: '...', language: 'python'}"
服务器 → 客户端:"请审查以下python代码:\n...\n要求:检查bug、性能问题、代码风格"
【第3步:使用提示词】
客户端拿到填充后的提示词 → 传给AI模型 → 生成回答
3.4 Prompts 的实际应用场景
场景1:代码审查提示词模板
作用:提供标准化的代码审查提示词
使用场景:
- PR 审查:自动生成代码审查提示词
- 代码质量检查:统一代码审查标准
- 团队协作:确保团队成员使用相同的审查标准
示例:
// 提示词模板
{
"name": "code_review",
"description": "代码审查提示词模板",
"arguments": [
{
"name": "code",
"description": "要审查的代码",
"required": true
},
{
"name": "language",
"description": "编程语言",
"required": true
},
{
"name": "focus",
"description": "审查重点(可选)",
"required": false
}
]
}
模板内容:
请审查以下{language}代码:
{code}
审查要求:
1. 检查潜在的bug和错误
2. 评估代码性能和效率
3. 检查代码风格和规范
4. 提供改进建议
{focus ? "重点关注:" + focus : ""}
实际应用:
用户:"帮我审查这段Python代码"
AI → MCP:获取 code_review 模板
AI → 填充参数:{code: 用户代码, language: 'python'}
AI → 使用填充后的提示词 → 生成审查报告
场景2:数据分析提示词模板
作用:提供标准化的数据分析提示词
使用场景:
- 数据报告:自动生成数据分析报告
- 业务洞察:从数据中提取业务洞察
- 可视化建议:提供数据可视化建议
示例:
// 提示词模板
{
"name": "data_analysis",
"description": "数据分析提示词模板",
"arguments": [
{
"name": "data",
"description": "要分析的数据",
"required": true
},
{
"name": "analysis_type",
"description": "分析类型(趋势/对比/预测)",
"required": true
}
]
}
模板内容:
请分析以下数据:
{data}
分析类型:{analysis_type}
分析要求:
1. 识别数据中的关键模式和趋势
2. 提供数据洞察和业务建议
3. 指出异常值和潜在问题
4. 建议下一步行动
实际应用:
用户:"分析一下销售数据"
AI → MCP:获取 data_analysis 模板
AI → 填充参数:{data: 销售数据, analysis_type: '趋势'}
AI → 使用填充后的提示词 → 生成分析报告
场景3:Bug 修复提示词模板
作用:提供标准化的 Bug 修复提示词
使用场景:
- 错误诊断:系统化诊断代码错误
- 修复建议:提供结构化的修复建议
- 测试验证:生成测试用例验证修复
示例:
// 提示词模板
{
"name": "bug_fix",
"description": "Bug修复提示词模板",
"arguments": [
{
"name": "error_message",
"description": "错误信息",
"required": true
},
{
"name": "code",
"description": "出错的代码",
"required": true
},
{
"name": "context",
"description": "上下文信息(可选)",
"required": false
}
]
}
模板内容:
请修复以下Bug:
错误信息:
{error_message}
出错代码:
{code}
{context ? "上下文信息:\n" + context : ""}
修复要求:
1. 分析错误原因
2. 提供修复方案
3. 解释修复原理
4. 提供测试建议
实际应用:
用户:"这个函数报错了,帮我修复"
AI → MCP:获取 bug_fix 模板
AI → 填充参数:{error_message: '...', code: '...'}
AI → 使用填充后的提示词 → 生成修复方案
3.5 Prompts 的优势
- 一致性:确保团队使用相同的提示词标准
- 效率:避免重复编写相似的提示词
- 质量:使用经过验证的模板,提高输出质量
- 维护性:集中管理提示词,便于更新和维护
4. Tools(工具)详解
4.1 什么是 Tools?
**Tools(工具)**是 MCP 服务器提供的可执行操作,类似于"功能按钮"。客户端可以调用这些工具执行搜索、计算、API 调用等操作,工具执行后返回结果。
4.2 Tools 的核心特点
- 可执行:工具是动态操作,不是静态数据
- 参数化:工具接受输入参数
- 有返回值:工具执行后返回结果
- 类型多样:查询工具、操作工具、计算工具等
4.3 Tools 的使用流程
【第1步:列出工具】
客户端 → 服务器:"有哪些工具?"
服务器 → 客户端:
- search_web:搜索网页
- read_file:读取文件
- calculate:计算
【第2步:调用工具】
客户端 → 服务器:"调用 search_web,参数:{query: 'MCP协议', num_results: 10}"
【第3步:执行】
服务器执行搜索 → 返回结果
【第4步:返回结果】
服务器 → 客户端:"搜索结果:..."
【第5步:使用结果】
客户端拿到结果 → 传给AI模型 → 生成回答
4.4 Tools 的实际应用场景
场景1:搜索工具
作用:提供网页、代码、文档搜索能力
使用场景:
- 信息检索:搜索最新信息和技术文档
- 代码搜索:在代码库中搜索相关实现
- 问题解答:搜索问题解决方案
示例:
// 工具定义
{
"name": "search_web",
"description": "搜索网页内容",
"inputSchema": {
"type": "object",
"properties": {
"query": {
"type": "string",
"description": "搜索关键词"
},
"num_results": {
"type": "number",
"description": "返回结果数量",
"default": 10
}
},
"required": ["query"]
}
}
实际应用:
用户:"MCP协议的最新发展是什么?"
AI → MCP:调用 search_web({query: 'MCP协议最新发展', num_results: 5})
AI → 获取搜索结果 → 分析总结 → 回答用户
场景2:文件操作工具
作用:提供文件读写、列表等操作能力
使用场景:
- 文件管理:读取、写入、列出文件
- 代码生成:生成代码文件
- 配置管理:读取和修改配置文件
示例:
// 工具定义
{
"name": "read_file",
"description": "读取文件内容",
"inputSchema": {
"type": "object",
"properties": {
"path": {
"type": "string",
"description": "文件路径"
}
},
"required": ["path"]
}
}
实际应用:
用户:"帮我看看这个配置文件的内容"
AI → MCP:调用 read_file({path: '/project/config.json'})
AI → 获取文件内容 → 分析配置 → 提供建议
场景3:计算工具
作用:提供数学计算、数据分析等能力
使用场景:
- 数值计算:执行复杂数学计算
- 数据分析:分析数据并生成统计信息
- 单位转换:进行单位换算
示例:
// 工具定义
{
"name": "calculate",
"description": "执行数学计算",
"inputSchema": {
"type": "object",
"properties": {
"expression": {
"type": "string",
"description": "数学表达式"
}
},
"required": ["expression"]
}
}
实际应用:
用户:"计算 (123 + 456) * 789 的结果"
AI → MCP:调用 calculate({expression: '(123 + 456) * 789'})
AI → 获取计算结果 → 返回给用户
4.5 Tools vs Resources 的区别
| 特性 | Tools(工具) | Resources(资源) |
|---|---|---|
| 性质 | 动态操作 | 静态数据 |
| 执行 | 需要调用执行 | 直接读取 |
| 参数 | 接受输入参数 | 通过 URI 标识 |
| 结果 | 执行后返回结果 | 返回资源内容 |
| 示例 | search_web、calculate | file:///path/to/file.txt |
选择原则:
- 需要执行操作 → 使用 Tools
- 需要读取数据 → 使用 Resources
5. Ping(连接测试)详解
5.1 什么是 Ping?
Ping是 MCP 协议中的连接测试功能,用于验证 MCP 服务器的连接状态和响应速度。类似于网络中的 ping 命令,用于检查服务器是否可用。
5.2 Ping 的核心特点
- 简单快速:轻量级测试,不执行复杂操作
- 状态检查:验证服务器是否在线
- 性能测试:测量响应时间
- 健康检查:用于监控和诊断
5.3 Ping 的使用流程
【第1步:发送 Ping 请求】
客户端 → 服务器:"ping"
【第2步:服务器响应】
服务器 → 客户端:"pong"(或包含时间戳的响应)
【第3步:计算延迟】
客户端计算请求-响应时间 → 判断连接质量
5.4 Ping 的实际应用场景
场景1:服务器健康检查
作用:定期检查 MCP 服务器是否正常运行
使用场景:
- 监控系统:自动化监控服务器状态
- 故障诊断:快速定位连接问题
- 负载均衡:选择响应最快的服务器
示例:
// 健康检查脚本
每30秒执行一次:
1. 发送 ping 请求
2. 检查响应时间
3. 如果响应时间 > 1秒 → 记录警告
4. 如果无响应 → 标记服务器为离线
实际应用:
监控系统 → MCP服务器:ping
MCP服务器 → 监控系统:pong (响应时间: 50ms)
监控系统 → 判断:服务器健康 ✓
场景2:连接初始化验证
作用:在建立连接后验证服务器可用性
使用场景:
- 连接建立:确认连接成功建立
- 配置验证:验证服务器配置正确
- 能力检查:确认服务器支持的功能
示例:
// 连接初始化流程
1. 建立连接
2. 发送 initialize 请求
3. 发送 ping 请求验证
4. 如果 ping 成功 → 连接就绪
5. 如果 ping 失败 → 重试或报错
实际应用:
AI应用启动 → 连接MCP服务器
AI应用 → MCP服务器:ping
MCP服务器 → AI应用:pong
AI应用 → 确认连接成功 → 开始使用
场景3:性能基准测试
作用:测量服务器响应性能
使用场景:
- 性能评估:评估服务器性能
- 优化指导:识别性能瓶颈
- 容量规划:规划服务器资源
示例:
// 性能测试脚本
执行100次ping:
1. 记录每次响应时间
2. 计算平均响应时间
3. 计算最大/最小响应时间
4. 生成性能报告
实际应用:
性能测试工具 → MCP服务器:ping (100次)
MCP服务器 → 性能测试工具:pong
性能测试工具 → 分析:
- 平均响应时间:45ms
- 最大响应时间:120ms
- 最小响应时间:30ms
- 性能评级:优秀 ✓
5.5 Ping 的优势
- 轻量级:不消耗大量资源
- 快速:几乎即时响应
- 标准化:所有 MCP 服务器都支持
- 诊断工具:快速定位连接问题
6. Sampling(采样)详解
6.1 什么是 Sampling?
**Sampling(采样)**是 MCP 协议中的高级功能,允许 MCP 服务器请求 AI 客户端进行推理。这是一个"反向"操作:不是客户端请求服务器,而是服务器请求客户端进行 AI 推理。
6.2 Sampling 的核心特点
- 反向请求:服务器请求客户端进行推理
- AI 推理:利用客户端的 AI 能力
- 复杂交互:实现更复杂的交互模式
- 双向通信:客户端和服务器可以互相请求
6.3 Sampling 的使用流程
【第1步:服务器请求采样】
服务器 → 客户端:"请对以下内容进行推理:{prompt: '...', model: '...'}"
【第2步:客户端执行推理】
客户端 → AI模型:执行推理请求
AI模型 → 客户端:返回推理结果
【第3步:客户端返回结果】
客户端 → 服务器:返回推理结果
【第4步:服务器使用结果】
服务器 → 处理推理结果 → 返回给客户端
6.4 Sampling 的实际应用场景
场景1:智能代码生成服务器
作用:服务器需要 AI 帮助生成代码
使用场景:
- 代码补全:根据上下文生成代码片段
- 代码重构:使用 AI 重构代码
- 代码解释:使用 AI 解释复杂代码
示例:
// 服务器请求采样
服务器 → 客户端:sampling({
prompt: "根据以下函数签名生成实现:\nfunction calculateTotal(items: Item[]): number",
model: "claude-3.5-sonnet",
temperature: 0.7
})
// 客户端执行推理
客户端 → AI模型:执行推理
AI模型 → 客户端:返回生成的代码
// 客户端返回结果
客户端 → 服务器:返回生成的代码
服务器 → 处理代码 → 返回给用户
实际应用:
用户:"帮我生成一个计算总价的函数"
代码生成服务器 → AI客户端:sampling({
prompt: "生成计算总价函数,参数:items数组"
})
AI客户端 → 生成代码 → 返回给服务器
服务器 → 返回生成的代码给用户
场景2:智能数据分析服务器
作用:服务器需要 AI 帮助分析数据
使用场景:
- 数据洞察:使用 AI 提取数据洞察
- 异常检测:使用 AI 识别数据异常
- 预测分析:使用 AI 进行预测
示例:
// 服务器请求采样
服务器 → 客户端:sampling({
prompt: "分析以下销售数据,找出关键趋势:\n{data: {...}}",
model: "claude-3.5-sonnet",
temperature: 0.3
})
// 客户端执行推理
客户端 → AI模型:分析数据
AI模型 → 客户端:返回分析结果
// 客户端返回结果
客户端 → 服务器:返回分析结果
服务器 → 格式化结果 → 返回给用户
实际应用:
用户:"分析一下这个月的销售数据"
数据分析服务器 → AI客户端:sampling({
prompt: "分析销售数据,找出趋势和异常"
})
AI客户端 → 分析数据 → 返回洞察
服务器 → 格式化洞察 → 返回给用户
场景3:智能问答服务器
作用:服务器需要 AI 帮助回答问题
使用场景:
- 知识问答:使用 AI 回答用户问题
- 文档理解:使用 AI 理解文档内容
- 上下文理解:使用 AI 理解对话上下文
示例:
// 服务器请求采样
服务器 → 客户端:sampling({
prompt: "根据以下文档内容回答问题:\n文档:{...}\n问题:什么是MCP?",
model: "claude-3.5-sonnet",
temperature: 0.5
})
// 客户端执行推理
客户端 → AI模型:理解文档并回答问题
AI模型 → 客户端:返回答案
// 客户端返回结果
客户端 → 服务器:返回答案
服务器 → 返回给用户
实际应用:
用户:"MCP协议是什么?"
问答服务器 → AI客户端:sampling({
prompt: "解释MCP协议,基于以下文档:{文档内容}"
})
AI客户端 → 生成答案 → 返回给服务器
服务器 → 返回答案给用户
6.5 Sampling 的优势
- 双向通信:服务器可以利用客户端的 AI 能力
- 复杂交互:实现更复杂的交互模式
- 灵活推理:服务器可以请求不同类型的推理
- 增强功能:扩展服务器的能力边界
6.6 Sampling 的注意事项
- 性能考虑:AI 推理可能较慢,需要合理使用
- 成本控制:AI 推理可能产生费用,需要控制调用频率
- 错误处理:需要处理推理失败的情况
- 权限管理:需要控制哪些服务器可以请求采样
7. Roots(根目录)详解
7.1 什么是 Roots?
**Roots(根目录)**是 MCP 协议中用于定义资源根目录的功能。它指定了服务器可以访问的文件系统路径范围,类似于文件系统的"工作目录"。
7.2 Roots 的核心特点
- 路径限制:定义服务器可以访问的路径范围
- 安全控制:限制服务器访问范围,提高安全性
- 资源组织:帮助组织和定位资源
- 多根支持:可以定义多个根目录
7.3 Roots 的使用流程
【第1步:定义根目录】
服务器配置:
roots: [
"/project/src",
"/project/docs"
]
【第2步:资源访问】
客户端请求资源:
- file:///project/src/main.py ✓ (在根目录内)
- file:///project/docs/readme.md ✓ (在根目录内)
- file:///etc/passwd ✗ (不在根目录内,拒绝访问)
【第3步:安全验证】
服务器检查资源路径是否在根目录内 → 允许或拒绝访问
7.4 Roots 的实际应用场景
场景1:项目文件系统服务器
作用:限制服务器只能访问项目目录
使用场景:
- 代码管理:只允许访问项目代码文件
- 安全隔离:防止访问系统文件
- 权限控制:控制文件访问范围
示例:
// 服务器配置
{
"roots": [
"/workspace/my-project/src",
"/workspace/my-project/docs",
"/workspace/my-project/config"
]
}
实际应用:
// 允许访问
file:///workspace/my-project/src/main.py ✓
file:///workspace/my-project/docs/readme.md ✓
// 拒绝访问
file:///etc/passwd ✗
file:///home/user/private.txt ✗
场景2:多项目支持
作用:支持访问多个项目目录
使用场景:
- 多项目开发:同时管理多个项目
- 共享资源:访问共享的配置和文档
- 模块化组织:按模块组织资源
示例:
// 服务器配置
{
"roots": [
"/workspace/project-a",
"/workspace/project-b",
"/workspace/shared"
]
}
实际应用:
// 可以访问所有根目录下的资源
file:///workspace/project-a/src/main.py ✓
file:///workspace/project-b/src/main.py ✓
file:///workspace/shared/config.json ✓
场景3:临时工作目录
作用:为临时文件定义工作目录
使用场景:
- 临时文件:管理临时生成的文件
- 缓存目录:管理缓存文件
- 日志目录:管理日志文件
示例:
// 服务器配置
{
"roots": [
"/tmp/mcp-workspace",
"/var/cache/mcp"
]
}
实际应用:
// 临时文件操作
file:///tmp/mcp-workspace/temp-file.txt ✓
file:///var/cache/mcp/cache.json ✓
7.5 Roots 的优势
- 安全性:限制访问范围,防止越权访问
- 组织性:帮助组织和定位资源
- 灵活性:支持多个根目录
- 清晰性:明确资源访问边界
7.6 Roots 的注意事项
- 路径规范:使用绝对路径,避免相对路径
- 权限检查:确保服务器有权限访问根目录
- 路径解析:正确处理路径分隔符(Windows vs Unix)
- 动态更新:支持运行时更新根目录(如果服务器支持)
8. 综合使用场景
8.1 场景1:智能代码审查工作流
需求:AI 帮助审查代码,提供改进建议
使用功能组合:
- Resources:读取代码文件
- Prompts:使用代码审查提示词模板
- Tools:调用代码分析工具
- Ping:验证服务器连接
工作流程:
1. Ping 验证连接
AI应用 → MCP服务器:ping
MCP服务器 → AI应用:pong
2. 获取代码审查提示词模板
AI应用 → MCP服务器:prompts/get(code_review)
MCP服务器 → AI应用:返回模板
3. 读取代码文件资源
AI应用 → MCP服务器:resources/read(file:///project/src/main.py)
MCP服务器 → AI应用:返回代码内容
4. 调用代码分析工具
AI应用 → MCP服务器:tools/call(analyze_code, {code: ...})
MCP服务器 → AI应用:返回分析结果
5. 使用提示词模板生成审查报告
AI应用 → 填充提示词模板 → 生成审查报告
8.2 场景2:智能数据分析工作流
需求:AI 分析数据,生成业务洞察
使用功能组合:
- Resources:读取数据资源
- Prompts:使用数据分析提示词模板
- Tools:调用计算和统计工具
- Sampling:请求 AI 进行深度分析
工作流程:
1. 读取数据资源
AI应用 → MCP服务器:resources/read(db://sales/recent)
MCP服务器 → AI应用:返回销售数据
2. 获取数据分析提示词模板
AI应用 → MCP服务器:prompts/get(data_analysis)
MCP服务器 → AI应用:返回模板
3. 调用统计工具
AI应用 → MCP服务器:tools/call(calculate_stats, {data: ...})
MCP服务器 → AI应用:返回统计结果
4. 请求 AI 深度分析(Sampling)
MCP服务器 → AI应用:sampling({
prompt: "分析销售数据,找出关键趋势和异常"
})
AI应用 → 执行推理 → 返回分析结果
5. 生成综合分析报告
AI应用 → 整合所有结果 → 生成报告
8.3 场景3:智能文档生成工作流
需求:AI 读取多个文档,生成综合报告
使用功能组合:
- Resources:读取多个文档资源
- Prompts:使用文档生成提示词模板
- Tools:调用文档处理工具
- Roots:限制文档访问范围
工作流程:
1. 列出可用文档资源(在根目录内)
AI应用 → MCP服务器:resources/list()
MCP服务器 → AI应用:返回资源列表(仅根目录内的)
2. 读取多个文档
AI应用 → MCP服务器:resources/read(file:///docs/doc1.md)
AI应用 → MCP服务器:resources/read(file:///docs/doc2.md)
AI应用 → MCP服务器:resources/read(file:///docs/doc3.md)
3. 获取文档生成提示词模板
AI应用 → MCP服务器:prompts/get(document_synthesis)
MCP服务器 → AI应用:返回模板
4. 调用文档处理工具
AI应用 → MCP服务器:tools/call(extract_key_points, {documents: ...})
MCP服务器 → AI应用:返回关键点
5. 生成综合报告
AI应用 → 使用提示词模板 → 生成综合报告
9. 最佳实践与注意事项
9.1 Resources 最佳实践
- 合理组织资源:使用清晰的 URI 命名规范
- 按需加载:只请求需要的资源,避免一次性加载大量资源
- 缓存策略:对不经常变化的资源进行缓存
- 错误处理:处理资源不存在或访问失败的情况
9.2 Prompts 最佳实践
- 模板标准化:使用统一的模板格式和命名规范
- 参数验证:验证模板参数的有效性
- 版本管理:对提示词模板进行版本控制
- 文档完善:为每个模板提供清晰的文档说明
9.3 Tools 最佳实践
- 工具命名:使用清晰、描述性的工具名称
- 参数设计:设计合理的参数结构,提供默认值
- 错误处理:提供详细的错误信息和错误码
- 性能优化:优化工具执行性能,避免长时间阻塞
9.4 Ping 最佳实践
- 定期检查:定期执行 ping 检查服务器健康状态
- 超时设置:设置合理的超时时间
- 重试机制:实现 ping 失败的重试机制
- 日志记录:记录 ping 结果用于监控和分析
9.5 Sampling 最佳实践
- 合理使用:只在必要时使用 sampling,避免过度调用
- 成本控制:监控 sampling 调用次数和成本
- 错误处理:处理 sampling 失败的情况
- 权限控制:限制哪些服务器可以请求 sampling
9.6 Roots 最佳实践
- 最小权限:只授予必要的访问权限
- 路径规范:使用绝对路径,避免相对路径
- 权限验证:确保服务器有权限访问根目录
- 安全审计:定期审计根目录配置
9.7 综合注意事项
- 性能优化:合理使用各种功能,避免性能瓶颈
- 错误处理:实现完善的错误处理机制
- 安全考虑:注意安全性和权限控制
- 文档完善:为服务器功能提供清晰的文档
- 测试验证:充分测试各种功能的使用场景
10. 真实案例详解:Prompts 和 Resources 的实际应用
10.1 Resources 真实案例:GitHub 代码审查助手
场景描述
假设你是一个开发团队,需要 AI 助手帮助审查 Pull Request。你希望 AI 能够:
- 读取 PR 中的所有代码文件
- 读取相关的配置文件
- 读取测试文件
- 基于这些文件内容进行代码审查
不使用 Resources 的方式(传统方式)
# 传统方式:AI 应用需要自己处理文件读取
import os
import json
def review_pr(pr_number):
# AI 应用需要自己实现文件读取逻辑
pr_files = get_pr_files(pr_number) # 需要自己调用 GitHub API
code_content = {}
for file_path in pr_files:
# 需要自己处理文件路径、权限、缓存等
if file_path.endswith('.py'):
with open(file_path, 'r', encoding='utf-8') as f:
code_content[file_path] = f.read()
# 需要自己处理错误、编码、大文件等问题
prompt = f"请审查以下代码:\n{code_content}"
result = ai_model.generate(prompt)
return result
问题:
- 需要重复实现文件读取逻辑
- 需要处理各种边界情况(权限、编码、大文件等)
- 代码耦合度高,难以复用
- 每个 AI 应用都要自己实现
使用 Resources 的方式(MCP 方式)
第一步:MCP 服务器提供资源
# MCP 服务器:github-resources-server.py
from mcp.server import Server
from mcp.types import Resource, TextContent
import requests
server = Server("github-resources")
@server.list_resources()
async def list_resources() -> list[Resource]:
"""列出 PR 中的所有文件作为资源"""
pr_files = get_pr_files_from_github()
resources = []
for file in pr_files:
resources.append(Resource(
uri=f"github://pr/123/{file['path']}",
name=file['path'],
description=f"PR #123 中的文件:{file['path']}",
mimeType="text/x-python" if file['path'].endswith('.py') else "text/plain"
))
return resources
@server.read_resource()
async def read_resource(uri: str) -> list[TextContent]:
"""读取资源内容"""
# 解析 URI:github://pr/123/src/main.py
parts = uri.replace("github://", "").split("/", 2)
pr_number = parts[1]
file_path = parts[2]
# 从 GitHub API 获取文件内容
content = get_file_content_from_github(pr_number, file_path)
return [TextContent(type="text", text=content)]
第二步:AI 应用使用资源
# AI 应用:使用 MCP Resources
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client
async def review_pr_with_resources(pr_number):
# 连接 MCP 服务器
params = StdioServerParameters(
command="python",
args=["github-resources-server.py"]
)
async with stdio_client(params) as (read, write):
async with ClientSession(read, write) as session:
await session.initialize()
# 1. 列出所有可用资源(PR 中的文件)
resources = await session.list_resources()
print(f"找到 {len(resources.resources)} 个文件")
# 2. 读取所有代码文件
code_files = []
for resource in resources.resources:
if resource.uri.endswith('.py'):
# 直接通过 URI 读取,不需要知道文件系统路径
content = await session.read_resource(resource.uri)
code_files.append({
'path': resource.name,
'content': content[0].text
})
# 3. 构建审查提示词
all_code = "\n\n".join([
f"文件:{f['path']}\n代码:\n{f['content']}"
for f in code_files
])
prompt = f"请审查以下 PR 中的代码:\n\n{all_code}"
result = ai_model.generate(prompt)
return result
对比优势:
| 特性 | 传统方式 | MCP Resources 方式 |
|---|---|---|
| 文件读取逻辑 | 每个应用自己实现 | MCP 服务器统一提供 |
| 代码复用 | 无法复用 | 多个 AI 应用共享 |
| 错误处理 | 每个应用自己处理 | MCP 服务器统一处理 |
| 权限管理 | 每个应用自己管理 | MCP 服务器统一管理 |
| 缓存策略 | 每个应用自己实现 | MCP 服务器统一实现 |
真实工作流程示例
【场景:审查 PR #123】
1. AI 应用启动,连接 GitHub Resources MCP 服务器
→ 连接成功 ✓
2. AI 应用列出资源
AI应用 → MCP服务器:resources/list()
MCP服务器 → AI应用:
- github://pr/123/src/main.py
- github://pr/123/src/utils.py
- github://pr/123/tests/test_main.py
- github://pr/123/.github/workflows/ci.yml
3. AI 应用读取代码文件
AI应用 → MCP服务器:resources/read(github://pr/123/src/main.py)
MCP服务器 → GitHub API:获取文件内容
GitHub API → MCP服务器:返回文件内容
MCP服务器 → AI应用:返回文件内容(已处理编码、权限等)
4. AI 应用读取其他相关文件
AI应用 → MCP服务器:resources/read(github://pr/123/src/utils.py)
AI应用 → MCP服务器:resources/read(github://pr/123/tests/test_main.py)
5. AI 应用基于所有文件内容进行审查
AI应用 → 整合所有文件内容 → 生成审查报告
实际代码示例:完整的 MCP 服务器实现
# github-resources-server.py - 完整的实现
from mcp.server import Server
from mcp.types import Resource, TextContent
import requests
import os
server = Server("github-resources")
# GitHub API 配置
GITHUB_TOKEN = os.getenv("GITHUB_TOKEN")
GITHUB_OWNER = os.getenv("GITHUB_OWNER", "your-org")
GITHUB_REPO = os.getenv("GITHUB_REPO", "your-repo")
def get_pr_files(pr_number: int):
"""从 GitHub API 获取 PR 文件列表"""
url = f"https://api.github.com/repos/{GITHUB_OWNER}/{GITHUB_REPO}/pulls/{pr_number}/files"
headers = {"Authorization": f"token {GITHUB_TOKEN}"}
response = requests.get(url, headers=headers)
return response.json()
def get_file_content(pr_number: int, file_path: str):
"""从 GitHub API 获取文件内容"""
url = f"https://api.github.com/repos/{GITHUB_OWNER}/{GITHUB_REPO}/contents/{file_path}"
headers = {"Authorization": f"token {GITHUB_TOKEN}"}
params = {"ref": f"pr/{pr_number}"}
response = requests.get(url, headers=headers, params=params)
data = response.json()
# GitHub API 返回 base64 编码的内容
import base64
content = base64.b64decode(data['content']).decode('utf-8')
return content
@server.list_resources()
async def list_resources() -> list[Resource]:
"""列出当前 PR 的所有文件作为资源"""
# 从环境变量获取当前 PR 号
pr_number = int(os.getenv("PR_NUMBER", "123"))
files = get_pr_files(pr_number)
resources = []
for file in files:
# 确定 MIME 类型
mime_type = "text/plain"
if file['filename'].endswith('.py'):
mime_type = "text/x-python"
elif file['filename'].endswith('.js'):
mime_type = "text/javascript"
elif file['filename'].endswith('.md'):
mime_type = "text/markdown"
elif file['filename'].endswith('.json'):
mime_type = "application/json"
resources.append(Resource(
uri=f"github://pr/{pr_number}/{file['filename']}",
name=file['filename'],
description=f"PR #{pr_number} 中的文件:{file['filename']} ({file['status']})",
mimeType=mime_type
))
return resources
@server.read_resource()
async def read_resource(uri: str) -> list[TextContent]:
"""读取资源内容"""
# 解析 URI:github://pr/123/src/main.py
if not uri.startswith("github://pr/"):
raise ValueError(f"Invalid URI: {uri}")
parts = uri.replace("github://pr/", "").split("/", 1)
pr_number = int(parts[0])
file_path = parts[1]
try:
content = get_file_content(pr_number, file_path)
return [TextContent(type="text", text=content)]
except Exception as e:
return [TextContent(
type="text",
text=f"Error reading file: {str(e)}"
)]
if __name__ == "__main__":
server.run()
使用方式:
# 设置环境变量
export GITHUB_TOKEN="ghp_xxxxx"
export GITHUB_OWNER="your-org"
export GITHUB_REPO="your-repo"
export PR_NUMBER="123"
# 启动 MCP 服务器
python github-resources-server.py
10.2 Prompts 真实案例:企业级代码审查系统
场景描述
假设你是一个大型科技公司,有 100+ 开发团队,每个团队都需要进行代码审查。你希望:
- 统一代码审查标准
- 确保所有团队使用相同的审查标准
- 便于更新和维护审查标准
- 支持不同编程语言的审查模板
不使用 Prompts 的方式(传统方式)
# 传统方式:每个 AI 应用自己写提示词
def review_code_traditional(code: str, language: str):
# 每个应用都要自己写提示词,容易不一致
if language == "python":
prompt = """
请审查以下Python代码:
{code}
检查点:
1. 是否有语法错误
2. 是否有逻辑错误
3. 性能是否优化
"""
elif language == "javascript":
prompt = """
请审查以下JavaScript代码:
{code}
检查点:
1. 是否有语法错误
2. 是否有逻辑错误
3. 性能是否优化
"""
# ... 更多语言
# 问题:
# 1. 提示词分散在各个应用中,难以统一更新
# 2. 每个应用可能写的不一样
# 3. 新团队需要重新写提示词
# 4. 标准更新需要修改所有应用
return ai_model.generate(prompt.format(code=code))
问题:
- 提示词分散,难以统一管理
- 每个团队可能写的不一样
- 更新标准需要修改所有应用
- 新团队需要从零开始
使用 Prompts 的方式(MCP 方式)
第一步:MCP 服务器提供提示词模板
# MCP 服务器:code-review-prompts-server.py
from mcp.server import Server
from mcp.types import Prompt, PromptMessage
server = Server("code-review-prompts")
# 定义标准化的代码审查提示词模板
CODE_REVIEW_TEMPLATES = {
"python": """
请审查以下Python代码:
{code}
审查标准(必须检查):
1. **语法和风格**
- 是否符合 PEP 8 规范
- 变量命名是否清晰
- 是否有未使用的导入
2. **逻辑和错误处理**
- 是否有潜在的逻辑错误
- 异常处理是否完善
- 边界条件是否考虑
3. **性能和最佳实践**
- 是否有性能瓶颈
- 是否使用了合适的数据结构
- 是否有内存泄漏风险
4. **安全性**
- 是否有 SQL 注入风险
- 是否有 XSS 风险
- 敏感信息是否硬编码
5. **可维护性**
- 代码是否易于理解
- 是否有足够的注释
- 函数是否过于复杂
请按照以下格式输出:
- 严重问题:[列出严重问题]
- 改进建议:[列出改进建议]
- 代码评分:[1-10分]
""",
"javascript": """
请审查以下JavaScript代码:
{code}
审查标准(必须检查):
1. **语法和风格**
- 是否符合 ESLint 规范
- 是否使用 const/let 而非 var
- 是否有未使用的变量
2. **逻辑和错误处理**
- 是否有潜在的逻辑错误
- Promise 错误处理是否完善
- 异步操作是否正确处理
3. **性能和最佳实践**
- 是否有性能瓶颈
- 是否有内存泄漏
- 事件监听器是否正确清理
4. **安全性**
- 是否有 XSS 风险
- 是否有 CSRF 风险
- API 密钥是否暴露
5. **可维护性**
- 代码是否易于理解
- 是否有足够的注释
- 函数是否过于复杂
请按照以下格式输出:
- 严重问题:[列出严重问题]
- 改进建议:[列出改进建议]
- 代码评分:[1-10分]
""",
"java": """
请审查以下Java代码:
{code}
审查标准(必须检查):
1. **语法和风格**
- 是否符合 Google Java Style Guide
- 命名是否规范
- 是否有未使用的导入
2. **逻辑和错误处理**
- 是否有潜在的逻辑错误
- 异常处理是否完善
- 空指针检查是否充分
3. **性能和最佳实践**
- 是否有性能瓶颈
- 集合使用是否高效
- 是否有内存泄漏
4. **安全性**
- 是否有 SQL 注入风险
- 输入验证是否充分
- 敏感信息是否硬编码
5. **可维护性**
- 代码是否易于理解
- 是否有足够的 JavaDoc
- 类和方法是否过于复杂
请按照以下格式输出:
- 严重问题:[列出严重问题]
- 改进建议:[列出改进建议]
- 代码评分:[1-10分]
"""
}
@server.list_prompts()
async def list_prompts():
"""列出所有可用的提示词模板"""
prompts = []
for language, template in CODE_REVIEW_TEMPLATES.items():
prompts.append(Prompt(
name=f"code_review_{language}",
description=f"标准化的{language}代码审查提示词模板",
arguments=[
{
"name": "code",
"description": "要审查的代码",
"required": True
},
{
"name": "focus",
"description": "重点关注领域(可选):security, performance, style",
"required": False
}
]
))
return prompts
@server.get_prompt()
async def get_prompt(name: str, arguments: dict):
"""获取填充后的提示词"""
# 解析提示词名称:code_review_python
if name.startswith("code_review_"):
language = name.replace("code_review_", "")
template = CODE_REVIEW_TEMPLATES.get(language)
if not template:
raise ValueError(f"Unsupported language: {language}")
# 填充模板
code = arguments.get("code", "")
focus = arguments.get("focus", "")
filled_prompt = template.format(code=code)
# 如果有重点关注,添加额外说明
if focus:
focus_instructions = {
"security": "\n\n特别注意:请重点关注安全性问题,包括注入攻击、XSS、CSRF等。",
"performance": "\n\n特别注意:请重点关注性能问题,包括算法复杂度、内存使用等。",
"style": "\n\n特别注意:请重点关注代码风格和规范问题。"
}
filled_prompt += focus_instructions.get(focus, "")
return PromptMessage(
role="user",
content=TextContent(type="text", text=filled_prompt)
)
raise ValueError(f"Unknown prompt: {name}")
第二步:AI 应用使用提示词模板
# AI 应用:使用 MCP Prompts
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client
async def review_code_with_prompts(code: str, language: str, focus: str = None):
# 连接 MCP 服务器
params = StdioServerParameters(
command="python",
args=["code-review-prompts-server.py"]
)
async with stdio_client(params) as (read, write):
async with ClientSession(read, write) as session:
await session.initialize()
# 1. 列出所有可用的提示词模板
prompts = await session.list_prompts()
print(f"可用提示词:{[p.name for p in prompts.prompts]}")
# 2. 获取标准化的代码审查提示词
prompt_name = f"code_review_{language}"
arguments = {"code": code}
if focus:
arguments["focus"] = focus
prompt_message = await session.get_prompt(prompt_name, arguments)
# 3. 使用标准化的提示词进行审查
result = ai_model.generate(prompt_message.content[0].text)
return result
真实工作流程示例
【场景:团队 A 审查 Python 代码】
1. 开发者提交代码到 PR
代码文件:src/utils.py
2. AI 审查系统启动
AI应用 → 连接 MCP Prompts 服务器 ✓
3. AI 应用获取标准化的审查提示词
AI应用 → MCP服务器:prompts/get(code_review_python, {code: "...", focus: "security"})
MCP服务器 → AI应用:返回填充后的标准化提示词
提示词内容:
"""
请审查以下Python代码:
[代码内容]
审查标准(必须检查):
1. 语法和风格...
2. 逻辑和错误处理...
...
特别注意:请重点关注安全性问题...
"""
4. AI 应用使用标准化提示词进行审查
AI应用 → AI模型:使用标准化提示词
AI模型 → AI应用:返回审查结果
审查结果:
- 严重问题:发现 SQL 注入风险
- 改进建议:使用参数化查询
- 代码评分:6/10
5. 审查结果返回给开发者
→ 开发者根据标准化审查结果修改代码
实际代码示例:完整的 MCP 服务器实现
# code-review-prompts-server.py - 完整的实现
from mcp.server import Server
from mcp.types import Prompt, PromptMessage, TextContent
import json
import os
server = Server("code-review-prompts")
# 从配置文件加载提示词模板(便于更新)
PROMPTS_CONFIG_PATH = os.getenv("PROMPTS_CONFIG", "prompts_config.json")
def load_prompts_config():
"""从配置文件加载提示词模板"""
if os.path.exists(PROMPTS_CONFIG_PATH):
with open(PROMPTS_CONFIG_PATH, 'r', encoding='utf-8') as f:
return json.load(f)
else:
# 默认配置
return {
"code_review_python": {
"template": """请审查以下Python代码:\n\n{code}\n\n审查标准...""",
"description": "标准化的Python代码审查提示词模板"
},
# ... 更多模板
}
PROMPTS_CONFIG = load_prompts_config()
@server.list_prompts()
async def list_prompts():
"""列出所有可用的提示词模板"""
prompts = []
for prompt_name, config in PROMPTS_CONFIG.items():
# 解析参数(从模板中提取 {变量名})
import re
args_in_template = re.findall(r'\{(\w+)\}', config['template'])
arguments = []
for arg_name in set(args_in_template):
if arg_name != 'code': # code 是必需的
arguments.append({
"name": arg_name,
"description": f"参数:{arg_name}",
"required": False
})
prompts.append(Prompt(
name=prompt_name,
description=config.get('description', ''),
arguments=arguments
))
return prompts
@server.get_prompt()
async def get_prompt(name: str, arguments: dict):
"""获取填充后的提示词"""
if name not in PROMPTS_CONFIG:
raise ValueError(f"Unknown prompt: {name}")
config = PROMPTS_CONFIG[name]
template = config['template']
# 填充模板
try:
filled_prompt = template.format(**arguments)
except KeyError as e:
raise ValueError(f"Missing required argument: {e}")
return PromptMessage(
role="user",
content=[TextContent(type="text", text=filled_prompt)]
)
# 支持热重载配置(便于更新)
@server.on_config_change()
async def reload_config():
"""当配置文件更新时重新加载"""
global PROMPTS_CONFIG
PROMPTS_CONFIG = load_prompts_config()
print("提示词配置已重新加载")
if __name__ == "__main__":
server.run()
配置文件示例(prompts_config.json):
{
"code_review_python": {
"template": "请审查以下Python代码:\n\n{code}\n\n审查标准(必须检查):\n1. 语法和风格...\n2. 逻辑和错误处理...\n{focus_instruction}",
"description": "标准化的Python代码审查提示词模板"
},
"code_review_javascript": {
"template": "请审查以下JavaScript代码:\n\n{code}\n\n审查标准(必须检查):\n1. 语法和风格...\n2. 逻辑和错误处理...\n{focus_instruction}",
"description": "标准化的JavaScript代码审查提示词模板"
}
}
对比优势
| 特性 | 传统方式 | MCP Prompts 方式 |
|---|---|---|
| 提示词管理 | 分散在各个应用中 | centralized 统一管理 |
| 标准一致性 | 每个团队可能不同 | 所有团队使用相同标准 |
| 更新维护 | 需要修改所有应用 | 只需更新 MCP 服务器 |
| 新团队上手 | 需要从零开始 | 直接使用标准模板 |
| 版本控制 | 难以追踪 | 可以版本化管理 |
| A/B 测试 | 难以实现 | 可以轻松测试不同版本 |
实际业务价值
场景:公司有 100 个开发团队
不使用 Prompts:
- 每个团队自己写提示词:100 个版本
- 更新标准需要通知 100 个团队:耗时且容易遗漏
- 新团队需要 2-3 天编写提示词
- 标准不一致导致审查质量参差不齐
使用 Prompts:
- 所有团队使用统一模板:1 个版本
- 更新标准只需更新 MCP 服务器:5 分钟完成
- 新团队直接使用:0 天
- 标准统一确保审查质量一致
效率提升:
- 维护时间:从 100 小时/次 → 5 分钟/次
- 新团队上手:从 2-3 天 → 0 天
- 标准一致性:从 60% → 100%
10.3 综合案例:Resources + Prompts 联合使用
场景描述
构建一个智能代码审查系统,需要:
- 从 GitHub PR 读取代码文件(Resources)
- 使用标准化的审查提示词(Prompts)
- 生成审查报告
完整实现
# 智能代码审查系统
async def intelligent_code_review(pr_number: int, language: str):
# 1. 连接两个 MCP 服务器
github_params = StdioServerParameters(
command="python",
args=["github-resources-server.py"]
)
prompts_params = StdioServerParameters(
command="python",
args=["code-review-prompts-server.py"]
)
async with stdio_client(github_params) as (gh_read, gh_write), \
stdio_client(prompts_params) as (pr_read, pr_write):
github_session = ClientSession(gh_read, gh_write)
prompts_session = ClientSession(pr_read, pr_write)
await github_session.initialize()
await prompts_session.initialize()
# 2. 从 GitHub Resources 读取所有代码文件
resources = await github_session.list_resources()
code_files = []
for resource in resources.resources:
if resource.uri.endswith(f'.{language}'):
content = await github_session.read_resource(resource.uri)
code_files.append({
'path': resource.name,
'content': content[0].text
})
# 3. 合并所有代码
all_code = "\n\n".join([
f"文件:{f['path']}\n代码:\n{f['content']}"
for f in code_files
])
# 4. 从 Prompts 获取标准化的审查提示词
prompt_message = await prompts_session.get_prompt(
f"code_review_{language}",
{"code": all_code, "focus": "security"}
)
# 5. 使用提示词进行审查
review_result = ai_model.generate(
prompt_message.content[0].text
)
return review_result
工作流程:
1. 连接两个 MCP 服务器
✓ GitHub Resources 服务器
✓ Code Review Prompts 服务器
2. 读取代码(Resources)
→ 列出 PR 中的所有文件
→ 读取每个代码文件的内容
3. 获取提示词(Prompts)
→ 获取标准化的审查提示词模板
→ 填充代码内容
4. 执行审查
→ 使用标准化提示词
→ 生成审查报告
5. 返回结果
→ 统一的审查报告格式
→ 包含严重问题、改进建议、评分
10.4 快速上手:5 分钟体验 Resources 和 Prompts
示例1:Resources - 读取项目文档
场景:AI 助手需要读取项目的 README 和配置文件来回答用户问题
步骤1:创建简单的 Resources MCP 服务器
# simple-resources-server.py
from mcp.server import Server
from mcp.types import Resource, TextContent
import os
server = Server("simple-resources")
# 定义项目根目录
PROJECT_ROOT = os.getenv("PROJECT_ROOT", "/workspace/my-project")
@server.list_resources()
async def list_resources() -> list[Resource]:
"""列出项目中的文档文件"""
resources = []
# 扫描项目目录
for root, dirs, files in os.walk(PROJECT_ROOT):
for file in files:
if file.endswith(('.md', '.txt', '.json', '.yaml', '.yml')):
file_path = os.path.join(root, file)
rel_path = os.path.relpath(file_path, PROJECT_ROOT)
resources.append(Resource(
uri=f"file://{rel_path}",
name=rel_path,
description=f"项目文件:{rel_path}",
mimeType="text/markdown" if file.endswith('.md') else "text/plain"
))
return resources
@server.read_resource()
async def read_resource(uri: str) -> list[TextContent]:
"""读取文件内容"""
# 解析 URI:file://README.md
file_path = uri.replace("file://", "")
full_path = os.path.join(PROJECT_ROOT, file_path)
try:
with open(full_path, 'r', encoding='utf-8') as f:
content = f.read()
return [TextContent(type="text", text=content)]
except Exception as e:
return [TextContent(
type="text",
text=f"Error: {str(e)}"
)]
if __name__ == "__main__":
server.run()
步骤2:使用 Resources
# use-resources.py
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client
import asyncio
async def main():
params = StdioServerParameters(
command="python",
args=["simple-resources-server.py"]
)
async with stdio_client(params) as (read, write):
async with ClientSession(read, write) as session:
await session.initialize()
# 列出所有资源
resources = await session.list_resources()
print(f"找到 {len(resources.resources)} 个文件:")
for resource in resources.resources:
print(f" - {resource.name}")
# 读取 README.md
if any(r.name == "README.md" for r in resources.resources):
content = await session.read_resource("file://README.md")
print(f"\nREADME.md 内容:\n{content[0].text[:200]}...")
if __name__ == "__main__":
asyncio.run(main())
运行:
export PROJECT_ROOT="/path/to/your/project"
python use-resources.py
示例2:Prompts - 使用代码审查模板
场景:统一团队的代码审查标准
步骤1:创建简单的 Prompts MCP 服务器
# simple-prompts-server.py
from mcp.server import Server
from mcp.types import Prompt, PromptMessage, TextContent
server = Server("simple-prompts")
# 定义提示词模板
TEMPLATES = {
"code_review": """请审查以下{language}代码:
{code}
审查清单:
✓ 语法错误检查
✓ 逻辑错误检查
✓ 性能优化建议
✓ 代码风格检查
✓ 安全性检查
请提供:
1. 发现的问题(严重程度:高/中/低)
2. 改进建议
3. 代码评分(1-10分)
""",
"explain_code": """请解释以下{language}代码的功能:
{code}
请说明:
1. 代码的主要功能
2. 关键逻辑流程
3. 可能的改进方向
"""
}
@server.list_prompts()
async def list_prompts():
"""列出所有提示词模板"""
return [
Prompt(
name="code_review",
description="代码审查提示词模板",
arguments=[
{"name": "code", "description": "要审查的代码", "required": True},
{"name": "language", "description": "编程语言", "required": True}
]
),
Prompt(
name="explain_code",
description="代码解释提示词模板",
arguments=[
{"name": "code", "description": "要解释的代码", "required": True},
{"name": "language", "description": "编程语言", "required": True}
]
)
]
@server.get_prompt()
async def get_prompt(name: str, arguments: dict):
"""获取填充后的提示词"""
if name not in TEMPLATES:
raise ValueError(f"Unknown prompt: {name}")
template = TEMPLATES[name]
filled = template.format(**arguments)
return PromptMessage(
role="user",
content=[TextContent(type="text", text=filled)]
)
if __name__ == "__main__":
server.run()
步骤2:使用 Prompts
# use-prompts.py
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client
import asyncio
async def main():
params = StdioServerParameters(
command="python",
args=["simple-prompts-server.py"]
)
async with stdio_client(params) as (read, write):
async with ClientSession(read, write) as session:
await session.initialize()
# 列出所有提示词
prompts = await session.list_prompts()
print("可用提示词:")
for prompt in prompts.prompts:
print(f" - {prompt.name}: {prompt.description}")
# 获取代码审查提示词
code = """
def calculate_total(items):
total = 0
for item in items:
total += item.price
return total
"""
prompt_msg = await session.get_prompt(
"code_review",
{"code": code, "language": "python"}
)
print(f"\n生成的提示词:\n{prompt_msg.content[0].text}")
if __name__ == "__main__":
asyncio.run(main())
运行:
python use-prompts.py
示例3:Resources + Prompts 联合使用
场景:读取项目代码文件,使用标准化模板进行审查
# combined-example.py
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client
import asyncio
async def review_project_code():
# 连接两个 MCP 服务器
resources_params = StdioServerParameters(
command="python",
args=["simple-resources-server.py"]
)
prompts_params = StdioServerParameters(
command="python",
args=["simple-prompts-server.py"]
)
async with stdio_client(resources_params) as (r_read, r_write), \
stdio_client(prompts_params) as (p_read, p_write):
resources_session = ClientSession(r_read, r_write)
prompts_session = ClientSession(p_read, p_write)
await resources_session.initialize()
await prompts_session.initialize()
# 1. 从 Resources 读取代码文件
resources = await resources_session.list_resources()
python_files = [r for r in resources.resources if r.name.endswith('.py')]
print(f"找到 {len(python_files)} 个 Python 文件")
# 2. 读取第一个 Python 文件
if python_files:
first_file = python_files[0]
content = await resources_session.read_resource(first_file.uri)
code = content[0].text
print(f"\n读取文件:{first_file.name}")
print(f"代码长度:{len(code)} 字符")
# 3. 使用 Prompts 获取审查提示词
prompt_msg = await prompts_session.get_prompt(
"code_review",
{"code": code, "language": "python"}
)
print(f"\n生成的审查提示词:\n{prompt_msg.content[0].text[:300]}...")
# 4. 这里可以调用 AI 模型进行实际审查
# result = ai_model.generate(prompt_msg.content[0].text)
# print(f"\n审查结果:\n{result}")
if __name__ == "__main__":
asyncio.run(review_project_code())
运行:
export PROJECT_ROOT="/path/to/your/project"
python combined-example.py
11. 总结
MCP 扩展功能为 AI 应用提供了强大的能力:
- Resources:提供数据访问能力,支持文件、数据库、API 等多种数据源
- Prompts:提供提示词模板库,提高提示词质量和一致性
- Tools:提供功能调用能力,支持搜索、计算、API 调用等操作
- Ping:提供连接测试能力,用于健康检查和性能监控
- Sampling:提供 AI 推理能力,实现服务器和客户端的双向通信
- Roots:提供路径管理能力,控制资源访问范围和安全
通过合理组合使用这些功能,可以构建强大的 AI 应用,提高开发效率和用户体验。
DAMO开发者矩阵,由阿里巴巴达摩院和中国互联网协会联合发起,致力于探讨最前沿的技术趋势与应用成果,搭建高质量的交流与分享平台,推动技术创新与产业应用链接,围绕“人工智能与新型计算”构建开放共享的开发者生态。
更多推荐



所有评论(0)