当前位置：首页 > news >正文

GenAI-MCP：大模型工具调用的标准化协议与实践指南

news 2026/5/7 6:17:11

1. 项目概述：一个连接大模型与外部工具的“万能适配器”

最近在折腾大语言模型应用开发的朋友，估计都遇到过同一个头疼的问题：怎么让模型去调用外部的工具和服务？比如，你想让模型帮你查查天气、发封邮件，或者分析一下数据库里的数据。直接让模型去干这些事，它办不到，因为它本质上是个“语言专家”，不是“行动专家”。这时候，你就需要一个可靠的“中间人”，把模型的“想法”翻译成具体的“动作指令”，再把执行结果“翻译”回模型能理解的语言。

adamydwang/genai-mcp这个项目，就是这样一个被设计出来的“中间人”，或者说“万能适配器”。它的全称是GenAI Model Context Protocol (MCP)，你可以把它理解为一套标准化的“插头”和“插座”规范。任何工具（比如计算器、文件系统、数据库客户端）只要按照 MCP 的标准做一个“插头”（即 MCP 服务器），就能无缝插入到支持 MCP 的“插座”（即 MCP 客户端，通常是你的 AI 应用框架）上。这样一来，你的大模型应用就能像搭积木一样，灵活地扩展各种能力。

这个项目的核心价值在于标准化和解耦。在没有 MCP 之前，每个 AI 应用框架（比如 LangChain、LlamaIndex）都要为每一个想集成的工具编写特定的适配器代码，工作重复且维护成本高。而 MCP 定义了一套通用的协议，工具开发者只需实现一次 MCP 服务器，就能被所有支持 MCP 的客户端使用。对于应用开发者来说，这意味着工具集成变得前所未有的简单和统一。

2. 核心设计思路：协议先行，实现分离

2.1 为什么是“协议”而不是“库”？

很多同类项目选择直接提供一个集成好的工具库，但genai-mcp选择了更底层、更彻底的路径：定义协议。这背后的考量非常深刻。

首先，生态兼容性。AI 应用框架层出不穷，今天流行 LangChain，明天可能就有新的框架崛起。如果genai-mcp只是一个为特定框架编写的工具库，那么它的生命力将和该框架深度绑定。而作为一个协议，它可以被任何框架实现和采纳，具备了成为行业事实标准的潜力。

其次，职责清晰。协议只规定“通信的语言和格式”，而不关心具体用什么编程语言实现服务器，也不关心客户端内部如何调度工具。这种关注点分离（Separation of Concerns）让工具开发者可以专注于工具本身的逻辑，应用开发者可以专注于提示工程和业务流程，双方通过协议接口交互，互不干扰。

最后，未来证明。工具的能力和种类是不断增长的。一个固定的库需要不断更新才能支持新工具，而一个设计良好的协议，可以通过扩展（如定义新的资源类型、工具参数）来适应未来需求，具有更好的可扩展性。

2.2 MCP 协议的核心组件解析

MCP 协议主要围绕几个核心概念构建，理解它们就掌握了整个项目的脉络。

资源（Resources）：这是模型可以“读取”或“感知”到的外部信息实体。比如，一个文件路径、一个数据库表的模式（Schema）、一个网页的 URL。资源本身不是可执行的动作，而是为模型提供上下文信息。例如，在让模型写 SQL 之前，先通过 MCP 把数据库表结构作为“资源”提供给模型，它才能写出正确的查询。

工具（Tools）：这是模型可以“调用”以执行具体操作的接口。每个工具都有明确的输入参数（input_schema）和输出格式。例如，“执行 SQL 查询”是一个工具，它的输入是数据库连接字符串和 SQL 语句，输出是查询结果集。“发送邮件”也是一个工具，输入是收件人、主题、正文等。

提示词模板（Prompts）：这是一些预定义的、可复用的对话开场白或指令模板。客户端可以获取这些模板并填充变量，快速引导模型进入特定任务场景。比如，一个“代码审查”提示词模板，可以预设好审查的维度和格式。

MCP 服务器通过标准输入输出（stdio）或 HTTP 等传输层，向客户端宣告自己提供了哪些资源、工具和提示词模板。客户端（你的 AI 应用）在运行时，根据需要调用这些工具，或读取资源来丰富模型的上下文。

2.3 传输层与会话管理

协议定义了“说什么”，还需要规定“怎么送”。MCP 支持多种传输方式：

