3步构建Python MCP服务器，轻松与数据对话

模型上下文协议（Model Context Protocol，简称 MCP）正在彻底改变我们与人工智能的互动方式，它使得从数据科学家到普通爱好者在内的所有人，都能让 AI 助手无缝连接真实数据，从而大幅提升其能力。例如，你可以像与同事自然对话一样，轻松地要求 AI 深入复杂的数据集、揭示隐藏模式、进行数据运算，甚至从网络获取实时市场趋势。MCP 充当一个通用桥梁，将现代 AI 模型的强大能力与动态多变的数据世界无缝连接起来。无论是分析销售数据、预测趋势，还是挖掘实时洞察，MCP 都能让 AI 成为你的强大智能数据伙伴，它不仅能对话，还能深入理解并处理你关注的信息。这项协议正在开启一个智能的、上下文感知的助手新时代，它们不仅能生成文本，还能进行推理、计算，并提供量身定制的、切实可行的结果。

什么是 MCP 服务器？

MCP 服务器就像是 AI 模型（如 Claude 或 GPT）与你的数据或工具之间的桥梁。它通过一个标准化协议，让 AI 能够访问你的数据集、执行计算或获取外部数据。你可以将其想象成 AI 的 USB-C 接口，它将你的数据科学工作流与 AI 连接起来，使其能力远超简单的文本生成。

对于数据科学家而言，MCP 服务器的功能非常实用，例如：

• • 读取并总结数据集（例如，CSV 文件）。
• • 计算均值、中位数或标准差等统计数据。
• • 从 API 获取实时数据（例如，股票价格或天气）。
• • 根据你的数据生成洞察或可视化。

在本文中，我们将构建一个 MCP 服务器，让你能够与 CSV 数据集对话、计算基本统计数据，并利用 Alpha Vantage API 获取股票市场数据。届时，你将能够提出诸如“我的数据集中平均销售额是多少？”或“苹果公司最新的股票价格是多少？”之类的问题。

项目概览：我们将构建什么

我们将创建一个基于 Python 的 MCP 服务器，具备以下功能：

1. 1. 读取包含销售数据的 CSV 文件。
2. 2. 提供计算统计数据（均值、中位数、标准差）的工具。
3. 3. 使用 Alpha Vantage API 获取实时股票价格。
4. 4. 连接到 Claude Desktop，让你能够与你的数据进行对话。

以下是其高层级工作流程图：

前提条件

要构建这个 MCP 服务器，你需要准备以下工具：

为何选择这些工具？

• • uv：一个现代、快速的包管理器，简化了虚拟环境和依赖项的管理。
• • pandas：在数据科学领域，处理 CSV 等表格数据的理想选择。
• • httpx：一个功能强大的库，用于向 API 发送 HTTP 请求。
• • Claude Desktop：一个用户友好的 AI 客户端，支持 MCP，让你能够轻松测试你的服务器。

设置开发环境

为了保持项目整洁并避免潜在冲突，让我们来设置一个干净独立的开发环境。

步骤 1：安装 Python

请确保系统已安装 Python 3.10 或更高版本，可通过以下命令检查：

python --version

如果未安装，可以从 python.org下载。

步骤 2：安装 uv

uv 是一个轻量级的 Python 包管理器，可以通过以下方式安装：

• • Mac/Linux：

  curl -LsSf https://astral.sh/uv/install.sh | sh

• • Windows：

  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 密钥

