尧图建网站 尧图建网站 YAOTU WEB BUILD 免费咨询
ARTICLE DETAIL

资讯详情

深耕网站建设与建站编程的一线实战洞察。

手搓MCP Server:用100行Python实现LLM安全文件读取工具

手搓MCP Server:用100行Python实现LLM安全文件读取工具 1. 项目概述为什么我们需要一个“文件读取”MCP Server最近在折腾大语言模型应用开发的朋友估计没少被一个痛点折磨怎么让LLM安全、可控地访问我们本地的文件系统无论是想让它帮你分析一份刚下载的财报PDF还是总结你电脑里那堆杂乱无章的会议纪要直接给模型“看”文件内容似乎是个刚需。但问题来了把文件内容一股脑塞进上下文Context不仅浪费宝贵的Token还可能因为文件太大直接超限。更关键的是从安全角度考虑你肯定不希望一个AI应用拥有对你整个硬盘的完全读写权限。这就是模型上下文协议Model Context Protocol简称MCP和MCP Server的价值所在。你可以把它理解为一个“安全的数据管道”或“模型的外挂大脑”。通过一个轻量级的MCP Server你可以精确地告诉LLM“嘿这是我的‘文档库’工具你只能通过这个特定的接口读取我指定目录下的文件并且每次只能读一小段。” 这样一来既解决了模型获取外部信息的需求又划清了安全边界。网上虽然有一些现成的MCP Server实现但要么功能太庞杂要么配置起来让人头大。今天我们就来干一件极客范儿十足的事用大约100行Python代码从头手搓一个专用于读取本地文本文件的MCP Server。这个Server将提供一个名为read_file的工具Tool让LLM比如通过Claude Desktop、Cursor等客户端能够按需读取你指定文件夹里的文件内容。整个实现过程清晰明了你会彻底搞懂MCP的核心通信机制、工具定义方法以及如何与客户端握手。完成后你将拥有一个完全受自己控制的、轻量级的文件助手。2. MCP 核心概念与我们的设计思路在动手写代码之前我们得先花几分钟把MCP的基本玩法搞清楚。这能让你后面的每一行代码都写得心里有底。2.1 MCP 是什么它如何工作MCP即模型上下文协议本质上是一套用于LLM大语言模型与外部资源如数据库、API、文件系统安全通信的开放标准。它采用客户端-服务器Client-Server架构MCP Client客户端通常是LLM应用本身比如Claude Desktop、Cursor IDE或者你自己写的AI应用。它负责发起请求说“我需要XXX信息”。MCP Server服务器就是我们今天要构建的东西。它托管着具体的“工具”Tools或“资源”Resources等待客户端的调用。当客户端说“用read_file工具读一下/home/me/report.txt”时Server就执行相应的操作并返回结果。它们之间通过标准输入输出stdin/stdout或SSEServer-Sent Events等方式使用JSON-RPC消息进行通信。这种设计非常巧妙它把AI的能力扩展和核心应用逻辑解耦了。你的AI应用Client不需要知道文件具体怎么读、数据库怎么查它只需要知道有一个叫read_file的工具可用然后去调用它就行。所有的具体实现和安全控制都封装在独立的MCP Server里。2.2 我们的 Server 要做什么设计蓝图我们的目标很明确构建一个单一功能的MCP Server它只做一件事——让LLM读取指定目录下的文本文件。基于这个目标我们的设计思路如下单一工具提供一个read_file工具。这是Server能力的核心。参数化路径read_file工具接受一个file_path参数。客户端需要提供想读取的文件的相对路径相对于我们Server设定的根目录。安全沙箱Server在启动时会设定一个“根目录”例如./data。所有文件读取操作都被限制在这个目录下防止LLM通过类似../../../etc/passwd的路径遍历攻击访问系统敏感文件。这是安全性的基石。错误处理当文件不存在、路径非法或读取失败时返回清晰的错误信息给客户端而不是让整个Server崩溃。标准化通信严格按照MCP的JSON-RPC协议格式来封装请求和响应确保能与标准的MCP客户端如Claude Desktop无缝对接。这个设计保证了Server的简单、专注和安全正好符合我们“100行代码搞定”的极简理念。2.3 技术栈与工具选型为了快速实现我们选择Python因为它语法简洁库生态丰富。核心只需要两个库jsonPython标准库用于处理JSON-RPC消息的序列化和反序列化。这是通信的“语言”。sysPython标准库用于从标准输入stdin读取客户端的请求并向标准输出stdout写入响应。这是通信的“管道”。我们刻意避免使用任何复杂的Web框架如FastAPI或额外的MCP SDK就是为了揭示最底层的原理。当你用最基础的工具实现后未来再换用更高级的封装比如官方的mcpPython SDK时你会理解得更加深刻。3. 从零开始构建 MCP Server 的骨架让我们打开代码编辑器创建一个新文件比如叫simple_file_server.py。我们从搭建最基本的通信骨架开始。3.1 初始化与协议握手MCP Client 和 Server 建立连接后的第一件事就是“握手”交换彼此的姓名、版本和支持的能力。这个过程通过initialize和initialized两个JSON-RPC通知来完成。#!/usr/bin/env python3 import sys import json def send_message(message): 向stdout发送一条JSON-RPC消息。 json.dump(message, sys.stdout) sys.stdout.write(\n) sys.stdout.flush() def main(): # Server主循环持续从stdin读取请求 for line in sys.stdin: try: request json.loads(line.strip()) except json.JSONDecodeError: continue # 忽略非JSON格式的输入 method request.get(method) params request.get(params, {}) request_id request.get(id) # 1. 处理客户端发来的初始化请求 if method initialize: # 构建初始化响应告知客户端我们的能力 response { jsonrpc: 2.0, id: request_id, result: { protocolVersion: 2024-11-05, capabilities: { tools: {} # 目前先声明支持tools具体列表在后面的initialized阶段提供 }, serverInfo: { name: simple-file-server, version: 0.1.0 } } } send_message(response) # 紧接着发送一个initialized通知这是一个没有id的通知 initialized_notification { jsonrpc: 2.0, method: initialized, params: {} } send_message(initialized_notification) if __name__ __main__: main()代码解读与注意事项send_message函数是我们的“发声器”。所有给客户端的回复都通过它遵循“一行一个JSON”的约定。main函数中的循环for line in sys.stdin:是我们的“耳朵”持续监听客户端发来的指令。initialize请求是客户端发起的第一个请求我们必须用对应的response回复其中id必须与请求中的id一致这是JSON-RPC协议的要求。在initialize响应之后我们主动发送一个method为initialized的通知。这是一个单向通知不需要客户端回复用于告知客户端“我准备好了你可以进行下一步了”。在capabilities里我们声明了支持tools但列表先留空。这是因为按照一些客户端的实现完整的工具列表可能在initialized通知之后通过tools/list请求来获取。这是一种常见的模式。注意这里的protocolVersion字段值2024-11-05是MCP协议的一个快照版本号你需要根据你使用的客户端要求来调整。Claude Desktop通常兼容这个版本。如果连接出错可以查阅客户端文档确认协议版本。3.2 声明我们的“工具”read_file握手完成后客户端会来询问“Server你都有哪些工具可用啊” 这时就需要响应tools/list请求。我们在main函数的循环里接着处理method tools/list的情况# ... 接在初始化处理代码之后 ... # 2. 处理客户端列出工具的请求 elif method tools/list: # 定义我们唯一的工具read_file tools_list { jsonrpc: 2.0, id: request_id, result: { tools: [ { name: read_file, description: 读取指定路径下的文本文件内容。文件路径需相对于服务器设定的根目录。, inputSchema: { type: object, properties: { file_path: { type: string, description: 相对于服务器根目录的文件路径例如 docs/meeting_notes.txt } }, required: [file_path] } } ] } } send_message(tools_list)关键点解析name: 工具的唯一标识符客户端调用时就靠这个名字。description: 工具的详细描述这很重要LLM客户端会根据这个描述来决定在什么场景下调用这个工具。描述得越清晰LLM用得就越准。inputSchema: 定义了调用这个工具时需要提供的参数。这里我们遵循JSON Schema格式。properties下定义了file_path参数类型是字符串。required数组指明file_path是必填参数。这个模式定义会帮助客户端尤其是LLM在调用时构造正确的参数结构。现在我们的Server已经能够完成握手和自我介绍。客户端知道了存在一个叫read_file的工具。接下来就是最核心的部分实现这个工具的具体逻辑。4. 核心功能实现安全地读取本地文件工具被“知道”了接下来就要能被“使用”。当客户端想要读取文件时它会发送一个tools/call请求。我们需要处理这个请求执行真正的文件读取操作并返回结果或错误。4.1 处理工具调用请求我们在主循环中增加对tools/call的处理# ... 接在tools/list处理代码之后 ... # 3. 处理客户端调用工具的请求 elif method tools/call: # 提取调用ID和参数 call_id params.get(callId) tool_name params.get(name) arguments params.get(arguments, {}) # 目前我们只处理 read_file 工具 if tool_name read_file: file_path arguments.get(file_path) result_content is_error False error_message # --- 安全校验与文件读取逻辑 --- # 设定一个安全的根目录例如当前目录下的 data 文件夹 import os BASE_DIR ./data # 确保BASE_DIR存在 os.makedirs(BASE_DIR, exist_okTrue) # 构造绝对路径并检查路径安全性防止目录遍历攻击 requested_path os.path.normpath(os.path.join(BASE_DIR, file_path)) # 关键安全步骤检查请求的路径是否仍在BASE_DIR之下 if not os.path.commonpath([BASE_DIR, requested_path]) BASE_DIR: is_error True error_message f安全错误请求的文件路径 {file_path} 试图访问根目录之外的范围。 else: # 路径安全尝试读取文件 try: with open(requested_path, r, encodingutf-8) as f: result_content f.read() except FileNotFoundError: is_error True error_message f文件未找到{file_path}。请检查路径是否正确。 except IsADirectoryError: is_error True error_message f路径是一个目录而非文件{file_path}。 except PermissionError: is_error True error_message f权限不足无法读取文件{file_path}。 except UnicodeDecodeError: # 尝试其他编码或简单提示 try: with open(requested_path, r, encodinggbk) as f: result_content f.read() except: is_error True error_message f无法解码文件{file_path}它可能不是纯文本文件或编码不被支持。 except Exception as e: is_error True error_message f读取文件时发生未知错误{str(e)} # --- 安全校验与文件读取逻辑结束 --- # 构造调用结果响应 if is_error: result_response { jsonrpc: 2.0, id: request_id, result: { callId: call_id, content: [ { type: text, text: f调用工具 read_file 失败。错误信息{error_message} } ] } } else: result_response { jsonrpc: 2.0, id: request_id, result: { callId: call_id, content: [ { type: text, text: result_content } ] } } send_message(result_response)安全与健壮性设计详解根目录BASE_DIR我们硬编码或通过配置设定一个根目录这里是./data。所有文件访问都被限制在此目录下。os.makedirs(BASE_DIR, exist_okTrue)确保了目录存在。路径规范化与遍历攻击防护这是安全的核心。os.path.normpath清理路径中的./和../。os.path.join(BASE_DIR, file_path)将用户输入的相对路径与根目录拼接。最关键的一行if not os.path.commonpath([BASE_DIR, requested_path]) BASE_DIR:。这行代码检查规范化后的请求路径requested_path是否仍然以BASE_DIR开头。如果用户传入../../../etc/passwd拼接并规范化后commonpath的比较就会失败从而阻止访问。这是防止目录遍历Path Traversal攻击的标准做法。全面的错误处理我们预判了多种错误情况文件不存在、路径是目录、权限不足、编码问题等。对于编码问题我们做了一个简单的回退尝试从UTF-8到GBK这在处理中文Windows系统生成的文本文件时很实用。清晰的错误信息能帮助LLM和终端用户理解问题所在。响应格式无论成功失败我们都按照MCP协议通过tools/call的响应返回结果。结果被包裹在content数组里类型为text。对于错误我们把错误信息放在文本内容中返回。4.2 完善 Server处理其他必要请求一个健壮的MCP Server还需要处理一些基本的生命周期请求。我们在主循环中补充# ... 接在tools/call处理代码之后 ... # 4. 处理客户端心跳或通知如notifications/initialized但我们已在初始化后主动发送 elif method notifications/initialized: # 我们已经主动发送过这里可以忽略或记录日志 pass # 5. 处理客户端关闭请求可选但建议实现 elif method shutdown: response {jsonrpc: 2.0, id: request_id, result: None} send_message(response) # 可以选择在此退出循环 break # 6. 处理客户端退出通知 elif method exit: # 客户端通知Server退出我们可以结束进程 sys.exit(0) # 7. 处理未知方法保持安静或返回错误 else: if request_id: # 如果是请求有id而非通知则返回方法未找到错误 error_response { jsonrpc: 2.0, id: request_id, error: { code: -32601, message: fMethod not found: {method} } } send_message(error_response)至此我们一个功能完整、具备基础安全性的文件读取MCP Server就编码完成了。算上注释和空行确实在100行左右的核心逻辑内。5. 如何运行与测试你的 MCP Server代码写好了怎么让它跑起来并且真正被LLM使用呢我们分两步走先进行基础功能测试再连接到真实的MCP客户端。5.1 基础功能测试模拟客户端通信在部署到复杂环境前我们可以写一个简单的Python脚本模拟客户端测试Server的核心逻辑是否通畅。创建一个test_server.py文件import subprocess import json import time # 启动我们的MCP Server进程 server_process subprocess.Popen( [python3, simple_file_server.py], stdinsubprocess.PIPE, stdoutsubprocess.PIPE, stderrsubprocess.PIPE, textTrue ) def send_to_server(message): 向Server进程发送一条JSON消息 server_process.stdin.write(json.dumps(message) \n) server_process.stdin.flush() def read_from_server(): 从Server进程读取一行响应 return server_process.stdout.readline() # 1. 发送初始化请求 init_request { jsonrpc: 2.0, id: 1, method: initialize, params: {} } send_to_server(init_request) print(Sent initialize request) response read_from_server() print(Received:, response) # 等待Server发送initialized通知我们模拟的客户端需要消费掉它 time.sleep(0.1) line server_process.stdout.readline() print(Received (notification):, line) # 2. 发送列出工具请求 list_request { jsonrpc: 2.0, id: 2, method: tools/list, params: {} } send_to_server(list_request) print(\nSent tools/list request) response read_from_server() print(Received:, response) # 3. 发送调用工具请求假设data目录下有一个test.txt文件 call_request { jsonrpc: 2.0, id: 3, method: tools/call, params: { callId: call-1, name: read_file, arguments: { file_path: test.txt # 确保在 ./data/test.txt 位置有这个文件 } } } send_to_server(call_request) print(\nSent tools/call request for test.txt) response read_from_server() print(Received:, response) # 4. 发送关闭请求 shutdown_request { jsonrpc: 2.0, id: 4, method: shutdown, params: {} } send_to_server(shutdown_request) response read_from_server() print(\nSent shutdown request) print(Received:, response) server_process.terminate() print(\nTest completed.)运行这个测试脚本前请确保在simple_file_server.py同级目录下创建data文件夹并在里面放一个test.txt文件写入一些内容。运行python3 test_server.py你应该能看到一系列成功的请求和响应并在最后看到test.txt文件的内容被打印出来。如果看到安全错误或文件未找到请检查路径和文件权限。5.2 连接至真实 MCP 客户端以 Claude Desktop 为例这是最激动人心的部分——让我们手搓的Server被真正的AI使用。这里以 Anthropic 的 Claude Desktop 应用为例。定位 Claude Desktop 配置macOS: 配置文件通常位于~/Library/Application Support/Claude/claude_desktop_config.json。Windows: 配置文件通常位于%APPDATA%\Claude\claude_desktop_config.json。编辑配置文件在配置文件中你需要添加一个mcpServers字段来注册我们的Server。如果该字段已存在就在数组里追加一个新对象。{ mcpServers: { simple-file-server: { command: python3, args: [ /ABSOLUTE/PATH/TO/YOUR/simple_file_server.py ] } } }重要提示将/ABSOLUTE/PATH/TO/YOUR/替换成你simple_file_server.py文件的绝对路径。command也可以直接写你的Python解释器全路径如/usr/local/bin/python3如果python3在系统PATH里这样写通常没问题。重启 Claude Desktop保存配置文件后完全退出并重启 Claude Desktop 应用。验证与使用重启后打开Claude Desktop新建一个对话。你应该能在输入框上方或侧边栏的工具图标处看到可用的工具列表。如果配置成功read_file工具应该会出现。现在你可以直接对Claude说“请使用 read_file 工具读取 data 目录下的 meeting_notes.txt 文件并总结其内容。” Claude 会识别出可用的工具并在后台调用你的Server将文件内容作为上下文获取然后进行总结。实操心得第一次配置时最容易出错的地方就是路径。务必使用绝对路径指向你的Python脚本。如果连接失败可以查看Claude Desktop的日志通常能在应用菜单中找到“查看日志”的选项里面会有详细的错误信息比如Python脚本语法错误、模块导入失败等这是排查问题的第一手资料。6. 进阶优化与扩展思路我们的基础版本已经能工作但作为一个可用的工具还有很大的优化和扩展空间。这里分享几个方向6.1 性能优化分块读取与流式传输目前我们是把整个文件一次性读入内存。如果文件很大比如几百MB的日志这会导致内存激增并且可能超出LLM上下文的处理能力。更优的方案是实现分块读取。改进思路修改read_file工具增加可选参数如chunk_size块大小例如1000字符和chunk_number第几块。Server每次只返回文件的一个片段。客户端或LLM可以多次调用通过标记或摘要来协调。更高级的做法是遵循MCP协议中关于“资源”Resources和“提示”Prompts的规范但这需要更复杂的实现。对于初版我们可以先提供一个简单的分页接口。6.2 功能扩展从“读”到“查”与“写”列表文件增加一个list_files工具接收一个directory_path参数返回指定目录下的文件列表。这能帮助LLM先“浏览”你的文件系统再决定读哪个。搜索文件内容增加一个search_in_files工具接收query搜索词和可选的文件扩展名过滤利用如grepLinux/macOS或内置的字符串搜索遍历文件返回匹配的行及其所在文件。这直接从“文件读取器”升级为“知识库检索器”。受限写入增加一个append_to_file工具允许LLM向指定文件追加内容例如记录对话要点。务必极度谨慎必须严格限制可写入的目录和文件类型并做好输入清洗防止注入攻击。6.3 提升安全性配置化与权限细分根目录可配置不要硬编码BASE_DIR ./data。可以通过环境变量、配置文件或启动参数来设置使其更灵活。访问控制列表ACL实现一个简单的ACL。例如在配置中指定哪些工具可以访问哪些路径模式如read_file只能访问./docs/*.txt。这提供了更细粒度的控制。请求审计日志将所有工具调用请求包括参数记录到日志文件中便于事后审查和调试。6.4 提升可用性错误信息与工具描述优化更友好的错误给LLM返回的错误信息可以更结构化。例如除了文本信息还可以在content里提供错误代码type: “error”方便客户端程序化处理。工具描述的“咒语”优化description字段是给LLM看的“说明书”。精心设计它的措辞能极大提升工具被准确调用的概率。例如“读取用户指定路径下的文本文件。路径必须是相对于服务器数据根目录的相对路径。如果文件不存在或无法访问将返回明确错误。此工具适用于获取文档、日志、笔记等内容以供分析。”7. 常见问题与排查技巧实录在实际搭建和运行过程中你几乎一定会遇到下面这些问题。这里是我踩过坑后总结的排查清单。问题现象可能原因排查步骤与解决方案Claude Desktop 中看不到read_file工具1. MCP Server 配置错误或未成功启动。2. Server 的tools/list响应格式不正确。3. Claude Desktop 未加载新配置。1.检查配置路径确认claude_desktop_config.json中的command和args绝对路径正确无误。2.查看日志在Claude Desktop中寻找日志输出如Help - View Logs看是否有Server启动失败的错误信息如Python语法错误、模块找不到。3.重启应用确保完全退出并重启Claude Desktop。4.手动测试Server使用前面的test_server.py脚本确保Server能正确响应tools/list请求。调用工具时返回“安全错误”或“文件未找到”1. 文件路径参数file_path格式错误。2. 文件确实不存在于data目录下。3. 路径遍历防护逻辑阻止了访问。1.检查路径确认LLM传递的file_path是类似“subfolder/file.txt”的相对路径且没有以/开头。2.确认文件存在登录服务器查看./data/目录下是否存在目标文件。3.打印调试信息在Server代码中临时打印requested_path变量查看最终解析出的绝对路径是什么是否符合预期。Server 进程启动后立即退出1. Python脚本存在语法错误。2. 缺少必要的模块虽然我们只用标准库。3. 配置文件中的命令执行权限不足。1.直接运行脚本在终端执行python3 simple_file_server.py观察是否有Python语法错误提示。2.检查Python环境确认使用的是正确的Python3解释器。3.检查文件权限确保Python脚本文件具有可执行权限或配置中指定的解释器路径正确。读取中文文件内容乱码文件编码与Python默认编码UTF-8不匹配。常见于Windows系统创建的GBK编码文件。1.代码已处理我们的代码中已经包含了对UnicodeDecodeError的捕获并尝试用GBK编码重试。2.统一编码最根本的解决方法是规范文件编码建议将所有文本文件保存为UTF-8格式。处理大文件时Server无响应或内存占用高一次性读取整个文件到内存遇到大文件时效率低下且资源消耗大。1.短期缓解在工具描述中注明“适用于中小型文本文件”。2.长期方案参考6.1 节实现分块读取功能。独家避坑技巧开发阶段使用Stdio调试在早期先别急着对接Claude Desktop。用我们写的test_server.py模拟客户端进行测试能更快地定位协议逻辑和业务代码的错误。善用日志输出在Server代码的关键节点如收到请求、开始处理、发生错误向标准错误sys.stderr打印日志。这些日志通常会被MCP客户端捕获并显示在其日志中是线上调试的利器。从官方示例入手如果你觉得从零实现协议太复杂Anthropic官方提供了Python的MCP SDK (anthropic-mcp)。先用SDK快速实现一个功能相同的Server再对比我们手搓的代码能帮你更好地理解SDK帮你封装了哪些底层细节。理解底层协议后使用SDK会更加得心应手。手搓这个MCP Server的过程本质上是一次对LLM如何安全、结构化地与外界交互的深度探索。它不仅仅是一个工具更是一个明确的能力边界定义。当你看到自己用百行代码搭建的“桥梁”让强大的LLM能够安全、按需地汲取你本地知识库的养分时那种创造力和控制感相结合的感觉正是开发者乐趣的核心所在。你可以在此基础上将它改造成连接数据库、调用内部API、管理云资源的强大网关真正成为你AI工作流的定制化中枢。
返回列表