MCP协议：构建AI智能体与外部工具的安全标准化桥梁

1. 项目概述MCP——连接AI与数字世界的“万能适配器”如果你最近在折腾AI应用开发特别是想让大语言模型LLM能像人类一样操作电脑、读取文件、调用API那你大概率已经听说过“MCP”这个词了。isteamhq/mcp这个GitHub仓库正是MCPModel Context Protocol协议的核心参考实现。简单来说它不是一个具体的应用而是一套标准、一套工具包旨在解决一个核心痛点如何让AI智能体Agent安全、标准化地接入和使用外部工具与数据。想象一下你开发了一个AI助手希望它能帮你分析服务器日志、从数据库拉取报表、甚至控制智能家居。在没有MCP之前你需要为每个功能单独编写一套复杂的对接代码处理认证、数据格式转换、错误处理还得担心安全问题。MCP的出现就是为了终结这种“手工作坊”式的集成。它定义了一套通用的“语言”协议让任何工具我们称之为“服务器”Server都能以一种AI能理解的方式向LLM我们称之为“客户端”Client宣告自己“能做什么”以及“如何调用”。isteamhq/mcp提供了实现这套协议的Python SDK和示例是开发者构建或集成MCP兼容工具的首选起点。这个项目适合所有正在或计划构建AI原生应用的开发者、希望将自己的服务或数据暴露给AI智能体的团队以及对AI Agent架构和工具调用生态感兴趣的工程师。通过它你可以快速理解MCP的核心思想并亲手搭建起连接AI与真实世界的第一座桥梁。2. MCP核心架构与设计哲学拆解2.1 为什么是MCP从“硬编码”到“动态发现”的范式转变在MCP之前AI工具调用的主流方式是“硬编码”或“预定义函数调用”。开发者需要预先在提示词Prompt或系统指令中以自然语言描述工具的功能和参数并在代码层面为每个工具编写固定的处理逻辑。这种方式存在几个致命缺陷扩展性差每增加一个新工具都需要修改客户端的提示词和代码重新部署无法实现工具的“即插即用”。维护成本高工具的功能、参数一旦变化所有依赖它的客户端都需要同步更新极易出现版本不一致导致的错误。安全性模糊工具权限控制往往与业务逻辑耦合缺乏统一的、声明式的权限管理机制。体验割裂不同的AI平台如Claude Desktop、Cursor、Windscope对工具的定义和调用方式各不相同开发者需要为每个平台做适配。MCP的设计哲学正是为了解决这些问题。它将工具提供方Server和工具使用方Client彻底解耦。Server负责以标准格式声明自己的能力称为“Tools”和“Resources”Client则在运行时动态发现并调用这些能力。这就像电脑的USB接口外设Server只要遵循USB协议插入电脑Client就能被自动识别和使用无需为每个外设重装系统。2.2 协议核心组件Tools、Resources与PromptsMCP协议定义了三种核心的上下文类型这是理解其能力的关键Tools工具代表可执行的操作。一个Tool由名称、描述、输入参数模式JSON Schema定义组成。例如一个“搜索天气”的Tool其参数模式可能要求传入city字符串和unit枚举celsius或fahrenheit。当Client如Claude调用这个Tool时需要提供符合模式的参数Server执行后返回结果。Resources资源代表可读取的静态或动态数据源。每个Resource有一个唯一的URI如file:///path/to/log.txt或sql://query/latest_users和一个MIME类型如text/plainapplication/json。Client可以“读取”Resource来获取其内容。这比单纯的文件访问更强大因为Resource可以是数据库查询的视图、API的封装甚至是实时生成的日志流。Prompts提示词模板这是MCP一个非常巧妙的设计。它允许Server预定义一些可复用的提示词片段或模板。Client可以获取这些Prompt并将其组合到自己的系统指令或用户提问中。例如一个代码库分析Server可以提供一个名为“code_review_ checklist”的Prompt里面包含针对特定编程语言的代码审查要点。这样AI在执行代码审查任务时就能动态注入更专业的上下文。这种设计使得AI的能力边界可以无限扩展。一个MCP Server可以同时提供查询数据库Resource、发送邮件Tool、获取代码审查模板Prompt等多种能力而AI Client只需连接这一个Server就能获得一个完整的“技能包”。2.3 通信与安全模型SSE与标准化授权MCP的通信基于HTTP和Server-Sent Events (SSE)。这是一种轻量级的、服务器向客户端单向推送事件的技术非常适合MCP这种由Server主动通知工具列表变更的场景。连接建立后Client和Server通过交换结构化的JSON消息进行通信。安全是MCP设计的重中之重。协议本身不强制规定具体的授权机制但强烈建议并提供了标准化的支持。核心安全理念包括最小权限原则Server声明的每个Tool和Resource都可以附带详细的权限说明。Client或最终用户可以明确知晓某个操作会访问哪些数据或执行什么命令。显式用户授权对于敏感操作应采用“请求-授权”流程。例如当AI试图调用“发送邮件”Tool时Client应用应该弹窗让用户确认收件人和内容用户批准后Client才将执行请求发送给Server。传输安全所有通信都应使用TLSHTTPS/ WSS加密。认证与审计Server可以实现各种认证方式API Key, OAuth等并为所有操作记录详细的审计日志。isteamhq/mcp的Python SDK内置了处理这些通信和安全协商的基础设施让开发者能聚焦于业务逻辑的实现。3. 使用isteamhq/mcpSDK 快速构建你的第一个MCP Server3.1 环境准备与项目初始化让我们从零开始构建一个简单的“系统信息查询”MCP Server。这个Server将提供一个Tool来获取当前服务器的负载情况并提供一个Resource来展示一个静态的欢迎信息。首先确保你的Python环境版本在3.8以上。然后创建一个新的项目目录并安装核心依赖mkdir my-mcp-server cd my-mcp-server python -m venv venv source venv/bin/activate # Windows: venv\Scripts\activate pip install mcpmcp这个包就是isteamhq/mcp提供的官方Python SDK。它非常轻量核心是提供了用于构建Server和Client的抽象类、协议消息的数据模型以及网络通信的处理逻辑。接下来我们创建主要的服务器文件server.py# server.py import asyncio import psutil # 我们需要这个库来获取系统信息先安装pip install psutil from mcp.server import Server from mcp.server.models import InitializationOptions import mcp.server.stdio from mcp.types import Tool, Resource, TextContent, Prompt # 初始化MCP Server app Server(system-info-server)3.2 定义Tools实现get_system_loadTools是MCP的核心。我们使用app.list_tools()装饰器来声明Server提供的工具列表使用app.call_tool()来处理具体的工具调用。# 在 server.py 中继续添加 app.list_tools() async def handle_list_tools() - list[Tool]: 返回服务器提供的工具列表 return [ Tool( nameget_system_load, description获取当前系统的CPU、内存和磁盘使用率。, inputSchema{ type: object, properties: {} # 这个工具不需要输入参数 } ) ] app.call_tool() async def handle_call_tool(name: str, arguments: dict | None) - list[TextContent]: 处理工具调用请求 if name get_system_load: # 使用psutil获取系统信息 cpu_percent psutil.cpu_percent(interval0.5) memory psutil.virtual_memory() disk psutil.disk_usage(/) result_text f ## 系统负载报告 - **CPU使用率**: {cpu_percent}% - **内存使用**: {memory.used / (1024**3):.2f} GB / {memory.total / (1024**3):.2f} GB ({memory.percent}%) - **磁盘使用 (根目录)**: {disk.used / (1024**3):.2f} GB / {disk.total / (1024**3):.2f} GB ({disk.percent}%) return [TextContent(typetext, textresult_text)] else: raise ValueError(f未知工具: {name})注意Tool定义中的inputSchema至关重要。它使用JSON Schema精确描述了Client调用时必须提供的参数格式。这里是空的因为我们的工具不需要参数。对于需要参数的Tool如search_file需要filename这里必须详细定义这能极大帮助LLM生成正确的调用参数。3.3 定义Resources提供静态欢迎信息Resources 用于提供数据。我们声明一个欢迎信息的Resource。# 在 server.py 中继续添加 app.list_resources() async def handle_list_resources() - list[Resource]: 返回服务器提供的资源列表 return [ Resource( urifile://welcome, name系统信息服务器欢迎页, description本MCP服务器的介绍和功能说明。, mimeTypetext/plain ) ] app.read_resource() async def handle_read_resource(uri: str) - list[TextContent]: 读取指定URI的资源内容 if uri file://welcome: content 欢迎使用系统信息查询MCP服务器 本服务器提供以下能力 1. **工具 (Tools)**: - get_system_load: 实时获取CPU、内存和磁盘使用率。 2. **资源 (Resources)**: - 本欢迎页面。 未来可能添加更多功能如进程列表、网络连接状态等。 使用方式在支持MCP的客户端如Claude Desktop中连接本服务器即可。 return [TextContent(typetext, textcontent)] else: raise ValueError(f资源未找到: {uri})3.4 运行与测试你的MCP ServerMCP Server通常通过标准输入输出stdio或HTTP与Client通信。对于开发和测试stdio方式最简单。在server.py末尾添加启动代码# 在 server.py 末尾添加 async def main(): async with mcp.server.stdio.stdio_server() as (read_stream, write_stream): await app.run( read_stream, write_stream, InitializationOptions( server_namesystem-info-server, server_version0.1.0 ) ) if __name__ __main__: asyncio.run(main())现在运行你的服务器python server.py。此时程序看起来会挂起因为它正在等待通过stdio接收来自Client的协议消息。为了测试我们需要一个MCP Client。一个快速测试的方法是使用isteamhq/mcp仓库中提供的示例Client脚本或者使用更直观的测试工具如MCP Inspector。这里介绍一个更开发者友好的方式使用SDK自带的简单测试客户端。创建一个test_client.py# test_client.py import asyncio import subprocess import json import sys async def test_server(): # 启动server.py子进程并连接到它的stdio proc await asyncio.create_subprocess_exec( sys.executable, server.py, stdinsubprocess.PIPE, stdoutsubprocess.PIPE, stderrsubprocess.PIPE ) # 这里需要模拟MCP协议握手和消息交换比较复杂 # 更推荐使用官方提供的cli工具或MCP Inspector进行测试 print(Server started. Use a proper MCP client (e.g., Claude Desktop) to connect.) # 为了演示我们只是保持进程运行 await proc.wait() if __name__ __main__: asyncio.run(test_server())实际上对于日常开发和集成测试最推荐的方法是直接连接到真实的MCP客户端环境例如Claude Desktop。你可以在Claude Desktop的设置中添加一个本地MCP Server配置其命令为python /path/to/your/server.py。连接成功后你就可以在Claude的对话中直接使用get_system_load来调用你的工具了。4. 高级实践构建一个实用的“项目文件检索”MCP Server4.1 设计支持复杂参数的工具让我们构建一个更实用的Server一个能检索当前项目目录下文件内容的工具。它需要支持按文件名模糊搜索、按文件类型过滤并能读取文件内容。首先定义更复杂的Tool Schema# project_file_server.py app.list_tools() async def handle_list_tools() - list[Tool]: return [ Tool( namesearch_project_files, description在指定项目根目录下搜索并列出文件。支持按名称模糊匹配和按扩展名过滤。, inputSchema{ type: object, properties: { root_path: { type: string, description: 要搜索的根目录路径。默认为当前工作目录。 }, name_pattern: { type: string, description: 文件名模糊匹配模式如*.py, config*。 }, extensions: { type: array, items: {type: string}, description: 文件扩展名过滤列表如[.py, .js]。 }, max_results: { type: integer, description: 返回的最大结果数量。, default: 20 } } } ), Tool( nameread_file_content, description读取指定文件的文本内容。, inputSchema{ type: object, properties: { file_path: { type: string, description: 要读取的文件的绝对路径或相对于项目根目录的路径。 }, max_lines: { type: integer, description: 最多返回的行数用于预览。默认为100。, default: 100 } }, required: [file_path] } ) ]4.2 实现工具逻辑与错误处理接下来实现工具调用的具体逻辑并加入健壮的错误处理。# project_file_server.py 继续 import os import glob from pathlib import Path app.call_tool() async def handle_call_tool(name: str, arguments: dict | None) - list[TextContent]: if name search_project_files: if not arguments: arguments {} root_path arguments.get(root_path, os.getcwd()) name_pattern arguments.get(name_pattern, *) extensions arguments.get(extensions, []) max_results arguments.get(max_results, 20) # 安全检查防止目录遍历攻击 try: root_path os.path.abspath(root_path) # 可以在这里添加路径白名单检查 if not os.path.isdir(root_path): return [TextContent(typetext, textf错误路径 {root_path} 不是一个有效的目录。)] except Exception as e: return [TextContent(typetext, textf路径解析错误{e})] search_pattern os.path.join(root_path, **, name_pattern) matched_files [] try: for file_path in glob.glob(search_pattern, recursiveTrue): if os.path.isfile(file_path): if extensions: if Path(file_path).suffix.lower() not in [ext.lower() for ext in extensions]: continue matched_files.append(file_path) if len(matched_files) max_results: break except Exception as e: return [TextContent(typetext, textf搜索文件时出错{e})] if not matched_files: result_text f在目录 {root_path} 下未找到匹配模式 {name_pattern} 的文件。 else: result_text f在 {root_path} 下找到 {len(matched_files)} 个文件\n\n for idx, fp in enumerate(matched_files, 1): rel_path os.path.relpath(fp, root_path) size os.path.getsize(fp) result_text f{idx}. {rel_path} ({size} 字节)\n return [TextContent(typetext, textresult_text)] elif name read_file_content: if not arguments or file_path not in arguments: return [TextContent(typetext, text错误缺少必要参数 file_path。)] file_path arguments[file_path] max_lines arguments.get(max_lines, 100) # 安全检查确保文件路径在允许的范围内 try: # 这里假设文件路径是相对于当前工作目录或一个预设的项目根目录 # 在实际应用中必须进行严格的规范化路径检查和访问控制 abs_path os.path.abspath(file_path) # 示例限制只能访问当前工作目录及其子目录 if not abs_path.startswith(os.path.abspath(.)): return [TextContent(typetext, text错误无权访问指定路径。)] if not os.path.isfile(abs_path): return [TextContent(typetext, textf错误路径 {file_path} 不是一个文件。)] # 检查文件大小防止读取过大文件 if os.path.getsize(abs_path) 1024 * 1024: # 1MB return [TextContent(typetext, text错误文件过大出于安全考虑拒绝读取。)] except Exception as e: return [TextContent(typetext, textf文件路径安全检查失败{e})] # 读取文件内容 try: with open(abs_path, r, encodingutf-8, errorsignore) as f: lines [] for i, line in enumerate(f): if i max_lines: lines.append(f... (文件超过{max_lines}行已截断)) break lines.append(line.rstrip(\n)) content \n.join(lines) file_name os.path.basename(abs_path) result_text f文件 {file_name} 的内容前{len(lines)}行\n\n{content}\n except UnicodeDecodeError: result_text f文件 {os.path.basename(abs_path)} 似乎是二进制文件无法以文本形式预览。 except Exception as e: result_text f读取文件时出错{e} return [TextContent(typetext, textresult_text)] else: raise ValueError(f未知工具: {name})实操心得在实现文件操作类MCP工具时安全是头等大事。必须对输入路径进行规范化、检查是否在允许的目录范围内、限制文件大小、避免读取二进制文件等。上述代码提供了基础的安全检查示例在生产环境中需要根据具体需求制定更严格的访问控制策略例如配置明确的白名单目录。4.3 动态Resource与Prompt的应用我们可以让Resource动态起来例如提供一个显示最近修改文件的Resource。# project_file_server.py 继续 import time app.list_resources() async def handle_list_resources() - list[Resource]: return [ Resource( uridynamic://recent_files, name最近修改的文件, description列出项目目录中最近5分钟内被修改过的文件。, mimeTypeapplication/json # 我们可以返回JSON格式的数据 ) ] app.read_resource() async def handle_read_resource(uri: str) - list[TextContent]: if uri dynamic://recent_files: current_time time.time() five_min_ago current_time - 300 recent_files [] for root, dirs, files in os.walk(.): for file in files: file_path os.path.join(root, file) try: mtime os.path.getmtime(file_path) if mtime five_min_ago: recent_files.append({ path: file_path, modified: time.ctime(mtime) }) except OSError: pass # 将列表转换为JSON字符串返回 import json content json.dumps(recent_files, indent2, ensure_asciiFalse) return [TextContent(typetext, textcontent)] else: raise ValueError(f资源未找到: {uri})同样我们可以提供一个用于代码审查的Prompt模板# project_file_server.py 继续 app.list_prompts() async def handle_list_prompts() - list[Prompt]: return [ Prompt( namepython_code_review, description针对Python代码的审查要点提示模板。, arguments[ { name: code_snippet, description: 需要审查的Python代码片段, required: True } ] ) ] app.get_prompt() async def handle_get_prompt(name: str, arguments: dict | None) - Prompt: if name python_code_review: if not arguments or code_snippet not in arguments: # 返回一个默认的提示模板 template 请对以下Python代码进行审查重点关注 1. **代码风格**是否符合PEP 8规范如命名、缩进 2. **潜在错误**是否有语法错误、未处理的异常或逻辑缺陷 3. **性能问题**是否存在低效的循环、重复计算或不必要的数据拷贝 4. **可读性与维护性**代码结构是否清晰函数和变量名是否具有描述性 请提供具体的修改建议。 else: code arguments[code_snippet] template f请对以下Python代码进行审查重点关注 1. **代码风格**是否符合PEP 8规范 2. **潜在错误**是否有语法错误或逻辑缺陷 3. **性能问题**是否存在可优化的地方 4. **可读性与维护性**代码结构是否清晰 待审查代码 python {code}请提供具体的修改建议。 # 注意get_prompt返回的是Prompt对象而不是TextContent列表 return Prompt( namename, descriptionPython代码审查模板, arguments[{name: code_snippet, description: Python代码, required: True}] ) else: raise ValueError(f未知提示词: {name})这个Prompt可以被AI客户端获取并动态插入到对话上下文中从而让AI具备更专业的代码审查能力。 ## 5. 集成、调试与生产部署指南 ### 5.1 与主流AI平台集成以Claude Desktop为例 构建好MCP Server后最大的价值在于被AI客户端使用。**Claude Desktop** 是目前对MCP支持最完善、体验最好的客户端之一。 集成步骤非常简单 1. 打开Claude Desktop进入 Settings - Developer - Edit Config。 2. 在打开的JSON配置文件中找到或添加 mcpServers 字段。它是一个对象键是自定义的服务器名值是该服务器的配置。 3. 添加你的服务器配置。对于本地运行的Python脚本使用 command 模式 json { mcpServers: { my-project-files: { command: /absolute/path/to/your/venv/bin/python, args: [/absolute/path/to/your/project_file_server.py], env: { PYTHONPATH: /absolute/path/to/your/project } } } }保存配置文件并重启Claude Desktop。重启后在Claude的输入框输入你应该能看到my-project-files这个服务器下的工具如search_project_files出现在自动补全列表中。选择它并输入参数Claude就会调用你的Server并返回结果。注意事项确保command中的Python解释器路径和脚本路径是绝对路径。环境变量PYTHONPATH有时是必要的特别是你的Server脚本依赖了项目内的其他自定义模块。5.2 调试与问题排查技巧实录在开发MCP Server时你可能会遇到各种问题。以下是一些常见问题及排查方法问题1Claude Desktop无法连接或找不到我的工具。检查点1Server进程是否正常启动在终端直接运行python your_server.py观察是否有报错。常见的错误包括缺少依赖如psutil、语法错误或导入错误。实操心得在Server脚本开头添加简单的日志输出如print(MCP Server starting..., filesys.stderr)。Claude Desktop会捕获stderr输出这有助于在客户端看到启动日志。检查点2Claude Desktop配置是否正确检查配置文件路径和格式。JSON格式必须严格正确最后一个条目后不能有逗号。确保command和args中的路径无误。一个快速验证的方法是在终端中手动执行配置中的完整命令如/path/to/python /path/to/server.py看脚本是否能正常运行并等待输入。检查点3协议初始化是否成功MCP协议要求Server在启动后立即发送initialize请求。确保你的Server正确实现了app.run()并处理了初始化握手。isteamhq/mcpSDK已经处理了大部分协议逻辑通常只要按照示例编写就不会有问题。问题2工具调用失败AI返回“调用工具时出错”。检查点1工具参数格式是否正确AI生成的参数必须完全匹配你定义的inputSchema。使用MCP Inspector这类工具可以直观地查看Server声明的所有Tool及其Schema并手动发起调用测试这比通过AI调试要直接得多。在你的handle_call_tool函数中加入详细的错误日志打印收到的name和arguments确保它们符合预期。检查点2工具函数本身是否有未处理的异常用try...except包裹你的核心业务逻辑并返回友好的错误信息给Client而不是让异常抛出导致整个协议会话中断。问题3Server运行一段时间后崩溃或无响应。检查点资源管理与超时。检查你的工具函数是否有可能陷入死循环或长时间阻塞的操作。考虑为耗时操作添加超时机制asyncio.wait_for。确保没有内存泄漏。对于需要访问大量文件或数据的工具使用生成器或分页查询避免一次性加载所有数据。5.3 性能优化与生产部署考量当你的MCP Server从原型走向生产环境时需要考虑以下几点长期运行与稳定性生产环境的Server不应是简单的脚本而应该作为一个守护进程或服务来运行。可以考虑使用systemd(Linux)、launchd(macOS) 或NSSM(Windows) 来管理进程实现开机自启和崩溃重启。连接管理与多路复用一个Server实例可能同时服务多个Client连接。isteamhq/mcp的SDK基于异步IO能够较好地处理并发。但要确保你的工具函数是线程安全或协程安全的避免在共享状态上产生竞态条件。认证与授权强化对于暴露敏感数据或操作的工具必须实现严格的认证。MCP支持在初始化阶段传递client_info和进行身份验证。你可以在Server端验证API Key或者集成OAuth等标准协议。所有敏感操作的调用都应在Client端如Claude Desktop得到用户的明确确认。监控与日志集成像structlog或loguru这样的日志库为所有工具调用和资源访问记录结构化的日志包括时间、调用者Client、工具名、参数摘要和结果状态。这对于审计、调试和用量分析至关重要。使用HTTP/WebSocket传输对于生产环境stdio可能不是最佳选择。MCP协议也支持HTTP/WebSocket。你可以使用mcp.server创建ASGI或WSGI应用然后使用像uvicorn这样的ASGI服务器来部署这样可以获得更好的可扩展性、负载均衡和监控能力。构建一个健壮的MCP Server其核心思想与构建任何微服务一致关注点分离、定义清晰的接口、实现健壮的错误处理、保障安全、并做好可观测性。isteamhq/mcp项目提供的协议和SDK为你解决了最困难的标准化通信问题让你可以专注于创造有价值的工具本身。