主题
MCP(Model Context Protocol)协议学习指南
📚 适合碎片化学习,预计总学习时长:4-5小时
MCP 被誉为「AI 世界的 USB-C 接口」—— 让 AI 模型能够标准化地连接一切外部工具和数据源
目录
1. 基础理论
🎯 核心知识点
1.1 什么是 MCP?
MCP(Model Context Protocol,模型上下文协议) 是 Anthropic 于 2024 年 11 月推出的开放标准协议,用于解决 AI 大模型与外部数据源、工具之间的标准化连接问题。
┌─────────────────────────────────────────────────────────────────┐
│ MCP 核心价值 │
├─────────────────────────────────────────────────────────────────┤
│ Before MCP: │
│ ┌──────┐ 自定义API ┌──────┐ │
│ │ LLM │ ←─────────────→ │ 工具A │ │
│ └──────┘ 自定义API └──────┘ │
│ ↑ │
│ └───── 自定义API ──→ ┌──────┐ │
│ │ 工具B │ (每个工具都需要单独适配) │
│ └──────┘ │
├─────────────────────────────────────────────────────────────────┤
│ After MCP: │
│ ┌──────┐ ┌──────┐ │
│ │ LLM │ ←── MCP ────→ │ 工具A │ │
│ └──────┘ └──────┘ │
│ ↑ │
│ └───── MCP ──────→ ┌──────┐ │
│ │ 工具B │ (统一协议,即插即用) │
│ └──────┘ │
└─────────────────────────────────────────────────────────────────┘一句话理解:MCP 就像是给 AI 装上了「万能插头」,让 AI 能够用统一的方式连接各种工具和数据源。
1.2 MCP 的核心能力
| 能力 | 说明 | 示例 |
|---|---|---|
| Resources(资源) | 向模型提供上下文数据 | 文件内容、数据库记录、API 响应 |
| Tools(工具) | 让模型调用外部功能 | 执行命令、发送请求、操作数据库 |
| Prompts(提示词) | 预定义的提示词模板 | 代码审查模板、翻译模板 |
| Sampling(采样) | 让服务器请求 LLM 补全 | 复杂工作流中的多轮对话 |
1.3 MCP 架构模型
三层架构:
| 层级 | 组件 | 职责 |
|---|---|---|
| Host(宿主) | Claude Desktop、IDE 插件等 | 用户交互界面,管理 Client |
| Client(客户端) | 协议实现层 | 与 Server 建立连接,转发请求 |
| Server(服务端) | 工具/资源提供者 | 暴露具体能力(工具、资源、提示词) |
1.4 MCP 通信传输方式
MCP 支持三种传输方式:
| 传输方式 | 特点 | 适用场景 |
|---|---|---|
| stdio | 通过标准输入/输出通信 | 本地工具,命令行程序 |
| SSE(HTTP + Server-Sent Events) | 基于 HTTP 的服务器推送 | 远程服务,Web 部署(逐步废弃) |
| Streamable HTTP | 新一代 HTTP 流式传输 | 远程服务,推荐使用 |
1.5 MCP 与 Function Calling 的关系
| 对比项 | Function Calling | MCP |
|---|---|---|
| 定义者 | OpenAI 等模型厂商 | Anthropic(开放标准) |
| 范围 | 单次函数调用 | 完整的上下文协议 |
| 能力 | 仅工具调用 | 工具 + 资源 + 提示词 + 采样 |
| 标准化 | 各家实现不同 | 统一开放协议 |
| 连接管理 | 无 | 有完整的连接生命周期 |
关系图:
┌─────────────────────────────────────────┐
│ MCP 协议 │
│ ┌───────────────────────────────────┐ │
│ │ Tools(包含 Function Call) │ │
│ └───────────────────────────────────┘ │
│ ┌───────────────────────────────────┐ │
│ │ Resources │ │
│ └───────────────────────────────────┘ │
│ ┌───────────────────────────────────┐ │
│ │ Prompts │ │
│ └───────────────────────────────────┘ │
│ ┌───────────────────────────────────┐ │
│ │ Sampling │ │
│ └───────────────────────────────────┘ │
└─────────────────────────────────────────┘✅ 实操任务 1:体验 MCP
在 Claude Desktop 或支持 MCP 的 IDE 中配置一个简单的 MCP Server:
json
// mcp.json 配置文件
{
"mcpServers": {
"demo-fetch": {
"command": "npx",
"args": ["-y", "@anthropics/mcp-server-fetch"]
}
}
}观察点:
- MCP Server 启动日志
- 工具列表的获取
- 调用工具时的请求/响应格式
2. 核心语法
🎯 核心知识点
2.1 MCP 配置格式
MCP 配置采用 JSON 格式,主要结构如下:
json
{
"mcpServers": {
"server-name": {
// 配置项...
}
}
}三种传输方式的配置:
2.1.1 stdio 模式(本地命令行)
json
{
"mcpServers": {
"git-server": {
"command": "uv",
"args": [
"--directory",
"/path/to/servers/src/git",
"run",
"mcp-server-git"
],
"env": {
"GIT_AUTHOR_NAME": "Your Name"
},
"timeout": 10000,
"transportType": "stdio"
}
}
}| 字段 | 说明 | 必填 |
|---|---|---|
command | 启动命令(如 npx、uv、python) | ✅ |
args | 命令参数数组(必须拆分) | ✅ |
env | 环境变量(key-value 都必须是字符串) | ❌ |
timeout | 超时时间(ms),默认 10000,最大 300000 | ❌ |
transportType | 传输类型,可省略 | ❌ |
⚠️ 注意:args 必须拆分成数组元素,不能写成单个字符串!
json
// ❌ 错误
"args": ["--directory /path/to/server run mcp-server"]
// ✅ 正确
"args": ["--directory", "/path/to/server", "run", "mcp-server"]2.1.2 SSE 模式(HTTP 推送)
json
{
"mcpServers": {
"remote-server": {
"url": "http://127.0.0.1:8081/sse",
"headers": {
"Authorization": "Bearer xxx"
},
"timeout": 30000,
"transportType": "sse"
}
}
}| 字段 | 说明 | 必填 |
|---|---|---|
url | 服务地址(通常以 /sse 结尾) | ✅ |
headers | 自定义请求头 | ❌ |
timeout | 超时时间(ms) | ❌ |
transportType | 可省略,有 url 时默认为 sse | ❌ |
2.1.3 Streamable HTTP 模式(推荐)
json
{
"mcpServers": {
"modern-server": {
"url": "http://127.0.0.1:8080/mcp",
"headers": {
"X-Custom-Header": "value"
},
"timeout": 30000,
"transportType": "streamable-http"
}
}
}| 字段 | 说明 | 必填 |
|---|---|---|
url | 服务地址(通常以 /mcp 结尾) | ✅ |
transportType | 必须填写 streamable-http | ✅ |
headers | 自定义请求头 | ❌ |
timeout | 超时时间(ms) | ❌ |
2.2 JSON-RPC 消息格式
MCP 基于 JSON-RPC 2.0 协议通信:
请求消息
json
{
"jsonrpc": "2.0",
"id": 1,
"method": "tools/call",
"params": {
"name": "read_file",
"arguments": {
"path": "/path/to/file.txt"
}
}
}响应消息(成功)
json
{
"jsonrpc": "2.0",
"id": 1,
"result": {
"content": [
{
"type": "text",
"text": "文件内容..."
}
]
}
}响应消息(错误)
json
{
"jsonrpc": "2.0",
"id": 1,
"error": {
"code": -32600,
"message": "Invalid Request"
}
}通知消息(无需响应)
json
{
"jsonrpc": "2.0",
"method": "notifications/progress",
"params": {
"progressToken": "token-123",
"progress": 50,
"total": 100
}
}2.3 核心方法(Methods)
| 方法 | 方向 | 说明 |
|---|---|---|
initialize | Client → Server | 初始化连接,交换能力信息 |
initialized | Client → Server | 确认初始化完成(通知) |
tools/list | Client → Server | 获取工具列表 |
tools/call | Client → Server | 调用指定工具 |
resources/list | Client → Server | 获取资源列表 |
resources/read | Client → Server | 读取指定资源 |
prompts/list | Client → Server | 获取提示词模板列表 |
prompts/get | Client → Server | 获取指定提示词模板 |
ping | 双向 | 心跳检测 |
2.4 工具定义格式(Tool Schema)
json
{
"name": "write_file",
"description": "将内容写入指定文件",
"inputSchema": {
"type": "object",
"properties": {
"path": {
"type": "string",
"description": "文件路径"
},
"content": {
"type": "string",
"description": "文件内容"
}
},
"required": ["path", "content"]
}
}inputSchema 遵循 JSON Schema 规范:
| 字段 | 说明 |
|---|---|
type | 参数类型(object、string、number、boolean、array) |
properties | 对象属性定义 |
required | 必填字段数组 |
description | 字段描述(帮助 LLM 理解用途) |
2.5 资源定义格式(Resource Schema)
json
{
"uri": "file:///path/to/document.md",
"name": "项目文档",
"description": "项目的 README 文档",
"mimeType": "text/markdown"
}✅ 实操任务 2:解析 MCP 消息
手动构造以下 MCP 请求,理解消息格式:
- 初始化请求:
json
{
"jsonrpc": "2.0",
"id": 1,
"method": "initialize",
"params": {
"protocolVersion": "2024-11-05",
"clientInfo": {
"name": "my-client",
"version": "1.0.0"
},
"capabilities": {}
}
}- 调用工具请求:
json
{
"jsonrpc": "2.0",
"id": 2,
"method": "tools/call",
"params": {
"name": "read_file",
"arguments": {
"path": "/tmp/test.txt"
}
}
}任务:在终端中使用 stdio 模式测试(如果有本地 MCP Server):
bash
# 启动 MCP Server 并发送初始化消息
echo '{"jsonrpc":"2.0","id":1,"method":"initialize","params":{"protocolVersion":"2024-11-05","clientInfo":{"name":"test","version":"1.0"},"capabilities":{}}}' | npx -y @anthropics/mcp-server-fetch3. 示例代码
🎯 核心知识点
3.1 Python MCP Server(使用官方 SDK)
python
# mcp_server.py
"""
一个简单的 MCP Server 示例,提供文件操作工具
安装依赖:pip install mcp
"""
import asyncio
import json
from pathlib import Path
from mcp.server import Server
from mcp.server.stdio import stdio_server
from mcp.types import (
Tool,
TextContent,
Resource,
LATEST_PROTOCOL_VERSION
)
# 创建 MCP Server 实例
server = Server("file-tools-server")
# 定义工具:读取文件
@server.tool()
async def read_file(path: str) -> str:
"""
读取指定路径的文件内容
Args:
path: 要读取的文件路径
"""
try:
file_path = Path(path)
if not file_path.exists():
return f"错误:文件不存在 - {path}"
content = file_path.read_text(encoding='utf-8')
return content
except Exception as e:
return f"读取文件失败:{str(e)}"
# 定义工具:写入文件
@server.tool()
async def write_file(path: str, content: str) -> str:
"""
将内容写入指定路径的文件
Args:
path: 要写入的文件路径
content: 要写入的内容
"""
try:
file_path = Path(path)
file_path.parent.mkdir(parents=True, exist_ok=True)
file_path.write_text(content, encoding='utf-8')
return f"成功写入文件:{path}"
except Exception as e:
return f"写入文件失败:{str(e)}"
# 定义工具:列出目录
@server.tool()
async def list_directory(path: str) -> str:
"""
列出指定目录下的文件和子目录
Args:
path: 目录路径
"""
try:
dir_path = Path(path)
if not dir_path.exists():
return f"错误:目录不存在 - {path}"
if not dir_path.is_dir():
return f"错误:不是目录 - {path}"
items = []
for item in dir_path.iterdir():
item_type = "📁" if item.is_dir() else "📄"
items.append(f"{item_type} {item.name}")
return "\n".join(items) if items else "目录为空"
except Exception as e:
return f"列出目录失败:{str(e)}"
# 定义资源:提供项目信息
@server.resource("project://info")
async def get_project_info() -> str:
"""返回项目基本信息"""
return json.dumps({
"name": "MCP File Tools Server",
"version": "1.0.0",
"description": "提供文件操作功能的 MCP Server"
}, ensure_ascii=False, indent=2)
# 主函数:启动服务器
async def main():
async with stdio_server() as (read_stream, write_stream):
await server.run(
read_stream,
write_stream,
server.create_initialization_options()
)
if __name__ == "__main__":
asyncio.run(main())配置使用:
json
{
"mcpServers": {
"file-tools": {
"command": "python",
"args": ["mcp_server.py"],
"transportType": "stdio"
}
}
}3.2 Python MCP Server(Streamable HTTP 模式)
python
# mcp_http_server.py
"""
基于 Streamable HTTP 的 MCP Server
安装依赖:pip install mcp fastapi uvicorn
"""
import json
from fastapi import FastAPI, Request, Response
from fastapi.responses import StreamingResponse
import asyncio
app = FastAPI()
# 工具定义
TOOLS = [
{
"name": "get_weather",
"description": "获取指定城市的天气信息",
"inputSchema": {
"type": "object",
"properties": {
"city": {
"type": "string",
"description": "城市名称"
}
},
"required": ["city"]
}
},
{
"name": "calculate",
"description": "执行数学计算",
"inputSchema": {
"type": "object",
"properties": {
"expression": {
"type": "string",
"description": "数学表达式,如 2+2"
}
},
"required": ["expression"]
}
}
]
# 服务器信息
SERVER_INFO = {
"name": "demo-mcp-server",
"version": "1.0.0"
}
# 工具执行函数
async def execute_tool(name: str, arguments: dict) -> dict:
if name == "get_weather":
city = arguments.get("city", "未知")
# 模拟天气数据
return {
"content": [
{
"type": "text",
"text": f"{city}天气:晴,温度25°C,湿度60%"
}
]
}
elif name == "calculate":
expression = arguments.get("expression", "0")
try:
# 安全计算(生产环境需要更严格的校验)
result = eval(expression, {"__builtins__": {}}, {})
return {
"content": [
{
"type": "text",
"text": f"计算结果:{expression} = {result}"
}
]
}
except Exception as e:
return {
"content": [
{
"type": "text",
"text": f"计算错误:{str(e)}"
}
],
"isError": True
}
else:
return {
"content": [
{
"type": "text",
"text": f"未知工具:{name}"
}
],
"isError": True
}
# 处理 JSON-RPC 请求
async def handle_jsonrpc(request_data: dict) -> dict:
method = request_data.get("method")
params = request_data.get("params", {})
request_id = request_data.get("id")
if method == "initialize":
return {
"jsonrpc": "2.0",
"id": request_id,
"result": {
"protocolVersion": "2024-11-05",
"serverInfo": SERVER_INFO,
"capabilities": {
"tools": {}
}
}
}
elif method == "tools/list":
return {
"jsonrpc": "2.0",
"id": request_id,
"result": {
"tools": TOOLS
}
}
elif method == "tools/call":
tool_name = params.get("name")
arguments = params.get("arguments", {})
result = await execute_tool(tool_name, arguments)
return {
"jsonrpc": "2.0",
"id": request_id,
"result": result
}
elif method == "ping":
return {
"jsonrpc": "2.0",
"id": request_id,
"result": {}
}
else:
return {
"jsonrpc": "2.0",
"id": request_id,
"error": {
"code": -32601,
"message": f"Method not found: {method}"
}
}
@app.post("/mcp")
async def mcp_endpoint(request: Request):
"""MCP Streamable HTTP 端点"""
body = await request.json()
# 处理请求
response_data = await handle_jsonrpc(body)
# 返回流式响应
async def generate():
yield json.dumps(response_data).encode('utf-8')
return StreamingResponse(
generate(),
media_type="application/json",
headers={
"Cache-Control": "no-cache",
"Connection": "keep-alive"
}
)
@app.get("/health")
async def health_check():
return {"status": "ok", "server": SERVER_INFO}
# 运行:uvicorn mcp_http_server:app --host 0.0.0.0 --port 8080配置使用:
json
{
"mcpServers": {
"demo-http": {
"url": "http://127.0.0.1:8080/mcp",
"transportType": "streamable-http"
}
}
}3.3 Go MCP Server(stdio 模式)
go
// mcp_server.go
package main
import (
"bufio"
"encoding/json"
"fmt"
"os"
)
// JSON-RPC 请求结构
type JSONRPCRequest struct {
JSONRPC string `json:"jsonrpc"`
ID interface{} `json:"id"`
Method string `json:"method"`
Params map[string]interface{} `json:"params,omitempty"`
}
// JSON-RPC 响应结构
type JSONRPCResponse struct {
JSONRPC string `json:"jsonrpc"`
ID interface{} `json:"id"`
Result interface{} `json:"result,omitempty"`
Error *RPCError `json:"error,omitempty"`
}
type RPCError struct {
Code int `json:"code"`
Message string `json:"message"`
}
// 工具定义
var tools = []map[string]interface{}{
{
"name": "echo",
"description": "回显输入的消息",
"inputSchema": map[string]interface{}{
"type": "object",
"properties": map[string]interface{}{
"message": map[string]interface{}{
"type": "string",
"description": "要回显的消息",
},
},
"required": []string{"message"},
},
},
{
"name": "get_time",
"description": "获取当前时间",
"inputSchema": map[string]interface{}{
"type": "object",
"properties": map[string]interface{}{},
},
},
}
// 处理请求
func handleRequest(req JSONRPCRequest) JSONRPCResponse {
switch req.Method {
case "initialize":
return JSONRPCResponse{
JSONRPC: "2.0",
ID: req.ID,
Result: map[string]interface{}{
"protocolVersion": "2024-11-05",
"serverInfo": map[string]string{
"name": "go-mcp-server",
"version": "1.0.0",
},
"capabilities": map[string]interface{}{
"tools": map[string]interface{}{},
},
},
}
case "tools/list":
return JSONRPCResponse{
JSONRPC: "2.0",
ID: req.ID,
Result: map[string]interface{}{
"tools": tools,
},
}
case "tools/call":
return handleToolCall(req)
case "ping":
return JSONRPCResponse{
JSONRPC: "2.0",
ID: req.ID,
Result: map[string]interface{}{},
}
default:
return JSONRPCResponse{
JSONRPC: "2.0",
ID: req.ID,
Error: &RPCError{
Code: -32601,
Message: "Method not found",
},
}
}
}
// 处理工具调用
func handleToolCall(req JSONRPCRequest) JSONRPCResponse {
params := req.Params
toolName, _ := params["name"].(string)
arguments, _ := params["arguments"].(map[string]interface{})
var resultText string
switch toolName {
case "echo":
message, _ := arguments["message"].(string)
resultText = fmt.Sprintf("Echo: %s", message)
case "get_time":
resultText = fmt.Sprintf("当前时间: %s", "2024-12-01 12:00:00")
default:
return JSONRPCResponse{
JSONRPC: "2.0",
ID: req.ID,
Result: map[string]interface{}{
"content": []map[string]string{
{"type": "text", "text": "未知工具: " + toolName},
},
"isError": true,
},
}
}
return JSONRPCResponse{
JSONRPC: "2.0",
ID: req.ID,
Result: map[string]interface{}{
"content": []map[string]string{
{"type": "text", "text": resultText},
},
},
}
}
func main() {
scanner := bufio.NewScanner(os.Stdin)
for scanner.Scan() {
line := scanner.Text()
if line == "" {
continue
}
var req JSONRPCRequest
if err := json.Unmarshal([]byte(line), &req); err != nil {
continue
}
resp := handleRequest(req)
respBytes, _ := json.Marshal(resp)
fmt.Println(string(respBytes))
}
}3.4 TypeScript MCP Client
typescript
// mcp_client.ts
/**
* MCP Client 示例 - 连接 MCP Server 并调用工具
* 安装依赖:npm install @modelcontextprotocol/sdk
*/
import { Client } from "@modelcontextprotocol/sdk/client/index.js";
import { StdioClientTransport } from "@modelcontextprotocol/sdk/client/stdio.js";
async function main() {
// 创建 stdio 传输层
const transport = new StdioClientTransport({
command: "python",
args: ["mcp_server.py"],
});
// 创建 MCP Client
const client = new Client(
{
name: "example-client",
version: "1.0.0",
},
{
capabilities: {},
}
);
// 连接到服务器
await client.connect(transport);
console.log("✅ 已连接到 MCP Server");
// 获取工具列表
const tools = await client.listTools();
console.log("📦 可用工具:");
tools.tools.forEach((tool) => {
console.log(` - ${tool.name}: ${tool.description}`);
});
// 调用工具
const result = await client.callTool({
name: "read_file",
arguments: {
path: "/tmp/test.txt",
},
});
console.log("🔧 工具执行结果:", result);
// 关闭连接
await client.close();
}
main().catch(console.error);3.5 MCP Server 目录结构最佳实践
my-mcp-server/
├── src/
│ ├── __init__.py
│ ├── server.py # 主服务器入口
│ ├── tools/ # 工具定义
│ │ ├── __init__.py
│ │ ├── file_tools.py
│ │ └── api_tools.py
│ ├── resources/ # 资源定义
│ │ ├── __init__.py
│ │ └── project_resources.py
│ └── utils/ # 工具函数
│ ├── __init__.py
│ └── helpers.py
├── tests/ # 测试
│ └── test_tools.py
├── pyproject.toml # 项目配置
├── README.md
└── mcp.json # 示例配置✅ 实操任务 3:搭建 MCP Server
- 使用上面的 Python 示例代码创建
mcp_server.py - 安装依赖:
pip install mcp - 在 IDE 的 MCP 配置中添加服务
- 测试工具调用:
- 让 AI 读取某个文件
- 让 AI 创建一个新文件
- 让 AI 列出目录内容
进阶任务:为 MCP Server 添加一个新工具 search_files,实现在目录中搜索包含指定关键词的文件。
4. 常见问题
🎯 核心知识点
4.1 Server 启动失败:uvx/npx not found in $PATH
问题描述:
Error: uvx/npx not found in $PATH解决方案:
bash
# 1. 确认工具已安装
which npx # Node.js
which uvx # Python uv
# 2. 检查 PATH 环境变量
echo $PATH
# 3. 添加到 shell 配置文件
# Linux/macOS
echo 'export PATH=$PATH:/usr/local/bin' >> ~/.bashrc
source ~/.bashrc
# macOS (zsh)
echo 'export PATH=$PATH:/usr/local/bin' >> ~/.zshrc
source ~/.zshrc
# 4. 远程开发时,需要在远程机器上安装相应工具环境变量配置技巧:
json
{
"mcpServers": {
"my-server": {
"command": "/full/path/to/npx", // 使用完整路径
"args": ["-y", "my-mcp-server"],
"env": {
"PATH": "/usr/local/bin:/usr/bin:/bin" // 显式设置 PATH
}
}
}
}4.2 调用超时:context deadline exceeded
问题描述:
Error: context deadline exceeded原因:MCP 工具执行时间超过默认超时时间(10秒)
解决方案:
json
{
"mcpServers": {
"slow-server": {
"url": "http://127.0.0.1:8081/mcp",
"timeout": 60000, // 增加到 60 秒
"transportType": "streamable-http"
}
}
}超时配置说明:
- 默认超时:10,000ms(10秒)
- 最大超时:300,000ms(5分钟)
- 超过最大值会被重置为默认值
4.3 连接失败:Server 状态为红色
排查步骤:
bash
# 1. 手动测试命令是否可执行
npx -y @anthropics/mcp-server-fetch
# 2. 发送初始化消息测试
echo '{"jsonrpc":"2.0","id":1,"method":"initialize","params":{"protocolVersion":"2024-11-05","clientInfo":{"name":"test","version":"1.0"},"capabilities":{}}}' | npx -y @anthropics/mcp-server-fetch
# 3. 检查网络(对于远程服务)
curl -v http://127.0.0.1:8081/mcp
# 4. 查看详细错误日志
# 在 IDE 的输出面板中查看 MCP 相关日志4.4 工具不生效:Agent 模式未开启
问题描述:配置了 MCP Server 但工具没有被调用
原因:MCP 只在 Agent 模式下生效
解决方案:
确保在对话时开启了 Agent 模式:
┌─────────────────────────────┐
│ 💬 对话框 │
│ │
│ [普通模式] [Agent模式 ✓] │
│ │
└─────────────────────────────┘4.5 参数格式错误:args 必须拆分
错误示例:
json
// ❌ 错误:args 写成单个字符串
{
"mcpServers": {
"git": {
"command": "uv",
"args": ["--directory /path/to/server run mcp-server-git"]
}
}
}正确示例:
json
// ✅ 正确:args 拆分成数组
{
"mcpServers": {
"git": {
"command": "uv",
"args": [
"--directory",
"/path/to/server",
"run",
"mcp-server-git"
]
}
}
}4.6 SSE/HTTP 模式的鉴权
对于 SSE 和 Streamable HTTP 模式,客户端会自动传递以下请求头:
| 请求头 | 说明 |
|---|---|
OAUTH-TOKEN | OAuth 鉴权令牌 |
X-Username | 用户名 |
Server 端验证示例(Python):
python
from fastapi import FastAPI, Request, HTTPException
app = FastAPI()
@app.post("/mcp")
async def mcp_endpoint(request: Request):
# 获取鉴权信息
oauth_token = request.headers.get("OAUTH-TOKEN")
username = request.headers.get("X-Username")
if not oauth_token:
raise HTTPException(status_code=401, detail="Unauthorized")
# 验证 token(调用鉴权服务)
# ...
# 处理 MCP 请求
body = await request.json()
# ...✅ 实操任务 4:排查 MCP 问题
创建以下场景并排查:
故意制造超时:
- 在工具中添加
time.sleep(15) - 观察超时错误
- 通过调整
timeout配置解决
- 在工具中添加
测试连接状态:
- 停止 MCP Server
- 观察 IDE 中的状态变化
- 重启 Server 观察自动重连
调试请求/响应:
- 在 Server 中打印收到的 JSON-RPC 消息
- 观察
initialize、tools/list、tools/call的调用顺序
5. 选型对比
🎯 核心知识点
5.1 MCP vs Function Calling vs LangChain Tools
| 特性 | MCP | Function Calling | LangChain Tools |
|---|---|---|---|
| 标准化 | ✅ 开放协议 | ❌ 各厂商不同 | ❌ 框架特定 |
| 连接管理 | ✅ 完整生命周期 | ❌ 无 | ❌ 无 |
| 资源系统 | ✅ 支持 | ❌ 不支持 | ⚠️ 部分支持 |
| 提示词模板 | ✅ 内置 | ❌ 需自己实现 | ✅ 内置 |
| 传输方式 | stdio/SSE/HTTP | HTTP | HTTP |
| 即插即用 | ✅ 统一配置 | ❌ 需要适配 | ❌ 需要编码 |
| 生态系统 | 快速增长 | 成熟 | 成熟 |
| 学习曲线 | 中等 | 低 | 中等 |
5.2 三种传输方式对比
| 传输方式 | stdio | SSE | Streamable HTTP |
|---|---|---|---|
| 协议 | 标准 I/O | HTTP + EventSource | HTTP Stream |
| 部署 | 本地 | 本地/远程 | 本地/远程 |
| 状态 | 推荐 | 逐步废弃 | 推荐 |
| 双向通信 | ✅ | 单向为主 | ✅ |
| 防火墙 | 无影响 | 需开放端口 | 需开放端口 |
| 适用场景 | 本地工具 | Web 服务 | Web 服务 |
5.3 何时选择 MCP?
✅ 推荐使用 MCP:
构建 AI Agent 应用
- 需要连接多个外部工具
- 希望工具即插即用
- 需要标准化的协议
开发 IDE 插件/扩展
- VS Code、Cursor 等 IDE
- 代码分析、生成工具
企业级 AI 应用
- 需要连接内部系统
- 统一的鉴权和权限管理
- 多团队协作开发
开源工具生态
- 希望工具能被广泛复用
- 贡献到 MCP 生态
❌ 不推荐使用 MCP:
简单的单次 API 调用
- 直接使用 Function Calling 更简单
已有成熟的 LangChain 项目
- 迁移成本高
不需要标准化的内部工具
- 自定义协议可能更灵活
5.4 MCP 生态系统
主流 MCP Server 市场:
| 市场 | 地址 | 特点 |
|---|---|---|
| 官方仓库 | github.com/modelcontextprotocol/servers | 官方维护,稳定可靠 |
| Smithery | smithery.ai | 社区市场,种类丰富 |
| OpenTools | opentools.com/registry | 搜索方便 |
| Knot MCP 市场 | knot.woa.com/mcp/market | 腾讯内网,企业工具 |
5.5 MCP 的优势与局限
优势:
| 优势 | 说明 |
|---|---|
| 标准化 | 统一协议,避免重复适配 |
| 即插即用 | 配置即可使用,无需编码 |
| 丰富生态 | 大量现成的 Server 可用 |
| 安全性 | 本地运行,数据不外传 |
| 可扩展 | 支持自定义工具和资源 |
| 跨平台 | 支持多种编程语言和 IDE |
局限:
| 局限 | 说明 | 解决方案 |
|---|---|---|
| 学习成本 | 需要理解协议和配置 | 参考官方文档和示例 |
| 调试困难 | stdio 模式不易调试 | 使用 HTTP 模式开发 |
| 生态较新 | 部分工具不够成熟 | 等待社区完善或自行开发 |
| 性能开销 | JSON-RPC 有序列化开销 | 对于高频调用需要优化 |
✅ 实操任务 5:技术选型练习
针对以下场景,选择最合适的方案并说明理由:
| 场景 | 你的选择 | 理由 |
|---|---|---|
| IDE 中集成代码分析工具 | ||
| 单次调用外部 API 获取数据 | ||
| 构建多工具协作的 AI Agent | ||
| 已有 LangChain 项目需要新增工具 | ||
| 开发可复用的企业内部工具 |
参考答案:
- IDE 代码分析 → MCP(IDE 原生支持,标准化集成)
- 单次 API 调用 → Function Calling(简单直接,无需额外协议)
- 多工具 Agent → MCP(统一管理多个工具,即插即用)
- LangChain 新增工具 → LangChain Tools(保持一致性,减少迁移成本)
- 企业内部工具 → MCP(标准化、可复用、便于团队协作)
6. 学习资源与工具
📚 权威学习资源
| 资源 | 链接 | 说明 |
|---|---|---|
| MCP 官方文档 | modelcontextprotocol.io | 最权威的协议规范 |
| MCP GitHub | github.com/modelcontextprotocol | 官方 SDK 和 Server |
| Anthropic MCP 博客 | anthropic.com/news/model-context-protocol | 官方发布公告 |
| MCP 规范文档 | spec.modelcontextprotocol.io | 详细协议规范 |
| Knot MCP 市场 | knot.woa.com/mcp/market | 腾讯内网 MCP 市场 |
🛠 常用调试工具
1. MCP Inspector
官方提供的可视化调试工具:
bash
# 安装并启动
npx @modelcontextprotocol/inspector
# 连接到本地 Server
npx @modelcontextprotocol/inspector --server "python mcp_server.py"功能:
- 可视化查看工具、资源、提示词
- 手动调用工具并查看结果
- 查看 JSON-RPC 消息日志
2. 命令行调试
bash
# 直接与 stdio Server 交互
echo '{"jsonrpc":"2.0","id":1,"method":"initialize","params":{"protocolVersion":"2024-11-05","clientInfo":{"name":"cli","version":"1.0"},"capabilities":{}}}' | python mcp_server.py
# 测试工具列表
echo '{"jsonrpc":"2.0","id":2,"method":"tools/list","params":{}}' | python mcp_server.py
# 调用工具
echo '{"jsonrpc":"2.0","id":3,"method":"tools/call","params":{"name":"read_file","arguments":{"path":"/tmp/test.txt"}}}' | python mcp_server.py3. HTTP 调试(对于 SSE/Streamable HTTP)
bash
# 使用 curl 测试
curl -X POST http://127.0.0.1:8080/mcp \
-H "Content-Type: application/json" \
-d '{"jsonrpc":"2.0","id":1,"method":"tools/list","params":{}}'
# 使用 httpie
http POST http://127.0.0.1:8080/mcp \
jsonrpc="2.0" id:=1 method="tools/list" params:='{}'4. IDE 内置工具
VS Code / Cursor / Tico:
- 输出面板:查看 MCP 日志
- MCP 配置面板:查看 Server 状态(绿色/红色)
- Agent 模式:测试工具调用📋 MCP 开发检查清单
markdown
## Server 开发检查清单
- [ ] 实现 `initialize` 方法,返回服务器能力
- [ ] 实现 `tools/list` 方法,返回工具列表
- [ ] 实现 `tools/call` 方法,处理工具调用
- [ ] 工具定义包含完整的 inputSchema
- [ ] 错误处理和异常捕获
- [ ] 超时处理
- [ ] 日志记录
## 配置检查清单
- [ ] command/url 路径正确
- [ ] args 已正确拆分为数组
- [ ] env 变量值都是字符串类型
- [ ] timeout 设置合理
- [ ] transportType 与实际传输方式匹配
## 调试检查清单
- [ ] Server 状态显示为绿色
- [ ] 能够获取工具列表
- [ ] 能够成功调用工具
- [ ] Agent 模式已开启🎯 综合实操项目
创建一个「智能文档助手」MCP Server,要求:
功能需求
工具(Tools):
read_document:读取指定文档内容search_documents:在文档中搜索关键词summarize_document:生成文档摘要(调用 LLM)create_document:创建新文档
资源(Resources):
documents://list:列出所有可用文档documents://recent:最近访问的文档
提示词(Prompts):
review_document:文档审查模板translate_document:文档翻译模板
技术要求
- 使用 Python + MCP SDK 开发
- 支持 stdio 和 Streamable HTTP 两种模式
- 实现完整的错误处理
- 添加日志记录
目录结构
doc-assistant-mcp/
├── src/
│ ├── __init__.py
│ ├── server.py
│ ├── tools/
│ │ ├── __init__.py
│ │ └── document_tools.py
│ ├── resources/
│ │ ├── __init__.py
│ │ └── document_resources.py
│ └── prompts/
│ ├── __init__.py
│ └── document_prompts.py
├── tests/
├── pyproject.toml
├── README.md
└── mcp.json📝 文档版本:v1.0
最后更新:2026-02-09
适用人群:AI 应用开发者、中级后端开发者