使用Python開發(fā)你的第一個AI工具服務MCP Server
更新時間：2026年04月14日 08:42:16 作者：平安的平安
MCP（Model Context Protocol）是 Anthropic 推出的開放協(xié)議,讓 AI 模型能夠安全地連接外部數(shù)據(jù)源和工具,本文將帶你從零理解 MCP 協(xié)議,并用 Python 開發(fā)一個完整的 MCP Server,讓 Claude 等 AI 助手直接調(diào)用你編寫的工具,需要的朋友可以參考下
一、MCP 協(xié)議是什么？
1.1 核心概念
MCP（Model Context Protocol）定義了一套標準協(xié)議，讓 AI 應用（Host）能夠通過統(tǒng)一的接口訪問外部工具和數(shù)據(jù)。
┌──────────────────────────────────────────────────────────────┐ │ MCP 架構(gòu)全景 │ │ │ │ ┌──────────┐ ┌──────────┐ ┌──────────────────────┐ │ │ │ AI 應用 │ │ MCP 客戶端 │ │ MCP 服務端 (Server) │ │ │ │ (Host) │?──?│ (Client) │?──?│ │ │ │ │ │ │ │ │ ┌────────┐ │ │ │ │ Claude │ │ 內(nèi)置于 │ │ │ Tools │ 工具調(diào)用 │ │ │ │ Cursor │ │ Host 應用 │ │ │ │ │ │ │ │ VS Code │ │ 之中 │ │ ├────────┤ │ │ │ │ │ │ │ │ │Resources│ 數(shù)據(jù)讀取 │ │ │ └──────────┘ └──────────┘ │ │ │ │ │ │ │ ├────────┤ │ │ │ │ │Prompts │ 提示模板 │ │ │ │ └────────┘ │ │ │ └──────────────────────┘ │ └──────────────────────────────────────────────────────────────┘
1.2 MCP 三大核心能力
| 能力 | 說明 | 示例 |
|---|---|---|
| Tools | AI 可調(diào)用的函數(shù) | 查詢數(shù)據(jù)庫、調(diào)用 API、執(zhí)行計算 |
| Resources | AI 可讀取的數(shù)據(jù)源 | 文件內(nèi)容、數(shù)據(jù)庫記錄、日志 |
| Prompts | 可復用的提示模板 | 代碼審查模板、文檔生成模板 |
1.3 通信方式
MCP 支持兩種傳輸方式：
方式一: stdio（標準輸入輸出） ┌──────────┐ stdin/stdout ┌──────────┐ │ AI 應用 │?───────────────?│MCP Server│ │ (Host) │ │(子進程) │ └──────────┘ └──────────┘ 方式二: SSE（Server-Sent Events） ┌──────────┐ HTTP + SSE ┌──────────┐ │ AI 應用 │?───────────────?│MCP Server│ │ (Host) │ (網(wǎng)絡通信) │(遠程服務) │ └──────────┘ └──────────┘
二、開發(fā)環(huán)境搭建
2.1 安裝依賴
pip install "mcp[cli]" httpx
2.2 驗證安裝
mcp --version
2.3 項目結(jié)構(gòu)
my-mcp-server/ ├── server.py # MCP Server 主文件 ├── tools/ │ ├── weather.py # 天氣查詢工具 │ ├── database.py # 數(shù)據(jù)庫查詢工具 │ └── calculator.py # 計算器工具 ├── resources/ │ └── system_info.py # 系統(tǒng)信息資源 ├── prompts/ │ └── code_review.py # 代碼審查提示模板 ├── pyproject.toml # 項目配置 └── README.md
三、第一個 MCP Server（快速上手）
3.1 最簡實現(xiàn)
# server.py
from mcp.server.fastmcp import FastMCP
# 創(chuàng)建 MCP Server 實例
mcp = FastMCP(
name="my-first-mcp-server",
version="1.0.0",
)
@mcp.tool()
def add(a: int, b: int) -> int:
"""兩個數(shù)字相加"""
return a + b
@mcp.tool()
def get_current_time(timezone: str = "Asia/Shanghai") -> str:
"""獲取當前時間
Args:
timezone: 時區(qū)名稱，默認為 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")
# 啟動 Server
if __name__ == "__main__":
mcp.run()3.2 運行測試
# 使用 MCP Inspector 測試（可視化調(diào)試工具） mcp dev server.py # 或直接運行 python server.py
四、實戰(zhàn)：開發(fā)完整的企業(yè)工具服務
下面我們構(gòu)建一個實用的 MCP Server，包含天氣查詢、數(shù)據(jù)庫操作、文件管理等功能。
4.1 天氣查詢工具
# 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（可選，默認使用內(nèi)置測試 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: 預報天數(shù)（最多 5 天）
api_key: OpenWeatherMap API Key
"""
params = {
"q": city,
"appid": api_key,
"units": "metric",
"lang": "zh_cn",
"cnt": min(days, 5) * 8, # 每天約 8 個數(shù)據(jù)點
}
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]: # 每天取一個數(shù)據(jù)點
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 數(shù)據(jù)庫查詢工具
# tools/database.py
import sqlite3
from pathlib import Path
from mcp.server.fastmcp import tool
DB_PATH = Path("./data/app.db")
def _get_connection():
"""獲取數(shù)據(jù)庫連接"""
DB_PATH.parent.mkdir(exist_ok=True)
return sqlite3.connect(str(DB_PATH))
@tool()
def query_database(sql: str) -> str:
"""執(zhí)行 SQL 查詢語句（僅支持 SELECT）
Args:
sql: SQL 查詢語句，僅允許 SELECT 操作
"""
# 安全檢查：只允許 SELECT
normalized = sql.strip().upper()
if not normalized.startswith("SELECT"):
return "錯誤：僅支持 SELECT 查詢，不允許修改數(shù)據(jù)"
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 "查詢結(jié)果為空"
# 格式化為 Markdown 表格
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\n{table}"
except Exception as e:
return f"查詢出錯: {str(e)}"
@tool()
def list_tables() -> str:
"""列出數(shù)據(jù)庫中的所有表及其結(jié)構(gòu)"""
conn = _get_connection()
cursor = conn.execute(
"SELECT name FROM sqlite_master WHERE type='table'"
)
tables = cursor.fetchall()
conn.close()
if not tables:
return "數(shù)據(jù)庫為空，暫無表"
result = []
for (table_name,) in tables:
result.append(f"- **{table_name}**")
return "數(shù)據(jù)庫中的表:\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:
"""確保文件路徑在安全目錄內(nèi)"""
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:
"""讀取文件內(nèi)容
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:
"""寫入文件內(nèi)容
Args:
filepath: 相對于工作目錄的文件路徑
content: 要寫入的內(nèi)容
"""
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（數(shù)據(jù)資源）
# resources/system_info.py
import platform
import psutil
from mcp.server.fastmcp import resource
@resource("system://info")
def get_system_info() -> str:
"""獲取系統(tǒng)信息"""
return f"""系統(tǒng)信息:
- 操作系統(tǒng): {platform.system()} {platform.release()}
- Python: {platform.python_version()}
- CPU 核心: {psutil.cpu_count()}
- 內(nèi)存總量: {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
# 按 CPU 使用率排序
procs.sort(key=lambda x: x["cpu_percent"] or 0, reverse=True)
lines = ["PID 名稱 CPU% 內(nèi)存%"]
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/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} 代碼進行專業(yè)審查:
```{language}
[code]
請從以下維度進行審查:
- 代碼質(zhì)量 — 可讀性、命名規(guī)范、代碼風格
- 潛在 Bug — 邏輯錯誤、邊界條件、空指針風險
- 安全漏洞 — 注入攻擊、數(shù)據(jù)泄露、權(quán)限問題
- 性能優(yōu)化 — 時間復雜度、內(nèi)存使用、I/O 效率
- 最佳實踐 — 設計模式、SOLID 原則、錯誤處理
請為每個維度給出 1-5 分評分和具體改進建議。“”"
@prompt()
def explain_code(code: str) -> str:
“”"代碼解釋提示模板
Args:
code: 需要解釋的代碼
"""
return f"""請用通俗易懂的語言解釋以下代碼的功能:
[code]
要求:
- 先用一句話概括整體功能
- 逐段解釋關(guān)鍵邏輯
- 說明輸入和輸出
- 指出可能的使用場景"“”
### 4.6 組裝完整 Server
```python
# server.py
from mcp.server.fastmcp import FastMCP
mcp = FastMCP(
name="enterprise-toolkit",
version="1.0.0",
description="企業(yè)級 MCP 工具服務：天氣查詢、數(shù)據(jù)庫操作、文件管理",
)
# ── 注冊 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 中配置
在項目根目錄創(chuàng)建 .cursor/mcp.json：
{
"mcpServers": {
"enterprise-toolkit": {
"command": "python",
"args": ["./my-mcp-server/server.py"]
}
}
}5.3 調(diào)試流程
┌─────────────────────────────────────────────────────────┐ │ MCP 調(diào)試流程 │ │ │ │ 1. 啟動 Inspector │ │ $ mcp dev server.py │ │ │ │ │ ▼ │ │ 2. 瀏覽器打開 http://localhost:5173 │ │ │ │ │ ▼ │ │ 3. 查看已注冊的 Tools / Resources / Prompts │ │ ┌─────────────────────────────────────┐ │ │ │ Tools: │ │ │ │ ? get_weather │ │ │ │ ? get_forecast │ │ │ │ ? query_database │ │ │ │ ? read_file │ │ │ │ ? write_file │ │ │ │ ? list_files │ │ │ │ │ │ │ │ Resources: │ │ │ │ ? system://info │ │ │ │ ? system://processes │ │ │ │ │ │ │ │ Prompts: │ │ │ │ ? code_review │ │ │ │ ? explain_code │ │ │ └─────────────────────────────────────┘ │ │ │ │ │ ▼ │ │ 4. 點擊工具 → 填入?yún)?shù) → 點擊 Run → 查看返回結(jié)果 │ │ │ │ │ ▼ │ │ 5. 確認無誤后，配置到 AI 客戶端正式使用 │ └─────────────────────────────────────────────────────────┘
六、進階：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-stopped6.3 客戶端配置（遠程 SSE）
{
"mcpServers": {
"enterprise-toolkit": {
"url": "http://your-server:8080/sse"
}
}
}七、MCP 協(xié)議消息流詳解
當用戶在 Claude 中說 “幫我查一下北京天氣” 時，完整的消息流如下：
用戶: "幫我查一下北京天氣"
│
▼
┌──────────┐
│ Claude │ 1. 分析意圖，決定調(diào)用 tool: get_weather
│ (Host) │
└────┬─────┘
│ MCP Request (JSON-RPC)
│ {
│ "jsonrpc": "2.0",
│ "method": "tools/call",
│ "params": {
│ "name": "get_weather",
│ "arguments": {
│ "city": "北京"
│ }
│ }
│ }
▼
┌──────────┐
│MCP Server│ 2. 執(zhí)行 get_weather("北京")
│ (Python) │
└────┬─────┘
│ MCP Response
│ {
│ "jsonrpc": "2.0",
│ "result": {
│ "content": [{
│ "type": "text",
│ "text": "【北京 天氣】\n天氣: 晴\n溫度: 22°C..."
│ }]
│ }
│ }
▼
┌──────────┐
│ Claude │ 3. 結(jié)合工具返回結(jié)果，生成自然語言回答
│ (Host) │
└────┬─────┘
│
▼
用戶看到: "北京今天天氣晴朗，氣溫 22°C，濕度 35%，風速 3 m/s。"
八、最佳實踐與安全注意事項
8.1 工具設計原則
┌───────────────────────────────────────────────────┐ │ MCP 工具設計原則 │ ├───────────────────────────────────────────────────┤ │ │ │ 1. 單一職責 ─── 每個工具只做一件事 │ │ │ │ 2. 清晰描述 ─── docstring 是 AI 理解工具的關(guān)鍵 │ │ │ │ 3. 類型標注 ─── 參數(shù)必須有明確的類型和默認值 │ │ │ │ 4. 安全沙箱 ─── 限制文件/數(shù)據(jù)庫訪問范圍 │ │ │ │ 5. 錯誤友好 ─── 返回清晰的錯誤信息，而非拋異常 │ │ │ │ 6. 冪等設計 ─── 相同輸入應產(chǎn)生相同輸出 │ │ │ └───────────────────────────────────────────────────┘
8.2 安全清單
# 安全示例：參數(shù)驗證與路徑限制
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)
九、總結(jié)
本文從零到一完成了一個完整的 MCP Server 開發(fā)，關(guān)鍵知識點回顧：
| 知識點 | 說明 |
|---|---|
| FastMCP | Python MCP Server 開發(fā)的核心框架 |
| @tool() | 裝飾器注冊 AI 可調(diào)用的工具函數(shù) |
| @resource() | 裝飾器注冊 AI 可讀取的數(shù)據(jù)資源 |
| @prompt() | 裝飾器注冊可復用的提示模板 |
| stdio 傳輸 | 本地開發(fā)首選，AI 應用直接啟動子進程 |
| SSE 傳輸 | 遠程部署方案，通過 HTTP 暴露服務 |
| MCP Inspector | 官方可視化調(diào)試工具，mcp dev 啟動 |
下一步探索
- 將 MCP Server 接入企業(yè)內(nèi)部 API（Jira、Confluence、飛書）
- 開發(fā)多用戶權(quán)限管理的 MCP 中間件
- 結(jié)合 RAG 構(gòu)建 MCP 知識庫服務
- 參與開源 MCP 生態(tài)，發(fā)布你的工具到 MCP Hub
MCP 協(xié)議正在快速成為 AI 工具調(diào)用的行業(yè)標準，掌握它意味著你能讓 AI 真正融入你的工作流。
以上就是使用Python開發(fā)你的第一個AI工具服務MCP Server的詳細內(nèi)容，更多關(guān)于Python開發(fā)AI工具MCP Server的資料請關(guān)注腳本之家其它相關(guān)文章！
相關(guān)文章
本文主要介紹了python實現(xiàn)春晚魔術(shù)的示例代碼,文中通過示例代碼介紹的非常詳細,對大家的學習或者工作具有一定的參考學習價值,需要的朋友們下面隨著小編來一起學習學習吧2025-02-02
這篇文章主要給大家介紹了關(guān)于利用Python過濾相似文本的簡單方法，文中通過示例代碼介紹的非常詳細，對大家的學習或者工作具有一定的參考學習價值，需要的朋友們下面隨著小編來一起學習學習吧2021-02-02
在Python編程中,嵌套字典是一種常見的數(shù)據(jù)結(jié)構(gòu),它可以以層次結(jié)構(gòu)的方式組織和存儲數(shù)據(jù),本文將詳細介紹如何在Python中訪問和操作嵌套字典,需要的可以參考下2024-02-02
Python控制臺輸出時刷新當前行內(nèi)容而不是輸出新行的實現(xiàn)
今天小編就為大家分享一篇Python控制臺輸出時刷新當前行內(nèi)容而不是輸出新行的實現(xiàn)，具有很好的參考價值，希望對大家有所幫助。一起跟隨小編過來看看吧2020-02-02
如何使用Python實現(xiàn)數(shù)據(jù)透視表、音頻文件格式轉(zhuǎn)換
這篇文章主要介紹了用Python實現(xiàn)數(shù)據(jù)透視表、音頻文件格式轉(zhuǎn)換,本文通過實例代碼給大家介紹的非常詳細，對大家的學習或工作具有一定的參考借鑒價值，需要的朋友可以參考下2023-10-10
這篇文章主要介紹了Python的加密模塊md5、sha、crypt使用實例,本文給出了MD5和crypt模塊的代碼實例,需要的朋友可以參考下2014-09-09
這篇文章主要介紹了Python實現(xiàn)的歸并排序算法,簡單描述了歸并排序算法的原理,并結(jié)合實例形式分析了Python實現(xiàn)歸并排序的具體操作技巧,需要的朋友可以參考下2017-11-11
Python實現(xiàn)代碼統(tǒng)計工具（終極篇）
這篇文章主要介紹了Python實現(xiàn)代碼統(tǒng)計工具的相關(guān)資料，供大家參考，感興趣的小伙伴們可以參考一下2016-07-07
對于Python程序員來說,處理彈出窗口似乎并不是一個常見的任務,這篇文章主要給大家介紹了關(guān)于Python圖形化界面基礎篇之如何使用彈出窗口和對話框的相關(guān)資料,需要的朋友可以參考下2024-03-03
這篇文章主要介紹了Django 請求Request的具體使用方法，文中通過示例代碼介紹的非常詳細，對大家的學習或者工作具有一定的參考學習價值，需要的朋友們下面隨著小編來一起學習學習吧2019-11-11