fastmcp做mcp工具服务

发布于:2025-09-04 ⋅ 阅读:(17) ⋅ 点赞:(0)
  1. 端到端通信与时序
Host(IDE/Agent)                    fastmcp 进程(STDIO)
─────────────── spawn ─────────────►  启动(附着 stdin/stdout 管道)
               Content-Length 帧化 + JSON-RPC 2.0

1) initialize ─────────────────────►  解析能力、返回 capabilities
2) initialized ◄────────────────────  Host 通知“完成初始化”
3) tools/list  ─────────────────────►  列出工具 (name, schema, desc)
4) resources/list(optional) ───────►  列出资源(如文件、URI模板)
5) tools/call  ─────────────────────►  执行工具函数(你的业务逻辑)
6)   result    ◄────────────────────  返回结果(contents 数组)
…(可多次调用)
7) shutdown    ─────────────────────►  优雅关闭
8) exit        ◄────────────────────  进程退出
•	传输层:MCP 规定可用多种传输;其中 STDIO 是标准且推荐的本地传输,客户端通过子进程与服务器的 stdin/stdout 双向通信。 
•	fastmcp 的默认运行方式:调用 run() 不带参数即走 STDIO;客户端通常“每会话拉起一个子进程”。 
•	编码与线格式:消息为 JSON-RPC 2.0,并用 Content-Length 头 + \r\n\r\n + JSON 做帧边界(和 LSP 一致)。例如:
Content-Length: 61\r\n
\r\n
{"jsonrpc":"2.0","id":1,"method":"initialize","params":{}}

(务必计算字节长度、用 CRLF,日志不要写到 stdout) 。    

  1. Host ↔ fastmcp 的核心调用语义
    • initialize / initialized:握手阶段,协商协议版本与能力。之后 Host 常会立即做一次工具发现。 
    • tools/list → tools/call:发现工具、调用工具(带参数),是最常见主路径。各 SDK 在客户端侧提供 list_tools() / call_tool()。  
    • resources/list / read_resource:用于暴露数据源(本地文件、HTTP、数据库等)并读取内容。 

  1. 可运行的最小示例(服务器:STDIO)

下面是一个 fastmcp 2.0 风格的最小服务器。直接运行后,它就等着 Host 通过 STDIO 连接、列工具、调工具。fastmcp 在 2.0 版本对 MCP 生态做了“更 Pythonic 的封装”。  

server.py

from typing import Annotated
from mcp.server.fastmcp import FastMCP, run

mcp = FastMCP("demo")

@mcp.tool()
def add(a: Annotated[int, "加数 A"], b: Annotated[int, "加数 B"]) -> str:
    """返回 a + b 的结果。"""
    return str(a + b)

if __name__ == "__main__":
    # 不带参数:默认用 STDIO 传输
    run(mcp)

要点:
• 不用自己处理 Content-Length,SDK 已内置;只专注写业务函数(工具)。 
• 不要 print 到 stdout(会破坏帧);日志写 stderr。 

  1. 客户端(调试/自测):用官方 Python SDK 走 STDIO

client.py

import asyncio
from mcp.client.stdio import stdio_client, StdioServerParameters
from mcp.client.session import ClientSession

async def main():
    params = StdioServerParameters(command="python", args=["server.py"])
    async with stdio_client(params) as (read, write):
        async with ClientSession(read, write) as session:
            await session.initialize()
            tools = await session.list_tools()
            print("Tools:", [t.name for t in tools.tools])
            result = await session.call_tool("add", {"a": 7, "b": 13})
            # MCP 工具结果通常是 contents 列表(文本/JSON等)
            print("Result:", result.content[0].text)

if __name__ == "__main__":
    asyncio.run(main())

这个示例演示了 握手 → 列工具 → 调工具 的完整流程。方法名与握手顺序与文档一致。 

  1. 手工诊断(原始帧)

你也可以在命令行手工发帧验证服务器是否按 STDIO 规范读消息:

发送 initialize(注意 \r\n\r\n、字节数、不要多空格)

printf 'Content-Length: 61\r\n\r\n{"jsonrpc":"2.0","id":1,"method":"initialize","params":{}}' \
 | python server.py

这是调试 STDIO 帧化最直观的方法(和文档示例一致)。 

  1. 生产化落地的关键细节

    1. 帧化与日志

    • Content-Length 必须是 UTF-8 字节长度;分隔必须 \r\n\r\n;
    • 严禁把业务日志写到 stdout(会和 JSON-RPC 混流导致解帧失败),请写到 stderr。

    1. 并发与重入

    • 客户端可能并行 call_tool;尽量让工具函数是无副作用/线程安全;
    • 某些 SDK 在内部会触发隐式的 list_tools()(曾引发竞争条件),建议使用较新的 SDK 版本并避免在工具执行期间动态重置能力表。

    1. 长任务与返回体

    • MCP 工具返回是内容列表(text/json 等),超大结果建议分页或资源化(resource://… 再 read_resource)。

    1. 何时选 STDIO

    • 本地 IDE、桌面客户端(如 Claude Desktop/Cursor 等)优先 STDIO;
    • 远程或需要浏览器接入时,用 Streamable HTTP / SSE 更合适;客户端与服务端都能无缝切换传输。

    1. 为什么 fastmcp 默认 STDIO

    • 本地子进程最简单(无需端口/防火墙);
    • 跨语言/跨平台(一切语言都有 stdin/stdout);
    • 低开销高安全(无网络面)。这些正是 fastmcp 的默认选择与官方建议。  

  1. 扩展示例:把“汇率查询”作为 MCP 工具(可接你之前的需求)

下面是一个“查询即期/历史汇率”的工具骨架(以 httpx 查询公共 API 为例);在真实环境里换成你企业内部汇率源即可。

fx_server.py

from typing import Annotated, Optional
from datetime import date
import httpx
from mcp.server.fastmcp import FastMCP, run

mcp = FastMCP("fx")

FX_BASE = "https://api.exchangerate.host"  # 仅示例

@mcp.tool()
def fx_rate(
    base: Annotated[str, "基准货币,如 USD"],
    quote: Annotated[str, "标价货币,如 CNY"],
    on: Annotated[Optional[str], "日期 YYYY-MM-DD,留空为今日"] = None,
) -> str:
    """查询即期/历史汇率:返回 1 {base} = ? {quote}"""
    d = on or date.today().isoformat()
    endpoint = f"{FX_BASE}/{d}"
    with httpx.Client(timeout=10) as client:
        r = client.get(endpoint, params={"base": base.upper(), "symbols": quote.upper()})
        r.raise_for_status()
        data = r.json()
        rate = data["rates"][quote.upper()]
        return f"1 {base.upper()} = {rate} {quote.upper()} @ {d}"

if __name__ == "__main__":
    run(mcp)  # 默认 STDIO
•	仍然通过 run(mcp) 走 STDIO;Host 侧 tools/list 能自动读到 fx_rate 的名称、入参 schema 和文档字符串。 

小结
• STDIO:MCP 的标准传输之一,fastmcp 默认使用,因其简单、通用、安全、低延迟。
• 线格式:Content-Length 帧 + JSON-RPC 2.0;严格遵守字节长度与 CRLF;日志走 stderr。
• 实现:fastmcp 让你专注于“写工具函数”,其余交给 SDK(握手、发现、调度、STDIO 管道)。


网站公告

今日签到

点亮在社区的每一天
去签到