基于Python從零構建一個MCP服務器
一、MCP協議:AI與工具的"USB-C接口"
想象你正在用AI助手處理工作:需要查詢天氣時,AI突然彈出"我需要調用天氣API"的提示;處理Excel數據時,它又卡在"如何讀取CSV文件"的步驟。傳統AI的局限性在于,它像一臺沒有外設的電腦——能思考卻無法操作現實世界。
MCP(Model Context Protocol)協議的出現解決了這個痛點。它定義了一套標準化通信機制,讓AI代理(如Claude、Cursor)能動態(tài)發(fā)現并調用服務器上的工具函數。就像給電腦插上USB-C線,AI瞬間獲得了訪問數據庫、調用API、操作文件系統的能力。
核心價值:
- 解耦設計:工具函數與AI模型分離,修改工具不影響模型訓練
- 安全沙箱:通過服務器中轉調用,避免直接暴露API密鑰
- 統一入口:用標準化協議整合分散的工具接口
二、環(huán)境搭建:3分鐘啟動開發(fā)環(huán)境
創(chuàng)建隔離環(huán)境
# 使用uv工具快速初始化項目(推薦) uv init mcp_weather_server && cd mcp_weather_server uv venv && source .venv/bin/activate # Linux/Mac # 或 .venv\Scripts\activate (Windows)
安裝核心依賴
# 基礎版(適合個人開發(fā)) uv add mcp>=1.9.1 requests python-dotenv # 企業(yè)版(需處理高并發(fā)) uv add fastmcp httpx uvicorn[standard]
驗證安裝
# 創(chuàng)建test_install.py
from mcp.server.fastmcp import FastMCP
print("MCP SDK安裝成功" if 'FastMCP' in globals() else "安裝失敗")
運行 python test_install.py 應輸出成功信息。
三、工具開發(fā):讓AI可調用的"魔法函數"
案例1:實時天氣查詢工具
# weather_tool.py
import requests
from mcp.server.fastmcp import tool
from dotenv import load_dotenv
import os
load_dotenv() # 加載環(huán)境變量
@tool()
def get_weather(city: str) -> dict:
"""獲取城市實時天氣(需配置API_KEY)
Args:
city: 城市名稱(如"北京")
Returns:
包含溫度、濕度、天氣狀況的字典
"""
api_key = os.getenv('WEATHER_API_KEY')
url = f"https://api.openweathermap.org/data/2.5/weather?q={city}&appid={api_key}&units=metric"
try:
response = requests.get(url, timeout=5)
data = response.json()
return {
"temperature": data["main"]["temp"],
"humidity": data["main"]["humidity"],
"condition": data["weather"][0]["description"]
}
except Exception as e:
return {"error": f"天氣查詢失敗: {str(e)}"}
關鍵設計原則:
- 類型注解:用 city: str 明確參數類型
- 錯誤處理:捕獲網絡請求異常
- 文檔字符串:詳細說明參數和返回值
- 環(huán)境變量:敏感信息(如API密鑰)通過 .env 文件管理
案例2:數據庫查詢工具
# db_tool.py
import sqlite3
from mcp.server.fastmcp import tool
from mcp.types import TableContent
@tool()
def query_sales_data(limit: int = 10) -> TableContent:
"""查詢銷售數據表(示例用)
Args:
limit: 返回記錄數上限
Returns:
表格數據(包含列名和行數據)
"""
conn = sqlite3.connect("sales.db")
cursor = conn.cursor()
cursor.execute(f"SELECT * FROM orders LIMIT {limit}")
columns = [desc[0] for desc in cursor.description]
rows = cursor.fetchall()
return TableContent(
type="table",
columns=columns,
rows=rows
)
企業(yè)級優(yōu)化:
- 使用參數化查詢防止SQL注入
- 添加連接池管理數據庫連接
- 通過Nacos動態(tài)配置數據庫地址
四、服務器構建:3種通信模式詳解
模式1:本地開發(fā)模式(stdio)
# server_stdio.py
from mcp.server.fastmcp import FastMCP
from weather_tool import get_weather
from db_tool import query_sales_data
mcp = FastMCP("Local Tools Server")
mcp.add_tool(get_weather)
mcp.add_tool(query_sales_data)
if __name__ == "__main__":
mcp.run(mode="stdio") # 通過標準輸入輸出通信
適用場景:
- 與Claude Desktop/Cursor本地集成
- 快速驗證工具功能
- 開發(fā)階段調試
模式2:云端部署模式(SSE)
# server_sse.py
from fastmcp import FastMCP
from weather_tool import get_weather
import uvicorn
mcp = FastMCP("Cloud Tools Server")
mcp.add_tool(get_weather)
if __name__ == "__main__":
# 通過HTTP事件流通信
uvicorn.run(
mcp.as_app(), # 轉換為FastAPI應用
host="0.0.0.0",
port=8000,
workers=4 # 多進程處理并發(fā)請求
)
關鍵配置:
- 添加JWT認證中間件
- 設置CORS允許前端調用
- 配置Nginx反向代理
模式3:自動化遷移模式(OpenAPI轉換)
# converter.py
import json
from jsonschema import validate
def convert_to_mcp_tool(api_spec):
"""將OpenAPI接口轉換為MCP工具
Args:
api_spec: OpenAPI規(guī)范字典
Returns:
MCP工具代碼模板字符串
"""
tool_template = f"""
@tool()
def {api_spec["operationId"]}(**kwargs):
"""
{api_spec["summary"]}
Parameters:
{json.dumps(api_spec["parameters"], indent=4)}
"""
# 這里添加實際調用邏輯
return盛{{}}
"""
return tool_template
# 示例:轉換天氣API
weather_api = {
"operationId": "get_weather",
"summary": "獲取城市天氣信息",
"parameters": [
{"name": "city", "in": "query", "required": True, "type": "string"}
]
}
print(convert_to_mcp_tool(weather_api))
企業(yè)應用:
- 批量遷移現有RESTful API
- 生成標準化工具文檔
- 與Higress網關集成實現協議轉換
五、AI集成:讓Claude調用你的工具
1.配置Claude Desktop
編輯配置文件 claude_desktop_config.json:
{
"mcpServers": {
"weather_service": {
"command": "python",
"args": ["server_stdio.py", "--mode", "stdio"]
}
}
}
2.自然語言調用示例
當用戶在Claude中輸入:
"查詢北京今天的天氣"
AI內部處理流程:
發(fā)現可用的 get_weather 工具
將自然語言轉換為工具調用:
{
"toolName": "get_weather",
"arguments": {"city": "北京"}
}
執(zhí)行工具并返回結果:
{
"temperature": 28,
"humidity": 65,
"condition": "多云"
}
將結果生成自然語言回復:
"北京今天氣溫28℃,濕度65%,天氣多云"
3.調試技巧
使用MCP Inspector可視化調試:
npx @modelcontextprotocol/inspector python server_stdio.py
瀏覽器打開 http://localhost:3000 可實時查看:
- 工具注冊情況
- 請求/響應數據流
- 調用耗時統計
六、安全與性能優(yōu)化
安全防護三板斧
敏感信息隔離:
# 使用Vault管理密鑰
from hvac import Client
def get_secret(path: str) -> str:
client = Client(url="http://vault-server:8200")
return client.secrets.kv.v2.read_secret_version(path=path)["data"]["data"]["key"]
參數校驗:
from pydantic import BaseModel, conint
class QueryParams(BaseModel):
limit: conint(ge=1, le=100) # 限制1-100的整數
@tool()
def safe_query(params: QueryParams) -> dict:
validated = params.dict() # 自動校驗參數
# 執(zhí)行查詢邏輯...
執(zhí)行超時控制:
from functools import partial
import asyncio
async def timed_fetch(url: str, timeout: float = 5.0):
try:
return await asyncio.wait_for(requests.get(url), timeout=timeout)
except asyncio.TimeoutError:
return {"error": "請求超時"}
性能優(yōu)化方案
緩存策略:
from functools import lru_cache
@lru_cache(maxsize=100)
@tool()
def cached_weather(city: str) -> dict:
# 首次調用執(zhí)行實際查詢
# 后續(xù)調用直接返回緩存結果
return _real_weather_query(city)
異步處理:
@tool()
async def async_data_processing(file_url: str):
async with httpx.AsyncClient() as client:
file_content = await client.get(file_url)
# 并行處理數據...
水平擴展:
# 使用Gunicorn部署FastMCP應用 gunicorn server_sse:app -w 8 -k uvicorn.workers.UvicornWorker
七、完整項目示例:股票數據查詢系統
項目結構
stock_mcp/
├── .env # API密鑰配置
├── tools/
│ ├── __init__.py
│ ├── stock_api.py # 股票查詢工具
│ └── data_analysis.py # 數據分析工具
├── server.py # 服務器入口
└── requirements.txt # 依賴列表
核心代碼實現
# tools/stock_api.py
import httpx
from mcp.server.fastmcp import tool
from dotenv import load_dotenv
import os
load_dotenv()
@tool()
def get_stock_price(symbol: str) -> dict:
"""獲取股票實時價格(Alpha Vantage API)
Args:
symbol: 股票代碼(如"AAPL")
Returns:
包含價格、漲跌幅的字典
"""
api_key = os.getenv('ALPHA_VANTAGE_KEY')
url = f"https://www.alphavantage.co/query?function=GLOBAL_QUOTE&symbol={symbol}&apikey={api_key}"
response = httpx.get(url)
data = response.json()
if "Global Quote" not in data:
return {"error": "未找到股票數據"}
quote = data["Global Quote"]
return {
"symbol": symbol,
"price": float(quote["05. price"]),
"change": float(quote["09. change"]),
"change_percent": float(quote["10. change percent"].strip('%'))
}
# server.py
from fastmcp import FastMCP
from tools.stock_api import get_stock_price
from tools.data_analysis import calculate_stats
mcp = FastMCP("Stock Analysis Server")
mcp.add_tool(get_stock_price)
mcp.add_tool(calculate_stats)
if __name__ == "__main__":
mcp.run(transport="streamable-http", host="0.0.0.0", port=8000)
部署與測試
# 安裝依賴
uv install -r requirements.txt
# 啟動服務器
python server.py
# 測試工具調用
curl -X POST http://localhost:8000/mcp/invoke \
-H "Content-Type: application/json" \
-d '{"toolName": "get_stock_price", "arguments": {"symbol": "AAPL"}}'
八、未來展望:MCP生態(tài)演進方向
- 工具市場:類似PyPI的MCP工具倉庫,實現"一鍵安裝"工具
- 協議擴展:支持gRPC、WebSocket等更多通信協議
- 智能路由:根據請求內容自動選擇最優(yōu)工具
- 邊緣計算:在IoT設備上部署輕量級MCP服務器
MCP協議正在重塑AI開發(fā)范式——它讓大模型從"封閉的大腦"進化為"可連接萬物的神經系統"。無論是個人開發(fā)者快速擴展AI能力,還是企業(yè)整合遺留系統,MCP都提供了標準化解決方案。當工具調用變得像呼吸一樣自然,AI才能真正成為生產力的延伸。
以上就是基于Python從零構建一個MCP服務器的詳細內容,更多關于Python MCP服務器的資料請關注腳本之家其它相關文章!

