MCP协议实战：用Python开发首个AI工具服务
来源：互联网 2026-07-19 06:33:22MCP协议赋予AI应用调用工具、读写数据等能力。文章介绍其核心概念、三大能力及stdio与SSE两种通信方式，并完整演示基于Python开发MCPServer的流程，涵盖环境搭建、工具实现及安全沙箱控制，为构建企业级AI工具服务提供实践参考。
大型语言模型的能力正在被赋予“双手”——通过MCP协议，AI应用能够真正执行代码、调用API、操作文件，而不仅仅是生成文本。这篇文章就从零开始，完整展示一个MCP Server的开发全流程。
一、MCP协议是什么？
1.1 核心概念
MCP（Model Context Protocol）定义了一套标准协议，让AI应用（Host）能够通过统一的接口访问外部工具和数据。这套架构看起来确实复杂，但拆解开来看其实很清晰：AI应用作为主机，中间层是MCP客户端，最外层是MCP服务端，三者通过一个标准化的通信管道相互协作。
长期稳定更新的攒劲资源： >>>点此立即查看<<<
从架构全景图可以看出，MCP服务端主要提供三类能力：Tools（工具调用）、Resources（数据读取）和Prompts（提示模板）。Host内部的MCP客户端就像是一个翻译官，负责在AI应用和服务端之间传递消息。
1.2 MCP三大核心能力
| 能力 | 说明 | 示例 |
|---|---|---|
| Tools | AI可调用的函数 | 查询数据库、调用API、执行计算 |
| Resources | AI可读取的数据源 | 文件内容、数据库记录、日志 |
| Prompts | 可复用的提示模板 | 代码审查模板、文档生成模板 |
1.3 通信方式
目前支持两种传输方式，怎么选？看场景。本地开发时，stdio模式是最直接的，AI应用直接把MCP Server当子进程启动，通过标准输入输出通信。远程部署则用SSE（Server-Sent Events），说白了就是通过HTTP进行网络通信。这两种方式各有千秋，本地开发图省事，生产环境则必须上SSE。
方式一: stdio（标准输入输出）
┌──────────┐ stdin/stdout ┌──────────┐
│ AI 应用 │──────────────│MCP Server│
│ (Host) │ │(子进程) │
└──────────┘ └──────────┘
方式二: SSE（Server-Sent Events）
┌──────────┐ HTTP + SSE ┌──────────┐
│ AI 应用 │──────────────│MCP Server│
│ (Host) │ (网络通信) │(远程服务)│
└──────────┘ └──────────┘
二、开发环境搭建
2.1 安装依赖
先装个包：pip install "mcp[cli]" httpx，一行命令搞定。httpx是用来替代requests的异步HTTP客户端，后面天气查询会用到。
2.2 验证安装
跑一下 mcp --version，能看到版本号就说明安装成功了。
2.3 项目结构
推荐的项目组织方式是这样的：
my-mcp-server/
├── server.py # MCP Server主文件
├── tools/
│ ├── weather.py # 天气查询工具
│ ├── database.py # 数据库查询工具
│ └── calculator.py # 计算器工具
├── resources/
│ └── system_info.py # 系统信息资源
├── prompts/
│ └── code_review.py # 代码审查提示模板
├── pyproject.toml # 项目配置
└── README.md
按功能模块拆分的好处是，每个文件职责清晰，后期维护起来也不用面对一个几千行的庞然大物。
三、第一个MCP Server（快速上手）
3.1 最简实现
先来个最简单的，感受一下MCP的魅力：
# server.py
from mcp.server.fastmcp import FastMCP
# 创建MCP Server实例
mcp = FastMCP(
name="my-first-mcp-server",
version="1.0.0",
)
@mcp.tool()
def add(a: int, b: int) -> int:
"""两个数字相加"""
return a + b
@mcp.tool()
def get_current_time(timezone: str = "Asia/Shanghai") -> str:
"""获取当前时间
Args:
timezone: 时区名称，默认为Asia/Shanghai
"""
from datetime import datetime
import pytz
tz = pytz.timezone(timezone)
now = datetime.now(tz)
return now.strftime("%Y-%m-%d %H:%M:%S %Z")
if __name__ == "__main__":
mcp.run()
看到 @mcp.tool() 这个装饰器了吗？这就是把普通Python函数变成AI可调用工具的关键记号。参数类型标注和docstring不是摆设——AI就是靠这些信息来理解工具的使用方式的。
3.2 运行测试
两种启动方式：
mcp dev server.py 会启动一个可视化调试工具（MCP Inspector），在浏览器里就能测试你的工具。或者直接 python server.py 跑起来也行。
四、实战：开发完整的企业工具服务
下面我们来点干货，构建一个实用的MCP Server，涵盖天气查询、数据库操作、文件管理等功能。
4.1 天气查询工具
调用OpenWeatherMap API，支持当前天气和天气预报：
# tools/weather.py
import httpx
from mcp.server.fastmcp import tool
BASE_URL = "https://api.openweathermap.org/data/2.5"
@tool()
async def get_weather(city: str, api_key: str = "") -> str:
"""查询指定城市的当前天气信息
Args:
city: 城市名称（中文或英文），如"北京"或"Beijing"
api_key: OpenWeatherMap API Key（可选，默认使用内置测试key）
"""
params = {
"q": city,
"appid": api_key,
"units": "metric",
"lang": "zh_cn",
}
async with httpx.AsyncClient() as client:
resp = await client.get(f"{BASE_URL}/weather", params=params)
data = resp.json()
if resp.status_code != 200:
return f"查询失败: {data.get('message', '未知错误')}"
return (
f"【{data['name']} 天气】\n"
f"天气: {data['weather'][0]['description']}\n"
f"温度: {data['main']['temp']}°C (体感 {data['main']['feels_like']}°C)\n"
f"湿度: {data['main']['humidity']}%\n"
f"风速: {data['wind']['speed']} m/s"
)
@tool()
async def get_forecast(city: str, days: int = 3, api_key: str = "") -> str:
"""查询未来几天的天气预报
Args:
city: 城市名称
days: 预报天数（最多5天）
api_key: OpenWeatherMap API Key
"""
params = {
"q": city,
"appid": api_key,
"units": "metric",
"lang": "zh_cn",
"cnt": min(days, 5) * 8, # 每天约8个数据点
}
async with httpx.AsyncClient() as client:
resp = await client.get(f"{BASE_URL}/forecast", params=params)
data = resp.json()
forecasts = []
for item in data["list"][::8]: # 每天取一个数据点
forecasts.append(
f"{item['dt_txt'][:10]}: "
f"{item['weather'][0]['description']}, "
f"{item['main']['temp']}°C"
)
return f"【{data['city']['name']} 未来天气预报】\n" + "\n".join(forecasts)
4.2 SQLite数据库查询工具
这个工具的设计要点是安全性——只允许SELECT操作，防止AI意外执行危险命令：
# tools/database.py
import sqlite3
from pathlib import Path
from mcp.server.fastmcp import tool
DB_PATH = Path("./data/app.db")
def _get_connection():
"""获取数据库连接"""
DB_PATH.parent.mkdir(exist_ok=True)
return sqlite3.connect(str(DB_PATH))
@tool()
def query_database(sql: str) -> str:
"""执行SQL查询语句（仅支持SELECT）
Args:
sql: SQL查询语句，仅允许SELECT操作
"""
# 安全检查：只允许SELECT
normalized = sql.strip().upper()
if not normalized.startswith("SELECT"):
return "错误：仅支持SELECT查询，不允许修改数据"
try:
conn = _get_connection()
cursor = conn.execute(sql)
columns = [desc[0] for desc in cursor.description]
rows = cursor.fetchall()
conn.close()
if not rows:
return "查询结果为空"
header = "| " + " | ".join(columns) + " |"
separator = "| " + " | ".join(["---"] * len(columns)) + " |"
data_rows = []
for row in rows[:50]: # 最多返回50行
data_rows.append("| " + " | ".join(str(v) for v in row) + " |")
table = "\n".join([header, separator] + data_rows)
return f"查询返回 {len(rows)} 行:\n{table}"
except Exception as e:
return f"查询出错: {str(e)}"
@tool()
def list_tables() -> str:
"""列出数据库中的所有表及其结构"""
conn = _get_connection()
cursor = conn.execute("SELECT name FROM sqlite_master WHERE type='table'")
tables = cursor.fetchall()
conn.close()
if not tables:
return "数据库为空，暂无表"
result = []
for (table_name,) in tables:
result.append(f"- **{table_name}**")
return "数据库中的表:\n" + "\n".join(result)
4.3 文件管理工具
文件操作最怕路径穿越攻击，所以需要做一个严格的沙箱限制：
# tools/file_manager.py
import os
from pathlib import Path
from mcp.server.fastmcp import tool
SAFE_DIR = Path("./data/workspace") # 安全沙箱目录
def _safe_path(filepath: str) -> Path:
"""确保文件路径在安全目录内"""
full_path = (SAFE_DIR / filepath).resolve()
if not str(full_path).startswith(str(SAFE_DIR.resolve())):
raise ValueError("路径超出安全范围")
return full_path
@tool()
def read_file(filepath: str) -> str:
"""读取文件内容
Args:
filepath: 相对于工作目录的文件路径
"""
path = _safe_path(filepath)
if not path.exists():
return f"文件不存在: {filepath}"
return path.read_text(encoding="utf-8")
@tool()
def write_file(filepath: str, content: str) -> str:
"""写入文件内容
Args:
filepath: 相对于工作目录的文件路径
content: 要写入的内容
"""
path = _safe_path(filepath)
path.parent.mkdir(parents=True, exist_ok=True)
path.write_text(content, encoding="utf-8")
return f"文件写入成功: {filepath} ({len(content)} 字符)"
@tool()
def list_files(directory: str = ".") -> str:
"""列出目录下的文件
Args:
directory: 相对于工作目录的目录路径
"""
path = _safe_path(directory)
if not path.exists():
return f"目录不存在: {directory}"
items = []
for item in sorted(path.iterdir()):
if item.is_dir():
items.append(f" {item.name}/")
else:
size = item.stat().st_size
items.append(f" {item.name} ({size} bytes)")
return "\n".join(items) if items else "目录为空"
4.4 定义Resources（数据资源）
Resources是AI可以直接读取的数据源，相当于只读接口：
# resources/system_info.py
import platform
import psutil
from mcp.server.fastmcp import resource
@resource("system://info")
def get_system_info() -> str:
"""获取系统信息"""
return f"""
系统信息:
- 操作系统: {platform.system()} {platform.release()}
- Python: {platform.python_version()}
- CPU核心: {psutil.cpu_count()}
- 内存总量: {psutil.virtual_memory().total // (1024**3)} GB
- 磁盘使用: {psutil.disk_usage('/').percent}%
"""
@resource("system://processes")
def get_top_processes() -> str:
"""获取占用资源最多的前10个进程"""
procs = []
for p in psutil.process_iter(["pid", "name", "cpu_percent", "memory_percent"]):
try:
info = p.info
procs.append(info)
except (psutil.NoSuchProcess, psutil.AccessDenied):
pass
procs.sort(key=lambda x: x["cpu_percent"] or 0, reverse=True)
lines = ["PID 名称 CPU% 内存%"]
lines.append("-" * 45)
for p in procs[:10]:
lines.append(f"{p['pid']:<8}{p['name']:<20}{p['cpu_percent']:.1f}% {p['memory_percent']:.1f}%")
return "\n".join(lines)
4.5 定义Prompts（提示模板）
Prompts本质上就是预设的上下文指令，让AI按照固定格式处理特定任务：
# prompts/code_review.py
from mcp.server.fastmcp import prompt
@prompt()
def code_review(code: str, language: str = "python") -> str:
"""代码审查提示模板
Args:
code: 需要审查的代码
language: 编程语言
"""
return f"""请对以下 {language} 代码进行专业审查:
```{language}
{code}
```
请从以下维度进行审查:
1. 代码质量 — 可读性、命名规范、代码风格
2. 潜在Bug — 逻辑错误、边界条件、空指针风险
3. 安全漏洞 — 注入攻击、数据泄露、权限问题
4. 性能优化 — 时间复杂度、内存使用、I/O效率
5. 最佳实践 — 设计模式、SOLID原则、错误处理
请为每个维度给出1-5分评分和具体改进建议。"""
@prompt()
def explain_code(code: str) -> str:
"""代码解释提示模板
Args:
code: 需要解释的代码
"""
return f"""请用通俗易懂的语言解释以下代码的功能:
```{code}
```
要求:
1. 先用一句话概括整体功能
2. 逐段解释关键逻辑
3. 说明输入和输出
4. 指出可能的使用场景"""
4.6 组装完整Server
把上面分散的模块全部注册到主Server中：
# server.py
from mcp.server.fastmcp import FastMCP
mcp = FastMCP(
name="enterprise-toolkit",
version="1.0.0",
description="企业级MCP工具服务：天气查询、数据库操作、文件管理",
)
# ── 注册Tools ──────────────────────────────────────
from tools.weather import get_weather, get_forecast
from tools.database import query_database, list_tables
from tools.file_manager import read_file, write_file, list_files
mcp.tool(get_weather)
mcp.tool(get_forecast)
mcp.tool(query_database)
mcp.tool(list_tables)
mcp.tool(read_file)
mcp.tool(write_file)
mcp.tool(list_files)
# ── 注册Resources ──────────────────────────────────
from resources.system_info import get_system_info, get_top_processes
mcp.resource("system://info")(get_system_info)
mcp.resource("system://processes")(get_top_processes)
# ── 注册Prompts ────────────────────────────────────
from prompts.code_review import code_review, explain_code
mcp.prompt(code_review)
mcp.prompt(explain_code)
# ── 启动 ────────────────────────────────────────────
if __name__ == "__main__":
mcp.run()
五、配置与使用
5.1 在Claude Desktop中配置
编辑Claude Desktop配置文件，路径因系统而异：
macOS: ~/Library/Application Support/Claude/claude_desktop_config.json
Windows: %APPDATA%\Claude\claude_desktop_config.json
{
"mcpServers": {
"enterprise-toolkit": {
"command": "python",
"args": ["C:/path/to/my-mcp-server/server.py"],
"env": {
"OPENWEATHER_API_KEY": "your-api-key"
}
}
}
}
5.2 在Cursor / VS Code中配置
项目根目录创建 .cursor/mcp.json：
{
"mcpServers": {
"enterprise-toolkit": {
"command": "python",
"args": ["./my-mcp-server/server.py"]
}
}
}
5.3 调试流程
整个调试链路非常清晰：启动MCP Inspector → 浏览器打开调试界面 → 查看已注册的各类工具 → 填入参数执行测试 → 确认结果后正式配置到AI客户端。可视化调试是MCP的一大亮点，省去了命令行反复测试的繁琐。
六、进阶：SSE模式部署为远程服务
6.1 使用SSE传输
# server_sse.py
from mcp.server.fastmcp import FastMCP
mcp = FastMCP(
name="enterprise-toolkit-remote",
host="0.0.0.0",
port=8080,
)
# ... 注册tools/resources/prompts（同上）...
if __name__ == "__main__":
# SSE模式：通过HTTP暴露服务
mcp.run(transport="sse")
6.2 Docker部署
# Dockerfile
FROM python:3.12-slim
WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY . .
EXPOSE 8080
CMD ["python", "server_sse.py"]
# docker-compose.yml
version: "3.8"
services:
mcp-server:
build: .
ports:
- "8080:8080"
environment:
- OPENWEATHER_API_KEY=${OPENWEATHER_API_KEY}
volumes:
- ./data:/app/data
restart: unless-stopped
6.3 客户端配置（远程SSE）
{
"mcpServers": {
"enterprise-toolkit": {
"url": "http://your-server:8080/sse"
}
}
}
七、MCP协议消息流详解
当用户在Claude中说"帮我查一下北京天气"，背后发生了什么？完整的消息流是这样的：
首先是用户输入 → Claude分析意图，决定调用get_weather工具 → 通过MCP发送JSON-RPC请求（method是tools/call，参数是工具名和参数）→ MCP Server执行实际函数 → 返回结果（包含天气数据）→ Claude结合结果生成自然语言回答 → 用户看到预期结果。
这套消息流看起来步骤多，但实际延迟很低，因为内部使用了高效的序列化协议。
八、最佳实践与安全注意事项
8.1 工具设计原则
设计MCP工具时，有几个原则必须刻在脑子里：单一职责每个工具只做一件事；docstring是AI理解工具的关键，写得越清晰AI就用得越准确；参数必须有明确的类型标注和默认值；文件/数据库访问一定要做沙箱限制；错误信息要友好可读而非直接抛异常；相同输入应产生相同输出，保持幂等性。
8.2 安全清单
安全永远是第一位的。SQL注入防护是基础，必须严格限定只允许SELECT操作。路径穿越攻击需要通过对路径做resolve()然后校验前缀来防御。用户输入中的危险字符要过滤掉。这些措施虽然会增加一点代码量，但在生产环境里少一个都可能出事。
# 安全示例：参数验证与路径限制
import re
from pathlib import Path
def validate_sql(sql: str) -> bool:
"""只允许SELECT查询"""
forbidden = ["DROP", "DELETE", "INSERT", "UPDATE", "ALTER", "CREATE"]
upper_sql = sql.upper()
return all(kw not in upper_sql for kw in forbidden)
def safe_filepath(base_dir: Path, user_path: str) -> Path:
"""防止路径穿越攻击"""
full_path = (base_dir / user_path).resolve()
if not str(full_path).startswith(str(base_dir.resolve())):
raise ValueError("非法路径访问")
return full_path
def sanitize_input(text: str) -> str:
"""清理用户输入"""
# 移除潜在的危险字符
return re.sub(r'[<>&\']', '', text)
九、总结
从零到一，我们完整走了一遍MCP Server的开发流程。关键知识点回顾：FastMCP是Python生态里最核心的开发框架；@tool()、@resource()、@prompt()三个装饰器分别对应不同能力；stdio适合本地开发，SSE适合远程部署；MCP Inspector是调式利器，不用命令行手动测试。下一阶段可以探索的包括将MCP Server接入企业内部API、开发多用户权限管理中间件、结合RAG构建知识库服务，甚至参与开源生态发布自己的工具。
MCP协议正在快速成为AI工具调用的行业标准，掌握它意味着你能让AI真正融入你的工作流。这不是锦上添花，而是从"对话式AI"到"行动式AI"的跨越。
侠游戏发布此文仅为了传递信息，不代表侠游戏网站认同其观点或证实其描述
同类更新
更多
热游推荐
更多-
- DreamStudio
- Android/ | AI绘图
- 2026-06-30
-
-
- Playground AI
- Android/ | AI绘图
- 2026-06-30
-
- Adobe Firefly
- Android/ | AI绘图
- 2026-06-30
-
-
-
- Leonardo AI
- Android/ | AI绘图
- 2026-06-30
-
- Stable Diffusion
- Android/ | AI绘图
- 2026-06-30
