☰
使用FastMCP构建AI工具服务器:从协议原理到工程实践
2026/9/25 15:43:05 网站建设 项目流程

在实际 AI 应用开发中,我们经常需要让大语言模型(LLM)与外部工具、数据源或系统进行安全、可控的交互。传统方法往往需要为每个工具编写复杂的适配代码,不仅开发效率低,还存在安全风险。Model Context Protocol(MCP)正是为了解决这一问题而设计的开放标准,而 FastMCP 则是 PrefectHQ 团队基于此协议推出的高性能 Python 实现,旨在简化 MCP 服务器的开发流程。

本文将以 Python 开发者为视角,带你从零理解 MCP 的核心概念,并使用 FastMCP 快速构建一个可与 Claude Code、Cursor 等 AI 助手集成的工具服务器。你将学会如何定义工具(Tools)和资源(Resources),如何启动 MCP 服务器,以及如何在实际项目中避免常见陷阱。

1. 理解 Model Context Protocol(MCP)的核心价值

MCP 是一个允许 LLM 安全、结构化地使用外部功能和数据的协议。你可以把它想象成 LLM 的“插件系统”或“驱动程序接口”。其核心价值在于:

  • 标准化交互:为工具集成提供了统一的通信规范,不同厂商的 MCP 服务器可以被支持该协议的客户端(如 Claude Code、Cursor)直接使用。
  • 安全性:通过严格的输入输出模式定义和权限控制,防止 LLM 执行危险操作或访问敏感数据。
  • 开发效率:开发者只需关注工具本身的逻辑,无需重复编写通信、认证、错误处理等底层代码。

一个典型的 MCP 架构包含三个核心组件:

  1. MCP 客户端:如 Claude Code、Cursor 等 AI 助手,负责向服务器发送请求并处理响应。
  2. MCP 服务器:提供具体工具和资源的后端服务,使用 FastMCP 等框架开发。
  3. 传输层:支持 STDIO、HTTP 等多种通信方式,确保客户端与服务器间的可靠通信。

2. 准备 FastMCP 开发环境

FastMCP 要求 Python 3.8 或更高版本。建议使用虚拟环境隔离项目依赖,避免与系统或其他项目的 Python 包发生冲突。

2.1 创建并激活虚拟环境

# 创建项目目录 mkdir fastmcp-demo cd fastmcp-demo # 创建虚拟环境(选择以下一种方式即可) # 方式一:使用 venv(Python 3.3+ 内置) python -m venv .venv # 方式二:使用 conda conda create -n fastmcp-demo python=3.9 conda activate fastmcp-demo # 激活虚拟环境 # Windows .venv\Scripts\activate # Linux/macOS source .venv/bin/activate

2.2 安装 FastMCP

激活虚拟环境后,使用 pip 安装 FastMCP:

pip install fastmcp

如果需要在开发中使用最新特性,可以从 GitHub 直接安装:

pip install git+https://github.com/PrefectHQ/fastmcp.git

2.3 验证安装

创建一个简单的验证脚本check_install.py:

import fastmcp print(f"FastMCP version: {fastmcp.__version__}")

运行脚本确认安装成功:

python check_install.py

正常输出应显示类似FastMCP version: 1.2.0的版本信息。

3. 构建第一个 MCP 服务器:计算器工具

让我们从一个简单的计算器 MCP 服务器开始,了解 FastMCP 的基本用法。

3.1 创建服务器文件

新建calculator_server.py文件:

from fastmcp import FastMCP # 创建 MCP 服务器实例 mcp = FastMCP("Calculator") # 注册一个加法工具 @mcp.tool() def add(a: float, b: float) -> float: """将两个数字相加。 Args: a: 第一个加数 b: 第二个加数 Returns: 两个数字的和 """ return a + b # 注册一个乘法工具 @mcp.tool() def multiply(a: float, b: float) -> float: """将两个数字相乘。 Args: a: 第一个乘数 b: 第二个乘数 Returns: 两个数字的乘积 """ return a * b if __name__ == "__main__": # 启动服务器(STDIO 模式,适用于 AI 助手集成) mcp.run(transport="stdio")

