网站导航
> 技术文档 > Python FastMCP：让你的AI工具链飞起来

Python FastMCP：让你的AI工具链飞起来

网友: 技术文档 2025-07-30 13:29:02

🔥 PythonFastMCP：让你的AI工具链飞起来

  • FastMCP
    • FastMCP 是什么？
  • 1. 工具 (Tools)：赋予LLM执行能力
  • 2. Resources（资源）：安全数据通道
  • 3. Prompts（提示模板）：标准化LLM交互
  • 4.组件协同：构建项目AI工具链
  • 5. 部署架构与性能优化
  • 博主热门文章推荐：

官方文档：

  • FastMCP官方文档 ：https://gofastmcp.com/
  • MCP协议规范：https://modelcontextprotocol.io/
  • github: https://github.com/jlowin/fastmcp

FastMCP

The fast, Pythonic way to build MCP servers and clients.
Python FastMCP：让你的AI工具链飞起来

FastMCP 是什么？

FastMCP 是一个基于 Python 的高效框架，用于构建符合 MCP 协议（Model Context Protocol）的服务器和客户端。MCP 协议由 Anthropic 提出，旨在为大语言模型（LLM）提供标准化接口，使其能安全访问本地工具、私有数据和外部服务 。

FastMCP将python MCP开发变成简洁优雅的代码，让你用装饰器就能构建AI生产力工具链！

# 3行开启from fastmcp import FastMCPmcp = FastMCP(\"Financial\", port=9000) # 取名+端口mcp.run(transport=\"sse\") # 一键启动服务

FastMCP 的核心价值在于：

  • 协议隐形化：把复杂的MCP协议压缩成@tool、@resource装饰器，告别协议文档噩梦
  • 开发闪电化：工具开发代码量直降90%，原本3天的集成现在3小时搞定
  • 生产就绪化：自带OAuth安全校验/资源缓存/多协议传输，出生即战士

“以前对接Claude要写200行胶水代码，现在FastMCP只需15行”

  • 核心功能（三大组件）：
    • 工具 (Tools)
    • 资源 (Resources)
    • 提示模板 (Prompts)

核心库：pip install fastmcp uvicorn

1. 工具 (Tools)：赋予LLM执行能力

功能：

  1. 动态执行层：处理写操作（POST/PUT）、计算、API调用, 支持同步/异步/多进程任务调度
  2. 协议映射：将Python函数转化为MCP协议的tools/call端点
  3. 安全沙箱：隔离LLM对本地系统的直接访问

