基于wet-mcp框架构建AI工具调用服务器：从MCP协议到实战应用

1. 项目概述：一个为AI应用量身定制的“湿”MCP服务器

最近在折腾AI应用开发，特别是想给大语言模型（LLM）接上各种外部工具和API时，发现了一个挺有意思的项目：n24q02m/wet-mcp。乍一看这个名字，可能会有点摸不着头脑。“wet”是“湿”的意思，MCP是“Model Context Protocol”的缩写，这“湿MCP”是个啥？

简单来说，你可以把它理解为一个“即插即用”的MCP服务器实现。MCP协议本身，是AI领域里一个新兴的、旨在标准化LLM与外部工具交互方式的规范。它有点像给LLM装上一个标准化的“USB接口”，让模型可以安全、可控地调用各种函数、查询数据、执行操作，而无需为每个工具都写一套复杂的适配代码。

那么，wet-mcp的“湿”体现在哪里？我的理解是，它强调的是一种“开箱即用”和“灵活可塑”的特性。就像一块干海绵，本身没什么用，但一旦“浸湿”（配置好），就能吸收并执行各种任务。这个项目提供了一个高度模块化的基础框架，开发者可以基于它，快速“浸润”自己的业务逻辑和数据源，构建出专属的、功能强大的MCP服务器。它不是给你一个僵硬的成品，而是给你一套好用的“乐高积木”和搭建手册。

对于正在构建AI智能体、希望为ChatGPT、Claude等模型扩展能力的开发者，或者任何想探索LLM工具调用标准化方案的工程师来说，深入理解并实践这个项目，能帮你省去大量从零搭建基础设施的麻烦，直接聚焦在核心的业务集成上。

2. MCP协议核心思想与wet-mcp的定位

在深入拆解wet-mcp之前，我们必须先搞清楚MCP协议到底要解决什么问题。当前，让LLM使用工具（Tools）或函数调用（Function Calling）虽然已经很常见，但存在几个痛点：

1. 协议碎片化：OpenAI有Function Calling，Anthropic有Tool Use，其他开源模型又有各自的格式。为同一个功能适配不同模型，需要写多套胶水代码。
2. 安全与权限控制薄弱：工具一旦暴露给LLM，其调用权限和参数往往缺乏细粒度的、声明式的控制。
3. 开发体验不统一：工具的描述、注册、调用流程各异，缺乏一个统一的开发框架。
4. 工具发现与管理困难：当工具数量增多时，如何让LLM动态地、按需地发现和理解可用工具，是个挑战。

MCP协议就是为了解决这些问题而生的。它定义了一套标准，涵盖了：

• 工具（Tools）：LLM可以调用的函数，有明确的输入输出模式（Schema）。
• 资源（Resources）：LLM可以读取的静态或动态数据，如文件、数据库查询结果、API状态等，用URI标识。
• 提示（Prompts）：可复用的对话模板或指令片段。
• 协议通信：基于JSON-RPC over STDIO/SSE，定义了服务器（提供工具/资源）和客户端（通常是LLM运行环境）之间的通信方式。

wet-mcp在这个生态中的定位非常清晰：它是一个用Python实现的、功能完整的MCP服务器框架。它的目标不是实现某个特定的工具集，而是让开发者能极其方便地基于它来构建自己的MCP服务器。它处理了所有枯燥的协议层细节——JSON-RPC消息的解析、路由、响应封装，以及生命周期管理。开发者只需要关心最核心的两件事：

1. 我要提供哪些工具（Tools）？
2. 我要暴露哪些资源（Resources）？

然后，用wet-mcp提供的装饰器和类，像写普通Python函数一样定义它们即可。框架会自动处理注册、Schema生成、调用分发和结果返回。这极大地降低了MCP服务器的开发门槛。

3.wet-mcp项目结构深度解析

要理解如何使用它，最好的方式是先看看它的代码仓库结构（虽然我们无法直接运行，但可以分析其设计）。一个典型的基于wet-mcp的项目会包含以下核心部分：

