Python 实现一个本地 MCP 工具服务器：让 AI 助手读取项目文件、搜索文本并分析 CSV

MCP 正在成为 AI Agent 连接外部工具和数据源的重要协议。本文不做空泛介绍，而是用 Python 实现一个本地 MCP Server，让支持 MCP 的 AI 客户端可以调用本机工具，完成项目文件列表、文本搜索、CSV 数据概览和分组统计等任务。

摘要

大模型本身擅长理解和生成文本，但它默认并不知道本地项目里有哪些文件，也不能直接查询你的 CSV 数据。过去要实现这些能力，通常需要为每个模型或每个应用单独写一套工具调用逻辑。

MCP 的价值在于提供一种标准化方式：开发者把本地能力封装成 MCP Server，AI 客户端通过统一协议发现工具、调用工具、读取结果。

本文会实现一个实用的本地 MCP 工具服务器，包含 5 个工具：

• workspace_info：查看当前工作区信息
• list_files：按规则列出项目文件
• search_text：在项目文件中搜索关键词
• csv_profile：分析 CSV 行数、字段、缺失值和数值统计
• csv_group_sum：按字段对 CSV 做分组求和

这个项目适合用来理解：

• MCP Server 的基本结构
• Tools 的定义方式
• 本地文件访问的安全边界
• AI Agent 如何通过工具获得外部上下文
• 为什么 MCP 比单纯 Function Calling 更适合跨客户端复用

目录

• 一、为什么这个选题值得做
• 二、MCP 到底解决什么问题
• 三、项目效果
• 四、项目目录
• 五、环境准备
• 六、完整代码
• 七、运行和调试
• 八、接入 MCP 客户端的配置思路
• 九、核心代码解析
• 十、安全边界设计
• 十一、常见问题
• 十二、扩展方向
• 十三、总结

一、为什么这个选题值得做

很多大模型应用教程停留在：

调用 API → 输入 Prompt → 输出回答

这个流程可以入门，但离真实 Agent 应用还有距离。真实任务往往需要模型调用外部能力，例如：

在项目里搜索某个函数

读取某个 Markdown 文档并总结

分析 CSV 文件中哪个分类金额最高

根据本地代码结构生成修改建议

这些事情不应该让模型“猜”，而应该让模型通过工具获得真实数据。

MCP 适合做这件事，因为它把工具暴露成标准协议。你写一次 MCP Server，就可以被不同的 MCP 客户端复用，而不是每换一个模型平台就重新写一套工具适配代码。

二、MCP 到底解决什么问题

MCP，全称 Model Context Protocol，可以理解为 AI 应用连接外部系统的标准协议。

一个简化理解：

AI 客户端
↓
MCP Client
↓
MCP Server
↓
本地文件 / 数据库 / API / 脚本工具

MCP Server 可以向客户端暴露三类能力：

类型	作用
Tools	可执行函数，例如搜索文件、查询数据库、调用接口
Resources	可读取资源，例如文件内容、数据库记录、API 响应
Prompts	可复用提示词模板

本文重点实现 Tools。

和普通 Function Calling 相比，MCP 的差异在于：

对比项	Function Calling	MCP
作用范围	通常绑定某个模型 API	面向多个 AI 客户端
工具发现	由应用代码传入 tools	客户端可向 MCP Server 查询工具列表
复用性	每个项目通常单独写	一个 Server 可被多个客户端复用
适合场景	单应用内部工具调用	本地工具、企业系统、数据库、开发工具集成

三、项目效果

本文最终实现一个名为 local-toolbox 的 MCP Server。

它可以完成以下任务：

1. 查看工作区

当前 MCP Server 允许访问哪个目录？
目录下有多少文件？

2. 搜索文件

列出当前项目下所有 .py 文件

3. 搜索文本

在 Markdown 和 Python 文件里搜索 "FastAPI"

4. 分析 CSV

分析 data/orders.csv 的字段、行数、缺失值和数值列统计

5. 分组求和

按 category 统计 sales.csv 中 amount 的总和

这些工具不是玩具功能。对于写代码、整理项目文档、分析小型数据文件，都有实际用途。

四、项目目录

建议新建项目目录：

mcp_local_toolbox
├── mcp_local_toolbox.py
├── requirements.txt
└── data
    └── sales.csv

其中：

• mcp_local_toolbox.py：MCP Server 主程序
• requirements.txt：Python 依赖
• data/sales.csv：测试用 CSV 文件

五、环境准备

建议使用 Python 3.10 或以上版本。