仅需通过@mcp.tool()装饰器, 实现函数到MCP端点的动态映射

 @mcp.tool( description=\"...\", # 功能说明（AI决策依据） tags=[\"text\", \"util\"] # 分类标签 # 可选thread/process/asyncIO执行器 ) def func(param: type) -> return_type: # FastMCP自动将类型转换为JSON Schema

实际案例：

能力维度 真实场景 安全机制 数据写入 数据库更新/API调用 Pydantic参数校验 复杂计算 风险价值分析/实时货币换算 异常结构化返回 异步操作 多源新闻聚合/批量文件处理 权限标签控制

代码实例：

from pydantic import BaseModel, Fieldfrom fastmcp import FastMCPmcp = FastMCP(\"FinanceTools\", port=9000)# 1. 基础工具：类型提示自动生成Schema@mcp.tooldef currency_converter(amount: float, from_currency: str, to_currency: str) -> dict: \"\"\"实时货币兑换（支持180种法定货币）\"\"\" # 实际对接ExchangeRateAPI return {\"converted\": amount * 0.85, \"rate\": 0.85} # 2. 复杂参数校验：Pydantic模型class TradeRequest(BaseModel): stock: str = Field(..., pattern=r\"^[A-Z]{1,5}$\") shares: int = Field(gt=0, le=1000, description=\"单次交易上限1000股\") action: Literal[\"buy\", \"sell\"]@mcp.tooldef execute_trade(request: TradeRequest) -> dict: \"\"\"执行证券交易（自动风控）\"\"\" if request.action == \"sell\": return {\"status\": \"SUCCESS\", \"order_id\": f\"SELL_{request.stock}_{uuid.uuid4()}\"} # Buy逻辑...# 3. 异步工具：处理IO密集型操作@mcp.toolasync def query_news(keywords: list[str]) -> list: \"\"\"聚合10家媒体实时新闻（支持关键词过滤）\"\"\" async with httpx.AsyncClient() as client: tasks = [client.get(f\"https://news.org/search?q={k}\") for k in keywords] return await asyncio.gather(*tasks)

技巧：

  1. 使用@mcp.tool(name=“custom_name”)覆盖默认函数名

  2. 异常处理返回结构化错误：raise mcp.ToolReturn(code=403, message=“Insufficient balance”)

  3. 敏感操作添加权限标签：@mcp.tool(tags=[“financial”])

2. Resources（资源）：安全数据通道

  • 只读数据源：类比GET请求，无副作用
  • 动态模板：URI参数化实现按需查询（如users://{user_id}/profile）
  • 自动编码：支持JSON/Text/Blob返回类型

URI模板引擎原理:

# 动态路由解析伪代码def resolve_resource(uri: str): if \"{var}\" in uri: pattern = uri.replace(\"{var}\", \"(?P[\\w-]+)\") match = re.match(pattern, request_uri) return handlers[uri].call(**match.groupdict())

金融数据服务案例：

# 1. 静态资源配置@mcp.resource(\"config://risk_policy\")def get_risk_policy() -> dict: \"\"\"获取当前风控阈值（JSON格式）\"\"\" return {\"max_trade\": 100000, \"currency_blacklist\": [\"IRR\", \"VES\"]}# 2. 动态模板资源（带参数验证）@mcp.resource(\"market://{symbol}/price\")def get_stock_price(symbol: str) -> float: \"\"\"查询实时股价（支持6000+美股）\"\"\" if not re.match(r\"^[A-Z]{1,5}$\", symbol): raise ValueError(\"Invalid stock symbol\") return yf.Ticker(symbol).fast_info.last_price# 3. 二进制资源（图片/PDF等）@mcp.resource( uri=\"reports://{date}/daily.pdf\", mime_type=\"application/pdf\")def generate_report(date: datetime) -> bytes: \"\"\"生成当日交易报告PDF\"\"\" return pdfkit.from_string(f\"
{date} Report
...\")

技巧：

  1. 缓存机制：@mcp.resource(uri=“…”, ttl=3600) 设置资源缓存时间
  2. 按需启用：动态控制资源可见性
 market_resource = mcp.resource(\"market://{symbol}\")(get_stock_price) market_resource.disable() # 收盘后自动禁用
  1. 上下文注入：访问请求元数据
 def get_logs(ctx: mcp.Context) -> str: return f\"Request from {ctx.client_id} at {ctx.timestamp}\"

3. Prompts（提示模板）：标准化LLM交互

核心：

  • 动态工作流：组合工具调用、资源加载、多轮对话
  • 企业级复用：统一客服/代码分析等场景的提示词规范
  • 客户端集成：支持Cursor/Sublime等IDE的快捷命令

动态生成原理：

@mcp.prompt(\"portfolio_risk\")def gen_prompt(user_id: str) -> list[dict]: # 实时获取资源数据 portfolio = mcp.resources.read(f\"portfolio://{user_id}\") return [ {\"role\": \"user\", \"content\": f\"分析{user_id}的资产\"}, {\"role\": \"assistant\", \"content\": { \"type\": \"tool_call\", # 关键！嵌入工具调用指令 \"tool\": \"calculate_var\", \"parameters\": {\"data\": portfolio} }} ]

实际案例：

# 1. 注册提示模板（服务端）mcp.register_prompt( name=\"analyze_portfolio\", description=\"投资组合风险评估\", arguments=[ {\"name\": \"user_id\", \"required\": True}, {\"name\": \"time_range\", \"enum\": [\"1d\", \"1w\", \"1m\"]} ])# 2. 动态提示生成器@mcp.prompt(\"analyze_portfolio\")def portfolio_prompt(user_id: str, time_range: str) -> list[dict]: \"\"\"构造包含实时数据的多步骤提示\"\"\" return [ { \"role\": \"system\", \"content\": \"你是一位CFA持证金融分析师，请评估用户投资组合风险\" }, { \"role\": \"user\", \"content\": { \"type\": \"resource\", # 嵌入资源 \"resource\": {  \"uri\": f\"portfolio://{user_id}?range={time_range}\",  \"mimeType\": \"application/json\" } } }, { \"role\": \"user\", \"content\": \"基于以上数据，分析：1) 最大回撤 2) 行业集中度 3) 建议调整方案\" } ]# 3. 客户端调用（Python示例）async def get_analysis_prompt(user: str): async with mcp.Client() as client: return await client.get_prompt( \"analyze_portfolio\", {\"user_id\": user, \"time_range\": \"1m\"} )

技巧：

  1. 多步骤工作流：在content中嵌入tools/call指令实现链式调用
  2. 条件分支：基于参数返回差异化提示结构
  3. 混合内容：同一消息内组合text+resource+tool_call

4.组件协同：构建项目AI工具链

金融行业实战：股票分析工作流

Python FastMCP：让你的AI工具链飞起来

# 全链路代码实现@mcp.tooldef calculate_var(portfolio: dict) -> float: \"\"\"计算投资组合VaR（风险价值）\"\"\"@mcp.resource(\"risk://models/{model_id}\")def get_risk_model(model_id: int) -> dict: ...# 组合提示模板：Prompt串联一切！@mcp.prompt(\"full_analysis\")def analysis_prompt(user_id: str): portfolio = await mcp.resources.read(f\"portfolio://{user_id}\") return { \"messages\": [ {\"role\": \"user\", \"content\": f\"用户{user_id}的持仓分析\"}, {\"role\": \"assistant\", \"content\": { \"type\": \"tool_call\", \"tool\": \"calculate_var\", \"parameters\": {\"portfolio\": portfolio} # 动态注入数据 }}, {\"role\": \"user\", \"content\": { \"type\": \"resource\", \"resource\": \"risk://models/1024\" # 调用风控模型 }} ] }

5. 部署架构与性能优化

调试与监控技巧:

# 1. 自动化测试工具fastmcp test server.py --tool \"stress_test\" --params \'{\"portfolio\":{}}\'# 2. 实时性能监控fastmcp monitor --metric qps --tool \"execute_trade\"# 3. 安全审计日志docker logs -f mcp-server | grep SECURITY_ALERT

多环境部署方案对比:
Python FastMCP：让你的AI工具链飞起来

博主热门文章推荐：

在这里插入图片描述

一篇读懂系列：

  • 一篇读懂无线充电技术（附方案选型及原理分析）
  • 一篇读懂：Android/iOS手机如何通过音频接口（耳机孔）与外设通信
  • 一篇读懂：Android手机如何通过USB接口与外设通信（附原理分析及方案选型）

LoRa Mesh系列：

  • LoRa学习：LoRa关键参数（扩频因子，编码率，带宽）的设定及解释
  • LoRa学习：信道占用检测原理（CAD）
  • LoRa/FSK 无线频谱波形分析（频谱分析仪测试LoRa/FSK带宽、功率、频率误差等）

网络安全系列：

  • ATECC508A芯片开发笔记（一）：初识加密芯片
  • SHA/HMAC/AES-CBC/CTR 算法执行效率及RAM消耗 测试结果
  • 常见加密/签名/哈希算法性能比较 (多平台 AES/DES, DH, ECDSA, RSA等)
  • AES加解密效率测试（纯软件AES128/256）–以嵌入式Cortex-M0与M3 平台为例

嵌入式开发系列：

  • 嵌入式学习中较好的练手项目和课题整理（附代码资料、学习视频和嵌入式学习规划）
  • IAR调试使用技巧汇总：数据断点、CallStack、设置堆栈、查看栈使用和栈深度、Memory、Set Next Statement等
  • Linux内核编译配置（Menuconfig）、制作文件系统 详细步骤
  • Android底层调用C代码（JNI实现）
  • 树莓派到手第一步：上电启动、安装中文字体、虚拟键盘、开启SSH等
  • Android/Linux设备有线&无线 双网共存（同时上内、外网）

AI / 机器学习系列：

  • AI: 机器学习必须懂的几个术语：Lable、Feature、Model…
  • AI：卷积神经网络CNN 解决过拟合的方法 （Overcome Overfitting）
  • AI: 什么是机器学习的数据清洗（Data Cleaning）
  • AI: 机器学习的模型是如何训练的？（在试错中学习）
  • 数据可视化：TensorboardX安装及使用（安装测试+实例演示）
网络标签：

相关问题