3.1 核心依赖与入口点

项目的pyproject.toml或requirements.txt会声明对mcp（官方协议SDK）和wet-mcp框架本身的依赖。真正的起点通常是一个Python脚本，比如server.py。

这个入口脚本的核心任务是创建一个McpServer实例。这个过程就像搭建一个舞台：

# 示例性代码，展示核心概念 from wet_mcp import McpServer import asyncio # 1. 创建服务器实例，给它起个名字 server = McpServer("my-awesome-tools") # 2. 注册工具和资源（这里先留空，后面详细讲） # ... # 3. 运行服务器，开始监听来自客户端的请求 async def main(): await server.run() if __name__ == "__main__": asyncio.run(main())

McpServer是这个框架的心脏，它管理着所有的工具、资源，并负责与MCP客户端进行通信。

3.2 工具（Tools）的定义与注册

工具是MCP的核心。在wet-mcp中，定义一个工具异常简单。你只需要写一个异步函数，并用@server.tool()装饰器来装饰它。

from wet_mcp import McpServer import httpx server = McpServer("weather-server") @server.tool( name="get_current_weather", description="获取指定城市的当前天气情况。", ) async def get_weather(city: str) -> str: """实际的工具实现。装饰器中的name和description用于生成LLM可理解的Schema。""" # 这里模拟一个API调用 async with httpx.AsyncClient() as client: # 假设我们调用一个天气API resp = await client.get(f"https://api.weather.example?city={city}") resp.raise_for_status() data = resp.json() return f"{city}的天气是{data['condition']}，温度{data['temp']}摄氏度。" # 更复杂的参数和Schema控制 from pydantic import BaseModel class WeatherInput(BaseModel): city: str unit: Literal["celsius", "fahrenheit"] = "celsius" @server.tool() async def get_weather_detail(params: WeatherInput) -> str: # 框架会自动将JSON参数转换为WeatherInput实例 # 这提供了强大的参数验证和文档生成能力 ...

关键点解析：

• 装饰器是灵魂：@server.tool()装饰器不仅注册了函数，更关键的是，它利用Python的类型注解（city: str）和description，自动生成了符合MCP标准的Tool Schema（JSON Schema）。这个Schema会被发送给LLM，LLM就知道这个工具叫什么、能干什么、需要什么参数。
• 异步是必须的：因为工具调用可能涉及网络I/O（调用API、查询数据库），所以工具函数必须是async的。wet-mcp内部基于asyncio，能高效处理并发请求。
• Pydantic集成：对于复杂参数，强烈推荐使用Pydantic的BaseModel。这能让参数验证、文档生成和错误处理更加清晰和强大。wet-mcp与Pydantic配合得非常好。

3.3 资源（Resources）的暴露与管理

资源是另一个核心概念。它允许LLM读取或“观察”数据。资源用URI来标识，例如file:///path/to/doc.txt或database://users/123。

在wet-mcp中，你可以通过实现一个资源“读取器”来暴露资源。

from wet_mcp import McpServer from wet_mcp.resource import ResourceTemplate server = McpServer("file-server") # 定义一个资源模板，允许动态路径 file_resource_template = ResourceTemplate("file://{path}") @server.resource(file_resource_template) async def read_file(path: str) -> str: """根据模板，当LLM请求 file:///etc/hosts 时，path 参数就是 ‘/etc/hosts‘""" try: with open(path, 'r', encoding='utf-8') as f: content = f.read() return content except FileNotFoundError: raise ValueError(f"文件未找到: {path}") except Exception as e: raise ValueError(f"读取文件失败: {e}") # 你也可以直接列出静态资源 @server.list_resources() async def list_my_resources(): return [ {"uri": "note://quick-idea", "name": "我的灵感笔记", "mimeType": "text/plain"}, {"uri": "config://app/settings", "name": "应用设置", "mimeType": "application/json"}, ]

