MCP协议深度解析：让AI连接一切的开放标准

MCP（Model Context Protocol，模型上下文协议）是由 Anthropic 在 2024 年底开源的协议，旨在建立一个开放标准，让 AI 模型能够与各类外部工具、数据源和服务平台进行标准化的双向通信。

你可以把 MCP 理解为”AI 世界的 USB 接口”——只要实现了 MCP 协议，任何 AI 应用就能像插入 USB 设备一样，即插即用地连接各种外部资源。

一、为什么需要 MCP？

大语言模型本身是”只读的”——它能推理，但无法主动获取实时数据、执行操作、连接外部系统。在 MCP 出现之前，AI 连接外部世界的方式是”定制开发”：每个 AI 平台都要为自己的工具写专门的集成代码。

1
2
3
4
5
6
7
8
# 传统方式：每个 AI 平台都要为每个工具单独适配
Claude → 自定义代码 → GitHub
Claude → 自定义代码 → 文件系统
Claude → 自定义代码 → 数据库

Cursor → 自定义代码 → GitHub
Cursor → 自定义代码 → 文件系统
...（重复造轮子）

MCP 改变了这个局面：

1
2
3
4
5
MCP 服务器（GitHub）
    ↕ MCP 协议
Claude / Cursor / 其他 AI 客户端
    ↕ MCP 协议
MCP 服务器（文件系统）

一个 MCP 服务器实现后，所有支持 MCP 的 AI 客户端都能用。

二、MCP 架构详解

2.1 核心组件

MCP 采用客户端-服务器架构，包含四个核心概念：

  • MCP Host：AI 应用本身（如 Claude Desktop、Cursor IDE）
  • MCP Client：运行在 Host 内的客户端，与 Server 保持一对一的长期连接
  • MCP Server：连接到特定数据源或工具的服务进程
  • Transport：客户端与服务器之间的通信层（支持 stdio 和 HTTP/SSE）

2.2 资源与工具

MCP 的核心抽象是两个概念：Resources（资源）和 Tools（工具）。

Resources 是 AI 可以读取的数据源，类似”只读 API”：

1
2
3
4
5
6
// 资源定义示例
{
  "uri": "file:///workspace/README.md",
  "name": "项目说明文档",
  "mimeType": "text/markdown"
}

Tools 是 AI 可以调用的函数，带输入参数和返回值：

1
2
3
4
5
6
7
8
9
10
11
12
13
// 工具定义示例
{
  "name": "search_code",
  "description": "在代码库中搜索关键词",
  "inputSchema": {
    "type": "object",
    "properties": {
      "query": {"type": "string", "description": "搜索关键词"},
      "language": {"type": "string", "description": "编程语言"}
    },
    "required": ["query"]
  }
}

2.3 通信协议

MCP 使用 JSON-RPC 2.0 作为消息格式：

1
2
3
4
5
6
7
8
请求：
{"jsonrpc": "2.0", "id": 1, "method": "tools/call", "params": {...}}

响应：
{"jsonrpc": "2.0", "id": 1, "result": {...}}

错误：
{"jsonrpc": "2.0", "id": 1, "error": {"code": -32600, "message": "..."}}

三、开发你的第一个 MCP 服务器

下面用 Python 从零构建一个 MCP 服务器，实现三个实用功能：搜索本地文件、读取代码文件内容、执行 Shell 命令。

3.1 项目结构

1
2
3
4
5
6
7
8
mcp-weather-server/
├── main.py              # 入口文件
├── weather_server.py    # MCP 服务器实现
├── tools/
│   ├── __init__.py
│   ├── file_tools.py    # 文件相关工具
│   └── system_tools.py # 系统相关工具
└── requirements.txt

3.2 安装依赖

1
pip install mcp

3.3 核心代码实现

main.py：

1
2
3
4
5
6
7
8
from mcp.server.fastmcp import FastMCP

# 初始化 MCP 服务器
mcp = FastMCP("开发助手工具集")

# 启动服务器
if __name__ == "__main__":
    mcp.run()

tools/file_tools.py：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
import os
import subprocess
from pathlib import Path

