MCP 协议实战：用 Python + Vue 3 搭建 AI 工具调用面板#

2026 年，AI Agent 的爆发让一个协议悄然成为基础设施——MCP（Model Context Protocol）。Alibaba 的 page-agent（18k stars）和 ByteDance 的 deer-flow（69k stars）都在用它。但市面上几乎没有中文实战教程。这篇文章补上这个缺口。

为什么你需要了解 MCP？#

想象一下这个场景：你写了一个 Python 脚本能查数据库、一个 Node.js 脚本能调 API、一个 Bash 脚本能管理文件。现在你想让 AI 调用这些能力——以前你得为每个能力写一套 API 适配，现在只需要一个 MCP Server。

MCP 的本质是一个标准化协议，让 AI 模型（Claude、GPT、本地 LLM）能发现并调用外部工具，就像 USB-C 统一了充电口一样，MCP 统一了 AI 与外部世界的接口。

MCP 的核心架构#

1
┌─────────────┐     MCP 协议      ┌──────────────┐
2
│  MCP Host   │ ◄──────────────► │  MCP Server  │
3
│  (AI 应用)   │   JSON-RPC 2.0   │  (工具提供者) │
4
│  Claude /   │                  │  Python/Node │
5
│  GPT / 本地  │                  │  /Rust/任意   │
6
└─────────────┘                  └──────────────┘
7
                                         │
8
                                  ┌──────┴──────┐
9
                                  │  工具/资源   │
10
                                  │  数据库/API  │
11
                                  │  文件系统    │
12
                                  └─────────────┘

MCP 定义了三种核心能力：

能力	作用	类比
Tools（工具）	可执行的函数，有输入输出	API 端点
Resources（资源）	只读数据源，支持订阅更新	REST 资源
Prompts（提示）	预定义的对话模板	宏/快捷指令

关键设计哲学：MCP 是 Client-Server 模型，不是 Server-Client。AI 应用是 Client（Host），你的代码是 Server。这意味着你的工具代码不需要知道 AI 的存在——它只是暴露能力，让 AI 来发现。

从零构建 Python MCP Server#

环境准备#

1
# 创建项目
2
mkdir mcp-tool-server && cd mcp-tool-server
3
python -m venv venv
4
source venv/bin/activate
5

6
# 安装 MCP SDK
7
pip install mcp httpx aiosqlite

mcp 是官方 Python SDK，httpx 用于异步 HTTP 调用，aiosqlite 提供轻量数据库支持。

第一个 MCP Server：系统监控工具#

我们从一个实用场景开始——让 AI 能查询服务器状态。

1
import platform
2
import psutil
3
from mcp.server import Server
4
from mcp.server.stdio import stdio_server
5
from mcp.types import Tool, TextContent
6

7
# 创建 MCP Server 实例
8
server = Server("system-monitor")
9

10
@server.tool()
11
async def get_system_info() -> list[TextContent]:
12
    """获取系统基本信息：OS、CPU 核心数、内存总量"""
13
    info = {
14
        "os": f"{platform.system()} {platform.release()}",
15
        "cpu_cores": psutil.cpu_count(logical=True),
16
        "memory_total_gb": round(psutil.virtual_memory().total / (1024**3), 2),
17
        "python_version": platform.python_version(),
18
    }
19
    return [TextContent(type="text", text=str(info))]
20

21
@server.tool()
22
async def get_disk_usage(path: str = "/") -> list[TextContent]:
23
    """查询指定路径的磁盘使用情况
24

25
    Args:
26
        path: 要查询的路径，默认为根目录
27
    """
28
    usage = psutil.disk_usage(path)
29
    result = {
30
        "path": path,
31
        "total_gb": round(usage.total / (1024**3), 2),
32
        "used_gb": round(usage.used / (1024**3), 2),
33
        "free_gb": round(usage.free / (1024**3), 2),
34
        "percent": usage.percent,
35
    }
36
    return [TextContent(type="text", text=str(result))]
37

38
@server.tool()
39
async def get_process_list(limit: int = 10) -> list[TextContent]:
40
    """列出占用 CPU 最多的进程
41

42
    Args:
43
        limit: 返回的进程数量，默认 10 个
44
    """