资源的价值：通过资源，你可以将内部数据源（数据库表、日志文件、系统状态）安全地、以结构化的方式暴露给LLM。LLM可以“读取”这些资源来获取上下文信息，然后再决定是否调用工具进行处理。这比将所有数据都塞进对话上下文要高效和可控得多。

3.4 配置与生命周期管理

一个健壮的MCP服务器还需要配置和管理。wet-mcp通常通过环境变量或配置文件来处理这些。

• 工具/资源权限：虽然MCP协议和基础框架层面支持声明权限，但在wet-mcp的具体实现中，你需要在工具函数内部实现业务逻辑层面的权限检查。例如，在工具实现里验证API Key或用户角色。
• 服务器元数据：你可以在创建服务器时指定版本、描述等信息，这些会通过协议告知客户端。
• 生命周期钩子：框架可能提供了on_startup,on_shutdown等钩子函数，让你可以在服务器启动或关闭时执行一些初始化或清理操作，比如连接数据库池、加载缓存等。

4. 从零构建一个实战项目：智能待办事项管理MCP服务器

理论说再多，不如动手做一个。假设我们要构建一个“智能待办事项（Todo）管理”MCP服务器，让LLM能帮我们查看、添加、完成待办事项。

4.1 项目初始化与环境搭建

首先，创建一个新的项目目录并设置虚拟环境。

mkdir mcp-todo-server && cd mcp-todo-server python -m venv venv # Windows: venv\Scripts\activate # Mac/Linux: source venv/bin/activate

然后，创建依赖文件requirements.txt：

wet-mcp>=0.5.0 pydantic>=2.0.0 # 我们使用SQLite作为简单数据库，并用aiosqlite进行异步操作 aiosqlite>=0.20.0

安装依赖：pip install -r requirements.txt。

4.2 数据模型与数据库层设计

我们使用SQLite来持久化数据。先创建一个models.py文件定义数据模型。

# models.py from pydantic import BaseModel from datetime import datetime from typing import Optional import enum class TodoStatus(str, enum.Enum): PENDING = "pending" COMPLETED = "completed" CANCELLED = "cancelled" class TodoItem(BaseModel): id: Optional[int] = None title: str description: Optional[str] = None status: TodoStatus = TodoStatus.PENDING created_at: datetime = datetime.utcnow() completed_at: Optional[datetime] = None class Config: use_enum_values = True # 让枚举在JSON序列化时使用其值

接着，创建database.py来管理数据库连接和操作。

# database.py import aiosqlite from pathlib import Path from models import TodoItem, TodoStatus from datetime import datetime DB_PATH = Path("todos.db") async def init_db(): """初始化数据库，创建表""" async with aiosqlite.connect(DB_PATH) as db: await db.execute(""" CREATE TABLE IF NOT EXISTS todos ( id INTEGER PRIMARY KEY AUTOINCREMENT, title TEXT NOT NULL, description TEXT, status TEXT NOT NULL DEFAULT 'pending', created_at TIMESTAMP NOT NULL, completed_at TIMESTAMP ) """) await db.commit() async def add_todo(item: TodoItem) -> int: """添加待办事项，返回ID""" async with aiosqlite.connect(DB_PATH) as db: cursor = await db.execute(""" INSERT INTO todos (title, description, status, created_at) VALUES (?, ?, ?, ?) """, (item.title, item.description, item.status, item.created_at)) await db.commit() return cursor.lastrowid async def get_todos(status: TodoStatus = None) -> list[TodoItem]: """获取待办列表，可筛选状态""" query = "SELECT * FROM todos" params = [] if status: query += " WHERE status = ?" params.append(status.value) query += " ORDER BY created_at DESC" async with aiosqlite.connect(DB_PATH) as db: db.row_factory = aiosqlite.Row # 以字典形式返回行 async with db.execute(query, params) as cursor: rows = await cursor.fetchall() return [TodoItem(**dict(row)) for row in rows] async def update_todo_status(todo_id: int, new_status: TodoStatus): """更新待办状态，如果是完成，则记录完成时间""" completed_at = datetime.utcnow() if new_status == TodoStatus.COMPLETED else None async with aiosqlite.connect(DB_PATH) as db: await db.execute(""" UPDATE todos SET status = ?, completed_at = ? WHERE id = ? """, (new_status.value, completed_at, todo_id)) await db.commit()