1. 创建项目目录

mkdir C:\Users\用户名\Desktop\mcp_local_toolbox
cd C:\Users\用户名\Desktop\mcp_local_toolbox

2. 创建虚拟环境

python -m venv .venv

激活虚拟环境：

.\.venv\Scripts\Activate.ps1

3. 安装依赖

新建 requirements.txt：

mcp[cli]>=1.2,<2

安装：

python -m pip install -r requirements.txt

这里将 mcp 限制在 1.x 版本，是为了使用当前稳定版 SDK，避免预发布版本接口变化影响代码运行。

4. 准备测试 CSV

新建 data 目录：

mkdir data

新建 data/sales.csv：

order_id,category,city,amount,quantity
O001,电脑配件,上海,199,2
O002,办公用品,广州,86,5
O003,电脑配件,北京,399,1
O004,图书资料,上海,128,3
O005,办公用品,北京,45,10
O006,电脑配件,广州,799,1

六、完整代码

新建 mcp_local_toolbox.py：

import csv
import os
from pathlib import Path
from statistics import mean
from typing import Any

from mcp.server.fastmcp import FastMCP


mcp = FastMCP("local-toolbox")

BASE_DIR = Path(os.getenv("MCP_WORKSPACE", Path.cwd())).resolve()

IGNORED_DIRS = {
    ".git",
    ".venv",
    "__pycache__",
    ".mypy_cache",
    ".pytest_cache",
    "node_modules",
}

TEXT_EXTENSIONS = {
    ".py",
    ".md",
    ".txt",
    ".json",
    ".csv",
    ".yaml",
    ".yml",
    ".toml",
    ".ini",
}


def resolve_safe_path(user_path: str = ".") -> Path:
    target = (BASE_DIR / user_path).resolve()

    if target != BASE_DIR and BASE_DIR not in target.parents:
        raise ValueError(f"只允许访问工作区目录：{BASE_DIR}")

    return target


def should_skip(path: Path) -> bool:
    return any(part in IGNORED_DIRS for part in path.parts)


def to_relative(path: Path) -> str:
    return str(path.relative_to(BASE_DIR)).replace("\\", "/")


def read_text_safely(path: Path, max_bytes: int = 1024 * 1024) -> str:
    if path.stat().st_size > max_bytes:
        raise ValueError(f"文件过大，已跳过：{to_relative(path)}")

    return path.read_text(encoding="utf-8", errors="ignore")


@mcp.tool()
def workspace_info() -> dict[str, Any]:
    """查看当前 MCP Server 允许访问的工作区信息。"""
    file_count = 0

    for path in BASE_DIR.rglob("*"):
        if path.is_file() and not should_skip(path):
            file_count += 1

    return {
        "workspace": str(BASE_DIR),
        "file_count": file_count,
        "ignored_dirs": sorted(IGNORED_DIRS),
    }


@mcp.tool()
def list_files(subdir: str = ".", pattern: str = "*", max_results: int = 50) -> list[dict[str, Any]]:
    """列出工作区中的文件。

    Args:
        subdir: 要搜索的子目录，默认是当前工作区
        pattern: 文件匹配规则，例如 *.py、*.md、*.csv
        max_results: 最多返回多少条结果
    """
    root = resolve_safe_path(subdir)

    if not root.exists():
        raise ValueError(f"目录不存在：{subdir}")

    if not root.is_dir():
        raise ValueError(f"目标不是目录：{subdir}")

    results = []

    for path in root.rglob(pattern):
        if len(results) >= max_results:
            break

        if not path.is_file() or should_skip(path):
            continue

        results.append({
            "path": to_relative(path),
            "size_kb": round(path.stat().st_size / 1024, 2),
            "suffix": path.suffix,
        })

    return results


@mcp.tool()
def search_text(
    keyword: str,
    subdir: str = ".",
    extensions: str = ".py,.md,.txt,.json,.csv",
    max_results: int = 20,
) -> list[dict[str, Any]]:
    """在文本文件中搜索关键词。

    Args:
        keyword: 要搜索的关键词
        subdir: 搜索子目录
        extensions: 逗号分隔的文件后缀，例如 .py,.md
        max_results: 最多返回多少条匹配结果
    """
    if not keyword.strip():
        raise ValueError("keyword 不能为空")

    root = resolve_safe_path(subdir)
    allowed_extensions = {
        item.strip().lower()
        for item in extensions.split(",")
        if item.strip()
    }

    results = []

    for path in root.rglob("*"):
        if len(results) >= max_results:
            break

        if not path.is_file() or should_skip(path):
            continue

        if path.suffix.lower() not in allowed_extensions:
            continue

        if path.suffix.lower() not in TEXT_EXTENSIONS:
            continue

        try:
            content = read_text_safely(path)
        except ValueError:
            continue

        for line_number, line in enumerate(content.splitlines(), start=1):
            if keyword.lower() in line.lower():
                results.append({
                    "path": to_relative(path),
                    "line": line_number,
                    "preview": line.strip()[:200],
                })

                if len(results) >= max_results:
                    break

    return results


