经管之家App
让优质教育人人可得
立即打开
模型上下文协议（Model Context Protocol，MCP）改变了大型语言模型（LLMs）与外部工具、数据源和服务的交互方式。然而，传统上从零构建MCP服务器需要处理复杂的样板代码和详细的协议规范。FastMCP消除了这一障碍，提供了一个基于装饰器的、Pythonic风格的框架，使开发者能够用最少的代码构建可用于生产环境的MCP服务器和客户端。
在本教程中，你将学习如何使用FastMCP构建MCP服务器和客户端，该框架功能全面且包含完整的错误处理，非常适合初学者和中级开发者。
开始本教程前，请确保你已具备以下条件：
Python 3.10或更高版本（推荐3.11+，以获得更好的异步性能）
pip或uv（推荐使用uv进行FastMCP部署，且uv是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服务通信一样。
标准化通信：使用JSON-RPC 2.0实现可靠、结构化的消息传递
双向通信：支持客户端向服务器发送请求，以及服务器向客户端返回响应
安全性：内置对认证和授权模式的支持
灵活的传输方式：可与任何传输机制配合使用（标准输入输出、HTTP、WebSocket、SSE）
MCP遵循清晰的客户端-服务器架构：
MCP服务器：暴露外部应用程序可使用的功能（工具、资源、提示词）。可以将其视为专门为LLM集成设计的后端API。
MCP客户端：嵌入在AI应用程序（如Claude Desktop、Cursor IDE或自定义应用程序）中，通过连接MCP服务器来访问其资源。
MCP服务器主要暴露三种类型的功能：
工具（Tools）：LLM可调用的可执行函数，用于执行特定操作。工具可以查询数据库、调用API、执行计算或触发工作流。
资源（Resources）：MCP客户端可获取并用作上下文的只读数据。资源可能是文件内容、配置数据或动态生成的内容。
提示词（Prompts）：可复用的消息模板，用于指导LLM的行为。提示词为多步骤操作或专业推理提供一致的指令。
FastMCP是一个高级Python框架，用于简化MCP服务器和客户端的构建过程。该框架旨在减少开发过程中的麻烦，具有以下特性：
基于装饰器的API：Python装饰器（@mcp.tool、@mcp.resource、@mcp.prompt）消除了样板代码
类型安全：利用Python的类型系统提供完整的类型提示和验证
支持Async/Await：采用现代异步Python，实现高性能操作
多种传输方式：支持标准输入输出、HTTP、WebSocket和SSE
内置测试：无需子进程复杂性，即可轻松进行客户端-服务器测试
可用于生产环境：具备错误处理、日志记录和生产部署配置等功能
FastMCP基于三个核心原则：
高层抽象：更少的代码，更快的开发周期
简洁性：最少的样板代码，让开发者专注于功能而非协议细节
Pythonic风格：采用自然的Python语法，让Python开发者感到熟悉易用
首先安装FastMCP及其必要的依赖项。推荐使用uv工具进行安装。
全部版块
我的主页
收藏
uv pip install fastmcp
如果你没有安装uv，可以使用pip安装uv：
pip install uv
或者直接使用pip安装FastMCP：
pip install fastmcp
验证FastMCP是否安装成功：
python -c "from fastmcp import FastMCP; print('FastMCP installed successfully')"
我们将创建一个实用的MCP服务器，演示工具、资源和提示词的使用。我们将构建一个计算器服务器，提供数学运算、配置资源和指令提示词。
首先需要创建一个项目目录并初始化环境。创建一个项目文件夹：
mkdir fastmcp-calculator
然后进入项目文件夹：
cd fastmcp-calculator
初始化项目并创建必要的文件：
uv init --python 3.11
我们的计算器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:
"""
将两个数字相加。
参数:
a: 第一个数字
b: 第二个数字
返回:
a和b的和
"""
try:
result = a + b
logger.info(f"执行加法运算: {a} + {b} = {result}")
return result
except TypeError as e:
logger.error(f"加法运算中出现类型错误: {e}")
raise ValueError(f"输入类型无效: {e}")
@mcp.tool
def subtract(a: float, b: float) -> float:
"""
用a减去b。
参数:
a: 第一个数字（被减数）
b: 第二个数字（减数）
返回:
a和b的差
"""
try:
result = a - b
logger.info(f"执行减法运算: {a} - {b} = {result}")
return result
except TypeError as e:
logger.error(f"减法运算中出现类型错误: {e}")
raise ValueError(f"输入类型无效: {e}")
我们定义了加法和减法函数，两个函数都包含try-catch块，用于抛出值错误、记录信息并返回结果。
@mcp.tool
def multiply(a: float, b: float) -> float:
"""
将两个数字相乘。
参数:
a: 第一个数字
b: 第二个数字
返回:
a和b的积
"""
try:
result = a * b
logger.info(f"执行乘法运算: {a} * {b} = {result}")
return result
except TypeError as e:
logger.error(f"乘法运算中出现类型错误: {e}")
raise ValueError(f"输入类型无效: {e}")
@mcp.tool
def divide(a: float, b: float) -> float:
"""
用a除以b。
参数:
a: 被除数（分子）
b: 除数（分母）
返回:
a除以b的商
抛出:
ValueError: 当尝试除以零时
"""
try:
if b == 0:
logger.warning(f"尝试除以零: {a} / {b}")
raise ValueError("不能除以零")
result = a / b
logger.info(f"执行除法运算: {a} / {b} = {result}")
return result
except (TypeError, ZeroDivision error) as e:
logger.error(f"除法运算中出现错误: {e}")
raise ValueError(f"除法错误: {e}")
四个被@mcp.tool装饰的函数暴露了数学运算功能。每个工具都包含：
参数和返回值的类型提示
详细的文档字符串（MCP将其用作工具描述）
带有try-except块的错误处理
用于调试和监控的日志记录
输入验证
接下来，我们构建资源：
@mcp.resource("config://calculator/settings")
def get_settings() -> Dict:
"""
提供计算器的配置信息和可用操作。
返回:
包含计算器配置和元数据的字典
"""
logger.debug("获取计算器配置")
return {
"version": "1.0.0",
"operations": ["add", "subtract", "multiply", "divide"],
"precision": "IEEE 754双精度",
"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:
"""
提供计算器服务器的用户指南。
返回:
包含使用指南和示例的字符串
"""
logger.debug("获取计算器指南")
guide = """
1. **add(a, b)**: 返回a + b
示例: add(5, 3) = 8
2. **subtract(a, b)**: 返回a - b
示例: subtract(10, 4) = 6
3. **multiply(a, b)**: 返回a * b
示例: multiply(7, 6) = 42
4. **divide(a, b)**: 返回a / b
示例: divide(20, 4) = 5.0
## 错误处理
- 除以零将抛出ValueError
- 非数字输入将抛出ValueError
- 所有输入应为有效的数字（int或float）
## 精度
计算器使用IEEE 754双精度浮点运算。
某些运算的结果可能包含微小的舍入误差。
"""
return guide
两个被@mcp.resource装饰的函数提供了静态和动态数据：
config://calculator/settings：返回计算器的元数据
docs://calculator/guide：返回格式化的用户指南
URI格式用于区分资源类型（约定：类型://类别/资源）。
接下来，我们构建提示词：
@mcp.prompt
def calculate_ex pression(ex pression: str) -> str:
"""
提供用于评估数学表达式的指令。
参数:
ex pression: 要评估的数学表达式
返回:
格式化的提示词，指导LLM如何评估表达式
"""
logger.debug(f"为表达式生成计算提示词: {ex pression}")
prompt = f"""
请逐步评估以下数学表达式：
表达式: {ex pression}
指令:
1. 将表达式分解为单个运算
2. 为每个运算使用合适的计算器工具
3. 遵循运算顺序（括号、乘除、加减）
4. 显示所有中间步骤
5. 提供最终结果
可用工具: add, subtract, multiply, divide
"""
return prompt.strip()
最后，添加服务器启动脚本：
if __name__ == "__main__":
logger.info("启动计算器MCP服务器...")
try:
# 使用stdio传输方式运行服务器（Claude Desktop默认方式）
mcp.run(transport="stdio")
except KeyboardInterrupt:
logger.info("服务器被用户中断")
sys.exit(0)
except Exception as e:
logger.error(f"致命错误: {e}", exc_info=True)
sys.exit(1)
@mcp.prompt装饰器创建了指导LLM处理复杂任务行为的指令模板。
此处包含的错误处理最佳实践：
特定异常捕获（TypeError、ZeroDivision error）
对用户有意义的错误消息
用于调试的详细日志
优雅的错误传播
在这一步中，我们将演示如何与上面创建的计算器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("初始化计算器客户端...")
try:
async with Client(server) as client:
logger.info("✓ 已连接到计算器服务器")
# 发现服务器功能
print("\n" + "="*60)
print("1. 发现服务器功能")
print("="*60)
# 列出可用工具
tools = await client.list_tools()
print(f"\n可用工具 ({len(tools)}):")
for tool in tools:
print(f" • {tool.name}: {tool.desc ription}")
# 列出可用资源
resources = await client.list_resources()
print(f"\n可用资源 ({len(resources)}):")
for resource in resources:
print(f" • {resource.uri}: {resource.name or resource.uri}")
# 列出可用提示词
prompts = await client.list_prompts()
print(f"\n可用提示词 ({len(prompts)}):")
for prompt in prompts:
print(f" • {prompt.name}: {prompt.desc ription}")
# 调用工具
print("\n" + "="*60)
print("2. 调用工具")
print("="*60)
# 简单加法
print("\n测试1: 15 + 27")
result = await client.call_tool("add", {"a": 15, "b": 27})
result_value = extract_tool_result(result)
print(f" 结果: 15 + 27 = {result_value}")
# 带错误处理的除法
print("\n测试2: 100 / 5")
result = await client.call_tool("divide", {"a": 100, "b": 5})
result_value = extract_tool_result(result)
print(f" 结果: 100 / 5 = {result_value}")
# 错误案例：除以零
print("\n测试3: 除以零（错误处理）")
try:
result = await client.call_tool("divide", {"a": 10, "b": 0})
print(f" 意外成功: {result}")
except Exception as e:
print(f" ✓ 正确捕获错误: {str(e)}")
# 读取资源
print("\n" + "="*60)
print("3. 读取资源")
print("="*60)
# 读取配置资源
print("\n获取计算器配置...")
settings_resource = await client.read_resource("config://calculator/settings")
print(f" 版本: {settings_resource[0].text}")
# 读取指南资源
print("\n获取计算器指南...")
guide_resource = await client.read_resource("docs://calculator/guide")
# 打印指南的前200个字符
guide_text = guide_resource[0].text[:200] + "..."
print(f" {guide_text}")
# 链式操作
print("\n" + "="*60)
print("4. 多操作链式调用")
print("="*60)
# 计算: (10 + 5) * 3 - 7
print("\n计算: (10 + 5) * 3 - 7")
# 步骤1: 加法
print(" 步骤1: 10 + 5")
add_result = await client.call_tool("add", {"a": 10, "b": 5})
step1 = extract_tool_result(add_result)
print(f" 结果: {step1}")
# 步骤2: 乘法
print(" 步骤2: 15 * 3")
mult_result = await client.call_tool("multiply", {"a": step1, "b": 3})
step2 = extract_tool_result(mult_result)
print(f" 结果: {step2}")
# 步骤3: 减法
print(" 步骤3: 45 - 7")
final_result = await client.call_tool("subtract", {"a": step2, "b": 7})
final = extract_tool_result(final_result)
print(f" 最终结果: {final}")
# 获取提示词模板
print("\n" + "="*60)
print("5. 使用提示词模板")
print("="*60)
ex pression = "25 * 4 + 10 / 2"
print(f"\n表达式的提示词模板: {ex pression}")
prompt_response = await client.get_prompt(
"calculate_ex pression",
{"ex pression": ex pression}
)
print(f" 模板:\n{prompt_response.messages[0].content.text}")
logger.info("✓ 客户端操作成功完成")
except Exception as e:
logger.error(f"客户端错误: {e}", exc_info=True)
sys.exit(1)
在上述代码中，客户端使用async with Client(server)进行安全的连接管理，这会自动处理连接的建立和清理。
我们还需要一个辅助函数来处理结果：
def extract_tool_result(response: Any) -> Any:
"""
从工具响应中提取实际的结果值。
MCP将结果包装在内容对象中，此辅助函数用于解包。
"""
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):
# 如果存在嵌套的result字段
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 res
return parsed
except Exception:
pass
return response
except Exception as e:
logger.warning(f"无法提取结果: {e}")
return response
if __name__ == "__main__":
logger.info("计算器客户端启动...")
asyncio.run(main())
从上述代码可以看出，在使用工具之前，客户端会列出可用的功能。await client.list_tools()获取所有工具的元数据（包括描述），await client.list_resources()发现可用的资源，最后，await client.list_prompts()会找到可用的提示词模板。
await client.call_tool()方法的功能如下：
接收工具名称和参数字典
返回包含结果的包装响应对象
集成工具失败的错误处理
在结果提取方面，extract_tool_result()辅助函数会解包MCP的响应格式以获取实际值，同时处理JSON和文本响应。
上面的链式操作演示了如何将一个工具的输出作为另一个工具的输入，从而实现跨多个工具调用的复杂计算。
最后，错误处理会捕获工具错误（如除以零）并进行优雅的日志记录，避免客户端崩溃。
你需要打开两个终端。在终端1中，启动服务器：
python calculator_server.py
你应该会看到：
在终端2中运行客户端：
python calculator_client.py
输出将显示：
虽然我们的计算器示例使用了基本逻辑，但FastMCP设计用于处理复杂的、可用于生产环境的场景。当你扩展MCP服务器时，可以利用以下特性：
异步操作：使用async def定义工具，用于执行数据库查询或API调用等I/O密集型任务
动态资源：资源可以接受参数（例如resource://users/{user_id}），用于动态获取特定数据点
复杂类型验证：使用Pydantic模型或复杂的Python类型提示，确保LLM以你的后端所需的确切格式发送数据
自定义传输方式：虽然我们使用了stdio，但FastMCP还支持SSE（服务器推送事件），用于基于Web的集成和自定义UI工具
FastMCP弥合了复杂的模型上下文协议与Python开发者期望的简洁、基于装饰器的开发体验之间的差距。通过消除与JSON-RPC 2.0和手动传输管理相关的样板代码，它让你能够专注于真正重要的事情：构建使LLM更强大的工具。
在本教程中，我们涵盖了：
MCP的核心架构（服务器与客户端）
如何定义用于执行操作的工具、用于提供数据的资源和用于提供指令的提示词
如何构建功能完善的客户端，用于测试和链式调用服务器逻辑
无论你是在构建简单的工具还是复杂的数据编排层，FastMCP都为你提供了通往可用于生产环境的智能体生态系统的最“Pythonic”路径。
你接下来会构建什么？查看FastMCP文档，探索更多高级部署策略和UI集成方式。
扫码加好友，拉您进群