注意：在生产环境中，你需要考虑数据库连接池（如aiosqlite的连接池封装或使用asyncpgfor PostgreSQL），而不是每次操作都建立新连接。这里为了示例清晰，使用了简单连接。

4.3 实现MCP工具：让LLM操作待办事项

现在，我们来创建核心的server.py，使用wet-mcp定义工具。

# server.py import asyncio from typing import List, Optional from wet_mcp import McpServer from pydantic import BaseModel, Field from models import TodoItem, TodoStatus import database # 初始化服务器 server = McpServer("todo-mcp-server", version="1.0.0") # 1. 工具：添加待办事项 class AddTodoInput(BaseModel): title: str = Field(..., description="待办事项的标题，必须简洁明了。") description: Optional[str] = Field(None, description="待办事项的详细描述，可选。") @server.tool() async def add_todo(params: AddTodoInput) -> str: """添加一个新的待办事项到列表中。""" new_item = TodoItem(title=params.title, description=params.description) todo_id = await database.add_todo(new_item) return f"✅ 待办事项已添加！ (ID: {todo_id}) 标题: {params.title}" # 2. 工具：列出所有待办事项 class ListTodosInput(BaseModel): status_filter: Optional[TodoStatus] = Field(None, description="按状态筛选，如 ‘pending‘, ‘completed‘。") @server.tool() async def list_todos(params: ListTodosInput) -> str: """获取待办事项列表，支持按状态筛选。""" todos = await database.get_todos(params.status_filter) if not todos: return "📭 当前没有待办事项。" if not params.status_filter else f"没有状态为 ‘{params.status_filter}‘ 的待办事项。" result_lines = [] for todo in todos: status_icon = "🟢" if todo.status == TodoStatus.PENDING else "✅" if todo.status == TodoStatus.COMPLETED else "❌" result_lines.append(f"{status_icon} [{todo.id}] {todo.title}") if todo.description: result_lines.append(f" 📝 {todo.description}") result_lines.append(f" 🕐 创建于: {todo.created_at.strftime('%Y-%m-%d %H:%M')}") if todo.completed_at: result_lines.append(f" 🎯 完成于: {todo.completed_at.strftime('%Y-%m-%d %H:%M')}") result_lines.append("") # 空行分隔 return "\n".join(result_lines) # 3. 工具：标记待办为完成 class CompleteTodoInput(BaseModel): todo_id: int = Field(..., description="要标记为完成的待办事项的ID。") @server.tool() async def complete_todo(params: CompleteTodoInput) -> str: """将指定ID的待办事项标记为已完成。""" # 先检查是否存在 todos = await database.get_todos() target_todo = next((t for t in todos if t.id == params.todo_id), None) if not target_todo: return f"❌ 错误：未找到ID为 {params.todo_id} 的待办事项。" if target_todo.status == TodoStatus.COMPLETED: return f"ℹ️ 待办事项 [{params.todo_id}] 已经是完成状态。" await database.update_todo_status(params.todo_id, TodoStatus.COMPLETED) return f"🎉 恭喜！待办事项 ‘{target_todo.title}‘ (ID: {params.todo_id}) 已完成。" async def main(): # 确保数据库已初始化 await database.init_db() print("MCP Todo Server 正在启动...") # 运行服务器，这将阻塞直到服务器关闭 await server.run() if __name__ == "__main__": asyncio.run(main())

4.4 运行与测试你的MCP服务器

