30 分钟搭一个数据分析的自定义 MCP 服务器

TL;DR — 自定义 MCP（Model Context Protocol）服务器把你的数据变成 Agent 可调用的工具。本文用 TypeScript 搭一个：加载 CSV、跑聚合、把结果交给任意 MCP 客户端（Claude Desktop、Cursor、自研 Agent）用。整个东西约 120 行。难点不是协议，而是设计出模型真能正确调用的工具 schema。

你要搭的东西

读完你会得到一个暴露三个工具的 MCP 服务器：load_dataset、query_stats、group_by。Agent 通过 stdio 连上来，问”各地区的平均订单金额是多少？“，服务器做掉 pandas 风格的活儿，把数字递回去。Agent 不需要懂 SQL，它只调工具。

这事重要是因为另一个选择——把 50,000 行的 CSV 倒进 context window——既贵又没用。模型没法在上下文里可靠地对几千行做算术。把计算推给工具，只返回答案。如果你完全没接触过把工具接进 Agent，先看我们的把 MCP 服务器接进 Agent 指南，再回来搭自己的。

为什么用 MCP 而不是直接 function calling

你也可以把这些函数硬编进一个 Agent。MCP 的价值在于服务器和客户端解耦。写一次，Claude Desktop、Cursor、你的 CLI Agent、你的生产应用全都用同一套工具，不用各自重新实现。取舍我们在 MCP vs function calling 里讲过，一句话：当同一个能力要跨多个 Agent 或入口复用时，MCP 赢。

准备

mkdir mcp-data-server && cd mcp-data-server
npm init -y
npm install @modelcontextprotocol/sdk csv-parse zod
npm install -D typescript @types/node tsx

官方 TypeScript SDK 处理 MCP 协议管道。csv-parse 读 CSV。zod 校验工具输入——这部分比看上去重要。

服务器

完整服务器如下。用 stdio transport，这是 Claude Desktop 和大多数本地客户端期望的。

// server.ts
import { McpServer } from "@modelcontextprotocol/sdk/server/mcp.js";
import { StdioServerTransport } from "@modelcontextprotocol/sdk/server/stdio.js";
import { parse } from "csv-parse/sync";
import { readFileSync } from "node:fs";
import { z } from "zod";

// 内存数据集存储。按名字索引，这样 Agent 能加载多个。
const datasets = new Map<string, Record<string, string>[]>();

const server = new McpServer({
  name: "data-analysis",
  version: "1.0.0",
});

// 工具 1：从磁盘加载 CSV 到内存
server.tool(
  "load_dataset",
  "Load a CSV file into memory so it can be queried. Returns row count and column names.",
  {
    name: z.string().describe("A short name to reference this dataset later"),
    path: z.string().describe("Absolute path to the CSV file"),
  },
  async ({ name, path }) => {
    const raw = readFileSync(path, "utf-8");
    const rows = parse(raw, { columns: true, skip_empty_lines: true });
    datasets.set(name, rows);
    const columns = rows.length ? Object.keys(rows[0]) : [];
    return {
      content: [{
        type: "text",
        text: `Loaded "${name}": ${rows.length} rows, columns: ${columns.join(", ")}`,
      }],
    };
  }
);

// 工具 2：单列数值统计
server.tool(
  "query_stats",
  "Compute count, sum, mean, min, and max for a numeric column.",
  {
    dataset: z.string(),
    column: z.string(),
  },
  async ({ dataset, column }) => {
    const rows = datasets.get(dataset);
    if (!rows) throw new Error(`Dataset "${dataset}" not loaded. Call load_dataset first.`);

    const nums = rows
      .map((r) => parseFloat(r[column]))
      .filter((n) => !Number.isNaN(n));

    if (nums.length === 0) throw new Error(`Column "${column}" has no numeric values.`);

    const sum = nums.reduce((a, b) => a + b, 0);
    const stats = {
      count: nums.length,
      sum,
      mean: sum / nums.length,
      min: Math.min(...nums),
      max: Math.max(...nums),
    };
    return { content: [{ type: "text", text: JSON.stringify(stats, null, 2) }] };
  }
);

// 工具 3：分组聚合
server.tool(
  "group_by",
  "Group rows by one column and aggregate (sum or mean) a numeric column.",
  {
    dataset: z.string(),
    groupColumn: z.string(),
    valueColumn: z.string(),
    agg: z.enum(["sum", "mean"]).default("sum"),
  },
  async ({ dataset, groupColumn, valueColumn, agg }) => {
    const rows = datasets.get(dataset);
    if (!rows) throw new Error(`Dataset "${dataset}" not loaded.`);

    const groups = new Map<string, number[]>();
    for (const row of rows) {
      const key = row[groupColumn] ?? "(null)";
      const val = parseFloat(row[valueColumn]);
      if (Number.isNaN(val)) continue;
      if (!groups.has(key)) groups.set(key, []);
      groups.get(key)!.push(val);
    }

    const result: Record<string, number> = {};
    for (const [key, vals] of groups) {
      const sum = vals.reduce((a, b) => a + b, 0);
      result[key] = agg === "mean" ? sum / vals.length : sum;
    }
    return { content: [{ type: "text", text: JSON.stringify(result, null, 2) }] };
  }
);