标准输入输出（stdio）：最常见的方式，服务器作为一个独立的子进程启动，通过管道与父进程（客户端）通信。这种方式部署简单，适合本地工具。
HTTP：服务器作为一个常驻的 Web 服务运行，客户端通过 HTTP/SSE 与之通信。这种方式适合远程服务或需要高可用性的场景。
其他：理论上，任何能满足双向、有序消息传递的通道都可以作为传输层。

会话初始化时，客户端和服务器会进行“握手”，交换各自的实现名称、版本和支持的协议版本。之后，客户端可以列出（list）可用的资源、工具和提示词，并通过调用（call）来使用工具，通过读取（read）来获取资源内容。

注意：MCP 协议本身不处理身份认证、授权或计费。这些是传输层或具体工具实现需要解决的问题。例如，一个需要 API 密钥的数据库工具，应该在服务器启动时或工具调用时，由客户端提供密钥。

3. 实操指南：从零构建一个 MCP 服务器

理解了设计思路，我们动手实现一个最简单的 MCP 服务器，这将让你对协议有最直观的认识。我们将创建一个“单位换算”服务器，它提供一个工具，可以将英里转换为公里。

3.1 环境准备与项目初始化

我们使用 Python 来实现，因为它有丰富的生态和相对简洁的 MCP 支持库。这里我们使用官方推荐的mcpSDK。

# 创建项目目录并进入 mkdir mcp-unit-converter && cd mcp-unit-converter # 创建虚拟环境（推荐） python -m venv .venv # 激活虚拟环境 # Windows: .venv\Scripts\activate # Linux/Mac: source .venv/bin/activate # 安装 mcp 库 pip install mcp

接下来，创建我们的服务器主文件server.py。

3.2 实现核心工具函数

首先，我们实现核心的业务逻辑函数。这个函数与 MCP 协议无关，就是一个普通的 Python 函数。

# server.py def convert_miles_to_km(miles: float) -> dict: """ 将英里转换为公里。 参数: miles (float): 英里数，必须为非负数。 返回: dict: 包含原始值、转换值和单位的字典。 """ if miles < 0: raise ValueError("英里数不能为负数") kilometers = miles * 1.60934 return { "original_value": miles, "original_unit": "miles", "converted_value": round(kilometers, 2), "converted_unit": "kilometers" }

3.3 集成 MCP 协议：创建服务器

现在，我们需要将这个函数包装成一个 MCP 工具。我们使用mcp库提供的Server类。

# server.py (续) from mcp.server import Server from mcp.server.models import InitializationOptions import mcp.server.stdio import asyncio from pydantic import BaseModel # 定义工具输入参数的模型，这会被用于生成 JSON Schema class MilesToKmInput(BaseModel): miles: float # 创建 MCP 服务器实例 server = Server("unit-converter-server") # 向服务器注册工具 @server.list_tools() async def handle_list_tools(): # 返回服务器提供的所有工具列表 return [ { "name": "convert_miles_to_kilometers", "description": "将英里转换为公里。输入为英里数（浮点数）。", "inputSchema": { "type": "object", "properties": { "miles": { "type": "number", "description": "需要转换的英里数，必须为非负数。" } }, "required": ["miles"] } } ] # 处理工具调用 @server.call_tool() async def handle_call_tool(name: str, arguments: dict): if name == "convert_miles_to_kilometers": # 验证并解析参数 try: input_data = MilesToKmInput(**arguments) except Exception as e: return { "content": [{"type": "text", "text": f"参数错误: {e}"}], "isError": True } # 调用业务逻辑函数 try: result = convert_miles_to_km(input_data.miles) result_text = f"{result['original_value']} {result['original_unit']} 等于 {result['converted_value']} {result['converted_unit']}" return { "content": [{"type": "text", "text": result_text}] } except ValueError as e: return { "content": [{"type": "text", "text": f"转换错误: {e}"}], "isError": True } # 如果工具名未找到 return { "content": [{"type": "text", "text": f"未知工具: {name}"}], "isError": True } # 主异步函数，启动服务器 async def main(): # 使用标准输入输出作为传输层运行服务器 async with mcp.server.stdio.stdio_server() as (read_stream, write_stream): await server.run( read_stream, write_stream, InitializationOptions( server_name="unit-converter", server_version="0.1.0" ) ) if __name__ == "__main__": asyncio.run(main())

3.4 测试你的 MCP 服务器

为了测试，我们可以使用一个简单的 MCP 客户端脚本，或者使用现有的支持 MCP 的客户端。这里我们用mcp库自带的简易测试方式。首先，确保服务器在运行：

python server.py

此时程序会挂起，等待标准输入。我们需要另一个终端或脚本来调用它。创建一个简单的测试客户端test_client.py：

