初次接触 MCP(Model Context Protocol)时,那一堆“给大模型接工具”、“协议标准化”的新词可能会让人感觉有点玄乎。但落到代码层面,它的核心思想其实很直接:将你本地的函数、文件或查询能力以标准化的方式暴露出去,让遵循 MCP 协议的客户端来调用。而 FastMCP 所做的,正是把这层繁琐的协议胶水代码给封装好了。目前官方推崇的写法也极其简洁,在 Python 中 from fastmcp import FastMCP,定义好服务后直接 run() 就能启动。
先别急着深入概念,让我们把第一个服务跑起来看看。
from fastmcp import FastMCP
from pathlib import Path
import json
from datetime import datetime
mcp = FastMCP("ops-helper")
LOG_FILE = Path("deploy.log")
@mcp.tool
def tail_error_lines(limit: int = 20) -> str:
"""读取最近的错误日志"""
if not LOG_FILE.exists():
return "deploy.log 不存在"
lines = LOG_FILE.read_text(encoding="utf-8").splitlines()
errors = [x for x in lines if "ERROR" in x or "Traceback" in x]
return "\n".join(errors[-limit:]) if errors else "最近没有错误日志"
@mcp.tool
def summarize_release(version: str, author: str = "unknown") -> dict:
"""生成一个简化版发布记录"""
return {
"version": version,
"author": author,
"released_at": datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
"status": "created"
}
@mcp.resource("config://app")
def app_config() -> str:
"""给客户端一个可读取的配置资源"""
return json.dumps({
"env": "test",
"db_pool": 20,
"feature_x": True
}, ensure_ascii=False, indent=2)
if __name__ == "__main__":
mcp.run()
这段代码刻意没有做成玩具式的教学示例。它包含一个读取错误日志的工具,一个生成发布摘要的工具,以及一个暴露应用配置的资源。因为在真实项目中,一个 MCP 服务器最先暴露出去的,往往不是那些听起来“高大上”的能力,而正是这些零碎、但实际工作中高频使用的功能:查询日志、获取配置、触发脚本、访问数据库或是执行一次兜底查询。
FastMCP 的优势,不在于“它能定义函数”,而在于它把 MCP 协议中的工具(tools)、资源(resources)等概念封装得非常顺滑。其官方文档也明确表示,FastMCP 用于将 Python 函数包装成符合 MCP 协议的 tools、resources 和 prompts,并自动处理协议细节、输入校验和连接管理。
安装过程也非常简单:
pip install fastmcp
官方的 Quickstart 指南正是这个思路:先安装库,然后定义 FastMCP("My Server") 实例,最后运行服务。
不过,当你真正准备落地自己的 MCP 服务器时,我建议先思考下面三个问题:
第一,哪些能力适合暴露成 Tool? 对于那些有副作用(Side Effects)的操作,比如“重新运行任务”、“补发消息”、“清理缓存”等,需要格外谨慎。我个人的习惯是从只读能力开始,例如查询订单状态、查看日志、获取某个功能开关的配置。先把能力边界和安全规范摸清楚,再逐步开放写操作,否则服务上线快,出问题的速度可能更快。
第二,返回值结构别设计得太复杂。 很多人一开始就想着返回嵌套多层、结构复杂的对象。其实没必要。字典、字符串、列表这些基本数据结构,在大多数场景下已经足够。大模型在调用工具时,最怕接收到一种“结构看似完整,但业务含义模糊不清”的数据。
第三,不要把数据库直连裸漏出去。 即使只是第一个 Demo,也建议在工具函数内部做一层封装,而不是在工具函数里随手拼接 SQL 语句。MCP 服务本质上是将内部能力开放给模型调用,开放面一旦扩大,任何不良的编码习惯都可能被放大。
再来看一个更贴近线上运维场景的例子。假设你想让大模型帮你查询某个定时任务(Job)最近的失败记录:
import sqlite3
from fastmcp import FastMCP
mcp = FastMCP("job-center")
@mcp.tool
def query_failed_jobs(job_name: str, limit: int = 10) -> list[dict]:
conn = sqlite3.connect("jobs.db")
conn.row_factory = sqlite3.Row
try:
rows = conn.execute(
"""
select job_name, run_time, status, error_msg
from job_history
where job_name = ? and status = 'FAILED'
order by run_time desc
limit ?
""",
(job_name, limit)
).fetchall()
return [dict(r) for r in rows]
finally:
conn.close()
if __name__ == "__main__":
mcp.run()
这种设计就很实用。客户端连接上来之后,既不需要了解背后复杂的表结构,也无需关心底层用的是 SQLite、PostgreSQL 还是其他数据库。它只需要知道:有一个叫 query_failed_jobs 的工具可以获取它想要的结果。
最后提一个容易被忽略但很重要的细节。官方 MCP 文档特别提醒,对于通过 STDIO(标准输入输出)通信的服务,不要随意向标准输出(stdout)打印日志,而应该输出到标准错误(stderr)或使用专门的 logging 模块。否则,你自己的调试输出很可能会污染 MCP 协议本身的通信数据。这个坑看似很小,但确实有人踩过。
所以,构建你的第一个 MCP 服务器时,别追求一步到位。先挂载一两个只读工具,暴露一个资源,确保能连接、能调用、能返回正确结果,这就已经成功了。后续再逐步考虑增加身份认证、切换传输方式(如 HTTP)、以及完善部署方案。
这东西真正有价值的地方,不在于“能把函数暴露出去”,而是你终于拥有了一种相对规范、可控的方式,让大模型能够与你的后端系统交互,而不是直接伸手进数据库里乱摸。掌握好这个分寸,非常重要。如果你在构建过程中有更多想法或问题,欢迎到 云栈社区 的开发者板块与大家交流探讨。