@mcp.tool()
def csv_profile(file_path: str, max_rows: int = 5000) -> dict[str, Any]:
    """分析 CSV 文件的字段、行数、缺失值和数值列统计。

    Args:
        file_path: CSV 文件路径，必须位于工作区内
        max_rows: 最多读取多少行，避免大文件拖慢工具
    """
    path = resolve_safe_path(file_path)

    if not path.exists():
        raise ValueError(f"文件不存在：{file_path}")

    if path.suffix.lower() != ".csv":
        raise ValueError("只支持 CSV 文件")

    rows = []

    with path.open("r", encoding="utf-8-sig", newline="") as file:
        reader = csv.DictReader(file)
        columns = reader.fieldnames or []

        for index, row in enumerate(reader):
            if index >= max_rows:
                break
            rows.append(row)

    null_counts = {column: 0 for column in columns}
    numeric_values: dict[str, list[float]] = {column: [] for column in columns}

    for row in rows:
        for column in columns:
            value = (row.get(column) or "").strip()

            if value == "":
                null_counts[column] += 1
                continue

            try:
                numeric_values[column].append(float(value))
            except ValueError:
                pass

    numeric_stats = {}
    for column, values in numeric_values.items():
        if not values:
            continue

        numeric_stats[column] = {
            "count": len(values),
            "min": min(values),
            "max": max(values),
            "mean": round(mean(values), 4),
            "sum": round(sum(values), 4),
        }

    return {
        "file": to_relative(path),
        "columns": columns,
        "row_count_read": len(rows),
        "null_counts": null_counts,
        "numeric_stats": numeric_stats,
    }


@mcp.tool()
def csv_group_sum(
    file_path: str,
    group_by: str,
    value_column: str,
    max_groups: int = 20,
) -> list[dict[str, Any]]:
    """对 CSV 文件按某个字段分组，并对数值列求和。

    Args:
        file_path: CSV 文件路径，必须位于工作区内
        group_by: 分组字段名
        value_column: 要求和的数值字段名
        max_groups: 最多返回多少个分组
    """
    path = resolve_safe_path(file_path)

    if not path.exists():
        raise ValueError(f"文件不存在：{file_path}")

    groups: dict[str, float] = {}

    with path.open("r", encoding="utf-8-sig", newline="") as file:
        reader = csv.DictReader(file)
        columns = reader.fieldnames or []

        if group_by not in columns:
            raise ValueError(f"分组字段不存在：{group_by}")

        if value_column not in columns:
            raise ValueError(f"数值字段不存在：{value_column}")

        for row in reader:
            key = (row.get(group_by) or "").strip() or "空值"
            raw_value = (row.get(value_column) or "").strip()

            try:
                value = float(raw_value)
            except ValueError:
                value = 0

            groups[key] = groups.get(key, 0) + value

    sorted_groups = sorted(groups.items(), key=lambda item: item[1], reverse=True)

    return [
        {
            group_by: key,
            f"{value_column}_sum": round(value, 4),
        }
        for key, value in sorted_groups[:max_groups]
    ]


def main():
    mcp.run(transport="stdio")


if __name__ == "__main__":
    main()

七、运行和调试

1. 直接启动 MCP Server

在项目目录下执行：

python mcp_local_toolbox.py

如果没有任何输出，不一定是错误。

因为这个 MCP Server 使用的是 stdio 通信方式，正常情况下它会等待 MCP 客户端通过标准输入输出和它通信。

注意：
对于 stdio 类型的 MCP Server，不要随便在代码里使用 print() 输出调试信息。标准输出会被 MCP 协议占用，乱输出可能破坏 JSON-RPC 消息。

2. 使用 MCP Inspector 调试

MCP Inspector 是官方提供的交互式调试工具，可以查看工具列表、输入参数并执行工具。

如果本机已经安装 Node.js，可以运行：

npx @modelcontextprotocol/inspector python mcp_local_toolbox.py