const transport = new StdioServerTransport();
await server.connect(transport);

运行：

npx tsx server.ts

它不会打印任何东西——stdio 服务器在客户端连上前是静默的。这是正常的。

连接到 Claude Desktop

把这段加到 Claude Desktop 配置（claude_desktop_config.json）：

{
  "mcpServers": {
    "data-analysis": {
      "command": "npx",
      "args": ["tsx", "/absolute/path/to/mcp-data-server/server.ts"]
    }
  }
}

重启 Claude Desktop。现在你可以说：“把 /data/orders.csv 加载为 ‘orders’，然后给我按地区分组的总营收。“Claude 会自动串起 load_dataset 再 group_by。

真正重要的部分：工具描述

上面的代码能跑。但模型能不能正确用它取决于你的描述和 schema，不是你的逻辑。从反复踩坑里学到的三条：

1. 描述工作流，不只是工具本身。 load_dataset 的描述写了 “so it can be queried”。就这一句教会模型其他工具依赖这个先跑。没有它，Agent 会对没加载的数据集调 query_stats 然后撞错误。

2. 错误就是 prompt。 注意错误信息写的是 “Call load_dataset first.”。MCP 错误会被喂回给模型。好的错误信息是一条恢复指令。烂的（“undefined is not a function”）会让 Agent 卡死。

3. 用 enum 约束。 agg 参数是 z.enum(["sum", "mean"])，不是自由字符串。你去掉的每一个自由度，就少一种模型能臆造出非法调用的方式。

对比：Schema 策略

方案	Token 成本	可靠性	何时用
自由格式字符串参数	低	差（模型乱编值）	结构化数据永远不用
Zod enum + 描述	中	高	默认选择
深度嵌套对象 schema	高	中（模型漏字段）	仅在真正需要时

中间这行是甜点区。带 enum 和一行描述的扁平参数给模型足够的结构去成功，又不会让 schema 膨胀——schema 大小直接推高每次调用的 token 成本。

生产加固

120 行的版本是 demo。在它碰真实数据前：

• 路径校验。 用 Agent 控制的路径调 readFileSync 是目录穿越风险。白名单一个数据目录，拒绝目录外的一切。
• 行数限制。 把 2GB 的 CSV 加载进 Map 会把进程 OOM。限制行数，大文件用流式。
• 设计成只读。 这个服务器只读。如果你加写工具，那是另一套威胁模型——隔离如何改变局面见为什么自主 Agent 需要安全沙箱。
• 无状态重启。 内存里的 datasets map 重启就没了。生产环境用真实存储兜底，或按需重载。

FAQ

这个需要 pandas 或数据库吗？

不需要，对中小型 CSV 不需要。示例用纯 JavaScript 做聚合。一旦数据集超过几十万行或你需要 join，就用 DuckDB 或 SQLite 替代内存数组兜底。工具接口完全不变，只换实现。

多个 Agent 能共享一个 MCP 服务器吗？

stdio 下不能，每个客户端会派生自己的进程。要共享访问，改用 Streamable HTTP transport，把服务器跑成长驻服务。工具定义不变，只换 transport。

这跟直接给 Agent SQL 权限有什么区别？

直接给 SQL 权限意味着 Agent 自己写查询，可能写错或写出破坏性的。MCP 工具约束了可能性：group_by 只能分组聚合，没法 DROP TABLE。你用灵活性换安全和可预测。

为什么用 TypeScript 不用 Python？

两者都有官方 SDK。TypeScript 作为单进程交付，好打包好分发。如果你的分析已经在 pandas 里，Python 更好。协议完全一样，选你数据工具链已经在用的语言。

关键要点

• MCP 服务器把数据操作变成 Agent 可调用的工具，把原始数据挡在 context window 之外——模型在上下文里没法可靠计算。
• 协议是简单的部分。工具描述、enum 约束的 schema、面向恢复的错误信息，决定了模型能不能正确调用你的工具。
• 本地用从 stdio + 内存存储起步，需要共享访问或更大数据时切到 Streamable HTTP + DuckDB/SQLite。
• 校验每一个 Agent 控制的输入。对任意路径的 readFileSync 是安全漏洞，不是功能。