现在，你可以运行这个服务器了：python server.py。服务器启动后，会等待MCP客户端的连接（通过标准输入输出）。

为了测试，我们通常不会直接运行它，而是通过一个MCP客户端来连接。一个最常用的测试方式是使用mcpCLI工具（需要单独安装：pip install mcp）或者与支持MCP的AI应用（如Claude Desktop、Cursor等）集成。

这里介绍一个简单的本地测试方法，使用一个模拟的Python客户端脚本：

# test_client.py (这是一个简化的模拟，用于理解交互) import asyncio import json import subprocess import sys async def test_tool_call(): # 启动MCP服务器子进程 proc = await asyncio.create_subprocess_exec( sys.executable, "server.py", stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE, ) # 模拟MCP客户端初始化握手（简化版） init_request = { "jsonrpc": "2.0", "id": 1, "method": "initialize", "params": {"protocolVersion": "0.1.0", "capabilities": {}} } proc.stdin.write((json.dumps(init_request) + "\n").encode()) await proc.stdin.drain() # 读取初始化响应 line = await proc.stdout.readline() print("Server init response:", line.decode().strip()) # 模拟调用 ‘list_tools‘ 方法，获取可用工具列表 list_tools_request = { "jsonrpc": "2.0", "id": 2, "method": "tools/list", } proc.stdin.write((json.dumps(list_tools_request) + "\n").encode()) await proc.stdin.drain() line = await proc.stdout.readline() tools_response = json.loads(line.decode().strip()) print("\nAvailable tools:", json.dumps(tools_response, indent=2, ensure_ascii=False)) # 模拟调用 ‘add_todo‘ 工具 call_tool_request = { "jsonrpc": "2.0", "id": 3, "method": "tools/call", "params": { "name": "add_todo", "arguments": {"title": "学习MCP协议", "description": "深入理解wet-mcp框架"} } } proc.stdin.write((json.dumps(call_tool_request) + "\n").encode()) await proc.stdin.drain() line = await proc.stdout.readline() call_response = json.loads(line.decode().strip()) print("\nTool call result:", json.dumps(call_response, indent=2, ensure_ascii=False)) # 关闭服务器 proc.terminate() await proc.wait() if __name__ == "__main__": asyncio.run(test_tool_call())

运行python test_client.py，你应该能看到服务器返回的工具列表和工具调用结果。这验证了你的MCP服务器在协议层面是正常的。

更实用的测试是集成到真实环境：例如，在Claude Desktop中，你可以通过配置claude_desktop_config.json来加载本地开发的MCP服务器，然后直接在Claude的对话窗口中用自然语言说“帮我把‘写周报’添加到待办列表”，Claude就会自动调用你的add_todo工具。

5. 高级特性与生产环境考量

基础功能跑通后，我们需要考虑如何让它更健壮、更实用。

5.1 错误处理与日志记录

在工具函数中，必须进行细致的错误处理，并向LLM返回友好的错误信息。

import logging logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) @server.tool() async def complete_todo(params: CompleteTodoInput) -> str: try: todos = await database.get_todos() target_todo = next((t for t in todos if t.id == params.todo_id), None) if not target_todo: # 返回对LLM友好的错误信息 return f"操作失败：未找到ID为 {params.todo_id} 的待办事项。请检查ID是否正确。" # ... 其余逻辑 except aiosqlite.Error as e: logger.error(f"数据库操作失败: {e}", exc_info=True) # 不要将内部数据库错误直接暴露给LLM/用户 return "操作失败：系统内部错误，请稍后重试。" except Exception as e: logger.exception(f"complete_todo 发生未知错误: {e}") return "操作失败：发生意外错误。"

5.2 工具权限与访问控制

MCP协议本身支持在工具Schema中声明required权限。在wet-mcp中，你可以在装饰器中声明，但更关键的是在工具函数内部实施业务逻辑校验。