在 Alpha Vantage (https://www.alphavantage.co/) 注册以获取免费 API 密钥（每日 25 次请求）。将其保存到项目根目录下的 .env 文件中：

ALPHA_VANTAGE_API_KEY=your_api_key_here

步骤 6：准备一个示例数据集

创建一个名为 data 的文件夹，并在其中添加一个名为 sales_data.csv 的示例 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 官网(https://www.anthropic.com/) 下载并安装 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"数据集有 {rows} 行和 {cols} 列。列名：{columns}"

# 工具 2：计算列的均值
@mcp.tool()
def compute_mean(column: str) -> float:
    """计算数字列的均值。"""
    if column not in df.columns:
        return f"错误：未找到列 '{column}'。"
    if not pd.api.types.is_numeric_dtype(df[column]):
        return f"错误：列 '{column}' 不是数字类型。"
    return float(df[column].mean())

# 工具 3：计算列的中位数
@mcp.tool()
def compute_median(column: str) -> float:
    """计算数字列的中位数。"""
    if column not in df.columns:
        return f"错误：未找到列 '{column}'。"
    if not pd.api.types.is_numeric_dtype(df[column]):
        return f"错误：列 '{column}' 不是数字类型。"
    


    
return float(df[column].median())

# 工具 4：计算列的标准差
@mcp.tool()
def compute_std(column: str) -> float:
    """计算数字列的标准差。"""
    if column not in df.columns:
        return f"错误：未找到列 '{column}'。"
    if not pd.api.types.is_numeric_dtype(df[column]):
        return f"错误：列 '{column}' 不是数字类型。"
    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?function=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 indata:
                return f"错误：未找到股票代码 '{symbol}' 的数据。"
            latest_time = list(data["Time Series (5min)"].keys())[0]
            price = data["Time Series (5min)"][latest_time]["4. close"]
            return f"股票代码 {symbol} 的最新价格：${price}"
        except httpx.HTTPError:
            return f"错误：无法获取股票代码 '{symbol}' 的价格。"

# 运行服务器
if __name__ == "__main__":
    mcp.run(transport="stdio")

代码解释

• • 导入（Imports）：我们导入 pandas 用于数据处理，httpx 用于 API 调用，FastMCP 用于 MCP 服务器，以及 dotenv 用于环境变量。
• • 服务器初始化（Server Initialization）：FastMCP("data_science_server") 创建了名为“data_science_server”的 MCP 服务器。
• • 数据集加载（Dataset Loading）：我们将 sales_data.csv 加载到一个 pandas DataFrame 中。
• • 工具（Tools）：
• • summarize_dataset：返回数据集的行数、列数和列名。
• • compute_mean、compute_median、compute_std：计算数字列（例如 price 或 quantity_sold）的统计数据。
• • get_stock_price：一个异步工具，使用 Alpha Vantage API 获取实时股票价格。
• • 运行服务器（Running the Server）：mcp.run(transport="stdio") 启动服务器，适用于本地测试。

步骤 2：配置 Claude Desktop

要将服务器连接到 Claude Desktop，请按照以下步骤操作：

1. 1. 在其数据目录中找到 Claude Desktop 的配置文件（claude_desktop_config.json）（具体路径请查阅 Anthropic 的文档）。
2. 2. 添加你的服务器配置：

   {
     "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。你应该能在“可用工具”下看到你的服务器。如果它显示一个绿色指示灯（🟢），则表示服务器已成功连接。

与数据对话

现在，让我们通过 Claude Desktop 与服务器进行互动。以下是一些查询示例及其预期输出：

对话工作流

1. 1. 启动 Claude Desktop：确保服务器正在运行。
2. 2. 查询数据集：

• • 输入：“总结我的数据集。”
• • Claude Desktop 将请求发送到 MCP 服务器，服务器会调用 summarize_dataset 工具。
• • 输出：“数据集有 4 行和 4 列。列名：product, category, price, quantity_sold。”

• • 输入：“平均价格是多少？”
• • 服务器调用 compute_mean("price") 工具。
• • 输出：“392.49。”

• • 输入：“获取 AAPL 的最新股票价格。”
• • 服务器调用 get_stock_price("AAPL") 工具，并从 Alpha Vantage 获取数据。
• • 输出：“股票代码 AAPL 的最新价格：$150.25”（实际价格会因时而异）。

Claude Desktop 中的示例输出

User: 总结我的数据集
Claude: 数据集有 4 行和 4 列。列名：product, category, price, quantity_sold

User: 平均价格是多少？
Claude: 392.49

User: 获取 AAPL 的最新股票价格
Claude: 股票代码 AAPL 的最新价格：$150.25

错误处理和最佳实践

• • 输入验证：compute_* 系列工具会检查列是否存在且是否为数字类型，如果不是，则返回清晰的错误消息。
• • API 错误处理：get_stock_price 工具能够妥善处理 HTTP 错误和无效的股票代码。
• • 日志记录：在 STDIO 模式下，请避免直接向 stdout 打印内容，以免干扰 JSON-RPC 消息。应改用 logging 模块：

  import logging
  logging.info("正在处理请求")

• • 安全性：API 密钥应存储在  .env 文件中，而不是直接写入代码。在生产环境中，请使用 HTTPS 和认证机制（例如，在请求头中传递 API 密钥）。

扩展服务器功能

以下是几种提升服务器功能的方式：

• • 更多工具：添加数据可视化工具（例如，使用 matplotlib 生成图表）或高级统计工具（例如，计算相关性）。
• • 数据库支持：将 CSV 文件替换为 SQLite 或 PostgreSQL 数据库，以支持更大的数据集。
• • 实时流媒体：使用 WebSockets 流式传输实时股票价格。
• • 认证：添加 API 密钥验证以实现安全访问。

示例：添加一个相关性计算工具：

@mcp.tool ()
def compute_correlation(col1: str, col2: str) -> float:
    """计算两个数字列之间的相关性。"""
    if col1 not in df.columns or col2 not in df.columns:
        return f"错误：未找到一个或两个列。"
    if not (pd.api.types.is_numeric_dtype(df[col1]) and pd.api.types.is_numeric_dtype(df[col2])):
        return f"错误：一个或两个列不是数字类型。"
    return float(df[col1].corr(df[col2]));

查询示例：“价格和销售数量之间的相关性是多少？” 输出：“-0.95”（示例，基于数据集）。

故障排除

以下是常见问题及其解决方法：

结论

本文指导你如何使用 Python 构建一个功能齐全的 MCP 服务器，让你能够与数据进行对话。这个服务器可以读取 CSV 数据集，计算统计数据，并获取实时股票价格，所有这些都与 Claude Desktop 进行了集成。通过遵循 MCP 标准，你的服务器可以与任何兼容 MCP 的客户端即插即用，使其成为数据科学工作流的强大工具。

资源：

• • 模型上下文协议（Model Context Protocol）文档(https://www.anthropic.com/)
• • Alpha Vantage API(https://www.alphavantage.co/)
• • FastMCP GitHub(https://github.com/anthropic/mcp-server-python)