def search_files(query: str, directory: str = ".") -> list[dict]:
    """在指定目录下搜索包含关键词的文件"""
    results = []
    query_lower = query.lower()

    for root, dirs, files in os.walk(directory):
        # 排除隐藏目录和常见无关目录
        dirs[:] = [d for d in dirs if not d.startswith('.') and d not in ['node_modules', '__pycache__', 'venv']]

        for filename in files:
            if query_lower in filename.lower():
                filepath = os.path.join(root, filename)
                try:
                    size = os.path.getsize(filepath)
                    results.append({
                        "path": filepath,
                        "name": filename,
                        "size": f"{size} bytes"
                    })
                except OSError:
                    pass

    return results[:20]  # 最多返回 20 个结果

def read_code_file(filepath: str, max_lines: int = 500) -> dict:
    """读取代码文件，带语法高亮信息"""
    path = Path(filepath).resolve()

    if not path.exists():
        return {"error": f"文件不存在: {filepath}"}

    try:
        with open(path, 'r', encoding='utf-8') as f:
            lines = f.readlines()

        content = ''.join(lines[:max_lines])
        total_lines = len(lines)

        return {
            "path": str(path),
            "total_lines": total_lines,
            "shown_lines": min(total_lines, max_lines),
            "truncated": total_lines > max_lines,
            "content": content
        }
    except Exception as e:
        return {"error": f"读取失败: {str(e)}"}

def get_file_tree(directory: str, max_depth: int = 3, current_depth: int = 0) -> list[str]:
    """生成目录树结构"""
    if current_depth >= max_depth:
        return []

    tree = []
    try:
        entries = sorted(Path(directory).iterdir(), key=lambda p: (not p.is_dir(), p.name))

        for entry in entries:
            if entry.name.startswith('.'):
                continue

            prefix = "  " * current_depth
            icon = "📁" if entry.is_dir() else "📄"
            tree.append(f"{prefix}{icon} {entry.name}")

            if entry.is_dir():
                subtree = get_file_tree(str(entry), max_depth, current_depth + 1)
                tree.extend(subtree)
    except PermissionError:
        tree.append(f"{'  ' * current_depth}⚠️ 权限不足")

    return tree

tools/system_tools.py：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
import subprocess
import shlex

def run_shell_command(command: str, timeout: int = 30) -> dict:
    """执行 Shell 命令并返回输出"""
    try:
        # 安全检查：限制可执行的命令类型
        allowed_prefixes = ['git ', 'ls ', 'cat ', 'grep ', 'find ', 'curl ', 'npm ', 'pip ', 'python ', 'node ']
        if not any(command.strip().startswith(prefix) for prefix in allowed_prefixes):
            return {"error": "此命令不在允许列表中，出于安全考虑被拒绝执行"}

        result = subprocess.run(
            command,
            shell=True,
            capture_output=True,
            text=True,
            timeout=timeout,
            cwd="/Users/xiatian"
        )

        return {
            "command": command,
            "returncode": result.returncode,
            "stdout": result.stdout[:2000],  # 限制输出长度
            "stderr": result.stderr[:500],
            "success": result.returncode == 0
        }
    except subprocess.TimeoutExpired:
        return {"error": f"命令超时（>{timeout}秒）", "command": command}
    except
    except Exception as e:
        return {"error": f"执行失败: {str(e)}", "command": command}

### 3.4 注册工具到 MCP 服务器