3.2 关键代码解释

  • FastMCP("Calculator"):创建名为 "Calculator" 的服务器实例,这个名称会在客户端中显示。
  • @mcp.tool():装饰器将普通 Python 函数注册为 MCP 工具,LLM 可以直接调用。
  • 类型注解:a: float, b: float和-> float不仅提供代码提示,还帮助 MCP 客户端理解参数的期望类型。
  • Docstring:工具的描述和参数说明至关重要,LLM 依赖这些信息来正确使用工具。
  • mcp.run(transport="stdio"):以 STDIO 模式启动服务器,这是与 Claude Code 等客户端集成的标准方式。

3.3 测试服务器

虽然 MCP 服务器主要设计为与专用客户端配合,但我们可以编写一个简单的测试脚本来验证功能:

# test_calculator.py import subprocess import json import time def test_mcp_server(): # 启动服务器进程 process = subprocess.Popen( ["python", "calculator_server.py"], stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True ) # 等待服务器启动 time.sleep(1) # 发送测试请求(模拟 MCP 协议格式) test_request = { "jsonrpc": "2.0", "id": 1, "method": "tools/call", "params": { "name": "add", "arguments": {"a": 5, "b": 3} } } request_str = json.dumps(test_request) + "\n" process.stdin.write(request_str) process.stdin.flush() # 读取响应 response = process.stdout.readline() print("Server response:", response) # 清理进程 process.terminate() if __name__ == "__main__": test_mcp_server()

这个测试脚本模拟了 MCP 客户端的基本通信流程,帮助我们在集成前验证服务器逻辑是否正确。

4. 实现实用的文件操作 MCP 服务器

现在让我们构建一个更实用的文件操作服务器,展示如何处理文件读写、错误管理和资源管理。

4.1 创建文件服务器

新建file_server.py:

import os import json from pathlib import Path from typing import Dict, List, Optional from fastmcp import FastMCP mcp = FastMCP("File Manager") # 定义安全的工作目录(避免意外操作系统文件) WORKSPACE_DIR = Path("./workspace") WORKSPACE_DIR.mkdir(exist_ok=True) @mcp.tool() def list_files(directory: str = "") -> List[Dict]: """列出指定目录下的文件和文件夹。 Args: directory: 相对路径(相对于工作空间),默认为根目录 Returns: 包含文件信息的字典列表 """ target_dir = WORKSPACE_DIR / directory if not target_dir.exists(): raise ValueError(f"目录不存在: {directory}") if not target_dir.is_dir(): raise ValueError(f"路径不是目录: {directory}") files = [] for item in target_dir.iterdir(): files.append({ "name": item.name, "type": "directory" if item.is_dir() else "file", "size": item.stat().st_size if item.is_file() else 0, "modified": item.stat().st_mtime }) return files @mcp.tool() def read_file(filename: str) -> str: """读取文本文件内容。 Args: filename: 相对路径(相对于工作空间) Returns: 文件内容字符串 """ file_path = WORKSPACE_DIR / filename if not file_path.exists(): raise ValueError(f"文件不存在: {filename}") if not file_path.is_file(): raise ValueError(f"路径不是文件: {filename}") try: with open(file_path, 'r', encoding='utf-8') as f: return f.read() except UnicodeDecodeError: raise ValueError("文件不是有效的文本文件,可能包含二进制数据") @mcp.tool() def write_file(filename: str, content: str, append: bool = False) -> str: """写入或追加内容到文本文件。 Args: filename: 相对路径(相对于工作空间) content: 要写入的内容 append: 是否为追加模式(默认覆盖) Returns: 操作结果信息 """ file_path = WORKSPACE_DIR / filename # 确保目录存在 file_path.parent.mkdir(parents=True, exist_ok=True) mode = 'a' if append else 'w' try: with open(file_path, mode, encoding='utf-8') as f: f.write(content) action = "追加到" if append else "写入" return f"成功{action}文件: {filename}" except Exception as e: raise ValueError(f"文件操作失败: {str(e)}") @mcp.resource("file://{path}") def file_resource(path: str) -> Optional[str]: """以资源形式提供文件内容。 Args: path: 文件路径 Returns: 文件内容或 None(如果文件不存在) """ file_path = WORKSPACE_DIR / path if file_path.exists() and file_path.is_file(): try: with open(file_path, 'r', encoding='utf-8') as f: return f.read() except: return None return None if __name__ == "__main__": mcp.run(transport="stdio")