45
    processes = []
46
    for proc in psutil.process_iter(["pid", "name", "cpu_percent", "memory_percent"]):
47
        try:
48
            processes.append(proc.info)
49
        except (psutil.NoSuchProcess, psutil.AccessDenied):
50
            continue
51

52
    # 按 CPU 使用率排序
53
    processes.sort(key=lambda p: p.get("cpu_percent", 0) or 0, reverse=True)
54
    return [TextContent(
55
        type="text",
56
        text=str(processes[:limit])
57
    )]
58

59
async def main():
60
    """启动 MCP Server，通过 stdio 通信"""
61
    async with stdio_server() as (read_stream, write_stream):
62
        await server.run(
63
            read_stream,
64
            write_stream,
65
            server.create_initialization_options(),
66
        )
67

68
if __name__ == "__main__":
69
    import asyncio
70
    asyncio.run(main())

为什么用 @server.tool() 装饰器？ 它自动完成三件事：

注册工具：把函数注册到 MCP 协议的工具列表中
提取文档：从 docstring 中提取工具描述，AI 靠这个理解工具用途
参数推断：从函数签名推断参数类型和默认值，生成 JSON Schema

这是 MCP SDK 最优雅的设计——你写 Python 函数，SDK 处理协议细节。

启动 Server#

1
# 安装 psutil
2
pip install psutil
3

4
# 通过 stdio 启动（MCP 标准通信方式）
5
python server.py

现在你的 Python 脚本已经是一个 MCP Server 了。Claude Desktop、Cursor、或者任何 MCP Client 都能发现并调用这三个工具。

用 SSE 让 Web 前端也能调用#

stdio 适合桌面应用，但我们的目标是 Web 前端。MCP 支持 SSE（Server-Sent Events）传输层，让浏览器也能通信。

1
import asyncio
2
from mcp.server import Server
3
from mcp.server.sse import SseServerTransport
4
from starlette.applications import Starlette
5
from starlette.routing import Route, Mount
6
import psutil
7

8
server = Server("system-monitor-sse")
9
sse = SseServerTransport("/messages/")
10

11
# 复用之前的工具定义
12
@server.tool()
13
async def get_system_info() -> list[TextContent]:
14
    info = {
15
        "os": f"{platform.system()} {platform.release()}",
16
        "cpu_cores": psutil.cpu_count(logical=True),
17
        "memory_total_gb": round(psutil.virtual_memory().total / (1024**3), 2),
18
    }
19
    return [TextContent(type="text", text=str(info))]
20

21
@server.tool()
22
async def get_disk_usage(path: str = "/") -> list[TextContent]:
23
    usage = psutil.disk_usage(path)
24
    result = {
25
        "path": path,
26
        "total_gb": round(usage.total / (1024**3), 2),
27
        "used_gb": round(usage.used / (1024**3), 2),
28
        "free_gb": round(usage.free / (1024**3), 2),
29
        "percent": usage.percent,
30
    }
31
    return [TextContent(type="text", text=str(result))]
32

33
async def handle_sse(request):
34
    """SSE 连接端点"""
35
    async with sse.connect_sse(
36
        request.scope, request.receive, request._send
37
    ) as streams:
38
        await server.run(
39
            streams[0], streams[1], server.create_initialization_options()
40
        )
41

42
async def handle_messages(request):
43
    """MCP 消息处理端点"""
44
    await sse.handle_post_message(
45
        request.scope, request.receive, request._send
46
    )
47

48
app = Starlette(
49
    debug=True,
50
    routes=[
51
        Route("/sse", endpoint=handle_sse),
52
        Mount("/messages/", app=sse.handle_post_message),
53
    ],
54
)
55

56
if __name__ == "__main__":
57
    import uvicorn
58
    uvicorn.run(app, host="0.0.0.0", port=8765)

SSE vs stdio 的选择逻辑：

stdio：本地 CLI 工具、桌面应用、单客户端场景
SSE：Web 应用、多客户端场景、需要跨网络通信

SSE 的优势是浏览器原生支持 EventSource API，不需要 WebSocket 的复杂握手。