# test_client.py import asyncio import json import subprocess import sys async def test_tool(): # 启动服务器进程 proc = await asyncio.create_subprocess_exec( sys.executable, 'server.py', stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE ) # 简单的测试协议：发送初始化请求 init_request = { "jsonrpc": "2.0", "id": 1, "method": "initialize", "params": { "protocolVersion": "0.1.0", "capabilities": {}, "clientInfo": {"name": "test-client"} } } proc.stdin.write((json.dumps(init_request) + '\n').encode()) await proc.stdin.drain() # 读取初始化响应（简化处理，实际协议更复杂） line = await proc.stdout.readline() print("Server response:", line.decode()) # 在实际完整实现中，这里需要继续完成握手，然后发送 tools/list 和 tools/call 请求 # 此处为演示，我们直接终止 proc.terminate() await proc.wait() if __name__ == "__main__": asyncio.run(test_tool())

更实用的方法是使用一个已经实现了 MCP 客户端的应用来测试，比如 Claude Desktop（如果其配置支持自定义 MCP 服务器）或者一些开发中的 CLI 工具。对于开发调试，关注服务器进程是否正常启动、有无报错信息是第一步。

4. 高级应用：构建一个文件系统浏览器 MCP 服务器

单一工具服务器展示了基础，但 MCP 的真正威力在于管理复杂的资源和工具集。让我们构建一个更实用的“文件系统浏览器”服务器，它向模型暴露指定目录下的文件列表（作为资源），并允许模型读取文件内容（作为工具）。

4.1 设计资源与工具

资源file://{path}：表示一个文件路径。客户端可以“列出”所有此类资源（即某个目录下的所有文件），从而让模型知道存在哪些文件。
工具read_file：输入一个文件路径，返回该文件的文本内容。这是模型“读取”资源内容的方式。

4.2 服务器实现代码

# filesystem_server.py import os from pathlib import Path from typing import List from mcp.server import Server from mcp.server.models import InitializationOptions import mcp.server.stdio import asyncio from pydantic import BaseModel # 配置：允许浏览的根目录（安全考虑！） ALLOWED_ROOT = Path.home() / "documents" / "ai_accessible" server = Server("filesystem-browser") class ReadFileInput(BaseModel): filepath: str @server.list_resources() async def handle_list_resources() -> List[dict]: """列出 ALLOWED_ROOT 目录下所有文件的资源 URI""" resources = [] try: for item in ALLOWED_ROOT.rglob("*"): if item.is_file(): # 构造资源 URI，例如 file:///home/user/documents/ai_accessible/note.txt uri = f"file://{item.absolute()}" resources.append({ "uri": uri, "name": item.name, "description": f"文件位于 {item.relative_to(ALLOWED_ROOT)}", "mimeType": "text/plain" # 简化处理，实际应根据后缀判断 }) except Exception as e: print(f"列出资源时出错: {e}") return resources @server.list_tools() async def handle_list_tools(): return [ { "name": "read_file", "description": "读取指定路径的文本文件内容。路径必须在允许的目录内。", "inputSchema": { "type": "object", "properties": { "filepath": { "type": "string", "description": "要读取的文件的绝对路径。" } }, "required": ["filepath"] } } ] @server.call_tool() async def handle_call_tool(name: str, arguments: dict): if name == "read_file": try: input_data = ReadFileInput(**arguments) file_path = Path(input_data.filepath) # 安全性检查：确保请求的文件在允许的根目录下 if not file_path.is_absolute(): file_path = ALLOWED_ROOT / file_path file_path = file_path.resolve() # 解析符号链接等 if not str(file_path).startswith(str(ALLOWED_ROOT.resolve())): return { "content": [{"type": "text", "text": "错误：无权访问指定路径。"}], "isError": True } if not file_path.is_file(): return { "content": [{"type": "text", "text": "错误：路径不是一个文件或不存在。"}], "isError": True } # 读取文件内容（假设是文本文件） content = file_path.read_text(encoding='utf-8', errors='ignore') # 限制返回内容长度，防止上下文爆炸 preview = content[:5000] + ("..." if len(content) > 5000 else "") return { "content": [{"type": "text", "text": preview}] } except Exception as e: return { "content": [{"type": "text", "text": f"读取文件时出错: {e}"}], "isError": True } return { "content": [{"type": "text", "text": f"未知工具: {name}"}], "isError": True } async def main(): # 确保允许的根目录存在 ALLOWED_ROOT.mkdir(parents=True, exist_ok=True) print(f"文件系统服务器启动，可访问根目录: {ALLOWED_ROOT}") async with mcp.server.stdio.stdio_server() as (read_stream, write_stream): await server.run( read_stream, write_stream, InitializationOptions( server_name="filesystem-browser", server_version="0.1.0" ) ) if __name__ == "__main__": asyncio.run(main())

