MCP 工具测试
项目结构概览
1
2
3
4
5
6
7
8
|
mcp-data-analyst/
├── .venv/ # 虚拟环境
├── data/ # 存放待分析的 CSV 文件（需手动创建）
├── store.py # 数据存储层
├── mcp_server.py # MCP 工具定义
├── app.py # FastAPI + MCP 挂载
├── requirements.txt # 依赖：mcp[cli], fastapi, uvicorn, pandas
└── data-analysis-skill.md # Skill 指南
|
第一步：环境准备
1
2
3
4
5
6
7
8
9
10
11
12
13
|
# 创建项目目录
mkdir mcp-data-analyst && cd mcp-data-analyst
# 创建虚拟环境
python -m venv .venv
# Windows:
.venv\Scripts\activate
# macOS/Linux:
source .venv/bin/activate
pip install "mcp[cli]" fastapi uvicorn
pip freeze > requirements.txt
pip install -r requirements.txt
|
官方 SDK 的核心是 mcp 包
第二步：创建数据存储层 (store.py)
先准备一个简单的数据存储，用来管理待分析的数据文件：
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
|
# store.py
import os
import pandas as pd
from pathlib import Path
from typing import Optional, Dict, Any
class DataStore:
"""本地数据存储管理器"""
def __init__(self, data_dir: str = "./data"):
self.data_dir = Path(data_dir)
self.data_dir.mkdir(exist_ok=True)
self._datasets: Dict[str, pd.DataFrame] = {}
def list_files(self) -> list:
"""列出数据目录下的所有文件"""
return [f.name for f in self.data_dir.iterdir() if f.is_file()]
def load_csv(self, filename: str) -> Optional[Dict[str, Any]]:
"""加载 CSV 文件并返回基本信息"""
filepath = self.data_dir / filename
if not filepath.exists():
return None
df = pd.read_csv(filepath)
self._datasets[filename] = df
return {
"filename": filename,
"rows": len(df),
"columns": list(df.columns),
"preview": df.head(5).to_dict(orient="records"),
}
def execute_query(self, filename: str, query: str) -> Optional[Dict]:
"""在指定数据集上执行 pandas 查询"""
if filename not in self._datasets:
self.load_csv(filename)
df = self._datasets.get(filename)
if df is None:
return None
try:
# 安全执行查询（生产环境需要更严格的沙箱）
result = df.eval(query)
if isinstance(result, pd.Series):
result = result.to_dict()
elif isinstance(result, pd.DataFrame):
result = result.head(100).to_dict(orient="records")
return {"result": result, "success": True}
except Exception as e:
return {"error": str(e), "success": False}
# 全局实例
store = DataStore()
|
第三步：定义 MCP 工具 (mcp_server.py)
这是核心部分——用官方 SDK 的 FastMCP 定义工具
关键点：设置 stateless_http=True 启用生产级无状态模式
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
|
# mcp_server.py
from mcp.server.fastmcp import FastMCP
from store import store
# 创建 MCP 服务器，启用无状态 HTTP 模式（生产部署关键）
mcp = FastMCP("LocalDataAnalyst", stateless_http=True)
# ========== 工具 1：列出数据文件 ==========
@mcp.tool()
async def list_data_files() -> list:
"""列出数据目录下的所有文件"""
return store.list_files()
# ========== 工具 2：加载并预览 CSV ==========
@mcp.tool()
async def preview_csv(filename: str) -> dict:
"""加载 CSV 文件并返回预览（前5行、列名、行数）
Args:
filename: CSV 文件名（需位于 ./data 目录下）
"""
result = store.load_csv(filename)
if result is None:
return {"error": f"文件 {filename} 不存在"}
return result
# ========== 工具 3：执行数据分析查询 ==========
@mcp.tool()
async def analyze_data(filename: str, query: str) -> dict:
"""在已加载的数据集上执行分析查询
Args:
filename: CSV 文件名
query: pandas 查询表达式，例如 "df['sales'].mean()" 或 "df[df['amount'] > 100]"
示例查询:
- "df['revenue'].mean()"
- "df.groupby('category')['sales'].sum()"
- "df[df['quantity'] > 10]"
"""
return store.execute_query(filename, query)
# ========== 工具 4：获取数据摘要统计 ==========
@mcp.tool()
async def get_summary_stats(filename: str, column: str = None) -> dict:
"""获取数据的描述性统计
Args:
filename: CSV 文件名
column: 可选，指定列名；不指定则返回所有数值列的统计
"""
result = store.load_csv(filename)
if result is None:
return {"error": f"文件 {filename} 不存在"}
# 重新获取 DataFrame（store 已缓存）
import pandas as pd
df = pd.read_csv(store.data_dir / filename)
if column:
if column not in df.columns:
return {"error": f"列 {column} 不存在"}
stats = df[column].describe().to_dict()
else:
stats = df.describe().to_dict()
return {"filename": filename, "statistics": stats}
if __name__ == "__main__":
mcp.run(transport="stdio") # stdio 模式用于本地测试
# mcp dev mcp_server.py
|
每个工具函数的文档字符串（docstring）和参数类型注解非常重要——AI 模型会读取它们来理解工具的作用和用法
第四步：Python 原生调试
创建 test_client.py：
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
|
# test_client.py
import asyncio
import json
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client
async def test_tools():
# 连接到你的 MCP 服务器
server_params = StdioServerParameters(
command="python",
args=["mcp_server.py"], # 你的服务器文件
)
async with stdio_client(server_params) as (read, write):
async with ClientSession(read, write) as session:
await session.initialize()
# 列出所有工具
tools = await session.list_tools()
print("可用工具：")
for tool in tools.tools:
print(f" - {tool.name}: {tool.description[:50]}...")
# 测试 list_data_files 工具
print("\n测试 list_data_files：")
result = await session.call_tool("list_data_files", arguments={})
print(f"结果: {result.model_dump_json()}")
print("\n测试 preview_csv")
result = await session.call_tool(
"preview_csv", arguments={"filename": "test.csv"}
)
print(f"结果: {result.model_dump_json()}")
print("\n测试 analyze_data")
result = await session.call_tool(
"analyze_data",
arguments={"filename": "test.csv", "query": "total = sales * quantity"},
)
print(f"结果: {result.model_dump_json()}")
print("\n测试 get_summary_stats")
result = await session.call_tool(
"get_summary_stats", arguments={"filename": "test.csv"}
)
print(f"结果: {result.model_dump_json()}")
if __name__ == "__main__":
asyncio.run(test_tools())
|
第五步：编写 Skill（操作流程指南）
在项目目录下创建 data-analysis-skill.md：
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
|
---
name: local-data-analysis
description: 分析本地 CSV 数据文件
---
# 本地数据分析技能
## 工作流程
1. **列出可用文件**：调用 `list_data_files` 查看数据目录下有哪些 CSV 文件
2. **预览文件**：调用 `preview_csv` 加载文件并查看前 5 行和列信息
3. **理解需求**：根据用户的问题，确定分析目标
4. **执行分析**：
- 简单统计：用 `get_summary_stats` 获取描述性统计
- 自定义分析：用 `analyze_data` 执行 pandas 表达式
5. **解释结果**：用自然语言向用户解释分析结果
## 查询示例
当用户问"销售额最高的前 5 个产品是什么"，你应该：
1. 调用 `preview_csv` 了解数据结构
2. 调用 `analyze_data`，query 参数设为 `"df.nlargest(5, 'sales')[['product', 'sales']]"`
3. 将返回的结果整理成表格或列表
## 注意事项
- 确保查询表达式使用 `df` 作为 DataFrame 变量名
- 复杂查询建议先小范围测试
|
其他
若需要交互式地测试你的工具：
创建 FastAPI 应用并挂载 MCP (app.py)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
|
# app.py
from contextlib import AsyncExitStack, asynccontextmanager
from fastapi import FastAPI
from fastapi.responses import JSONResponse
from mcp_server import mcp
@asynccontextmanager
async def lifespan(app: FastAPI):
"""管理 MCP 会话生命周期"""
async with AsyncExitStack() as stack:
await stack.enter_async_context(mcp.session_manager.run())
yield
app = FastAPI(lifespan=lifespan)
# 将 MCP 服务器挂载到根路径
app.mount("/", mcp.streamable_http_app())
# 健康检查端点（用于容器部署）
@app.get("/health")
async def health():
return JSONResponse({"status": "healthy"})
# 可选：添加一个简单的 Web 界面提示
@app.get("/")
async def root():
return {
"message": "MCP Data Analyst Server is running",
"mcp_endpoint": "/mcp",
"tools": [
"list_data_files",
"preview_csv",
"analyze_data",
"get_summary_stats",
],
}
# uvicorn app:app --reload --port 7891
|
官方 SDK 的 streamable_http_app() 方法返回一个标准的 FastAPI 应用，MCP 端点默认在 /mcp 子路径
-
测试服务器
1
2
|
# 启动服务器
uvicorn app:app --reload --port 8080
|
你应该看到类似这样的输出：
1
2
|
INFO: Uvicorn running on http://127.0.0.1:8080
INFO: Application startup complete.
|
-
用 MCP Inspector 测试（官方推荐）
这会打开一个 Web 界面，你可以：
- 查看所有已注册的工具
- 手动调用工具并查看返回结果
- 调试参数传递
记录生活