构建 Vue 3 可视化面板#

有了 MCP Server，我们搭建一个 Vue 3 前端来可视化调用这些工具。

项目初始化#

1
npm create vite@latest mcp-dashboard -- --template vue-ts
2
cd mcp-dashboard
3
npm install
4
npm install mcp-client  # MCP 官方客户端库
5
npm run dev

核心组件：工具调用面板#

1
<script setup lang="ts">
2
import { ref, onMounted, computed } from 'vue'
3

4
interface Tool {
5
  name: string
6
  description: string
7
  inputSchema: Record<string, any>
8
}
9

10
interface ToolResult {
11
  tool: string
12
  args: Record<string, any>
13
  output: string
14
  timestamp: Date
15
  error?: string
16
}
17

18
const tools = ref<Tool[]>([])
19
const results = ref<ToolResult[]>([])
20
const loading = ref(false)
21
const mcpClient = ref<any>(null)
22

23
// 参数输入表单
24
const paramForm = ref<Record<string, string>>({})
25

26
// 当前选中的工具
27
const selectedTool = ref<string>('')
28

29
// 连接 MCP Server
30
async function connect() {
31
  try {
32
    // 使用 SSE 传输连接到 MCP Server
33
    const { SSEClientTransport } = await import('mcp-client')
34
    const { Client } = await import('mcp-client')
35

36
    const transport = new SSEClientTransport(
37
      new URL('http://localhost:8765/sse')
38
    )
39
    const client = new Client({
40
      name: 'mcp-dashboard',
41
      version: '1.0.0',
42
    })
43

44
    await client.connect(transport)
45
    mcpClient.value = client
46

47
    // 获取工具列表
48
    const { tools: toolList } = await client.listTools()
49
    tools.value = toolList
50
  } catch (err) {
51
    console.error('MCP 连接失败:', err)
52
  }
53
}
54

55
// 调用工具
56
async function callTool(toolName: string) {
57
  if (!mcpClient.value) return
58

59
  loading.value = true
60
  const tool = tools.value.find(t => t.name === toolName)
61

62
  try {
63
    const args = { ...paramForm.value }
64
    const result = await mcpClient.value.callTool({
65
      name: toolName,
66
      arguments: args,
67
    })
68

69
    const output = result.content
70
      .filter((c: any) => c.type === 'text')
71
      .map((c: any) => c.text)
72
      .join('\n')
73

74
    results.value.unshift({
75
      tool: toolName,
76
      args,
77
      output,
78
      timestamp: new Date(),
79
    })
80
  } catch (err: any) {
81
    results.value.unshift({
82
      tool: toolName,
83
      args: paramForm.value,
84
      output: '',
85
      timestamp: new Date(),
86
      error: err.message,
87
    })
88
  } finally {
89
    loading.value = false
90
  }
91
}
92

93
// 动态生成参数表单
94
const currentParams = computed(() => {
95
  const tool = tools.value.find(t => t.name === selectedTool.value)
96
  if (!tool) return []
97

98
  const props = tool.inputSchema.properties || {}
99
  return Object.entries(props).map(([key, schema]: [string, any]) => ({
100
    name: key,
101
    type: schema.type || 'string',
102
    description: schema.description || '',
103
    required: tool.inputSchema.required?.includes(key) || false,
104
  }))
105
})
106

107
onMounted(() => {
108
  connect()
109
})
110
</script>
111

112
<template>
113
  <div class="tool-panel">
114
    <header class="panel-header">
115
      <h1>🔧 MCP 工具面板</h1>
116
      <span class="connection-status" :class="{ connected: mcpClient }">
117
        {{ mcpClient ? '● 已连接' : '○ 未连接' }}
118
      </span>
119
    </header>
120

121
    <!-- 工具选择 -->
122
    <div class="tool-selector">
123
      <select v-model="selectedTool" :disabled="!mcpClient">
124
        <option value="" disabled>选择工具...</option>
125
        <option v-for="tool in tools" :key="tool.name" :value="tool.name">
126
          {{ tool.name }}
127
        </option>
128
      </select>
129