4.3 安全性与生产环境考量

这个文件服务器示例揭示了一个 MCP 服务器在生产中必须面对的核心问题：安全。

路径遍历攻击防护：代码中通过resolve()和startswith()检查，确保请求的文件路径严格位于ALLOWED_ROOT目录之下。这是防止攻击者通过../../../etc/passwd这类路径访问系统文件的关键。
资源消耗控制：我们限制了返回的文件内容长度（前5000字符）。对于大文件，更好的做法是提供分页读取工具，或者只返回元数据。
错误处理：将详细的系统错误信息（如PermissionError的具体路径）吞掉，只返回用户友好的错误提示，避免信息泄露。
传输层安全：如果使用 HTTP 传输，务必启用 HTTPS。对于 stdio，要确保启动服务器的进程本身是可信的。

实操心得：在实现任何提供系统访问能力的 MCP 服务器（如文件、数据库、命令执行）时，必须遵循“最小权限原则”。服务器进程应该以一个低权限的系统用户运行，并且只能访问完成其功能所必需的最少资源和系统调用。永远不要相信来自客户端的输入，必须进行严格的验证和清理。

5. 客户端集成：在 AI 应用中使用 MCP 工具

服务器准备好了，我们还需要一个客户端来使用它。这里以在 LangChain 中集成我们的单位换算器为例，展示如何将 MCP 工具接入现有的 AI 应用工作流。

5.1 使用`mcp`客户端库进行集成

首先，确保安装了包含客户端功能的mcp库。然后，我们可以编写一个脚本，动态加载 MCP 服务器并将其工具转化为 LangChain 可以使用的工具对象。

# langchain_mcp_client.py import asyncio from typing import List from mcp import ClientSession, StdioServerParameters from mcp.client.stdio import stdio_client from langchain.tools import Tool from langchain.agents import initialize_agent, AgentType from langchain.llms import OpenAI # 或使用其他兼容的LLM import os async def load_mcp_tools(server_command: List[str]) -> List[Tool]: """ 启动一个 MCP 服务器进程，并获取其所有工具，转换为 LangChain Tool 对象。 参数: server_command: 启动服务器的命令列表，如 ["python", "/path/to/server.py"] 返回: List[Tool]: LangChain 可用的工具列表。 """ tools = [] # 创建服务器参数 server_params = StdioServerParameters(command=server_command[0], args=server_command[1:]) # 创建客户端会话 async with stdio_client(server_params) as (read, write): async with ClientSession(read, write) as session: # 初始化会话 await session.initialize() # 列出服务器提供的所有工具 listed_tools = await session.list_tools() for tool_info in listed_tools: # 为每个工具定义一个异步调用函数 async def tool_func(**kwargs): async with stdio_client(server_params) as (r, w): async with ClientSession(r, w) as inner_session: await inner_session.initialize() result = await inner_session.call_tool(tool_info.name, arguments=kwargs) # 提取结果中的文本内容 for content in result.content: if content.type == "text": return content.text return str(result) # 包装成 LangChain Tool # 注意：这里需要处理工具函数的序列化问题，实际生产环境可能需要更复杂的包装 # 例如使用 langchain.tools.StructuredTool.from_function langchain_tool = Tool( name=tool_info.name, func=lambda **kw: asyncio.run(tool_func(**kw)), # 简化处理，实际需注意事件循环 description=tool_info.description, ) tools.append(langchain_tool) return tools # 示例：在主程序中使用 async def main(): # 1. 加载 MCP 工具 mcp_tools = await load_mcp_tools(["python", "server.py"]) # 我们的单位换算服务器 # 2. 初始化 LLM llm = OpenAI(temperature=0, openai_api_key=os.getenv("OPENAI_API_KEY")) # 3. 创建代理 agent = initialize_agent( mcp_tools, llm, agent=AgentType.ZERO_SHOT_REACT_DESCRIPTION, verbose=True ) # 4. 运行一个查询 result = await agent.arun("请问 26.2 英里是多少公里？") print(f"代理的最终回答: {result}") if __name__ == "__main__": asyncio.run(main())

5.2 集成中的挑战与解决方案

上面的示例是一个概念验证，在实际集成中你会遇到几个关键挑战：

