免费获取学习方案
ARTICLE DETAIL

资讯详情

深耕编程基础知识与建站技术分享的一线实战洞察。

手搓MCP Server:百行Python代码让LLM安全读取本地文件

手搓MCP Server:百行Python代码让LLM安全读取本地文件 1. 项目缘起为什么需要让 LLM 直接读本地文件最近在折腾各种大语言模型LLM应用时我遇到了一个挺普遍的痛点我想让 AI 帮我分析一下本地的项目日志、整理一下电脑里的文档草稿或者让它基于我私有的代码库给出建议。但你会发现无论是 ChatGPT 的 Web 界面还是大多数开源的 LLM 应用框架它们处理外部知识的主要方式要么是你手动复制粘贴一大段文本很快就超上下文长度了要么就是需要你先费劲地把文件内容“喂”进一个专门的向量数据库里。这个过程不仅繁琐而且不实时。我电脑里的文件是动态变化的难道每次更新一个文档我都要重新跑一遍嵌入Embedding和索引流程吗这显然不符合我们“随时问随时答”的直觉。就在我琢磨有没有更轻量、更直接的方法时我注意到了MCPModel Context Protocol这个协议。它本质上定义了一套标准让 LLM 能够通过一组定义好的“工具”Tools安全、可控地访问外部数据和系统。一个典型的应用就是让 LLM 获得“读取我指定目录下文件内容”的能力。于是一个想法冒了出来与其等待某个大型框架集成这个功能不如自己动手用最少的代码实现一个专属于我自己的、能让 LLM 直接读取本地文件的“桥梁”。目标很明确代码控制在 100 行左右功能聚焦只读文件但足够健壮和安全。这就是本次“手搓”MCP Server 的由来。接下来我会带你从零开始一步步理解 MCP 的核心概念并用大约 100 行 Python 代码实现这个实用的工具。你会发现给 LLM 装上“眼睛”去查看本地世界并没有想象中那么复杂。2. MCP 协议速览LLM 与外部世界的安全通道在开始写代码之前我们得先搞清楚 MCP 到底是什么以及它如何工作。你可以把 MCP 想象成 LLM 世界里的“USB 协议”或“驱动程序框架”。在没有 MCP 之前LLM 就像一个只有大脑但没有感官和手脚的智者它所有的知识都来自于训练时灌进去的静态数据。而 MCP 的目的就是为这个“大脑”安全地接上各种“感官”读取数据和“手脚”执行操作。MCP 的核心架构基于客户端-服务器Client-Server模型但这里的角色和我们平常理解的不太一样MCP Server服务器这是我们即将要构建的部分。它扮演“能力提供者”的角色。我们的文件读取 Server就是一个提供了read_file和list_files等工具Tools的服务端。它启动后就静静地等待来自客户端的指令。MCP Client客户端这通常是LLM 应用本身。比如 Claude Desktop、Cursor IDE或者任何集成了 MCP 客户端库的应用。客户端的职责是管理 LLM 的对话并在 LLM 想要使用某个工具时向对应的 Server 发出请求。LLM它作为“决策大脑”运行在 Client 内部。当用户提出“帮我看看~/projects/logs/error.log里最近有什么错误”时LLM 会判断“要回答这个问题我需要读取一个文件。” 于是它通过 Client 向我们的 File Server 发出调用read_file工具的请求。它们之间的通信使用的是JSON-RPC 2.0协议这是一种轻量级的远程过程调用规范消息通过标准输入输出stdio或网络套接字进行传输。整个流程可以简化为用户向 LLM 应用Client提问。LLM分析后决定调用工具Client 生成一个 JSON-RPC 请求。请求通过 stdio 发送给我们手搓的MCP Server。Server执行对应的 Python 函数如读取文件然后将结果封装成 JSON-RPC 响应。响应返回给 ClientClient 将文件内容作为上下文提供给 LLM。LLM基于获取到的文件内容生成最终的回答给用户。这个协议的关键优势在于标准化和安全性。标准化意味着只要我们的 Server 按照 MCP 的规范来暴露工具和处理消息它就能被任何兼容 MCP 的客户端如 Claude Desktop直接使用。安全性则体现在 Server 运行在独立的进程中拥有明确的权限边界。我们的文件读取 Server 只能访问我们允许它访问的目录而无法触及系统的其他部分或网络这比让 LLM 应用本身直接拥有文件系统权限要安全得多。3. 环境准备与项目初始化理论清楚了我们开始动手。首先确保你的开发环境就绪。本项目主要依赖 Python推荐使用 3.8 及以上版本。我们将使用官方推荐的mcp库来简化开发它处理了 JSON-RPC 通信的底层细节让我们能专注于工具的逻辑。第一步创建项目目录并初始化虚拟环境。我强烈建议使用虚拟环境来隔离依赖避免污染全局的 Python 环境。mkdir mcp-file-server cd mcp-file-server python -m venv venv # 激活虚拟环境 # 在 macOS/Linux 上 source venv/bin/activate # 在 Windows 上 # venv\Scripts\activate第二步安装核心依赖。我们只需要安装mcp这个包。它非常轻量没有复杂的次级依赖。pip install mcp安装完成后你可以通过pip list确认mcp库已就位。第三步创建主程序文件。在项目根目录下创建一个名为file_server.py的文件。这将是我们的 MCP Server 的全部代码所在。用你喜欢的编辑器如 VSCode、PyCharm打开它我们准备开始编码。注意在开始编码前请思考一下安全边界。我们这个 Server 的核心能力是读文件因此我们必须在一开始就划定一个“安全沙箱”。一个常见的做法是限制 Server 只能访问某个特定的“根目录”比如~/mcp_accessible。在本教程中为了演示的灵活性我会将根目录设置为一个可通过环境变量配置的路径默认是当前用户的家目录下的一个子目录。在实际部署时务必将其设置为一个仅包含你愿意向 LLM 公开文件的目录。4. 核心代码逐行解析构建文件读取工具现在打开file_server.py让我们开始编写核心的 100 行代码。我会将代码分成几个逻辑块并逐部分解释。第一部分导入依赖与配置初始化#!/usr/bin/env python3 一个简单的 MCP 服务器提供读取本地文件的能力。 import os import sys from pathlib import Path from typing import Any, List import mcp.server as mcp from mcp.server.models import InitializationOptions import mcp.server.stdio from mcp.types import TextContent, Tool # 配置定义服务器可访问的根目录 # 出于安全考虑强烈建议将其设置为一个受限目录 ROOT_PATH Path(os.getenv(MCP_FILE_ROOT, Path.home() / mcp_accessible)).resolve()第1-2行标准的 Shebang 和模块文档字符串。第4-9行导入必要的库。os,sys,pathlib,typing是 Python 标准库。核心是mcp包我们从其中导入服务器模块、类型定义和 stdio 通信工具。第12-13行定义安全核心ROOT_PATH。我们使用pathlib.Path来处理路径它比传统的os.path更现代、更易用。os.getenv(“MCP_FILE_ROOT”, default)允许我们通过环境变量MCP_FILE_ROOT来动态配置根目录。如果未设置则默认指向用户家目录下的mcp_accessible文件夹。.resolve()方法将路径转换为绝对路径消除任何符号链接或..的影响确保路径明确。第二部分创建 Server 实例与工具定义# 创建 MCP 服务器实例 app mcp.Server(“file-server”) # 定义工具列出目录下的文件 app.list_tools() async def list_tools() - List[Tool]: 返回服务器提供的工具列表 return [ Tool( name“list_files”, description“列出指定目录下的文件和子目录。如果路径为空则列出根目录的内容。”, inputSchema{ “type”: “object”, “properties”: { “path”: { “type”: “string”, “description”: “相对于服务器根目录的目录路径。例如‘docs’ 或 ‘projects/logs’。留空表示根目录。” } }, }, ), Tool( name“read_file”, description“读取指定文件的内容。”, inputSchema{ “type”: “object”, “properties”: { “path”: { “type”: “string”, “description”: “相对于服务器根目录的文件路径。例如‘notes/meeting.txt’。” } }, “required”: [“path”], }, ), ]第16行实例化一个 MCP Server并给它起个名字叫”file-server”。这个名字会在客户端连接时被识别。第19-48行使用app.list_tools()装饰器定义一个异步函数list_tools。这个函数是 MCP 协议的必需部分当客户端查询服务器有哪些能力时就会调用这个函数。第24-47行我们返回一个包含两个Tool对象的列表。每个Tool对象定义了一个工具。name: 工具的唯一标识符客户端将通过这个名字来调用它。description: 工具的详细描述。这个描述至关重要因为 LLM 会根据这个描述来决定在什么情况下使用这个工具。描述要清晰、准确。inputSchema: 定义了调用这个工具时需要提供的参数遵循 JSON Schema 格式。对于list_files我们定义了一个可选的path参数对于read_filepath是必需的 (”required”: [“path”])。注意这里的路径都是相对于我们之前定义的ROOT_PATH的。第三部分实现list_files工具app.call_tool() async def call_tool(name: str, arguments: dict[str, Any]) - list[TextContent]: 根据工具名称和参数执行相应的操作 if name “list_files”: # 处理列出文件的请求 rel_path_str arguments.get(“path”, “”) target_dir ROOT_PATH / rel_path_str if rel_path_str else ROOT_PATH # 安全检查确保目标路径仍在 ROOT_PATH 之下 try: target_dir_resolved target_dir.resolve() if not str(target_dir_resolved).startswith(str(ROOT_PATH)): return [TextContent(type“text”, text“错误试图访问根目录之外的路径。”)] except Exception: return [TextContent(type“text”, text“错误提供的路径无效。”)] # 检查路径是否存在且为目录 if not target_dir_resolved.exists(): return [TextContent(type“text”, textf“错误目录 ‘{rel_path_str}’ 不存在。”)] if not target_dir_resolved.is_dir(): return [TextContent(type“text”, textf“错误‘{rel_path_str}’ 不是一个目录。”)] # 列出目录内容 items [] for item in target_dir_resolved.iterdir(): item_type “(目录)” if item.is_dir() else “(文件)” items.append(f”{item.name} {item_type}“) items.sort() # 排序便于阅读 result_text f”目录 ‘{target_dir_resolved.relative_to(ROOT_PATH)}’ 下的内容\n“ “\n”.join(items) if items else “空目录” return [TextContent(type“text”, textresult_text)]第51行使用app.call_tool()装饰器定义call_tool函数。这是所有工具调用的总入口。参数name对应工具名arguments是客户端传来的参数字典。第53-78行处理list_files工具。第55行获取相对路径参数默认为空字符串根目录。第56行构造绝对路径target_dir。第60-66行关键的安全检查。通过resolve()解析可能包含..或符号链接的路径然后检查解析后的绝对路径是否以ROOT_PATH的绝对路径开头。这是防止路径遍历攻击Path Traversal的核心。例如如果ROOT_PATH是/home/user/data而用户传入path”../../../etc/passwd”resolve()后会得到/etc/passwd它不以/home/user/data开头因此会被拒绝。第69-73行检查路径是否存在以及是否为目录。第76-78行使用Path.iterdir()遍历目录区分文件和文件夹排序后格式化输出。relative_to(ROOT_PATH)用于在结果显示中去除根目录部分让输出更清晰。第四部分实现read_file工具elif name “read_file”: # 处理读取文件的请求 rel_path_str arguments.get(“path”) if not rel_path_str: return [TextContent(type“text”, text“错误必须提供 ‘path’ 参数。”)] target_file ROOT_PATH / rel_path_str # 同样的安全检查 try: target_file_resolved target_file.resolve() if not str(target_file_resolved).startswith(str(ROOT_PATH)): return [TextContent(type“text”, text“错误试图访问根目录之外的路径。”)] except Exception: return [TextContent(type“text”, text“错误提供的路径无效。”)] # 检查路径是否存在且为文件 if not target_file_resolved.exists(): return [TextContent(type“text”, textf“错误文件 ‘{rel_path_str}’ 不存在。”)] if not target_file_resolved.is_file(): return [TextContent(type“text”, textf“错误‘{rel_path_str}’ 不是一个文件。”)] # 读取文件内容注意编码和文件大小 try: # 简单限制文件大小避免读取超大文件导致内存问题 if target_file_resolved.stat().st_size 1024 * 1024: # 1MB return [TextContent(type“text”, text“错误文件过大超过1MB出于性能考虑拒绝读取。”)] content target_file_resolved.read_text(encoding“utf-8”) # 在内容前加上文件路径信息 result_text f”文件 ‘{rel_path_str}’ 的内容\n\n{content}“ return [TextContent(type“text”, textresult_text)] except UnicodeDecodeError: return [TextContent(type“text”, text“错误文件不是 UTF-8 文本格式无法读取。”)] except Exception as e: return [TextContent(type“text”, textf“读取文件时出错{e}”)]第80-118行处理read_file工具。第82-85行获取必需的path参数。第88-95行执行与list_files相同的路径解析和安全检查。第98-102行检查路径是否存在以及是否为普通文件。第106-118行读取文件内容。第108-110行重要的工程实践检查文件大小。LLM 的上下文长度有限无节制地读取大文件如视频、数据库文件会浪费资源甚至导致崩溃。这里我们设置了一个简单的 1MB 限制。在实际应用中你可能需要根据 LLM 客户端的上下文窗口来调整这个值或者实现分块读取。第111行使用read_text(encoding”utf-8″)以 UTF-8 编码读取文本文件。这是最通用的文本编码。第113行在返回的内容前附加文件路径帮助 LLM 理解上下文。第115-118行异常处理。捕获UnicodeDecodeError来处理二进制文件捕获其他通用异常并返回友好错误信息。第五部分处理未知工具与 Server 启动else: # 如果收到未知的工具名返回错误 return [TextContent(type“text”, textf“错误未知工具 ‘{name}’。”)] # 主函数启动服务器 async def main(): 运行 MCP 服务器 # 打印根目录信息便于调试 print(f”文件服务器已启动根目录为{ROOT_PATH}“, filesys.stderr) print(f”请确保您的 MCP 客户端如 Claude Desktop配置正确。“, filesys.stderr) # 通过标准输入输出与客户端通信 async with mcp.server.stdio.stdio_server() as (read_stream, write_stream): await app.run(read_stream, write_stream, InitializationOptions()) if __name__ “__main__”: import asyncio asyncio.run(main())第120-122行兜底逻辑处理客户端可能请求的、我们未定义的工具。第125-134行定义主异步函数main()和程序入口。第128-129行将启动信息打印到标准错误输出stderr这样不会干扰与客户端的 JSON-RPC 通信通过 stdout。这是一个有用的调试信息。第131行mcp.server.stdio.stdio_server()创建了一个通过标准输入输出进行通信的服务器传输层。这是 MCP Server 最常见的运行方式由父进程MCP Client启动并管理其生命周期。第132行app.run()启动服务器开始监听来自read_stream的请求并通过write_stream发送响应。第134-135行标准的 Python 异步程序入口。至此一个功能完整、具备基础安全防护的 MCP 文件读取服务器就完成了。算上注释和空行代码大约在 110 行左右完全符合我们“百行代码”的目标。5. 运行、测试与客户端配置代码写好了怎么让它跑起来并真正被 LLM 使用呢我们需要两步先独立测试 Server 是否正常工作再将其配置到 MCP 客户端如 Claude Desktop中。第一步独立测试 Server我们可以手动模拟一个 JSON-RPC 请求来测试。创建一个简单的测试脚本test_server.py这不是必须的但有助于调试import subprocess import json import time # 启动服务器进程 proc subprocess.Popen( [‘python’, ‘file_server.py’], stdinsubprocess.PIPE, stdoutsubprocess.PIPE, stderrsubprocess.PIPE, textTrue ) # 构建一个 JSON-RPC 请求列出根目录文件 request { “jsonrpc”: “2.0”, “id”: 1, “method”: “tools/call”, “params”: { “name”: “list_files”, “arguments”: {} } } # 发送请求 proc.stdin.write(json.dumps(request) ‘\n’) proc.stdin.flush() time.sleep(0.5) # 等待一下 # 读取响应 response_line proc.stdout.readline() print(“Server Response:”, json.loads(response_line)) # 再测试一个读取文件的请求假设根目录下有个 test.txt request2 { “jsonrpc”: “2.0”, “id”: 2, “method”: “tools/call”, “params”: { “name”: “read_file”, “arguments”: {“path”: “test.txt”} # 请确保这个文件存在 } } proc.stdin.write(json.dumps(request2) ‘\n’) proc.stdin.flush() time.sleep(0.5) response_line2 proc.stdout.readline() print(“Server Response 2:”, json.loads(response_line2)) proc.terminate()运行这个测试脚本你应该能看到 Server 返回的目录列表或文件内容。这验证了我们的 Server 逻辑和 JSON-RPC 接口是正确的。第二步配置到 Claude Desktop以 macOS 为例这是最关键的一步让我们的 Server 被真正的 LLM 应用调用。找到 Claude Desktop 的配置文件。在 macOS 上它通常位于~/Library/Application Support/Claude/claude_desktop_config.json。在 Windows 上可能在%APPDATA%\Claude\claude_desktop_config.json。编辑这个 JSON 文件。如果文件不存在就创建它。我们需要在mcpServers部分添加我们的 Server 配置。{ “mcpServers”: { “my-file-server”: { “command”: “/path/to/your/venv/bin/python”, “args”: [“/path/to/your/mcp-file-server/file_server.py”], “env”: { “MCP_FILE_ROOT”: “/Users/yourname/Documents/mcp_files” } } } }”my-file-server”: 给这个 Server 起个名字可以任意。”command”: Python 解释器的绝对路径。这里必须使用虚拟环境中的 Python以确保mcp库可用。你可以通过which python在激活的虚拟环境中命令找到它。”args”: 一个列表第一个元素就是我们主程序file_server.py的绝对路径。”env”: 可选但推荐设置环境变量。这里我们覆盖了代码中的默认ROOT_PATH将其指向一个更具体的、安全的目录/Users/yourname/Documents/mcp_files。请确保这个目录存在。保存配置文件并重启 Claude Desktop。验证连接。重启后打开 Claude Desktop新建一个对话。如果你在输入框里输入“/” 理论上应该能看到可用的工具列表。如果配置成功你应该能看到list_files和read_file这两个工具。你可以直接尝试输入“请使用 list_files 工具看看我能访问哪些目录。” Claude 应该会调用该工具并返回结果。重要提示不同的 MCP 客户端配置方式可能略有不同。例如一些客户端可能支持通过图形界面添加 Server或者配置文件的位置和格式有差异。请务必查阅你所使用客户端的官方文档。核心原理不变告诉客户端如何启动你的 Server 进程命令、参数、环境变量。6. 安全加固与生产环境考量我们之前已经植入了一些基础的安全措施比如路径解析和边界检查。但要让这个 Server 真正可靠地运行还需要考虑更多。1. 更严格的路径安全我们的startswith检查在大多数情况下是有效的但在某些边缘情况下比如符号链接、硬链接、网络路径映射可能需要更复杂的处理。一个更稳健的方法是使用pathlib的is_relative_to()方法Python 3.9if not target_path_resolved.is_relative_to(ROOT_PATH): return [TextContent(type“text”, text“错误路径越界访问。”)]这个方法能更语义化地判断一个路径是否位于另一个路径之下。2. 文件类型与内容过滤避免读取二进制文件我们虽然用UTF-8解码捕获了错误但更好的做法是在读取前通过文件扩展名或magic number文件头字节进行初步过滤避免尝试读取图片、视频、可执行文件等。内容清理对于文本文件也要警惕。文件内容中可能包含敏感信息如密码、密钥、恶意代码或超长乱码。虽然 LLM 本身不会“执行”这些内容但将其灌入上下文可能引发意外行为或泄露隐私。在生产环境中可以考虑对读取的内容进行简单的关键字过滤或长度截断。3. 资源限制与超时控制内存限制我们限制了单个文件大小1MB。对于目录列表如果目录下文件极多iterdir()也可能消耗较多内存和 IO。可以考虑对返回的条目数量进行限制。超时机制文件 IO 操作可能会因为磁盘问题而挂起。应该为工具调用设置超时防止一个慢速请求阻塞整个 Server。这可以在call_tool函数内部用asyncio.wait_for包装实现。4. 认证与授权进阶目前的 Server 是完全信任启动它的客户端的。在某些多用户或网络部署场景下这可能不够。基础认证可以在 Server 启动时读取一个预共享的令牌客户端在请求中必须携带该令牌。这需要修改 Server 的初始化流程和请求处理逻辑检查 JSON-RPC 请求的authorization头。更细粒度的权限可以为不同的工具或不同的路径前缀设置不同的访问权限。例如定义一个配置文件指定哪些路径可读、哪些可写如果我们未来扩展了写功能。这会将 Server 从一个简单的文件读取器升级为一个具备初步权限管理的能力网关。5. 日志与监控对于长期运行的服务添加日志记录至关重要。记录工具调用时间、工具名、路径参数、执行结果成功/失败、以及任何异常。这有助于后期审计和故障排查。可以使用 Python 标准的logging模块将日志输出到文件或标准错误。将这些考量点融入代码后我们的 Server 会从一个演示原型进化为一个可在个人或小团队内部安全使用的生产级工具。记住安全是一个持续的过程需要根据实际使用场景不断评估和调整。7. 功能扩展思路超越“只读”一旦核心的“读”通道建立起来你会发现 MCP 的想象力远不止于此。我们的百行代码是一个完美的起点可以在此基础上轻松扩展更多实用功能让 LLM 成为你数字世界更得力的助手。1. 文件搜索与内容过滤list_files只展示了文件名。我们可以新增一个search_in_files工具。功能在指定目录及其子目录下递归搜索包含特定关键词的文本文件。实现要点使用pathlib.Path.rglob(“*.txt”)来递归遍历特定扩展名的文件。在读取每个文件时同样施加大小和编码限制使用if keyword in content进行简单搜索或者集成正则表达式进行更复杂的模式匹配。返回匹配的文件名、路径和包含关键词的上下文片段。LLM 提示词“在我的项目笔记目录下搜索所有提到‘用户认证’的 Markdown 文件。”2. 文件信息获取read_file返回全部内容。有时我们只需要元信息。新增工具get_file_info:输入文件路径。输出文件大小、最后修改时间、创建时间、文件类型通过扩展名或mimetypes库判断。应用场景LLM 可以帮你回答“我上周修改的那个最大的日志文件是什么”这类问题。3. 结构化数据读取很多本地文件是结构化的如 CSV、JSON、YAML。新增工具read_csv或read_json:输入文件路径可选参数如 CSV 的分隔符JSON 的某个特定键。实现使用 Python 内置的csv、json库或pyyaml来安全地解析文件。输出将数据转换为清晰的文本描述例如“该 CSV 文件有 3 列姓名、年龄、邮箱共 100 行。前 5 行数据如下…”或格式化的 JSON 字符串。务必注意解析不可信的 CSV/JSON 文件存在风险如 CSV 注入、JSON 解析消耗大量内存需要施加严格的限制如最大行数、最大深度。4. 简单的文件操作需极度谨慎在充分评估风险后可以考虑增加受严格限制的写操作。新增工具append_to_file:输入文件路径要追加的文本内容。实现以追加模式”a”打开文件写入内容。必须进行比读操作更严格的路径和权限检查最好限制在某个特定的“草稿”或“输出”目录内。新增工具create_markdown_summary:输入目录路径。实现调用list_files和read_file针对文本文件然后请求 LLM 生成一个该目录内容的摘要最后自动调用append_to_file将摘要写入该目录下的README.md文件。这是一个组合工具的例子展示了如何将多个基础工具串联起来让 LLM 完成一个多步骤的任务。扩展时务必牢记“最小权限原则”。每增加一个工具就增加了一份风险。始终问自己这个功能是否必需它的操作边界是否清晰且可被安全地限制对于写操作尤其是删除操作必须设计二次确认机制例如LLM 生成操作描述需用户明确同意后才执行或者完全避免在自动化工具中提供。通过以上扩展你的 MCP Server 将从“文件阅读器”演变为一个“智能文件助手”真正成为连接 LLM 与你本地知识库的高效、安全的桥梁。这 100 行代码打开的是一个充满可能性的世界。
返回列表