130
      <p v-if="selectedTool" class="tool-desc">
131
        {{ tools.find(t => t.name === selectedTool)?.description }}
132
      </p>
133
    </div>
134

135
    <!-- 参数表单 -->
136
    <div v-if="currentParams.length" class="param-form">
137
      <div v-for="param in currentParams" :key="param.name" class="param-field">
138
        <label>
139
          {{ param.name }}
140
          <span v-if="param.required" class="required">*</span>
141
          <span class="param-type">({{ param.type }})</span>
142
        </label>
143
        <input
144
          v-model="paramForm[param.name]"
145
          :placeholder="param.description"
146
        />
147
      </div>
148
    </div>
149

150
    <!-- 执行按钮 -->
151
    <button
152
      class="execute-btn"
153
      :disabled="!selectedTool || loading"
154
      @click="callTool(selectedTool)"
155
    >
156
      {{ loading ? '执行中...' : '执行工具' }}
157
    </button>
158

159
    <!-- 结果展示 -->
160
    <div class="results">
161
      <div
162
        v-for="(result, index) in results"
163
        :key="index"
164
        class="result-card"
165
        :class="{ error: result.error }"
166
      >
167
        <div class="result-header">
168
          <span class="tool-name">{{ result.tool }}</span>
169
          <span class="timestamp">
170
            {{ result.timestamp.toLocaleTimeString() }}
171
          </span>
172
        </div>
173
        <pre v-if="result.output" class="result-output">{{ result.output }}</pre>
174
        <div v-if="result.error" class="error-msg">
175
          ❌ {{ result.error }}
176
        </div>
177
      </div>
178
    </div>
179
  </div>
180
</template>
181

182
<style scoped>
183
.tool-panel {
184
  max-width: 800px;
185
  margin: 0 auto;
186
  padding: 24px;
187
  font-family: 'Inter', system-ui, sans-serif;
188
}
189

190
.panel-header {
191
  display: flex;
192
  justify-content: space-between;
193
  align-items: center;
194
  margin-bottom: 24px;
195
}
196

197
.panel-header h1 {
198
  font-size: 1.5rem;
199
  margin: 0;
200
}
201

202
.connection-status {
203
  font-size: 0.875rem;
204
  color: #999;
205
}
206

207
.connection-status.connected {
208
  color: #22c55e;
209
}
210

211
.tool-selector {
212
  margin-bottom: 16px;
213
}
214

215
.tool-selector select {
216
  width: 100%;
217
  padding: 10px 14px;
218
  border: 1px solid #e2e8f0;
219
  border-radius: 8px;
220
  font-size: 1rem;
221
  background: white;
222
}
223

224
.tool-desc {
225
  margin-top: 8px;
226
  color: #64748b;
227
  font-size: 0.875rem;
228
}
229

230
.param-form {
231
  display: flex;
232
  flex-direction: column;
233
  gap: 12px;
234
  margin-bottom: 16px;
235
}
236

237
.param-field label {
238
  display: block;
239
  font-size: 0.875rem;
240
  font-weight: 500;
241
  margin-bottom: 4px;
242
}
243

244
.required {
245
  color: #ef4444;
246
}
247

248
.param-type {
249
  color: #94a3b8;
250
  font-weight: 400;
251
}
252

253
.param-field input {
254
  width: 100%;
255
  padding: 8px 12px;
256
  border: 1px solid #e2e8f0;
257
  border-radius: 6px;
258
  font-size: 0.875rem;
259
}
260

261
.execute-btn {
262
  width: 100%;
263
  padding: 12px;
264
  background: #3b82f6;
265
  color: white;
266
  border: none;
267
  border-radius: 8px;
268
  font-size: 1rem;
269
  font-weight: 500;
270
  cursor: pointer;
271
  transition: background 0.2s;
272
}
273

274
.execute-btn:disabled {
275
  background: #94a3b8;
276
  cursor: not-allowed;
277
}
278

279
.execute-btn:not(:disabled):hover {
280
  background: #2563eb;
281
}
282

283
.results {
284
  margin-top: 24px;
285
  display: flex;
286
  flex-direction: column;
287
  gap: 12px;
288
}
289

