大型语言模型的能力正在被赋予“双手”——借助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作为异步HTTP客户端，用于替代requests，后续天气查询功能会用到它。
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"
