📢 转载信息
原文链接：https://www.kdnuggets.com/fastmcp-the-pythonic-way-to-build-mcp-servers-and-clients
原文作者：Shittu Olumide
Image by Author
# 引言
模型上下文协议（Model Context Protocol, MCP）改变了大型语言模型（LLMs）与外部工具、数据源和服务的交互方式。然而，从零开始构建MCP服务器传统上需要处理复杂的样板代码和详细的协议规范。FastMCP 消除了这一障碍，它提供了一个基于装饰器的、Pythonic的框架，使开发人员只需少量代码就能构建可投入生产的MCP服务器和客户端。
在本教程中，您将学习如何使用FastMCP构建MCP服务器和客户端，该框架功能全面，并内置了错误处理机制，非常适合初学者和中级开发人员。
// 先决条件
在开始本教程之前，请确保您已准备好以下条件：
- Python 3.10 或更高版本（建议使用 3.11+ 以获得更好的异步性能）
- pip 或 uv（推荐使用 uv 进行 FastMCP 部署，CLI 工具也需要它）
- 代码编辑器（我使用 VS Code，但您可以使用任何喜欢的编辑器）
- 用于运行 Python 脚本的终端/命令行基础知识
拥有良好的 Python 编程知识（函数、装饰器、类型提示）、对 async/await 语法有一些了解（对于高级示例是可选的，但有帮助）、熟悉 JSON 和 REST API 概念以及基本的命令行终端使用技巧也是非常有益的。
在 FastMCP 出现之前，构建 MCP 服务器需要深入了解 MCP JSON-RPC 规范、大量的协议处理样板代码、手动连接和传输管理，以及复杂的错误处理和验证逻辑。
FastMCP 通过直观的装饰器和简洁的 Pythonic API 解决了这些问题，让您可以专注于业务逻辑，而不是协议实现。
# 什么是模型上下文协议？
模型上下文协议（MCP）是 Anthropic 创建的一个开放标准。它提供了一个通用的接口，供 AI 应用程序安全地连接到外部工具、数据源和服务。MCP 规范了 LLM 如何与外部系统交互，这很像 Web API 规范化了 Web 服务通信。
// MCP 的关键特性
- 标准化通信： 使用 JSON-RPC 2.0 进行可靠、结构化的消息传递
- 双向性： 支持从客户端到服务器的请求以及来自服务器的响应
- 安全性： 内置支持身份验证和授权模式
- 灵活的传输： 适用于任何传输机制（stdio、HTTP、WebSocket、SSE）
// MCP 架构：服务器与客户端
MCP 遵循清晰的客户端-服务器架构：
Image by Author
- MCP 服务器： 暴露外部应用程序可以使用的能力（工具、资源、提示词）。可以将其视为专为 LLM 集成而设计的后端 API。
- MCP 客户端： 嵌入在 AI 应用程序中（如 Claude Desktop、Cursor IDE 或自定义应用程序），用于连接到 MCP 服务器以访问其资源。
// MCP 的核心组件
MCP 服务器暴露三种主要的“能力”：
- 工具（Tools）： LLM 可以调用的、用于执行操作的可执行函数。工具可以查询数据库、调用 API、执行计算或触发工作流。
- 资源（Resources）： MCP 客户端可以获取并用作上下文的只读数据。资源可能是文件内容、配置文件或动态生成的内容。
- 提示词（Prompts）： 用于指导 LLM 行为的可重用消息模板。提示词为多步操作或专业推理提供一致的指令。
# 什么是 FastMCP？
FastMCP 是一个高级 Python 框架，它简化了构建 MCP 服务器和客户端的过程。FastMCP 为减少开发痛苦而创建，具有以下特点：
- 基于装饰器的 API： Python 装饰器（
@mcp.tool、@mcp.resource、@mcp.prompt）消除了样板代码 - 类型安全： 使用 Python 的类型系统实现完整的类型提示和验证
- Async/Await 支持： 现代异步 Python，用于高性能操作
- 多种传输方式： 支持 stdio、HTTP、WebSocket 和 SSE
- 内置测试： 轻松进行客户端-服务器测试，无需复杂的子进程管理
- 生产就绪： 具备错误处理、日志记录和配置等功能，适用于生产部署
// FastMCP 理念
FastMCP 依赖于三个核心原则：
- 高级抽象： 编写更少的代码，实现更快的开发周期
- 简单： 极少的样板代码，让您可以专注于功能，而非协议细节
- Pythonic： 使用自然 Python 惯用法，对 Python 开发人员来说非常熟悉
# 安装
首先安装 FastMCP 和必要的依赖项。我推荐使用 uv。
uv pip install fastmcp
如果您没有 uv，可以使用 pip 安装：
pip install uv
或者直接使用 pip 安装 FastMCP：
pip install fastmcp
验证 FastMCP 是否已安装：
python -c "from fastmcp import FastMCP; print('FastMCP installed successfully')"# 构建您的第一个 MCP 服务器
我们将创建一个实用的 MCP 服务器，演示工具、资源和提示词。我们将构建一个计算器服务器，它提供数学运算、配置资源和指令提示词。
// 步骤 1：设置项目结构
我们首先需要创建一个项目目录并初始化您的环境。为您的项目创建一个文件夹：
mkdir fastmcp-calculator
然后导航到您的项目文件夹：
cd fastmcp-calculator
使用必要的文初始化您的项目：
uv init --python 3.11
// 步骤 2：创建 MCP 服务器
我们的计算器 MCP 服务器是一个简单的 MCP 服务器，演示了工具、资源和提示词。在项目文件夹中，创建一个名为 calculator_server.py 的文件，并添加以下代码。
import logging import sys from typing import Dict from fastmcp import FastMCP # 配置日志记录到 stderr（对 MCP 协议完整性至关重要） logging.basicConfig( level=logging.DEBUG, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s', stream=sys.stderr ) logger = logging.getLogger(__name__) # 创建 FastMCP 服务器实例 mcp = FastMCP(name="CalculatorServer")
服务器导入 FastMCP 并将日志配置为输出到 stderr。MCP 协议要求除协议消息外的所有输出都应定向到 stderr，以避免破坏通信。FastMCP(name="CalculatorServer") 调用创建了服务器实例。这会自动处理所有协议管理。
现在，让我们创建我们的工具。
@mcp.tool def add(a: float, b: float) -> float: """ 将两个数字相加。 Args: a: 第一个数字 b: 第二个数字 Returns: a 和 b 的和 """ try: result = a + b logger.info(f"Addition performed: {a} + {b} = {result}") return result except TypeError as e: logger.error(f"Type error in add: {e}") raise ValueError(f"Invalid input types: {e}") @mcp.tool def subtract(a: float, b: float) -> float: """ 从 a 中减去 b。 Args: a: 第一个数字 (被减数) b: 第二个数字 (减数) Returns: a 和 b 的差 """ try: result = a - b logger.info(f"Subtraction performed: {a} - {b} = {result}") return result except TypeError as e: logger.error(f"Type error in subtract: {e}") raise ValueError(f"Invalid input types: {e}")
我们为加法和减法定义了函数。两者都包装在 try-catch 块中，以引发值错误、记录信息并返回结果。
@mcp.tool def multiply(a: float, b: float) -> float: """ 两个数相乘。 Args: a: 第一个数字 b: 第二个数字 Returns: a 和 b 的乘积 """ try: result = a * b logger.info(f"Multiplication performed: {a} * {b} = {result}") return result except TypeError as e: logger.error(f"Type error in multiply: {e}") raise ValueError(f"Invalid input types: {e}") @mcp.tool def divide(a: float, b: float) -> float: """ 将 a 除以 b。 Args: a: 被除数 (分子) b: 除数 (分母) Returns: a 除以 b 的商 Raises: ValueError: 如果尝试除以零 """ try: if b == 0: logger.warning(f"Division by zero attempted: {a} / {b}") raise ValueError("Cannot divide by zero") result = a / b logger.info(f"Division performed: {a} / {b} = {result}") return result except (TypeError, ZeroDivisionError) as e: logger.error(f"Error in divide: {e}") raise ValueError(f"Division error: {e}")
四个装饰函数（@mcp.tool）暴露了数学运算。每个工具都包含：
- 参数和返回值的类型提示
- 完整的文档字符串（MCP 使用这些作为工具描述）
- 带有 try-except 块的错误处理
- 用于调试和监控的日志记录
- 输入验证
让我们继续构建资源。
@mcp.resource("config://calculator/settings") def get_settings() -> Dict: """ 提供计算器配置和可用操作。 Returns: 包含计算器设置和元数据的字典 """ logger.debug("Fetching calculator settings") return { "version": "1.0.0", "operations": ["add", "subtract", "multiply", "divide"], "precision": "IEEE 754 double precision", "max_value": 1.7976931348623157e+308, "min_value": -1.7976931348623157e+308, "supports_negative": True, "supports_decimals": True } @mcp.resource("docs://calculator/guide") def get_guide() -> str: """ 提供计算器服务器的用户指南。 Returns: 包含用法指南和示例的字符串 """ logger.debug("Retrieving calculator guide") guide = """ 1. **add(a, b)**: Returns a + b Example: add(5, 3) = 8 2. **subtract(a, b)**: Returns a - b Example: subtract(10, 4) = 6 3. **multiply(a, b)**: Returns a * b Example: multiply(7, 6) = 42 4. **divide(a, b)**: Returns a / b Example: divide(20, 4) = 5.0 ## Error Handling - Division by zero will raise a ValueError - Non-numeric inputs will raise a ValueError - All inputs should be valid numbers (int or float) ## Precision The calculator uses IEEE 754 double precision floating-point arithmetic. Results may contain minor rounding errors for some operations. """ return guide.strip()
两个装饰函数（@mcp.resource）提供静态和动态数据：
config://calculator/settings：返回有关计算器的元数据docs://calculator/guide：返回格式化的用户指南- URI 格式区分资源类型（约定：
type://category/resource）
让我们构建我们的提示词。
@mcp.prompt def calculate_expression(expression: str) -> str: """ 提供评估数学表达式的说明。 Args: expression: 要评估的数学表达式 Returns: 格式化的提示词，指示 LLM 如何评估表达式 """ logger.debug(f"Generating calculation prompt for: {expression}") prompt = f""" Please evaluate the following mathematical expression step by step: Expression: {expression} Instructions: 1. Break down the expression into individual operations 2. Use the appropriate calculator tool for each operation 3. Follow order of operations (parentheses, multiplication/division, addition/subtraction) 4. Show all intermediate steps 5. Provide the final result Available tools: add, subtract, multiply, divide """ return prompt.strip()
最后，添加服务器启动脚本。
if __name__ == "__main__": logger.info("Starting Calculator MCP Server...") try: # 使用 stdio 传输运行服务器（Claude Desktop 的默认设置） mcp.run(transport="stdio") except KeyboardInterrupt: logger.info("Server interrupted by user") sys.exit(0) except Exception as e: logger.error(f"Fatal error: {e}", exc_info=True) sys.exit(1)
@mcp.prompt 装饰器创建指令模板，指导 LLM 执行复杂任务的行为。
此处包含的错误处理最佳实践有：
- 具体的异常捕获（TypeError、ZeroDivisionError）
- 为用户提供有意义的错误消息
- 用于调试的详细日志记录
- 优雅的错误传播
// 步骤 3：构建 MCP 客户端
在此步骤中，我们将演示如何与上面创建的计算器 MCP 服务器进行交互。创建一个名为 calculator_client.py 的新文件。
import asyncio import logging import sys from typing import Any from fastmcp import Client, FastMCP logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s', stream=sys.stderr ) logger = logging.getLogger(__name__) async def main(): """ 主客户端函数，演示服务器交互。 """ from calculator_server import mcp as server logger.info("Initializing Calculator Client...") try: async with Client(server) as client: logger.info("✓ Connected to Calculator Server") # DISCOVER CAPABILITIEs print("\n" + "="*60) print("1. DISCOVERING SERVER CAPABILITIES") print("="*60) # 列出可用工具 tools = await client.list_tools() print(f"\nAvailable Tools ({len(tools)}):") for tool in tools: print(f" • {tool.name}: {tool.description}") # 列出可用资源 resources = await client.list_resources() print(f"\nAvailable Resources ({len(resources)}):") for resource in resources: print(f" • {resource.uri}: {resource.name or resource.uri}") # 列出可用提示词 prompts = await client.list_prompts() print(f"\nAvailable Prompts ({len(prompts)}):") for prompt in prompts: print(f" • {prompt.name}: {prompt.description}") # CALL TOOLS print("\n" + "="*60) print("2. CALLING TOOLS") print("="*60) # 简单加法 print("\nTest 1: Adding 15 + 27") result = await client.call_tool("add", {"a": 15, "b": 27}) result_value = extract_tool_result(result) print(f" Result: 15 + 27 = {result_value}") # 带错误处理的除法 print("\nTest 2: Dividing 100 / 5") result = await client.call_tool("divide", {"a": 100, "b": 5}) result_value = extract_tool_result(result) print(f" Result: 100 / 5 = {result_value}") # 错误案例：除以零 print("\nTest 3: Division by Zero (Error Handling)") try: result = await client.call_tool("divide", {"a": 10, "b": 0}) print(f" Unexpected success: {result}") except Exception as e: print(f" ✓ Error caught correctly: {str(e)}") # READ RESOURCES print("\n" + "="*60) print("3. READING RESOURCES") print("="*60) # 读取设置资源 print("\nFetching Calculator Settings...") settings_resource = await client.read_resource("config://calculator/settings") print(f" Version: {settings_resource[0].text}") # 读取指南资源 print("\nFetching Calculator Guide...") guide_resource = await client.read_resource("docs://calculator/guide") # 打印指南的前 200 个字符 guide_text = guide_resource[0].text[:200] + "..." print(f" {guide_text}") # CHAINING OPERATIONS print("\n" + "="*60) print("4. CHAINING MULTIPLE OPERATIONS") print("="*60) # 计算: (10 + 5) * 3 - 7 print("\nCalculating: (10 + 5) * 3 - 7") # Step 1: Add print(" Step 1: Add 10 + 5") add_result = await client.call_tool("add", {"a": 10, "b": 5}) step1 = extract_tool_result(add_result) print(f" Result: {step1}") # Step 2: Multiply print(" Step 2: Multiply 15 * 3") mult_result = await client.call_tool("multiply", {"a": step1, "b": 3}) step2 = extract_tool_result(mult_result) print(f" Result: {step2}") # Step 3: Subtract print(" Step 3: Subtract 45 - 7") final_result = await client.call_tool("subtract", {"a": step2, "b": 7}) final = extract_tool_result(final_result) print(f" Final Result: {final}") # GET PROMPT TEMPLATE print("\n" + "="*60) print("5. USING PROMPT TEMPLATES") print("="*60) expression = "25 * 4 + 10 / 2" print(f"\nPrompt Template for: {expression}") prompt_response = await client.get_prompt( "calculate_expression", {"expression": expression} ) print(f" Template:\n{prompt_response.messages[0].content.text}") logger.info("✓ Client operations completed successfully") except Exception as e: logger.error(f"Client error: {e}", exc_info=True) sys.exit(1)
从上面的代码可以看出，客户端使用 async with Client(server) 进行安全的连接管理。这会自动处理连接的建立和清理。
我们还需要一个辅助函数来处理结果。
def extract_tool_result(response: Any) -> Any: """ 从工具响应中提取实际结果值。MCP 将结果包装在 content 对象中，此助手将其解包。 """ try: if hasattr(response, 'content') and response.content: content = response.content[0] # 当可用时，优先使用明确的文本内容 (TextContent) if hasattr(content, 'text') and content.text is not None: # 如果文本是 JSON，尝试解析并提取 'result' 字段 import json as _json text_val = content.text try: parsed_text = _json.loads(text_val) # 如果 JSON 包含 result 字段，则返回它 if isinstance(parsed_text, dict) and 'result' in parsed_text: return parsed_text.get('result') return parsed_text except _json.JSONDecodeError: # 尝试将纯文本转换为数字 try: if '.' in text_val: return float(text_val) return int(text_val) except Exception: return text_val # 尝试通过模型 .json() 或类似字典的 .json 来提取 JSON 结果 if hasattr(content, 'json'): try: if callable(content.json): json_str = content.json() import json as _json try: parsed = _json.loads(json_str) except _json.JSONDecodeError: return json_str else: parsed = content.json # 如果解析结果是字典，尝试常见的结构 if isinstance(parsed, dict): # 如果存在嵌套结果 if 'result' in parsed: res = parsed.get('result') elif 'text' in parsed: res = parsed.get('text') else: res = parsed # 如果 res 是看起来像数字的字符串，则转换 if isinstance(res, str): try: if '.' in res: return float(res) return int(res) except Exception: return res return parsed except Exception: pass return response except Exception as e: logger.warning(f"Could not extract result: {e}") return response if __name__ == "__main__": logger.info("Calculator Client Starting...") asyncio.run(main())
在调用工具之前，客户端会列出可用能力。await client.list_tools() 获取所有工具元数据，包括描述。await client.list_resources() 发现可用资源。最后，await client.list_prompts() 会找到可用的提示词模板。
await client.call_tool() 方法执行以下操作：
- 接收工具名称和参数（字典形式）
- 返回一个包含结果的包装响应对象
- 与工具失败的错误处理集成
在结果提取方面，extract_tool_result() 辅助函数会解包 MCP 的响应格式以获取实际值，处理 JSON 和文本响应。
上面展示的链式操作演示了如何将一个工具的输出用作另一个工具的输入，从而实现跨多个工具调用的复杂计算。
最后，错误处理会优雅地捕获工具错误（如除以零）并记录下来，而不会导致程序崩溃。
// 步骤 4：运行服务器和客户端
您需要打开两个终端。在终端 1 中，您将启动服务器：
python calculator_server.py
您应该会看到：
Image by Author
在终端 2 中运行客户端：
python calculator_client.py
输出将显示：
Image by Author
# FastMCP 的高级模式
虽然我们的计算器示例使用了基本逻辑，但 FastMCP 旨在处理复杂的、可投入生产的场景。随着 MCP 服务器的扩展，您可以利用：
- 异步操作： 对执行 I/O 密集型任务（如数据库查询或 API 调用）的工具使用
async def - 动态资源： 资源可以接受参数（例如，
resource://users/{user_id}）以便动态获取特定数据点 - 复杂类型验证： 使用 Pydantic 模型或复杂的 Python 类型提示来确保 LLM 发送的数据格式完全符合后端要求
- 自定义传输： 虽然我们使用了
stdio，但 FastMCP 也支持 SSE（服务器发送事件）用于基于 Web 的集成和自定义 UI 工具
# 结论
FastMCP 弥合了复杂模型上下文协议与 Python 程序员所期望的干净、基于装饰器的开发体验之间的鸿沟。通过消除与 JSON-RPC 2.0 和手动传输管理相关的样板代码，它使您能够专注于最重要的事情：构建使 LLM 更强大的工具。
在本教程中，我们涵盖了：
- MCP 的核心架构（服务器与客户端）
- 如何定义用于操作的工具、用于数据的资源和用于指令的提示词
- 如何构建一个功能齐全的客户端来测试和链接服务器逻辑
无论您是构建一个简单的实用工具还是一个复杂的数据编排层，FastMCP 都提供了实现可投入生产的代理生态系统最“Pythonic”的路径。
您下一步将构建什么？ 请查阅 FastMCP 文档，探索更高级的部署策略和 UI 集成。
Shittu Olumide 是一位软件工程师和技术作家，热衷于利用尖端技术来构建引人入胜的叙事，对细节一丝不苟，并擅长简化复杂的概念。您也可以在 Twitter 上找到 Shittu。
🚀 想要体验更好更全面的AI调用？
欢迎使用青云聚合API，约为官网价格的十分之一，支持300+全球最新模型，以及全球各种生图生视频模型，无需翻墙高速稳定，文档丰富，小白也可以简单操作。
评论区