290
.result-card {
291
  background: #f8fafc;
292
  border: 1px solid #e2e8f0;
293
  border-radius: 8px;
294
  padding: 16px;
295
}
296

297
.result-card.error {
298
  border-color: #fecaca;
299
  background: #fef2f2;
300
}
301

302
.result-header {
303
  display: flex;
304
  justify-content: space-between;
305
  margin-bottom: 8px;
306
  font-size: 0.875rem;
307
}
308

309
.tool-name {
310
  font-weight: 600;
311
  color: #1e40af;
312
}
313

314
.timestamp {
315
  color: #94a3b8;
316
}
317

318
.result-output {
319
  background: #1e293b;
320
  color: #e2e8f0;
321
  padding: 12px;
322
  border-radius: 6px;
323
  font-size: 0.8125rem;
324
  overflow-x: auto;
325
  white-space: pre-wrap;
326
  word-break: break-word;
327
}
328

329
.error-msg {
330
  color: #dc2626;
331
  font-size: 0.875rem;
332
}
333
</style>

这个面板的核心价值：它不需要知道任何具体工具的实现细节。新增一个 MCP 工具，面板自动发现、自动渲染参数表单、自动执行。这就是 MCP 协议的力量——工具发现和调用是标准化的。

进阶：构建数据库查询工具#

系统监控只是热身，让我们构建一个更实用的场景——让 AI 查询 SQLite 数据库。

1
import aiosqlite
2
from mcp.server import Server
3
from mcp.types import Tool, TextContent
4

5
server = Server("database-query")
6

7
DB_PATH = "app.db"
8

9
@server.tool()
10
async def list_tables() -> list[TextContent]:
11
    """列出数据库中所有用户表"""
12
    async with aiosqlite.connect(DB_PATH) as db:
13
        cursor = await db.execute(
14
            "SELECT name FROM sqlite_master WHERE type='table' AND name NOT LIKE 'sqlite_%'"
15
        )
16
        tables = [row[0] for row in await cursor.fetchall()]
17
    return [TextContent(type="text", text=f"表列表: {', '.join(tables)}")]
18

19
@server.tool()
20
async def describe_table(table_name: str) -> list[TextContent]:
21
    """查看表结构
22

23
    Args:
24
        table_name: 表名
25
    """
26
    async with aiosqlite.connect(DB_PATH) as db:
27
        cursor = await db.execute(f"PRAGMA table_info({table_name})")
28
        columns = await cursor.fetchall()
29

30
        schema = []
31
        for col in columns:
32
            schema.append({
33
                "cid": col[0],
34
                "name": col[1],
35
                "type": col[2],
36
                "not_null": bool(col[3]),
37
                "default": col[4],
38
            })
39

40
    return [TextContent(
41
        type="text",
42
        text=f"表 {table_name} 的结构:\n" + "\n".join(
43
            f"  {c['name']} ({c['type']})" +
44
            (" NOT NULL" if c["not_null"] else "") +
45
            (f" DEFAULT {c['default']}" if c["default"] else "")
46
            for c in schema
47
        )
48
    )]
49

50
@server.tool()
51
async def query_table(
52
    table_name: str,
53
    limit: int = 20,
54
    where: str = "",
55
    order_by: str = "",
56
) -> list[TextContent]:
57
    """查询表数据（只读）
58

59
    Args:
60
        table_name: 表名
61
        limit: 返回行数，默认 20
62
        where: WHERE 条件（不含 WHERE 关键字）
63
        order_by: ORDER BY 子句（不含 ORDER BY 关键字）
64
    """
65
    # ⚠️ 安全：只允许 SELECT，拒绝危险操作
66
    # 实际生产环境应该用参数化查询或白名单机制
67

68
    sql = f"SELECT * FROM {table_name}"
69
    params = []
70

71
    if where:
72
        sql += f" WHERE {where}"
73
    if order_by:
74
        sql += f" ORDER BY {order_by}"
75
    sql += f" LIMIT {limit}"
76

77
    async with aiosqlite.connect(DB_PATH) as db:
78
        cursor = await db.execute(sql)
79
        columns = [desc[0] for desc in cursor.description]