from functools import wraps def require_auth(func): """一个简单的权限检查装饰器示例""" @wraps(func) async def wrapper(*args, **kwargs): # 这里假设我们从某个上下文中获取了用户信息（实际可能来自会话） # 例如，通过环境变量或初始握手参数传递的API Key if not current_user_is_admin(): # 假设的检查函数 raise PermissionError("权限不足，需要管理员角色。") return await func(*args, **kwargs) return wrapper @server.tool() @require_auth # 先应用权限装饰器 async def delete_all_todos() -> str: """（危险操作）清空所有待办事项。""" # ... 实现 return "所有待办事项已清空。"

5.3 性能优化：连接池与缓存

对于生产环境，数据库连接池是必须的。我们可以使用aiosqlite的池化功能，或者换用asyncpg（PostgreSQL）或aiomysql。

# 使用 databases 库（支持连接池和多种数据库） # pip install databases[aiosqlite] from databases import Database DATABASE_URL = "sqlite:///./todos.db" database = Database(DATABASE_URL) async def init_db(): await database.connect() await database.execute("CREATE TABLE IF NOT EXISTS ...") # 在工具中直接使用 database 对象进行查询 async def get_todos(): query = "SELECT * FROM todos" return await database.fetch_all(query)

对于频繁读取、变化不频繁的数据（如配置、城市列表），可以考虑在内存中使用缓存（如functools.lru_cache或cachetools），并在数据更新时使缓存失效。

5.4 部署与进程管理

你的MCP服务器最终是一个长期运行的程序。你需要考虑如何部署它。

• 容器化：使用Docker是标准做法。创建一个Dockerfile，将你的代码、依赖和环境打包成镜像。

  FROM python:3.11-slim WORKDIR /app COPY requirements.txt . RUN pip install --no-cache-dir -r requirements.txt COPY . . CMD ["python", "server.py"]

• 进程管理：使用systemd（Linux）、supervisord或pm2来管理进程，确保服务器崩溃后能自动重启。
• 日志收集：将日志输出到标准输出（stdout），方便被Docker或进程管理器捕获，并集成到ELK、Loki等日志系统中。

6. 常见问题与调试技巧实录

在实际开发和集成过程中，你肯定会遇到各种问题。以下是我踩过的一些坑和总结的技巧。

6.1 问题排查清单

6.2 调试与开发心得

1. 从最简单的“回声”工具开始：在开发复杂工具前，先实现一个最简单的工具，如@server.tool() async def echo(text: str) -> str: return text。这能最快验证你的服务器基础功能是否正常。
2. 充分利用日志：在服务器启动、工具注册、收到请求、处理请求、返回响应等关键节点添加logging.info()或logging.debug()。这比用print更专业，且可以方便地控制日志级别。
3. 模拟客户端进行单元测试：为你的工具函数编写单元测试，模拟输入参数，测试其核心逻辑。这能保证业务代码的健壮性，与MCP协议层解耦。
4. 注意异步上下文：记住所有工具函数都是异步的。如果你需要在其中调用同步库，务必使用asyncio.to_thread()或loop.run_in_executor()将其放到线程池中执行，避免阻塞整个事件循环，导致服务器失去响应。
5. Schema就是文档：你为工具定义的description和参数Field(..., description="...")非常重要。LLM（尤其是GPT-4、Claude 3等）能很好地理解这些描述，生成更准确的调用参数。写得越清晰，LLM的调用成功率越高。
6. 版本化你的服务器：在McpServer初始化时指定version参数。当你有多个版本的客户端或服务器在运行时，这有助于排查兼容性问题。

通过wet-mcp构建MCP服务器，本质上是在为LLM打造一套标准化的“手”和“眼”。这个过程迫使你以LLM能理解的方式去思考和封装你的业务逻辑。当你的服务器稳定运行，并看到AI助手能流畅地使用你提供的工具完成任务时，那种成就感是非常独特的。这不仅仅是实现了一个功能，更是搭建了一座连接人类意图与数字世界能力的桥梁。