**weather_server.py**：
```python
from mcp.server.fastmcp import FastMCP
from tools.file_tools import search_files, read_code_file, get_file_tree
from tools.system_tools import run_shell_command

mcp = FastMCP("开发助手工具集")

# 注册文件搜索工具
@mcp.tool()
def search_files_tool(query: str, directory: str = ".") -> str:
    """在目录下搜索文件名包含关键词的文件"""
    results = search_files(query, directory)
    if not results:
        return "没有找到匹配的文件"
    return "\n".join([f"- {r['name']}: {r['path']}" for r in results])

# 注册文件读取工具
@mcp.tool()
def read_file_tool(filepath: str, max_lines: int = 500) -> str:
    """读取代码文件内容"""
    result = read_code_file(filepath, max_lines)
    if "error" in result:
        return f"错误: {result['error']}"
    suffix = "（已截断）" if result["truncated"] else ""
    return f"文件: {result['path']}\n行数: {result['total_lines']}（显示{result['shown_lines']}）{suffix}\n\n{result['content']}"

# 注册目录树工具
@mcp.tool()
def file_tree_tool(directory: str, max_depth: int = 3) -> str:
    """生成目录树结构"""
    tree = get_file_tree(directory, max_depth)
    return "\n".join(tree)

# 注册 Shell 命令工具
@mcp.tool()
def shell_tool(command: str, timeout: int = 30) -> str:
    """执行安全的白名单 Shell 命令"""
    result = run_shell_command(command, timeout)
    if "error" in result:
        return f"错误: {result['error']}"
    status = "✅ 成功" if result["success"] else f"❌ 失败（退出码: {result['returncode']}）"
    output = result["stdout"] or result["stderr"] or "(无输出)"
    return f"{status}\n\n输出:\n{output}"

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

四、在 Claude Desktop 中使用 MCP

4.1 安装 MCP Server

1
2
3
4
5
6
7
# 克隆你的 MCP 服务器
git clone https://github.com/yourname/mcp-dev-tools.git
cd mcp-dev-tools
pip install -e .

# 验证安装
mcp --help

4.2 配置 Claude Desktop

在 Claude Desktop 的 MCP 设置中添加服务器：

1
2
3
4
5
6
7
8
9
10
11
// macOS: ~/Library/Application Support/Claude/claude_desktop_config.json
// Windows: %APPDATA%\Claude\claude_desktop_config.json

{
  "mcpServers": {
    "dev-tools": {
      "command": "python",
      "args": ["/path/to/weather_server.py"]
    }
  }
}

重启 Claude Desktop 后，你就能在对话中直接调用这些工具了：

1
2
3
> 帮我看看 /workspace 项目里最近改了什么文件
> 执行 git status 看看当前 Git 状态
> 搜索项目里所有包含 "RAG" 的文件名

五、官方 MCP 服务器生态

Anthropic 官方维护了一批高质量的 MCP 服务器，可以直接集成使用：

1
2
3
4
5
6
7
8
9
10
# 文件系统服务器
npx -y @modelcontextprotocol/server-filesystem \
  /Users/xiatian/Documents

# GitHub 服务器
npx -y @modelcontextprotocol/server-github

# PostgreSQL 数据库服务器
npx -y @modelcontextprotocol/server-postgres \
  postgresql://localhost:5432/mydb

配合 Claude Desktop 配置：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
{
  "mcpServers": {
    "filesystem": {
      "command": "npx",
      "args": ["-y", "@modelcontextprotocol/server-filesystem",
               "/Users/xiatian/workspace"]
    },
    "github": {
      "command": "npx",
      "args": ["-y", "@modelcontextprotocol/server-github"],
      "env": {
        "GITHUB_PERSONAL_ACCESS_TOKEN": "ghp_xxxx"
      }
    }
  }
}

六、MCP 与其他协议对比

协议 厂商 定位 生态
MCP Anthropic AI 工具连接标准 快速扩张中
OpenAI Plugins OpenAI ChatGPT 扩展 成熟但封闭
LangChain Tools LangChain Python AI 工具 广泛但分散
Tool USE (Anthropic) Anthropic 基础函数调用 简单但强大

MCP 的优势在于它是开放协议，不绑定特定 AI 平台。随着 Google、Meta 等厂商的跟进，MCP 有望成为 AI 工具互联的事实标准。

总结

MCP 解决了 AI 工具生态碎片化的核心问题。通过标准化的协议，任何工具开发者只需要实现一次，就能被所有支持 MCP 的 AI 应用使用。

对于开发者来说，现在是好时机：可以基于 MCP 构建自己的 AI 工具生态，也可以将现有的 API 和服务包装成 MCP 服务器，供 AI 助手直接调用。