80
        rows = await cursor.fetchall()
81

82
    # 格式化为表格
83
    lines = [" | ".join(columns)]
84
    lines.append("-" * len(lines[0]))
85
    for row in rows:
86
        lines.append(" | ".join(str(v) for v in row))
87

88
    return [TextContent(type="text", text="\n".join(lines))]
89

90
# 初始化数据库（示例数据）
91
async def init_db():
92
    async with aiosqlite.connect(DB_PATH) as db:
93
        await db.execute("""
94
            CREATE TABLE IF NOT EXISTS users (
95
                id INTEGER PRIMARY KEY,
96
                name TEXT NOT NULL,
97
                email TEXT,
98
                role TEXT DEFAULT 'user',
99
                created_at TEXT DEFAULT CURRENT_TIMESTAMP
100
            )
101
        """)
102
        await db.execute("""
103
            INSERT OR IGNORE INTO users (id, name, email, role)
104
            VALUES (1, '张三', 'zhangsan@example.com', 'admin'),
105
                   (2, '李四', 'lisi@example.com', 'user'),
106
                   (3, '王五', 'wangwu@example.com', 'user')
107
        """)
108
        await db.commit()

安全设计要点：

只读查询：这个工具只暴露 SELECT，不暴露 INSERT/UPDATE/DELETE
LIMIT 限制：防止全表扫描导致性能问题
生产环境建议：用参数化查询代替字符串拼接，或者实现 SQL 白名单机制

Docker 部署：让一切可复现#

开发环境搞定了，接下来用 Docker 把整个系统打包。

MCP Server 的 Dockerfile#

1
FROM python:3.12-slim
2

3
WORKDIR /app
4

5
# 安装依赖
6
COPY requirements.txt .
7
RUN pip install --no-cache-dir -r requirements.txt
8

9
# 复制代码
10
COPY server.py sse_server.py db_tool.py ./
11
COPY app.db ./
12

13
# 暴露 SSE 端口
14
EXPOSE 8765
15

16
# 启动
17
CMD ["python", "sse_server.py"]

Vue 前端的 Dockerfile#

1
FROM node:20-alpine AS builder
2

3
WORKDIR /app
4
COPY package*.json ./
5
RUN npm ci
6
COPY . .
7
RUN npm run build
8

9
# 生产镜像用 Nginx 托管
10
FROM nginx:alpine
11
COPY --from=builder /app/dist /usr/share/nginx/html
12
COPY nginx.conf /etc/nginx/conf.d/default.conf
13
EXPOSE 80

docker-compose 编排#

1
version: '3.8'
2

3
services:
4
  mcp-server:
5
    build:
6
      context: ./mcp-server
7
      dockerfile: Dockerfile.mcp-server
8
    ports:
9
      - "8765:8765"
10
    volumes:
11
      - db-data:/app/data
12
    environment:
13
      - DB_PATH=/app/data/app.db
14
    restart: unless-stopped
15

16
  frontend:
17
    build:
18
      context: ./mcp-dashboard
19
      dockerfile: Dockerfile.frontend
20
    ports:
21
      - "3000:80"
22
    depends_on:
23
      - mcp-server
24
    environment:
25
      - VITE_MCP_URL=http://mcp-server:8765
26
    restart: unless-stopped
27

28
volumes:
29
  db-data:

多阶段构建的价值：前端镜像最终只有 Nginx + 静态文件，不包含 Node.js、npm、源码，镜像体积从几百 MB 降到 20MB 左右。

Nginx 反向代理配置#

1
server {
2
    listen 80;
3
    server_name _;
4

5
    # 前端静态文件
6
    location / {
7
        root /usr/share/nginx/html;
8
        try_files $uri $uri/ /index.html;
9
    }
10

11
    # 反向代理 MCP Server 的 SSE 端点
12
    location /mcp/ {
13
        proxy_pass http://mcp-server:8765/;
14
        proxy_http_version 1.1;
15
        proxy_set_header Upgrade $http_upgrade;
16
        proxy_set_header Connection "upgrade";
17
        proxy_set_header Host $host;
18
        # SSE 需要禁用缓冲
19
        proxy_buffering off;
20
        proxy_cache off;
21
    }
22
}

