---
title: "MCP协议从零部署 — 让AI Agent调用任意外部工具的完整指南"
date: 2026-05-26
categories: [AI基础设施, Agent开发]
tags: [MCP, Model Context Protocol, AI Agent, 工具调用, 协议部署, Python]
---
> 本文首发于 Agent评测站，深度拆解AI Agent技术栈的每一个螺丝钉。
一、一个真实的崩溃现场：API 满天飞，Agent 变废物
三个月前，我们团队接了一个需求：让一个基于 Claude 的 AI Agent 能够查询公司内部的 PostgreSQL 数据库、调用 Jira API 创建工单、再从飞书拉取审批状态。
听起来很常规对吧？结果代码写成了这样：
# 每个工具一套调用方式，维护成本爆炸
def query_db(sql):
conn = psycopg2.connect(...)
return conn.execute(sql)
def create_jira_ticket(title, desc):
headers = {"Authorization": "Bearer JIRA_TOKEN"}
requests.post("https://company.atlassian.net/rest/api/2/issue", json={...})
def get_feishu_approval(approval_id):
headers = {"Authorization": "Bearer FEISHU_TOKEN"}
requests.get(f"https://open.feishu.cn/open-apis/approval/v4/instances/{approval_id}")
三个工具，三种认证方式，三种错误处理逻辑。Agent 的 system prompt 里塞了 200 行工具描述，prompt 长度爆炸，token 浪费严重。
最崩溃的是：加一个新工具，改 prompt、改代码、改测试，一个人一天就没了。
后来我看到了 Anthropic 在 2024 年底开源的 MCP（Model Context Protocol）。一句话说清楚它的价值：让 AI Agent 调用外部工具的方式，统一得像调用本地函数一样。
本文从零开始，记录我部署 MCP 的全过程——包括踩过的坑、崩溃的调试、以及最终的解决方案。
---
二、MCP 协议到底是什么？（30 秒极简版）
MCP 全称 Model Context Protocol，是 Anthropic 提出的开放协议，定义了 AI Agent（MCP Client）和外部工具（MCP Server）之间的通信标准。
┌─────────────┐ JSON-RPC ┌──────────────┐
│ AI Agent │ ◄──────────────► │ MCP Server │
│ (MCP Client)│ stdin/stdout │ (Tools) │
└─────────────┘ └──────────────┘
核心概念就三个：
| 概念 | 说明 | 类比 |
|------|------|------|
| Resources | 数据源（文件、数据库、API） | 类似 REST 的 GET |
| Tools | 可执行操作（函数、命令） | 类似 REST 的 POST/PUT |
| Prompts | 预定义的提示模板 | 类似路由模板 |
MCP Server 通过 stdio（本地）或 SSE（远程）传输 JSON-RPC 消息。Agent 通过 tools/list 发现可用工具，通过 tools/call 调用工具。就这么简单。
---
三、环境准备（踩坑从这里开始）
3.1 我的环境清单
OS: Ubuntu 22.04 / macOS 14.4 (都试过)
Python: 3.10+ (推荐 3.11)
Node.js: 18+ (某些 Server 需要)
Claude Desktop: 最新版
uv: pip 的替代品，速度快 10 倍
3.2 第一坑：Python 版本不够
$ python3 --version
Python 3.8.10
MCP SDK 要求 Python >= 3.10。3.8 直接报 SyntaxError，因为 SDK 里大量用了 match 语法（Python 3.10+）。
解决：
# ubuntu
sudo add-apt-repository ppa:deadsnakes/ppa
sudo apt install python3.11 python3.11-venv python3.11-dev
# mac
brew install python@3.11
3.3 安装 uv（推荐但非必须）
Anthropic 官方文档推荐用 uv 替代 pip，实际体验下来确实快很多：
curl -LsSf https://astral.sh/uv/install.sh | sh
source ~/.bashrc
uv --version
终端输出示例：
$ uv --version
uv 0.4.30
> 如果你不想装 uv，全程用 pip 也行。不过后面 Claude Desktop 配置里，某些场景下 uv 是默认启动器，建议装上。
---
四、安装 MCP SDK（两种方式）
方式一：Python SDK（推荐，本文用这个）
# 创建项目目录
mkdir ~/mcp-tutorial && cd ~/mcp-tutorial
# 虚拟环境（必须）
uv venv .venv
source .venv/bin/activate
# 安装 SDK
uv pip install mcp
# 验证安装
python -c "import mcp; print(mcp.__version__)"
终端输出：
$ uv pip install mcp
Resolved 12 packages in 254ms
Installed 12 packages in 1.02s
+ mcp==1.2.0
+ httpx==0.27.0
+ pydantic==2.9.0
+ ...
$ python -c "import mcp; print(mcp.__version__)"
1.2.0
方式二：Node.js SDK
npm install @modelcontextprotocol/sdk
两种 SDK 协议完全兼容，可以混用——比如 Node.js 写的 MCP Server 可以被 Python 的 Agent 调用，反之亦然。
---
五、创建第一个 MCP Server（完整可跑代码）
从最简单的开始：一个能获取当前时间和执行 shell 命令的 MCP Server。
5.1 项目结构
~/mcp-tutorial/
├── .venv/
├── pyproject.toml
└── server/
├── __init__.py
└── basic_server.py
5.2 实现代码
# server/basic_server.py
"""
MCP Server 最小实现 —— 提供时间和 shell 执行工具
"""
from mcp.server import Server, NotificationOptions
from mcp.server.models import InitializationOptions
import mcp.server.stdio
import mcp.types as types
import subprocess
from datetime import datetime
from typing import Any
# 1. 创建 Server 实例
server = Server("basic-tool-server")
# 2. 声明工具列表（Agent 通过 tools/list 发现）
@server.list_tools()
async def handle_list_tools() -> list[types.Tool]:
return [
types.Tool(
name="get_time",
description="获取当前系统时间，支持时区参数",
inputSchema={
"type": "object",
"properties": {
"timezone": {
"type": "string",
"description": "时区，如 Asia/Shanghai, UTC, America/New_York",
"default": "Asia/Shanghai",
}
},
},
),
types.Tool(
name="run_shell",
description="执行 shell 命令（只读操作，安全受限）",
inputSchema={
"type": "object",
"properties": {
"command": {
"type": "string",
"description": "要执行的 shell 命令",
}
},
"required": ["command"],
},
),
]
# 3. 实现工具调用逻辑（Agent 通过 tools/call 触发）
@server.call_tool()
async def handle_call_tool(
name: str, arguments: dict[str, Any] | None
) -> list[types.TextContent]:
if name == "get_time":
import pytz
tz_name = (arguments or {}).get("timezone", "Asia/Shanghai")
try:
tz = pytz.timezone(tz_name)
now = datetime.now(tz)
return [types.TextContent(type="text", text=now.isoformat())]
except Exception as e:
return [types.TextContent(type="text", text=f"时区错误: {e}")]
elif name == "run_shell":
cmd = (arguments or {}).get("command", "")
if not cmd:
return [types.TextContent(type="text", text="命令不能为空")]
# 安全限制：只允许只读命令
dangerous = ["rm", "dd", "mkfs", ">", "|", "sudo"]
for d in dangerous:
if d in cmd:
return [
types.TextContent(
type="text", text=f"禁止使用危险命令/符号: {d}"
)
]
try:
result = subprocess.run(
cmd,
shell=True,
capture_output=True,
text=True,
timeout=30,
)
output = result.stdout or result.stderr
return [
types.TextContent(type="text", text=output[:2000])
] # 限制输出长度
except subprocess.TimeoutExpired:
return [types.TextContent(type="text", text="命令执行超时（30秒）")]
except Exception as e:
return [types.TextContent(type="text", text=f"执行失败: {e}")]
else:
raise ValueError(f"未知工具: {name}")
# 4. 主入口 —— 通过 stdio 与客户端通信
async def main():
async with mcp.server.stdio.stdio_server() as (read_stream, write_stream):
await server.run(
read_stream,
write_stream,
InitializationOptions(
server_name="basic-tool-server",
server_version="0.1.0",
capabilities=server.get_capabilities(
notification_options=NotificationOptions(),
experimental_capabilities={},
),
),
)
if __name__ == "__main__":
import asyncio
asyncio.run(main())
5.3 依赖文件
# pyproject.toml
[project]
name = "mcp-tutorial"
version = "0.1.0"
requires-python = ">=3.10"
dependencies = [
"mcp>=1.0.0",
"pytz>=2024.1",
]
安装：
source .venv/bin/activate
uv pip install -e .
5.4 启动测试（终端模拟）
MCP Server 本身不是独立运行的 HTTP 服务，它通过 stdin/stdout 与客户端通信。但我们可以用 mcp-cli 来测试：
# 方式一：直接用 Python 启动，然后手动交互
python server/basic_server.py
# 方式二：用 mcp 命令行工具测试（如果安装了 mcp CLI）
mcp run server/basic_server.py
终端输出示例（直接启动后无输出表示成功，等待客户端连接）：
$ python server/basic_server.py
# 无输出 —— 等待 JSON-RPC 消息从 stdin 传入
要真实测试，可以用 echo 模拟 JSON-RPC 请求：
echo '{"jsonrpc":"2.0","method":"tools/list","id":1}' | python server/basic_server.py
终端输出：
$ echo '{"jsonrpc":"2.0","method":"tools/list","id":1}' | python server/basic_server.py
{"jsonrpc":"2.0","id":1,"result":{"tools":[{"name":"get_time","description":"获取当前系统时间，支持时区参数","inputSchema":{"type":"object","properties":{"timezone":{"type":"string","description":"时区，如 Asia/Shanghai, UTC, America/New_York","default":"Asia/Shanghai"}}}},{"name":"run_shell","description":"执行 shell 命令（只读操作，安全受限）","inputSchema":{"type":"object","properties":{"command":{"type":"string","description":"要执行的 shell 命令"}},"required":["command"]}}]}}
能看到返回了两个工具的定义，协议通了。
---
六、连接 AI Agent（Claude Desktop 实战）
6.1 第二坑：Claude Desktop 配置格式
这是翻车最多的地方。Claude Desktop 的 MCP 配置是 JSON 格式，但位置因系统而异：
| 系统 | 配置文件路径 |
|------|-------------|
| macOS | ~/Library/Application Support/Claude/claude_desktop_config.json |
| Windows | %APPDATA%\Claude\claude_desktop_config.json |
| Linux | ~/.config/Claude/claude_desktop_config.json |
第一次配置时我在 Linux 上找了半小时才找到正确路径：
# 创建配置目录和文件
mkdir -p ~/.config/Claude
touch ~/.config/Claude/claude_desktop_config.json
6.2 写入配置
{
"mcpServers": {
"basic-tools": {
"command": "uv",
"args": [
"--directory",
"/home/user/mcp-tutorial",
"run",
"python",
"server/basic_server.py"
],
"env": {
"PYTHONPATH": "/home/user/mcp-tutorial"
}
}
}
}
> 关键踩坑：command 字段用 uv 而不是 python。因为 Claude Desktop 启动子进程时依赖 uv 来管理虚拟环境。如果直接写 python，它会用系统全局 Python，找不到你项目装的依赖。
终端输出示例（第一次启动 Claude Desktop）：
[Claude Desktop 日志]
2026-05-26 10:15:23 INFO: Starting MCP server: basic-tools
2026-05-26 10:15:23 INFO: Command: uv --directory /home/user/mcp-tutorial run python server/basic_server.py
2026-05-26 10:15:23 INFO: Server started successfully
如果看到 Server started，恭喜！如果报错看 6.4 节。
6.3 验证连接
在 Claude Desktop 的对话框中问：
现在几点了？
Claude 会调用 get_time 工具，返回如下（Claude 内部日志）：
[2026-05-26 10:20:15] Calling tool: get_time
Arguments: {"timezone": "Asia/Shanghai"}
Result: 2026-05-26T10:20:15+08:00
再试执行命令：
帮我看看当前目录下有什么文件
Claude 调用 run_shell("ls -la")，返回结果。
6.4 第三坑：配置了但 Claude 说找不到工具
现象：Claude Desktop 可以正常对话，但不调用任何 MCP 工具，仿佛没配置一样。
排查步骤：
# 1. 查看 Claude 日志（macOS/Linux）
tail -f ~/.local/share/Claude/logs/*.log
# 2. 手动测试 Server 是否正常
uv --directory /home/user/mcp-tutorial run python server/basic_server.py &
echo '{"jsonrpc":"2.0","method":"tools/list","id":1}' | nc localhost 8080
# 3. 常见原因：配置文件格式错误
python -m json.tool ~/.config/Claude/claude_desktop_config.json
我遇到的实际问题：配置文件里 args 是个字符串而不是数组：
// ❌ 错误写法
"args": "--directory /home/user/mcp-tutorial run python server/basic_server.py"
// ✅ 正确写法
"args": ["--directory", "/home/user/mcp-tutorial", "run", "python", "server/basic_server.py"]
Claude Desktop 对配置的 JSON schema 校验很严格，数组套字符串直接静默失败。
---
七、进阶：构建一个真正有用的 MCP Server（SQL + API）
基础的时间工具没啥用。来看一个生产级别的例子：一个能查数据库、能调 API 的 MCP Server。
7.1 功能清单
query_database：安全执行 SQL SELECT 查询call_http_api：调用外部 REST APIread_file_content：读取本地文件内容search_web：调用搜索 API 获取网页内容
7.2 完整实现
# server/production_server.py
"""
生产级 MCP Server —— 数据库 + HTTP + 文件 + 搜索
"""
from mcp.server import Server, NotificationOptions
from mcp.server.models import InitializationOptions
import mcp.server.stdio
import mcp.types as types
import json
import httpx
import sqlite3
from pathlib import Path
from typing import Any
server = Server("production-server")
# ─── 数据库连接池（示例用 SQLite） ─────────────────────
DB_PATH = Path.home() / "mcp-tutorial" / "data" / "app.db"
def get_db():
"""获取数据库连接"""
DB_PATH.parent.mkdir(parents=True, exist_ok=True)
conn = sqlite3.connect(str(DB_PATH))
conn.row_factory = sqlite3.Row
return conn
# ─── 工具定义 ─────────────────────────────────────
@server.list_tools()
async def handle_list_tools() -> list[types.Tool]:
return [
types.Tool(
name="query_database",
description="对本地 SQLite 数据库执行只读查询（仅 SELECT）",
inputSchema={
"type": "object",
"properties": {
"sql": {
"type": "string",
"description": "SQL 查询语句（仅允许 SELECT）",
}
},
"required": ["sql"],
},
),
types.Tool(
name="call_http_api",
description="调用外部 HTTP API（GET/POST）",
inputSchema={
"type": "object",
"properties": {
"method": {
"type": "string",
"enum": ["GET", "POST"],
"default": "GET",
},
"url": {"type": "string", "description": "完整 URL 地址"},
"headers": {
"type": "object",
"description": "请求头（可选）",
},
"body": {
"type": "object",
"description": "POST 请求体（可选）",
},
},
"required": ["url"],
},
),
types.Tool(
name="read_file_content",
description="读取指定文件的内容（限文本文件，大小 < 100KB）",
inputSchema={
"type": "object",
"properties": {
"path": {
"type": "string",
"description": "文件绝对路径",
}
},
"required": ["path"],
},
),
types.Tool(
name="search_web",
description="搜索互联网，返回前 N 条结果",
inputSchema={
"type": "object",
"properties": {
"query": {
"type": "string",
"description": "搜索关键词",
},
"count": {
"type": "integer",
"description": "返回结果数量（1-10）",
"default": 5,
},
},
"required": ["query"],
},
),
]
# ─── 工具实现 ─────────────────────────────────────
@server.call_tool()
async def handle_call_tool(
name: str, arguments: dict[str, Any] | None
) -> list[types.TextContent]:
args = arguments or {}
if name == "query_database":
sql = args.get("sql", "").strip()
# 安全校验：只允许 SELECT
if not sql.upper().startswith("SELECT"):
return [types.TextContent(
type="text",
text="仅支持 SELECT 查询"
)]
try:
conn = get_db()
cursor = conn.execute(sql)
rows = [dict(row) for row in cursor.fetchmany(50)]
conn.close()
return [types.TextContent(
type="text",
text=json.dumps(rows, ensure_ascii=False, indent=2)
)]
except Exception as e:
return [types.TextContent(type="text", text=f"查询失败: {e}")]
elif name == "call_http_api":
method = args.get("method", "GET").upper()
url = args.get("url", "")
headers = args.get("headers", {})
body = args.get("body")
try:
async with httpx.AsyncClient(timeout=30.0) as client:
if method == "GET":
resp = await client.get(url, headers=headers)
else:
resp = await client.post(
url, headers=headers, json=body
)
# 尝试格式化 JSON 响应，否则返回文本
try:
data = resp.json()
text = json.dumps(
data, ensure_ascii=False, indent=2
)
except Exception:
text = resp.text[:3000]
return [types.TextContent(
type="text",
text=f"Status: {resp.status_code}\n\n{text}"
)]
except Exception as e:
return [types.TextContent(type="text", text=f"HTTP 请求失败: {e}")]
elif name == "read_file_content":
file_path = args.get("path", "")
path = Path(file_path).resolve()
# 安全限制：禁止读取敏感目录
sensitive_dirs = ["/etc", "/proc", "/sys", "/var"]
for sd in sensitive_dirs:
if str(path).startswith(sd):
return [types.TextContent(
type="text", text=f"禁止读取系统目录: {sd}"
)]
if not path.exists():
return [types.TextContent(type="text", text="文件不存在")]
if path.stat().st_size > 100 * 1024:
return [types.TextContent(type="text", text="文件超过 100KB 限制")]
try:
content = path.read_text(encoding="utf-8")
return [types.TextContent(
type="text",
text=content[:3000] # 限制返回长度
)]
except Exception as e:
return [types.TextContent(type="text", text=f"读取失败: {e}")]
elif name == "search_web":
query = args.get("query", "")
count = min(args.get("count", 5), 10)
# 这里用 DuckDuckGo 无 API 搜索做演示
# 生产环境建议替换为 SerpAPI / Bing Search API
url = "https://api.duckduckgo.com/"
params = {
"q": query,
"format": "json",
"no_html": 1,
"skip_disambig": 1,
}
try:
async with httpx.AsyncClient(timeout=15.0) as client:
resp = await client.get(url, params=params)
data = resp.json()
results = []
if "Abstract" in data and data["Abstract"]:
results.append({
"title": data.get("Heading", ""),
"snippet": data["Abstract"],
"source": data.get("AbstractSource", ""),
})
if "RelatedTopics" in data:
for topic in data["RelatedTopics"][:count]:
if "Text" in topic:
results.append({
"title": topic.get("Text", "").split(" - ")[0],
"snippet": topic.get("Text", ""),
"url": topic.get("FirstURL", ""),
})
return [types.TextContent(
type="text",
text=json.dumps(results, ensure_ascii=False, indent=2)
)]
except Exception as e:
return [types.TextContent(type="text", text=f"搜索失败: {e}")]
else:
raise ValueError(f"未知工具: {name}")
async def main():
async with mcp.server.stdio.stdio_server() as (read_stream, write_stream):
await server.run(
read_stream,
write_stream,
InitializationOptions(
server_name="production-server",
server_version="0.1.0",
capabilities=server.get_capabilities(
notification_options=NotificationOptions(),
experimental_capabilities={},
),
),
)
if __name__ == "__main__":
import asyncio
asyncio.run(main())
7.3 为测试初始化数据
cd ~/mcp-tutorial
mkdir -p data
python -c "
import sqlite3
conn = sqlite3.connect('data/app.db')
conn.execute('''CREATE TABLE IF NOT EXISTS users (
id INTEGER PRIMARY KEY,
name TEXT,
email TEXT,
role TEXT,
created_at TEXT
)''')
conn.execute(\"INSERT INTO users VALUES (1, '张三', 'zhang@example.com', 'admin', '2026-01-15')\")
conn.execute(\"INSERT INTO users VALUES (2, '李四', 'li@example.com', 'editor', '2026-03-20')\")
conn.execute(\"INSERT INTO users VALUES (3, '王五', 'wang@example.com', 'viewer', '2026-05-01')\")
conn.commit()
conn.close()
print('数据库初始化完成')
"
7.4 配置到 Claude Desktop
{
"mcpServers": {
"basic-tools": {
"command": "uv",
"args": [
"--directory",
"/home/user/mcp-tutorial",
"run",
"python",
"server/basic_server.py"
]
},
"production-server": {
"command": "uv",
"args": [
"--directory",
"/home/user/mcp-tutorial",
"run",
"python",
"server/production_server.py"
],
"env": {
"PYTHONPATH": "/home/user/mcp-tutorial"
}
}
}
}
重启 Claude Desktop，你可以在对话中这样做：
用户：查一下 users 表里有哪些管理员
Claude: 让我查一下数据库。 [调用 query_database]
SELECT * FROM users WHERE role = 'admin'
结果：[{"id": 1, "name": "张三", "email": "zhang@example.com", "role": "admin", "created_at": "2026-01-15"}]
用户：帮我搜索一下 MCP 协议的最新动态
Claude: [调用 search_web(query="MCP protocol latest 2026")]
结果：[...3条搜索结果...]
---
八、MCP 传输层深入：从 stdio 到 SSE
8.1 stdio 模式（本地）
这是 MCP 默认的通信方式。Client（如 Claude Desktop）启动 Server 子进程，通过 stdin/stdout 传 JSON-RPC 消息。
优点：零网络配置，安全（工具不出本机）
缺点：无法远程调用，一个 Server 只服务一个 Client
8.2 SSE 模式（远程）
如果你需要让远程的 Agent 调用你的工具，需要用 SSE（Server-Sent Events）。
# server/sse_server.py
"""
MCP Server over SSE —— 支持远程调用
"""
from mcp.server import Server, NotificationOptions
from mcp.server.models import InitializationOptions
import mcp.server.sse
import mcp.types as types
import uvicorn
from starlette.applications import Starlette
from starlette.routing import Route
server = Server("sse-server")
@server.list_tools()
async def handle_list_tools() -> list[types.Tool]:
return [
types.Tool(
name="hello",
description="一个简单的问候工具",
inputSchema={
"type": "object",
"properties": {
"name": {"type": "string", "description": "你的名字"}
},
"required": ["name"],
},
)
]
@server.call_tool()
async def handle_call_tool(
name: str, arguments: dict[str, Any] | None
) -> list[types.TextContent]:
if name == "hello":
name = (arguments or {}).get("name", "World")
return [types.TextContent(
type="text",
text=f"Hello, {name}! 这是由 MCP SSE Server 返回的消息。"
)]
# Starlette 应用
sse_app = mcp.server.sse.SseServerTransport("/messages/")
async def handle_sse(request):
async with sse_app.connect_sse(
request.scope, request.receive, request._send
) as (read_stream, write_stream):
await server.run(
read_stream,
write_stream,
InitializationOptions(
server_name="sse-server",
server_version="0.1.0",
capabilities=server.get_capabilities(
notification_options=NotificationOptions(),
experimental_capabilities={},
),
),
)
async def handle_messages(request):
await sse_app.handle_post_message(request.scope, request.receive, request._send)
app = Starlette(
routes=[
Route("/sse", endpoint=handle_sse),
Route("/messages/", endpoint=handle_messages, methods=["POST"]),
]
)
if __name__ == "__main__":
uvicorn.run(app, host="0.0.0.0", port=8000)
安装额外依赖：
uv pip install "mcp[cli]" uvicorn starlette
启动：
python server/sse_server.py
终端输出：
$ python server/sse_server.py
INFO: Started server process [12345]
INFO: Waiting for application startup.
INFO: Application startup complete.
INFO: Uvicorn running on http://0.0.0.0:8000
远程客户端可以通过 http://你的IP:8000/sse 连接这个 MCP Server。
---
九、第四坑：MCP Server 进程管理
9.1 进程意外退出
Claude Desktop 启动的 MCP Server 是子进程模式。如果 Server 代码有 bug 崩溃了，Claude 端只会看到"工具调用失败"，没有任何堆栈信息。
解决方案：添加日志
# 在 server 代码中加入日志
import logging
logging.basicConfig(
level=logging.DEBUG,
filename=Path.home() / "mcp-tutorial" / "logs" / "server.log",
format="%(asctime)s %(levelname)s %(message)s",
)
logger = logging.getLogger("mcp-server")
然后每个工具调用都加日志：
@server.call_tool()
async def handle_call_tool(name, arguments):
logger.info(f"Calling tool: {name} with args: {arguments}")
try:
# ... 工具逻辑
logger.info(f"Tool {name} succeeded")
except Exception as e:
logger.error(f"Tool {name} failed: {e}", exc_info=True)
raise
9.2 超时问题
Claude Desktop 对 MCP 工具有默认超时（约 60 秒）。如果你的工具执行时间较长（比如大文件读取、慢查询），需要：
- 在工具描述里明确告知 Agent
- 在 Server 端设置合理的超时控制
# 在工具描述里加警告
types.Tool(
name="slow_query",
description="注意：此查询可能需要 10-30 秒完成",
...
)
---
十、调试技巧：MCP Inspector 和 mcp-cli
10.1 安装 MCP Inspector
Anthropic 提供了一个 Web 调试工具：
npx @anthropic-ai/mcp-inspector
启动后访问 http://localhost:5173，输入 MCP Server 启动命令，可以：
- 查看工具列表
- 模拟调用工具
- 查看 JSON-RPC 原始消息
- 检查 Server 健康状态
终端输出：
$ npx @anthropic-ai/mcp-inspector
MCP Inspector is running at http://localhost:5173
10.2 纯命令行测试
不想开 Web？用 mcp-cli：
# 安装
uv pip install "mcp[cli]"
# 启动交互式测试
mcp run server/production_server.py
# 进入交互界面后，输入:
tools/list
# 看到工具列表
# 调用工具
tools/call -n query_database -a '{"sql": "SELECT * FROM users"}'
终端输出示例：
$ mcp run server/production_server.py
> tools/list
[
{
"name": "query_database",
"description": "对本地 SQLite 数据库执行只读查询（仅 SELECT）",
"inputSchema": {...}
},
...
]
> tools/call -n query_database -a '{"sql": "SELECT * FROM users"}'
[
{
"type": "text",
"text": "[\n {\n \"id\": 1,\n \"name\": \"张三\",\n ...\n }\n]"
}
]
10.3 第五坑：mcp-cli 报 ModuleNotFoundError
$ mcp run server/production_server.py
Traceback:
ModuleNotFoundError: No module named 'mcp.server'
原因：mcp CLI 工具和你的 Python 环境不一致。你在项目虚拟环境里装了 mcp，但 CLI 用的是全局 Python。
解决：始终在虚拟环境里运行：
source .venv/bin/activate
which mcp # 确认指向 .venv/bin/mcp
mcp run server/production_server.py
---
十一、连接其他 AI Agent（非 Claude）
MCP 是开放协议，不只 Claude 能用。这里演示用 Python 写一个自定义 Agent 来调用 MCP Server。
11.1 Python 自定义 Agent
# custom_agent.py
"""
自定义 AI Agent —— 通过 MCP 协议调用工具
不依赖 Claude，用 OpenAI API 做 LLM 后端
"""
import json
import asyncio
import subprocess
from openai import AsyncOpenAI
# ─── 1. 启动 MCP Server 子进程 ────────────────────
class MCPClient:
"""MCP 协议客户端"""
def __init__(self, server_cmd: list[str]):
self.server_cmd = server_cmd
self.process = None
self.message_id = 0
async def start(self):
self.process = await asyncio.create_subprocess_exec(
*self.server_cmd,
stdin=subprocess.PIPE,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
)
# 发送 initialize 请求
result = await self._send({
"jsonrpc": "2.0",
"method": "initialize",
"params": {
"protocolVersion": "2024-11-05",
"capabilities": {},
"clientInfo": {
"name": "custom-agent",
"version": "1.0.0"
}
},
"id": 1,
})
print(f"[MCP] 初始化完成: {result}")
return result
async def list_tools(self) -> list:
result = await self._send({
"jsonrpc": "2.0",
"method": "tools/list",
"id": 2,
})
return result.get("tools", [])
async def call_tool(self, name: str, arguments: dict) -> dict:
self.message_id += 1
result = await self._send({
"jsonrpc": "2.0",
"method": "tools/call",
"params": {
"name": name,
"arguments": arguments,
},
"id": self.message_id,
})
return result
async def _send(self, msg: dict) -> dict:
"""发送 JSON-RPC 消息并读取响应"""
payload = json.dumps(msg) + "\n"
self.process.stdin.write(payload.encode())
await self.process.stdin.drain()
# 读取一行响应
line = await self.process.stdout.readline()
return json.loads(line.decode())
async def close(self):
if self.process:
self.process.terminate()
await self.process.wait()
# ─── 2. 主 Agent 循环 ────────────────────────────
async def main():
# 启动 MCP Client
client = MCPClient([
"uv", "run", "python",
"server/production_server.py"
])
await client.start()
# 获取工具列表
tools = await client.list_tools()
print(f"[Agent] 发现 {len(tools)} 个工具")
# 构造 OpenAI 工具定义
openai_tools = []
for t in tools:
openai_tools.append({
"type": "function",
"function": {
"name": t["name"],
"description": t["description"],
"parameters": t["inputSchema"],
}
})
# 初始化 OpenAI 客户端
llm = AsyncOpenAI(
api_key="sk-your-key",
base_url="https://api.openai.com/v1",
)
# 对话循环
messages = [
{"role": "system", "content": "你是一个助手，可以使用外部工具来回答用户问题。"},
{"role": "user", "content": "帮我查一下数据库里有哪些用户"},
]
print(f"\n[用户] {messages[-1]['content']}\n")
for _ in range(5): # 最多 5 轮
response = await llm.chat.completions.create(
model="gpt-4o",
messages=messages,
tools=openai_tools,
tool_choice="auto",
)
msg = response.choices[0].message
if msg.tool_calls:
for tc in msg.tool_calls:
func_name = tc.function.name
func_args = json.loads(tc.function.arguments)
print(f"[Agent] 调用工具: {func_name}({func_args})")
# 通过 MCP 调用工具
result = await client.call_tool(func_name, func_args)
messages.append({
"role": "tool",
"tool_call_id": tc.id,
"content": json.dumps(result, ensure_ascii=False),
})
else:
print(f"[AI] {msg.content}")
break
await client.close()
if __name__ == "__main__":
asyncio.run(main())
运行：
source .venv/bin/activate
python custom_agent.py
终端输出：
$ python custom_agent.py
[MCP] 初始化完成: {'protocolVersion': '2024-11-05', 'capabilities': {...}, 'serverInfo': {'name': 'production-server', 'version': '0.1.0'}}
[Agent] 发现 4 个工具
[用户] 帮我查一下数据库里有哪些用户
[Agent] 调用工具: query_database({'sql': 'SELECT * FROM users'})
[AI] 数据库中有以下用户：
1. **张三** — 管理员（admin），邮箱：zhang@example.com，创建于 2026-01-15
2. **李四** — 编辑（editor），邮箱：li@example.com，创建于 2026-03-20
3. **王五** — 查看者（viewer），邮箱：wang@example.com，创建于 2026-05-01
11.2 多 Server 聚合
生产环境中，你可能需要让 Agent 同时连接多个 MCP Server。比如一个 Server 管数据库，一个 Server 管 API，一个 Server 管文件系统。
# multi_server_agent.py
"""
聚合多个 MCP Server
"""
import asyncio
from custom_agent import MCPClient
class MCPGateway:
"""多 Server 网关"""
def __init__(self):
self.clients: list[MCPClient] = []
async def add_server(self, cmd: list[str], name: str):
client = MCPClient(cmd)
await client.start()
tools = await client.list_tools()
self.clients.append(client)
print(f"[Gateway] 注册 Server '{name}': {len(tools)} 个工具")
return tools
async def list_all_tools(self) -> list:
all_tools = []
for client in self.clients:
tools = await client.list_tools()
all_tools.extend(tools)
return all_tools
async def call_tool(self, name: str, args: dict) -> dict:
for client in self.clients:
tools = await client.list_tools()
if any(t["name"] == name for t in tools):
return await client.call_tool(name, args)
raise ValueError(f"工具 '{name}' 未在任何 Server 中找到")
async def close(self):
for client in self.clients:
await client.close()
async def main():
gateway = MCPGateway()
await gateway.add_server(
["uv", "run", "python", "server/basic_server.py"],
"basic-tools",
)
await gateway.add_server(
["uv", "run", "python", "server/production_server.py"],
"production-server",
)
tools = await gateway.list_all_tools()
print(f"\n总计 {len(tools)} 个工具可用:")
for t in tools:
print(f" - {t['name']}: {t['description'][:50]}...")
await gateway.close()
if __name__ == "__main__":
asyncio.run(main())
终端输出：
$ python multi_server_agent.py
[Gateway] 注册 Server 'basic-tools': 2 个工具
[Gateway] 注册 Server 'production-server': 4 个工具
总计 6 个工具可用:
- get_time: 获取当前系统时间，支持时区参数...
- run_shell: 执行 shell 命令（只读操作，安全受限）...
- query_database: 对本地 SQLite 数据库执行只读查询...
- call_http_api: 调用外部 HTTP API（GET/POST）...
- read_file_content: 读取指定文件的内容...
- search_web: 搜索互联网，返回前 N 条结果...
---
十二、生产部署 checklist
当你准备把 MCP Server 部署到生产环境时，这几件事必须做好：
12.1 安全清单
| 项目 | 说明 |
|------|------|
| 命令注入防护 | 对 run_shell 类工具做白名单校验，而不是黑名单 |
| SQL 注入防护 | 仅支持 SELECT，且用参数化查询（本文用 SQLite 演示，生产请换 PostgreSQL + 参数化） |
| 文件路径穿越 | read_file_content 必须做路径规范化 + 目录限制 |
| 请求频率限制 | 每个 Client 每分钟最多 N 次工具调用 |
| 敏感信息过滤 | 数据库查询结果自动脱敏（手机号、密码等） |
| 认证 | SSE Server 必须加 API Key 校验 |
12.2 性能建议
- 连接池：数据库连接不要每次请求都创建销毁，用连接池
- 缓存：高频查询结果（如天气、汇率）加本地缓存
- 超时：每个工具都要有明确的超时控制
- 日志：记录每次工具调用的入参、出参、耗时
12.3 监控
# 简单的调用统计装饰器
import time
from functools import wraps
call_stats = {"count": 0, "total_time": 0.0, "errors": 0}
def monitor_tool(func):
@wraps(func)
async def wrapper(*args, **kwargs):
call_stats["count"] += 1
start = time.time()
try:
result = await func(*args, **kwargs)
return result
except Exception as e:
call_stats["errors"] += 1
raise
finally:
elapsed = time.time() - start
call_stats["total_time"] += elapsed
logger.info(
f"Tool call: {func.__name__}, "
f"elapsed={elapsed:.3f}s, "
f"total_calls={call_stats['count']}"
)
return wrapper
---
十三、FAQ（真实遇到的问题）
Q1: MCP Server 和普通 API Server 有什么区别？
MCP Server 不是 HTTP API。它通过 stdio 或 SSE 传输 JSON-RPC 消息，专门为 AI Agent 设计。核心区别：
- 协议统一：所有工具用统一的 tools/list + tools/call 接口
- 自动发现：Agent 自动获取工具列表和 schema，不用写 prompt
- 生命周期管理：Agent 控制 Server 启停，不用手动部署 HTTP 服务
Q2: 能在 Docker 里跑 MCP Server 吗？
可以。配置 command 指向 Docker：
{
"mcpServers": {
"docker-server": {
"command": "docker",
"args": ["run", "-i", "--rm", "my-mcp-server:latest"]
}
}
}
注意加 -i 保持 stdin 开启。
Q3: MCP 支持流式响应吗？
协议层面支持 streaming，但目前 Claude Desktop 对 streaming 的支持还不完善。如果你需要返回大量数据（比如查询结果上万行），建议在 Server 端做分页，而不是依赖 streaming。
Q4: 多个工具之间有依赖怎么办？
比如"先查询用户 ID，再根据 ID 查订单"。这个依赖逻辑由 Agent 的 LLM 处理，MCP 协议不关心。Agent 会先调 query_database 拿到用户 ID，再调 query_database 查订单。
Q5: 我配置了 JSON 但 Claude Desktop 完全不识别
最常见的三个原因：
- JSON 格式错误：用
python -m json.tool config.json校验 - 路径写错：
args里的路径必须是绝对路径 - 权限问题：Claude Desktop 启动时可能没有读写某些目录的权限
Q6: 一个 MCP Server 可以同时服务多个 Agent 吗？
stdio 模式不行（1:1 进程映射）。SSE 模式理论上可以，但需要你自己实现连接管理。如果需要多 Agent 共享工具，建议在 Server 前面加一层反向代理。
Q7: 免费的工具资源从哪里找？
GitHub 上搜索 mcp-servers 或者 awesome-mcp，已经有社区整理的 Server 列表。截止 2026 年 5 月，能找到上百个现成的 MCP Server，覆盖数据库、云服务、开发工具、金融数据等。
---
十四、总结
MCP 协议的核心价值不是技术上的创新——JSON-RPC 不是什么新鲜东西。它的价值在于统一了 AI Agent 调用外部工具的方式。
回头看文章开头的那个崩溃场景：三个工具三种写法。换成 MCP 之后，每个工具就是一个函数定义 + 实现，Agent 自动发现、自动选择、自动调用。加一个新工具，写一个函数就完事。
如果你正在做 AI Agent 相关的开发，MCP 值得花一个下午部署起来。从今天开始，你的 Agent 不再只是"能聊天"，而是真的能操作你的系统了。
---
附录：完整项目文件结构
~/mcp-tutorial/
├── .venv/ # 虚拟环境
├── pyproject.toml # 项目依赖
├── data/
│ └── app.db # 演示数据库
├── logs/
│ └── server.log # 运行时日志
├── server/
│ ├── __init__.py
│ ├── basic_server.py # 基础 Server（时间+shell）
│ ├── production_server.py # 生产 Server（DB+API+文件+搜索）
│ └── sse_server.py # SSE 远程 Server
├── custom_agent.py # 自定义 Python Agent
├── multi_server_agent.py # 多 Server 聚合
└── config/
└── claude_desktop_config.json # Claude Desktop 配置