MCP開發從入門到實戰，少走99%彎路
模型上下文協議（Model Context Protocol，MCP）正在徹底改變我們與人工智能的交互方式，它讓從數據科學家到業余愛好者的每個人都能輕松地為 AI 助手接入真實世界的數據，從而大幅提升其能力。舉個例子，你可以像和同事聊天一樣與 AI 對話，毫不費力地讓它深入分析復雜數據集、發現隱藏模式、進行數值計算，甚至實時抓取網絡上的市場趨勢——整個過程自然流暢，就像普通對話一樣。MCP 就像一個通用連接器，彌合了現代 AI 模型強大能力與動態變化的數據世界之間的鴻溝。無論是分析銷售數據、預測趨勢，還是探索實時洞察，MCP 都能把 AI 轉變成一個強大的數據伙伴，它不僅能對話，更能真正理解并交互與你關心的重要信息。這個協議正在開啟智能、上下文感知助手的新紀元，它們不再只是生成文本，而是能夠推理、計算，并交付符合你需求的可執行結果。
什么是 MCP 服務器？
MCP 服務器就像一座橋梁，把 AI 模型（如 Claude 或 GPT）與您的數據或工具連接起來。它允許 AI 通過標準化協議訪問數據集、執行計算或獲取外部數據。你可以把它想象成 AI 的 USB-C 接口——它把數據科學的工作流與 AI 連接，使其功能遠超文本生成。
對于數據科學家來說，MCP 服務器可以：
- 讀取并總結數據集（例如 CSV 文件）。
- 計算統計量（如均值、中位數或標準差）。
- 從 API 獲取實時數據（如股票價格或天氣）。
- 基于數據生成洞察或可視化結果。
我們將在本文創建一個基于 Python 的 MCP 服務器，它可以：
- 讀取包含銷售數據的 CSV 文件。
- 提供計算統計量的工具（均值、中位數、標準差）。
- 使用 Alpha Vantage API 獲取實時股票價格。
- 連接到 Claude Desktop，讓你能與數據進行對話。
以下是高層次的工作流圖：
設置開發環境
步驟 1：安裝 Python
確保已安裝 Python 3.10 或更高版本。檢查方式：
python --version步驟 2：安裝 uv
uv 是一個輕量級的 Python 包管理器。安裝方法：
Mac/Linux
curl -LsSf https://astral.sh/uv/install.sh | shWindows
powershell -ExecutionPolicy ByPass -c "irm https://astral.sh/uv/install.ps1 | iex"安裝完成后，重啟終端以確保 ??uv?? 命令可用。
步驟 3：創建項目目錄
創建一個項目文件夾，并設置虛擬環境：
mkdir data_science_mcp
cd data_science_mcp
uv init data_science_mcp
uv venv
source .venv/bin/activate # Mac/Linux
.venv\Scripts\activate # Windows步驟 4：安裝依賴
創建一個 ??requirements.txt?? 文件，列出依賴：
mcp[cli]>=1.4.0
pandas>=2.0.0
httpx>=0.23.0
python-dotenv>=1.0.0安裝依賴：
uv pip install -r requirements.txt步驟 5：獲取 Alpha Vantage API Key
在 Alpha Vantage 注冊并獲取免費的 API key（每天 25 次請求）。將其保存到項目根目錄下的 ??.env?? 文件：
ALPHA_VANTAGE_API_KEY=your_api_key_here步驟 6：準備示例數據集
創建一個 ??data??? 文件夾，并添加一個示例 CSV 文件 ??sales_data.csv??：
product,category,price,quantity_sold
Laptop,Electronics,999.99,50
Phone,Electronics,499.99,120
T-Shirt,Clothing,19.99,200
Jeans,Clothing,49.99,80這個數據集包含了產品的銷售數據，我們將用它來做分析。
步驟 7：安裝 Claude Desktop
從 Anthropic 官網 下載并安裝 Claude Desktop。它支持 macOS 和 Windows（Linux 用戶可以構建自定義 MCP 客戶端）。
環境準備完成！
項目結構應如下
data_science_mcp/
├── .venv/
├── data/
│ └── sales_data.csv
├── .env
├── requirements.txt構建 MCP 服務器
接下來我們來創建 MCP 服務器。我們將使用 FastMCP 庫來定義數據分析和 API 調用的工具。
步驟 1：創建服務器文件
在項目根目錄下創建 ??server.py?? 文件，寫入以下代碼：
import pandas as pd
import httpx
from mcp.server.fastmcp import FastMCP
from dotenv import load_dotenv
import os
# 加載環境變量
load_dotenv()
# 初始化 MCP 服務器
mcp = FastMCP("data_science_server")
# 加載示例數據集
DATA_PATH = "data/sales_data.csv"
df = pd.read_csv(DATA_PATH)
# 工具 1：獲取數據集摘要
@mcp.tool()
def summarize_dataset() -> str:
"""總結數據集：行數、列數和列名。"""
rows, cols = df.shape
columns = ", ".join(df.columns)
return f"Dataset has {rows} rows and {cols} columns. Columns: {columns}"
# 工具 2：計算列均值
@mcp.tool()
def compute_mean(column: str) -> float:
"""計算數值列的均值。"""
if column not in df.columns:
return f"Error: Column '{column}' not found."
if not pd.api.types.is_numeric_dtype(df[column]):
return f"Error: Column '{column}' is not numeric."
return float(df[column].mean())
# 工具 3：計算列中位數
@mcp.tool()
def compute_median(column: str) -> float:
"""計算數值列的中位數。"""
if column not in df.columns:
return f"Error: Column '{column}' not found."
if not pd.api.types.is_numeric_dtype(df[column]):
return f"Error: Column '{column}' is not numeric."
return float(df[column].median())
# 工具 4：計算列標準差
@mcp.tool()
def compute_std(column: str) -> float:
"""計算數值列的標準差。"""
if column not in df.columns:
return f"Error: Column '{column}' not found."
if not pd.api.types.is_numeric_dtype(df[column]):
return f"Error: Column '{column}' is not numeric."
return float(df[column].std())
# 工具 5：獲取股票價格
@mcp.tool()
async def get_stock_price(symbol: str) -> str:
"""獲取指定股票代碼的最新價格。"""
api_key = os.getenv("ALPHA_VANTAGE_API_KEY")
url = f"https://www.alphavantage.co/query?functinotallow=TIME_SERIES_INTRADAY&symbol={symbol}&interval=5min&apikey={api_key}"
async with httpx.AsyncClient() as client:
try:
response = await client.get(url)
response.raise_for_status()
data = response.json()
if "Time Series (5min)" not in data:
return f"Error: No data found for symbol '{symbol}'."
latest_time = list(data["Time Series (5min)"].keys())[0]
price = data["Time Series (5min)"][latest_time]["4. close"]
return f"Latest price for {symbol}: ${price}"
except httpx.HTTPError:
return f"Error: Failed to fetch stock price for '{symbol}'."
# 運行服務器
if __name__ == "__main__":
mcp.run(transport="stdio")代碼解釋
- 導入：使用 pandas 處理數據，httpx 調用 API，FastMCP 創建 MCP 服務器，dotenv 管理環境變量。
- 服務器初始化：FastMCP("data_science_server")創建一個名為 “data_science_server” 的服務器。
- 加載數據集：讀取 sales_data.csv 到 Pandas DataFrame。
- 工具：
summarize_dataset：返回行數、列數和列名。
compute_mean / compute_median / compute_std：計算數值列的基本統計量（如價格或銷量）。
get_stock_price：異步獲取實時股票價格（Alpha Vantage API）。
- 運行服務器：mcp.run(transport="stdio")啟動服務器（STDIO 模式），適合本地與 Claude Desktop 測試。
步驟 2：配置 Claude Desktop
要將服務器連接到 Claude Desktop：
- 找到 Claude Desktop 的配置文件 claude_desktop_config.json（路徑見官方文檔）。
- 添加服務器配置：
{
"mcpServers": {
"data_science_server": {
"command": "/path/to/your/venv/bin/python",
"args": ["/path/to/data_science_mcp/server.py"]
}
}
}將 ??/path/to/your/venv/bin/python??? 和 ??/path/to/data_science_mcp/server.py?? 替換為你本地的實際路徑。
步驟 3：測試服務器
運行服務器：
uv run python server.py打開 Claude Desktop。你應該能在可用工具中看到該服務器。如果顯示綠色指示符（??），說明連接成功。
本文轉載自????????PyTorch研習社????，作者：南七無名士