会话管理：每次调用工具都重新启动服务器进程和会话是极其低效的。解决方案是实现一个连接池或会话管理器，保持服务器进程长连接，并在多个工具调用间复用。一些 MCP 客户端库（如 Claude Desktop 内置的）已经处理了这部分。
工具描述与提示工程：MCP 工具的描述 (description) 至关重要，它是模型决定是否以及如何调用该工具的主要依据。描述必须清晰、准确，包含输入参数的预期格式和示例。糟糕的描述会导致模型无法正确使用工具。
错误处理与重试：网络波动、服务器暂时无响应、工具执行超时等情况都需要在客户端有稳健的重试和降级机制。LangChain 本身提供了一些工具调用时的错误处理回调，需要合理配置。
多个 MCP 服务器的管理：一个复杂的 AI 应用可能同时需要连接文件服务器、数据库服务器、计算服务器等。客户端需要能够管理多个服务器连接，并根据工具名或功能动态路由请求。

6. 生态现状、最佳实践与未来展望

6.1 现有生态与工具

MCP 协议虽然较新，但生态正在快速成长。除了自行开发，已经有许多高质量的开源 MCP 服务器可供直接使用或参考：

官方示例与模板：modelcontextprotocol/servers仓库提供了大量参考实现，包括时钟、文件系统、搜索引擎、SQLite 数据库等。
第三方服务器：社区已经贡献了用于连接 GitHub、JIRA、Notion、Slack、各种数据库（PostgreSQL, MySQL）乃至智能家居平台（如 Home Assistant）的 MCP 服务器。
客户端支持：
- Claude Desktop：Anthropic 的 Claude 桌面应用直接支持配置 MCP 服务器，是体验 MCP 最便捷的方式。
- Cline：一个专为程序员设计的 IDE 插件，深度集成 MCP，让 AI 助手能直接操作代码库。
- 其他框架：LangChain 和 LlamaIndex 社区正在积极集成 MCP 支持，未来有望成为标准配置。

6.2 开发与部署最佳实践

从简单开始：先实现一个功能单一、边界清晰的服务器。验证协议通信和基础功能无误后，再逐步增加复杂功能。
完善的日志：在服务器中增加详细日志，记录收到的请求、调用的参数、执行结果和错误。这对于调试分布式问题至关重要。可以使用structlog或loguru等库。
配置化：将服务器监听的地址、端口、允许访问的路径、API 密钥等所有可变参数通过环境变量或配置文件管理，而不是硬编码在代码中。
健康检查：如果部署为 HTTP 服务，实现一个/health端点，供容器编排系统（如 Kubernetes）或负载均衡器检查服务状态。
版本化：为你发布的 MCP 服务器定义清晰的版本号（遵循语义化版本控制）。在协议初始化时声明版本，便于客户端兼容性处理。

6.3 常见问题排查实录

在实际开发和集成中，我遇到过一些典型问题，这里分享排查思路：

问题一：客户端连接服务器后，无法列出任何工具。

排查步骤：
1. 检查服务器启动日志，确认无报错且正常进入了server.run循环。
2. 在服务器的@server.list_tools()装饰函数内添加打印语句，确认函数被调用。
3. 检查返回的工具列表格式是否正确，必须是一个字典列表，每个字典包含name,description,inputSchema等必需字段。
4. 使用mcp库提供的mcp inspectCLI 工具（如果有）直接测试服务器，隔离客户端问题。
根本原因：最常见的是工具列表的返回格式不符合协议规范，或者inputSchema的 JSON Schema 定义有误。

问题二：工具调用成功，但模型无法理解返回的结果。

排查步骤：
1. 检查工具调用返回的content字段。它必须是一个列表，列表中的每一项是一个包含type和text（或image等）字段的对象。type: "text"是模型最擅长处理的。
2. 确保返回的文本内容是结构化的、清晰的。避免返回冗长的、未经处理的 JSON 或二进制数据。
3. 在工具的description中说明返回值的格式，例如“返回一个包含‘结果’和‘单位’字段的 JSON 对象”。
根本原因：MCP 协议规定了返回格式，但内容的质量和可读性由服务器开发者负责。模型对结构良好的文本解析能力更强。

问题三：服务器进程在工具调用一次后退出或无响应。

排查步骤：
1. 检查服务器代码中是否有未捕获的异常导致进程崩溃。确保所有工具函数都有try...except包裹，并返回格式化的错误信息而非抛出异常。
2. 确认使用的是异步函数 (async def) 并在需要时正确使用了await。在同步代码中执行耗时 I/O 操作会阻塞整个事件循环。
3. 检查资源泄漏。例如，在文件服务器中，是否在读取文件后正确关闭了文件句柄？数据库连接是否及时归还到连接池？
根本原因：通常是服务器端的代码健壮性问题或资源管理不当。