当前位置：首页 > news >正文

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

news 2025/7/19 11:39:45

🔥 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.
在这里插入图片描述

FastMCP 是什么？

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

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

# 3行开启
from fastmcp import FastMCP
mcp = 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执行能力

功能：

动态执行层：处理写操作（POST/PUT）、计算、API调用, 支持同步/异步/多进程任务调度
协议映射：将Python函数转化为MCP协议的tools/call端点
安全沙箱：隔离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, Field
from fastmcp import FastMCP
mcp = FastMCP("FinanceTools", port=9000)# 1. 基础工具：类型提示自动生成Schema
@mcp.tool
def currency_converter(amount: float, from_currency: str, to_currency: str) -> dict:"""实时货币兑换（支持180种法定货币）"""# 实际对接ExchangeRateAPIreturn {"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.tool
def execute_trade(request: TradeRequest) -> dict:"""执行证券交易（自动风控）"""if request.action == "sell":return {"status": "SUCCESS", "order_id": f"SELL_{request.stock}_{uuid.uuid4()}"}# Buy逻辑...# 3. 异步工具：处理IO密集型操作
@mcp.tool
async 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)

技巧：

使用@mcp.tool(name=“custom_name”)覆盖默认函数名
异常处理返回结构化错误：raise mcp.ToolReturn(code=403, message=“Insufficient balance”)
敏感操作添加权限标签：@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<var>[\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"<h1>{date} Report</h1>...")

技巧：

缓存机制：@mcp.resource(uri=“…”, ttl=3600) 设置资源缓存时间
按需启用：动态控制资源可见性

  market_resource = mcp.resource("market://{symbol}")(get_stock_price)market_resource.disable()  # 收盘后自动禁用

上下文注入：访问请求元数据

  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"})

技巧：

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

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

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

在这里插入图片描述

# 全链路代码实现
@mcp.tool
def 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