如果提示确认安装，输入 y。

打开 Inspector 页面后，可以在 Tools 里看到：

workspace_info
list_files
search_text
csv_profile
csv_group_sum

可以依次测试：

{}

调用 workspace_info。

调用 list_files：

{
  "subdir": ".",
  "pattern": "*.py",
  "max_results": 20
}

调用 search_text：

{
  "keyword": "mcp",
  "subdir": ".",
  "extensions": ".py,.md,.txt",
  "max_results": 10
}

调用 csv_profile：

{
  "file_path": "data/sales.csv",
  "max_rows": 5000
}

调用 csv_group_sum：

{
  "file_path": "data/sales.csv",
  "group_by": "category",
  "value_column": "amount",
  "max_groups": 10
}

预期返回：

[
  {
    "category": "电脑配件",
    "amount_sum": 1397.0
  },
  {
    "category": "办公用品",
    "amount_sum": 131.0
  },
  {
    "category": "图书资料",
    "amount_sum": 128.0
  }
]

八、接入 MCP 客户端的配置思路

不同 MCP 客户端的配置格式略有差异，但核心思路一致：

{
  "mcpServers": {
    "local-toolbox": {
      "command": "python",
      "args": [
        "C:/Users/mobai/Desktop/mcp_local_toolbox/mcp_local_toolbox.py"
      ],
      "env": {
        "MCP_WORKSPACE": "C:/Users/mobai/Desktop/mcp_local_toolbox"
      }
    }
  }
}

这里有两个关键点：

1. command 和 args 告诉客户端如何启动 MCP Server。
2. MCP_WORKSPACE 限定工具服务器能访问的工作区目录。

如果希望访问另一个项目，只需要把 MCP_WORKSPACE 改成对应目录。

例如：

"MCP_WORKSPACE": "D:/projects/my_python_project"

不要直接把整个 C 盘暴露给工具服务器。MCP 工具越强，越要注意访问边界。

九、核心代码解析

1. 创建 MCP Server

mcp = FastMCP("local-toolbox")

FastMCP 会根据 Python 函数的类型注解和 docstring 自动生成工具定义。也就是说，开发者不需要手写复杂的 JSON Schema。

2. 使用装饰器注册工具

@mcp.tool()
def list_files(...):
    ...

被 @mcp.tool() 标记的函数会暴露给 MCP 客户端。客户端可以先发现工具，再决定是否调用工具。

这也是 MCP 的一个重要特点：工具不是写死在某次请求里，而是由 Server 统一提供。

3. 限制工作区访问

BASE_DIR = Path(os.getenv("MCP_WORKSPACE", Path.cwd())).resolve()

如果设置了 MCP_WORKSPACE，就使用该目录作为工作区；否则使用当前运行目录。

路径校验在这里完成：

def resolve_safe_path(user_path: str = ".") -> Path:
    target = (BASE_DIR / user_path).resolve()

    if target != BASE_DIR and BASE_DIR not in target.parents:
        raise ValueError(f"只允许访问工作区目录：{BASE_DIR}")

    return target

这段代码可以防止用户通过类似下面的路径逃逸到工作区外：

../../Windows/System32

做本地工具时，这种边界非常重要。

4. 跳过敏感目录

IGNORED_DIRS = {
    ".git",
    ".venv",
    "__pycache__",
    "node_modules",
}

.venv 和 node_modules 通常文件数量巨大，没有必要交给 AI 检索。.git 里可能包含历史信息和远程仓库地址，也不建议默认暴露。

5. 搜索文本

if keyword.lower() in line.lower():
    results.append({
        "path": to_relative(path),
        "line": line_number,
        "preview": line.strip()[:200],
    })

这个工具返回：

• 文件路径
• 行号
• 命中的文本预览

这对 AI 编程助手很有价值。它可以先定位代码或文档位置，再基于命中内容回答问题。

6. CSV 概览分析

numeric_stats[column] = {
    "count": len(values),
    "min": min(values),
    "max": max(values),
    "mean": round(mean(values), 4),
    "sum": round(sum(values), 4),
}

这个工具能快速给出 CSV 的基础画像，包括：

• 有哪些列
• 读取了多少行
• 每列缺失值数量
• 数值列的最小值、最大值、均值、求和

对小型业务数据、实验数据、日志导出文件都比较实用。

7. 分组求和

groups[key] = groups.get(key, 0) + value

这个工具可以回答类似问题：

按品类统计销售额

按城市统计订单金额

如果接入 AI 客户端，用户可以用自然语言描述需求，AI 再选择调用 csv_group_sum。