4.2 资源(Resources)与工具(Tools)的区别

这个示例展示了 MCP 中两个核心概念:

  • 工具(Tools):主动执行的操作,如读写文件、调用 API。需要显式调用并传递参数。
  • 资源(Resources):被动的数据源,如文件内容、数据库查询结果。客户端可以通过统一资源标识符(URI)直接访问。

在文件服务器中:

  • list_files、read_file、write_file是工具,需要主动调用。
  • file_resource是资源,客户端可以通过file://notes.txt这样的 URI 直接获取内容。

4.3 错误处理最佳实践

注意示例中的错误处理模式:

# 好的做法:提供具体的错误信息 if not target_dir.exists(): raise ValueError(f"目录不存在: {directory}") # 好的做法:限制操作范围,避免安全风险 WORKSPACE_DIR = Path("./workspace") # 好的做法:处理编码问题 try: with open(file_path, 'r', encoding='utf-8') as f: return f.read() except UnicodeDecodeError: raise ValueError("文件不是有效的文本文件")

这种错误处理方式既保证了用户体验,又避免了暴露系统敏感信息。

5. 配置 AI 客户端使用 MCP 服务器

5.1 配置 Claude Code

在 Claude Code 中配置 MCP 服务器,需要编辑配置文件(通常位于~/.config/claude-desktop/config.json):

{ "mcpServers": { "calculator": { "command": "python", "args": ["/path/to/your/calculator_server.py"] }, "filemanager": { "command": "python", "args": ["/path/to/your/file_server.py"] } } }

配置完成后重启 Claude Code,即可在对话中使用注册的工具。

5.2 配置 Cursor

Cursor 的配置类似,编辑~/.cursor/mcp/servers.json:

{ "servers": { "calculator": { "command": "python", "args": ["/path/to/your/calculator_server.py"] } } }

5.3 验证集成

配置完成后,在 AI 助手中尝试以下对话:

用户:请计算 15 乘以 27 助手:我将使用计算器工具来计算 15 × 27。 (调用 multiply 工具) 结果是 405。 用户:请列出工作空间中的文件 助手:我将查看文件管理器中的文件列表。 (调用 list_files 工具) 当前工作空间包含以下文件...

6. 高级特性与生产环境实践

6.1 使用依赖注入管理复杂服务

对于需要数据库连接、API 客户端等依赖的复杂工具,可以使用 FastMCP 的依赖注入功能:

from fastmcp import FastMCP import sqlite3 from typing import Annotated mcp = FastMCP("Database Manager") # 定义数据库依赖 def get_db_connection(): conn = sqlite3.connect("example.db") conn.row_factory = sqlite3.Row return conn # 在工具中使用依赖 @mcp.tool() def query_users( conn: Annotated[sqlite3.Connection, get_db_connection], min_id: int = 0 ) -> list: """查询用户信息。 Args: min_id: 最小用户ID Returns: 用户列表 """ cursor = conn.execute( "SELECT * FROM users WHERE id >= ?", (min_id,) ) return [dict(row) for row in cursor.fetchall()]

6.2 性能优化与缓存策略

对于耗时的操作,可以添加缓存机制:

from functools import lru_cache from fastmcp import FastMCP mcp = FastMCP("Weather Service") @lru_cache(maxsize=100) def expensive_weather_lookup(city: str) -> dict: # 模拟耗时的天气查询 import time time.sleep(1) return {"city": city, "temperature": 22.5} @mcp.tool() def get_weather(city: str) -> dict: """获取城市天气信息(带缓存)。 Args: city: 城市名称 Returns: 天气信息字典 """ return expensive_weather_lookup(city)

6.3 生产环境部署考虑

在实际部署 MCP 服务器时,需要考虑以下方面:

日志记录