为什么需要反向代理？ 浏览器前端和 MCP Server 在不同端口（甚至不同域名），直接跨域调用会被浏览器拦截。Nginx 把 /mcp/ 路径代理到 MCP Server，前端只需要同源调用即可。

实战经验：MCP 开发的常见坑#

坑 1：工具描述决定 AI 调用准确率#

AI 模型靠工具的 description 字段决定何时调用哪个工具。描述模糊会导致 AI 选错工具。

1
# ❌ 不好的描述
2
@server.tool()
3
async def get_info():
4
    """获取信息"""  # AI 不知道什么时候该调用这个
5

6
# ✅ 好的描述
7
@server.tool()
8
async def get_system_info():
9
    """获取服务器系统信息：操作系统、CPU 核心数、内存总量。
10
    当用户询问服务器状态、系统配置、硬件信息时调用。"""

经验法则：描述中应该包含”当用户 XXX 时调用”的触发条件。

坑 2：SSE 连接的超时处理#

SSE 连接长时间空闲可能被代理服务器断开。需要实现心跳机制。

1
// 前端 SSE 重连逻辑
2
const eventSource = new EventSource('/mcp/sse')
3

4
eventSource.onmessage = (event) => {
5
  // 处理 MCP 消息
6
  handleMessage(JSON.parse(event.data))
7
}
8

9
eventSource.onerror = () => {
10
  // SSE 断开，延迟重连
11
  setTimeout(() => {
12
    eventSource.close()
13
    reconnect()
14
  }, 3000)
15
}

坑 3：参数类型不匹配#

MCP 协议用 JSON Schema 定义参数类型。Python 的 int 对应 JSON 的 integer，但前端传过来的是字符串。

1
# SDK 会自动做类型转换，但你要确保：
2
# 1. 函数签名有类型注解
3
# 2. 有默认值（让参数可选）
4
@server.tool()
5
async def query_table(
6
    table_name: str,      # 必填
7
    limit: int = 20,      # 可选，有默认值
8
    where: str = "",      # 可选，空字符串默认值
9
) -> list[TextContent]:

坑 4：错误处理要返回结构化信息#

不要直接抛异常，MCP 协议有标准的错误格式。

1
@server.tool()
2
async def query_table(table_name: str, limit: int = 20) -> list[TextContent]:
3
    try:
4
        # 查询逻辑
5
        ...
6
    except aiosqlite.Error as e:
7
        return [TextContent(
8
            type="text",
9
            text=f"数据库错误: {str(e)}",
10
            # 可以标记为错误状态
11
        )]

MCP vs 传统 API：什么时候用哪个？#

场景	推荐方案	原因
给 AI 模型提供工具调用	MCP	工具发现、参数 Schema 自动生成
给其他服务提供数据接口	REST/GraphQL	人类开发者更熟悉
内部微服务通信	gRPC	性能更好，类型安全
浏览器端实时推送	SSE/WebSocket	MCP 的 SSE 传输层也是基于这个
CLI 工具	MCP stdio	零配置，管道通信

核心判断标准：如果消费者是 AI 模型，用 MCP。如果消费者是人类开发者或其他服务，用传统 API。

总结#

MCP 不是另一个 API 框架，它是 AI 时代的”设备驱动层”。就像 USB 让鼠标、键盘、摄像头都能插到电脑上一样，MCP 让任何工具都能被 AI 调用。

这篇文章覆盖了：

✅ MCP 协议的核心概念和架构
✅ Python MCP Server 的完整实现（系统监控 + 数据库查询）
✅ SSE 传输层让 Web 前端能调用
✅ Vue 3 可视化面板的完整代码
✅ Docker 多容器编排部署
✅ 4 个实战踩坑经验

下一步：把你手头的 Python 脚本包装成 MCP Server，你的 AI 助手就能直接调用它们了。不需要改脚本逻辑，只需要加一个装饰器。

如果你对 MCP 协议有更多疑问，或者想讨论 AI 工具链的最佳实践，欢迎在评论区交流。

音乐

音乐