十、安全边界设计

MCP 工具服务器一旦接入本地文件，就必须认真考虑安全问题。

本文做了几层限制：

风险	处理方式
访问任意目录	使用 MCP_WORKSPACE 限定根目录
路径穿越	使用 resolve() 后检查是否仍在工作区内
扫描大量依赖文件	默认跳过 .venv、.git、node_modules
读取超大文件	read_text_safely 限制单文件大小
执行危险操作	本文只提供读取和分析工具，不提供删除、移动、写入

本地工具的原则是：

先只读，再考虑写入；
先限定目录，再扩展范围；
先人工确认，再自动执行。

不要一上来就给 AI 暴露删除文件、执行任意命令、修改数据库这类工具。

十一、常见问题

1. 运行后为什么没有输出？

正常。stdio MCP Server 是给 MCP 客户端调用的，不是普通命令行程序。

建议使用 MCP Inspector 调试：

npx @modelcontextprotocol/inspector python mcp_local_toolbox.py

2. 报错 No module named mcp 怎么办？

说明当前 Python 环境没有安装 MCP SDK，或者没有激活虚拟环境。

执行：

.\.venv\Scripts\Activate.ps1
python -m pip install -r requirements.txt

然后再运行。

3. 为什么不要在代码里 print？

因为 stdio 模式下，标准输入输出是 MCP 协议通信通道。随便 print() 可能污染协议消息。

如果需要调试，建议写入日志文件，或者输出到 stderr：

import sys

print("debug message", file=sys.stderr)

4. 能不能读取整个电脑？

技术上可以，但不建议。

更合理的做法是只给 MCP Server 暴露某个项目目录：

$env:MCP_WORKSPACE="D:/projects/my_project"

这样即使 AI 客户端误调用，也只能访问限定目录。

5. CSV 大文件怎么办？

本文代码用 max_rows 限制读取行数。大文件场景可以继续优化：

• 使用 Pandas 分块读取
• 使用 DuckDB 查询 CSV
• 将数据导入 SQLite / PostgreSQL
• 对常用指标做缓存

十二、扩展方向

1. 增加 SQLite 查询工具

可以增加：

query_sqlite(db_path, sql)

让 AI 查询本地 SQLite 数据库。但要注意 SQL 安全，最好限制只允许 SELECT。

2. 增加 Markdown 文档摘要

可以增加：

read_markdown(file_path)

让 AI 读取指定文档，再由客户端模型完成总结。

3. 增加项目结构分析

可以增加：

project_tree(max_depth)

用于快速生成项目目录结构，辅助代码理解。

4. 增加日志分析

可以增加：

search_log_error(log_path, keyword)

用于定位异常日志、统计错误次数。

5. 接入企业系统

MCP Server 不只能读本地文件，还可以连接：

• MySQL / PostgreSQL
• Elasticsearch
• Git 仓库
• 内部 HTTP API
• 工单系统
• 知识库系统

这也是 MCP 真正有价值的地方：把组织里的工具和数据，以统一方式暴露给 AI Agent。

十三、总结

本文用 Python 实现了一个本地 MCP 工具服务器，提供文件检索和 CSV 分析能力。

核心流程是：

定义 Python 函数
↓
使用 @mcp.tool() 暴露工具
↓
MCP 客户端发现工具
↓
AI 根据任务选择工具
↓
工具返回真实数据
↓
AI 基于结果生成回答

相比只会调用大模型 API，MCP 更接近 AI Agent 的真实工程形态。因为它解决的不是“怎么让模型回答”，而是“怎么让模型安全、标准化地使用外部工具”。

如果要继续深入，可以重点学习：

• MCP Tools、Resources、Prompts 的区别
• MCP Inspector 调试方式
• 本地 Server 和远程 Server 的部署差异
• 工具权限控制
• 数据库和企业 API 接入
• Agent 工作流编排

掌握这些之后，大模型应用就不再只是聊天窗口，而可以逐步变成能读取上下文、调用工具、处理任务的智能系统。

参考资料

• Model Context Protocol 官方介绍：https://modelcontextprotocol.io/docs/getting-started/intro
• MCP 架构说明：https://modelcontextprotocol.io/docs/learn/architecture
• MCP Server 构建教程：https://modelcontextprotocol.io/docs/develop/build-server
• MCP Inspector 文档：https://modelcontextprotocol.io/docs/tools/inspector
• MCP Python SDK：https://github.com/modelcontextprotocol/python-sdk