import logging from fastmcp import FastMCP # 配置日志 logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s' ) mcp = FastMCP("Production Server") logger = logging.getLogger(__name__) @mcp.tool() def critical_operation(data: str) -> str: """关键操作示例。""" logger.info(f"执行关键操作,数据长度: {len(data)}") try: # 业务逻辑 result = process_data(data) logger.info("操作成功完成") return result except Exception as e: logger.error(f"操作失败: {str(e)}") raise

健康检查

@mcp.tool() def health_check() -> dict: """服务器健康状态检查。""" return { "status": "healthy", "timestamp": time.time(), "version": "1.0.0" }

7. 常见问题排查与调试技巧

7.1 服务器启动问题

问题现象可能原因解决方案
客户端无法连接服务器Python 路径错误使用绝对路径配置 command 和 args
服务器立即退出代码语法错误单独运行 Python 文件检查错误信息
工具列表为空装饰器使用错误确认使用@mcp.tool()而非@mcp.tool

7.2 工具调用问题

工具未找到错误

  • 检查工具名称拼写(区分大小写)
  • 确认工具已正确注册(装饰器位置正确)
  • 验证服务器重启后配置生效

参数验证失败

# 错误示例:缺少类型提示 @mcp.tool() def bad_example(a, b): # 缺少类型提示 return a + b # 正确示例:明确的类型提示 @mcp.tool() def good_example(a: float, b: float) -> float: return a + b

权限问题

  • 文件操作时确保工作目录存在且可写
  • 网络操作时检查防火墙和代理设置
  • 数据库操作时验证连接字符串和权限

7.3 调试技巧

添加详细日志

import logging logging.basicConfig(level=logging.DEBUG) @mcp.tool() def debug_tool(input: str) -> str: logging.debug(f"工具被调用,输入: {input}") # 业务逻辑 result = process_input(input) logging.debug(f"工具完成,输出: {result}") return result

使用测试模式

if __name__ == "__main__": # 开发时使用更详细的日志 import logging logging.basicConfig(level=logging.DEBUG) mcp.run(transport="stdio")

8. 安全最佳实践

8.1 输入验证与清理

import re from fastmcp import FastMCP mcp = FastMCP("Safe Server") @mcp.tool() def safe_filename_operation(filename: str) -> str: """安全的文件名操作示例。""" # 验证文件名格式 if not re.match(r'^[a-zA-Z0-9_\-\.]+$', filename): raise ValueError("文件名包含非法字符") # 防止路径遍历攻击 if '..' in filename or filename.startswith('/'): raise ValueError("非法文件路径") # 限制文件扩展名 if not filename.endswith(('.txt', '.md', '.json')): raise ValueError("不支持的文件类型") return process_file(filename)

8.2 权限控制

# 模拟基于上下文的权限检查 def check_permission(operation: str, user_context: dict) -> bool: allowed_operations = user_context.get('permissions', []) return operation in allowed_operations @mcp.tool() def restricted_operation(data: str, user_context: dict) -> str: """需要权限验证的操作。""" if not check_permission("restricted_operation", user_context): raise PermissionError("没有执行此操作的权限") return process_restricted_data(data)

8.3 资源限制

import resource from fastmcp import FastMCP mcp = FastMCP("Resource Limited Server") def set_memory_limit(): """设置内存使用限制。""" # 限制为 256MB memory_limit = 256 * 1024 * 1024 resource.setrlimit(resource.RLIMIT_AS, (memory_limit, memory_limit)) @mcp.tool() def memory_intensive_operation() -> str: """内存密集型操作。""" set_memory_limit() try: return perform_memory_intensive_task() except MemoryError: raise ValueError("操作超出内存限制")

通过遵循这些安全实践,可以确保 MCP 服务器在生产环境中的稳定性和安全性。记住,任何时候都要假设 LLM 可能产生意外的输入,因此防御性编程在 MCP 开发中尤为重要。

FastMCP 为 Python 开发者提供了构建高质量 MCP 服务器的高效路径。从简单的计算器工具到复杂的业务系统集成,这个框架都能提供良好的开发体验。在实际项目中,建议先从简单的工具开始,逐步增加复杂度,并在每个阶段进行充分的测试和验证。

需要专业的网站建设服务?

联系我们获取免费的网站建设咨询和方案报价,让我们帮助您实现业务目标

立即咨询