在AI智能体开发中,Function Calling(FC) 是大模型调用外部工具的事实标准,但手写FC的JSON格式繁琐、易出错、维护成本高。
FastMCP 作为MCP协议的Python极简实现,底层核心就是封装标准化Function Calling,通过极简装饰器语法,全自动完成FC的JSON构造、解析、响应全流程,让开发者无需关心底层协议细节,快速搭建可被LLM调用的工具服务。

一、MCP核心本质:底层基于Function Calling

MCP并非全新的调用机制,其工具调用的底层通信逻辑完全依托标准Function Calling实现

简单来说:

  • Function Calling:LLM与外部工具交互的底层标准协议,基于固定JSON格式完成指令传递;
  • MCP:在FC基础上扩展的上下文管理、多工具协同上层规范;

最常用的实现框架,FastMCP:

将MCP规范+FC的JSON格式全自动化封装,用Python装饰器替代手写JSON,零门槛实现工具服务。

二、FastMCP架构(核心的客户端和服务端,不介绍host)

  1. Client(客户端):调用方,即支持Function Calling的LLM(通义千问/GPT/Claude等)、本地测试程序;
  2. FastMCP Server(服务端):框架核心,自动完成FC-JSON封装、参数校验、工具执行、响应格式化。

三、核心重点:FastMCP ↔ Function Calling ↔ JSON 映射关系

FastMCP的核心原理,先明确原生Function Calling的JSON格式,再对应FastMCP的封装逻辑,理清三者关系。

3.1 原生Function Calling标准JSON格式

Function Calling的交互分为工具定义调用请求两个核心JSON结构,这是LLM识别的底层语法:

(1)工具定义JSON(告诉LLM有哪些工具可用)
[
  {
    "type": "function",
    "function": {
      "name": "calculator",
      "description": "计算器工具,支持加减乘除",
      "parameters": {
        "type": "object",
        "properties": {
          "num1": {"type": "number", "description": "第一个数字"},
          "num2": {"type": "number", "description": "第二个数字"},
          "operator": {"type": "string", "description": "运算符+-*/"}
        },
        "required": ["num1", "num2", "operator"]
      }
    }
  }
]
(2)调用请求JSON(LLM让程序执行工具)
{
  "role": "assistant",
  "content": null,
  "tool_calls": [
    {
      "function": {
        "name": "calculator",
        "arguments": "{\"num1\":10,\"num2\":6.5,\"operator\":\"*\"}"
      }
    }
  ]
}

3.2 FastMCP与FC-JSON的自动映射(核心)

FastMCP通过@mcp.tool()装饰器,将Python函数自动转换为上述FC标准JSON,全程无需手写JSON,映射关系如下:

Python代码元素

FastMCP封装后

对应Function Calling JSON字段

函数名 calculator

工具唯一标识

function.name

函数文档字符串 """计算器工具..."""

工具功能描述

function.description

参数类型注解 num1: float

参数数据类型

properties.num1.type

参数名 num1/num2/operator

参数键名

properties 下级键

必传参数

必填约束

required 数组

函数返回值

工具执行结果

LLM整合的FC响应内容

一句话总结
你写的Python函数 → FastMCP自动转成FC标准JSON → LLM解析JSON并下发调用指令 → FastMCP解析指令执行函数 → 返回结果给LLM。

四、FastMCP传输方式:stdio 与 SSE

FastMCP基于Function Calling的JSON指令,提供两种传输方式,用于传递FC格式数据,适配不同场景:

4.1 stdio(标准输入输出)

  • 原理:基于本地进程的标准输入/输出流,直接传输FC-JSON数据;
  • 特点:无网络开销、轻量高效、仅本地通信;
  • 适用场景:LLM与FastMCP服务在同一台机器/同一进程,本地开发、测试首选;
  • 底层:通过stdin/stdout传递FC格式的JSON字符串。

4.2 SSE(Server-Sent Events)

  • 原理:基于HTTP长连接的服务器推送,将FC-JSON数据通过网络流式传输;
  • 特点:跨网络、支持远程调用、兼容云端LLM;
  • 适用场景:LLM部署在云端,FastMCP服务在本地/服务器,生产环境首选;
  • 底层:通过HTTP接口传输FC格式的JSON数据,实现远程工具调用。

五、FastMCP与LLM的完整交互流程

基于Function Calling底层协议,FastMCP与LLM的交互全流程如下,每一步都围绕FC-JSON展开:

  1. 启动FastMCP服务
    服务端扫描所有@mcp.tool()装饰的函数,自动生成FC工具定义JSON,等待LLM连接。
  2. LLM连接并获取工具列表
    LLM通过stdio/SSE连接FastMCP,获取FC格式的工具定义JSON,知晓可用工具。
  3. 用户发起业务请求
    例:10乘以6.5等于多少?
  4. LLM生成FC调用JSON
    LLM解析用户意图,生成FC调用请求JSON,通过传输方式发送给FastMCP。
  5. FastMCP解析并执行工具
    服务端解析FC-JSON中的工具名、参数,自动调用对应的Python函数,执行业务逻辑。
  6. 返回FC响应结果
    FastMCP将函数执行结果封装为FC标准响应格式,回传给LLM。
  7. LLM生成最终回答
    LLM结合FC工具返回结果,整理为自然语言,回复给用户。

六、完整可运行实战代码

6.1 FastMCP 三大核心 API(装饰器)

FastMCP 封装了 3 个开发必备的装饰器,全自动对接 MCP/FC 协议:

  1. @mcp.tool() → 工具注册(底层对应 Function Calling)
    用于注册计算器、查天气等业务工具,LLM 核心调用对象。
  2. @mcp.prompt() → 动态提示词模板
    用于定义系统角色、交互规则,约束 LLM 行为。
  3. @mcp.resource() → 静态资源 / 知识库
    注册固定文本、知识库、配置信息,供 LLM 读取使用。

6.2 代码实现(服务端)

# server.py
from fastmcp import FastMCP
from datetime import datetime
import socket


def demo_log(message: str) -> None:
    """演示日志:统一时间戳,便于投屏讲解调用流程。"""
    now = datetime.now().strftime("%H:%M:%S")
    print(f"[MCP-DEMO {now}] {message}")


def pick_available_port(start_port: int = 8000, max_tries: int = 20) -> int:
    """从 start_port 开始找可用端口,演示时避免端口冲突中断。"""
    for port in range(start_port, start_port + max_tries):
        with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock:
            sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
            if sock.connect_ex(("127.0.0.1", port)) != 0:
                return port
    raise RuntimeError(f"未找到可用端口(尝试范围: {start_port}-{start_port + max_tries - 1})")

# ===================== 1. 创建FastMCP服务端 =====================
mcp_server = FastMCP(
    name="FastMCP完整服务",
    version="1.0.0",
    instructions="包含tool/prompt/resource,支持客户端工具发现"
)

# ===================== 2. 注册工具(@tool) =====================
@mcp_server.tool()
def calculator(num1: float, num2: float, operator: str) -> str:
    """
    计算器工具,支持加减乘除
    :param num1: 第一个数字
    :param num2: 第二个数字
    :param operator: 运算符 + - * /
    """
    demo_log(f"tool.calculator 收到请求: num1={num1}, num2={num2}, operator='{operator}'")
    try:
        if operator == "+":
            result = f"计算结果:{num1 + num2}"
        elif operator == "-":
            result = f"计算结果:{num1 - num2}"
        elif operator == "*":
            result = f"计算结果:{num1 * num2}"
        elif operator == "/":
            result = f"计算结果:{num1 / num2}"
        else:
            result = "错误:仅支持 + - * /"
        demo_log(f"tool.calculator 返回结果: {result}")
        return result
    except Exception as e:
        error_msg = f"工具执行失败:{str(e)}"
        demo_log(f"tool.calculator 发生异常: {error_msg}")
        return error_msg

@mcp_server.tool()
def weather_query(city: str) -> str:
    """查询指定城市天气"""
    demo_log(f"tool.weather_query 收到请求: city='{city}'")
    weather_data = {
        "北京": "晴,25℃,微风",
        "上海": "多云,28℃,湿度65%",
        "广州": "小雨,26℃"
    }
    result = weather_data.get(city, "暂无该城市天气")
    demo_log(f"tool.weather_query 返回结果: {result}")
    return result

# ===================== 3. 注册提示词(@prompt) =====================
@mcp_server.prompt()
def system_prompt(user: str = "用户") -> str:
    """系统提示词模板"""
    prompt_text = f"你是{user}的智能助手,必须使用工具回答,禁止编造信息"
    demo_log(f"prompt.system_prompt 被调用: user='{user}'")
    return prompt_text

# ===================== 4. 注册静态资源(@resource) =====================
@mcp_server.resource("info://bot")
def bot_info() -> str:
    """机器人基础信息"""
    data = "我是基于FastMCP开发的智能助手,支持工具调用与知识库查询"
    demo_log("resource.bot_info 被读取")
    return data

# ===================== 5. 启动服务(SSE模式,支持远程客户端连接) =====================
if __name__ == "__main__":
    demo_log("服务初始化完成")
    demo_log("已注册: tools=[calculator, weather_query]")
    demo_log("已注册: prompt=[system_prompt]")
    demo_log("已注册: resource=[info://bot]")
    port = pick_available_port(start_port=8000, max_tries=30)
    if port != 8000:
        demo_log(f"检测到 8000 端口被占用,自动切换到 {port}")
    demo_log(f"启动 transport=sse,监听 http://0.0.0.0:{port}/sse")
    mcp_server.run(transport="sse", host="0.0.0.0", port=port)
[MCP-DEMO 17:37:01] 服务初始化完成
[MCP-DEMO 17:37:01] 已注册: tools=[calculator, weather_query]
[MCP-DEMO 17:37:01] 已注册: prompt=[system_prompt]
[MCP-DEMO 17:37:01] 已注册: resource=[info://bot]
[MCP-DEMO 17:37:03] 检测到 8000 端口被占用,自动切换到 8002
[MCP-DEMO 17:37:03] 启动 transport=sse,监听 http://0.0.0.0:8002/sse
[04/06/26 17:37:03] INFO     Starting MCP server 'FastMCP完整服务' with transport 'sse' on       server.py:1358
                             http://0.0.0.0:8002/sse/
D:\Python\Python312\Lib\site-packages\websockets\legacy\__init__.py:6: DeprecationWarning: websockets.legacy is deprecated; see https://websockets.readthedocs.io/en/stable/howto/upgrade.html for upgrade instructions
  warnings.warn(  # deprecated in 14.0 - 2024-11-09
D:\Python\Python312\Lib\site-packages\uvicorn\protocols\websockets\websockets_impl.py:17: DeprecationWarning: websockets.server.WebSocketServerProtocol is deprecated
  from websockets.server import WebSocketServerProtocol
INFO:     Started server process [27296]
INFO:     Waiting for application startup.
INFO:     Application startup complete.
INFO:     Uvicorn running on http://0.0.0.0:8002 (Press CTRL+C to quit)

6.3 代码实现(客户端)

客户端通过 list_tools() 方法自动获取服务端所有工具,模拟 LLM 的工具发现逻辑

import asyncio
from datetime import datetime
from fastmcp.client import Client


def demo_log(message: str) -> None:
    now = datetime.now().strftime("%H:%M:%S")
    print(f"[MCP-CLIENT {now}] {message}")


# ===================== 1. 创建FastMCP客户端(SSE) =====================
async def create_client(start_port: int = 8000, max_tries: int = 30):
    """
    自动探测可连接的 SSE MCP 服务端(与 MCP-server.py 的端口策略匹配)。
    """
    last_error = None
    for port in range(start_port, start_port + max_tries):
        url = f"http://127.0.0.1:{port}/sse"
        client = Client(url, timeout=8, init_timeout=8)
        try:
            await client._connect()
            demo_log(f"已连接服务端: {url}")
            return client, url
        except Exception as e:
            last_error = e
            await client.close()
    raise RuntimeError(f"SSE 服务端连接失败,尝试端口 {start_port}-{start_port + max_tries - 1}") from last_error

# ===================== 2. 核心:客户端自动发现服务端工具 =====================
async def discover_tools(client):
    """
    工具自动发现:客户端获取服务端所有注册的工具
    对应MCP协议:tools/list 方法
    底层:获取服务端生成的FC-JSON工具列表
    """
    print("=" * 60)
    demo_log("客户端开始自动发现服务端工具...")
    tools = await client.list_tools()  # 核心方法:工具发现

    # 打印所有工具信息
    for tool in tools:
        print(f"\n工具名称:{tool.name}")
        print(f"工具描述:{tool.description}")
        print(f"输入参数:{tool.inputSchema}")
    print("=" * 60)
    return tools

# ===================== 3. 客户端调用服务端工具 =====================
async def call_server_tools(client):
    print("\n客户端调用服务端工具:")

    # 调用计算器工具
    calc_result = await client.call_tool(
        "calculator",
        {"num1": 10, "num2": 6.5, "operator": "*"}
    )
    print(f"1) 计算器结果:{calc_result}")

    # 调用天气查询工具
    weather_result = await client.call_tool(
        "weather_query",
        {"city": "北京"}
    )
    print(f"2) 天气查询结果:{weather_result}")

# ===================== 4. 客户端调用提示词/资源 =====================
async def call_prompt_and_resource(client):
    print("\n客户端调用提示词与资源:")
    prompt = await client.get_prompt("system_prompt", {"user": "小明"})
    print(f"1) 系统提示词:{prompt.messages}")

    resource = await client.read_resource("info://bot")
    print(f"2) 静态资源:{resource}")

# ===================== 主流程 =====================
async def main():
    client, url = await create_client()
    try:
        demo_log(f"当前连接地址: {url}")
        # 1. 自动发现工具(核心)
        await discover_tools(client)
        # 2. 调用工具
        await call_server_tools(client)
        # 3. 调用提示词/资源
        await call_prompt_and_resource(client)
    finally:
        await client.close()
        demo_log("客户端已断开")

if __name__ == "__main__":
    asyncio.run(main())
[MCP-CLIENT 17:40:59] 已连接服务端: http://127.0.0.1:8000/sse
[MCP-CLIENT 17:40:59] 当前连接地址: http://127.0.0.1:8000/sse
============================================================
[MCP-CLIENT 17:40:59] 客户端开始自动发现服务端工具...

工具名称:calculator
工具描述:计算器工具,支持加减乘除
:param num1: 第一个数字
:param num2: 第二个数字
:param operator: 运算符 + - * /
输入参数:{'properties': {'num1': {'title': 'Num1', 'type': 'number'}, 'num2': {'title': 'Num2', 'type': 'number'}, 'operator': {'title': 'Operator', 'type': 'string'}}, 'required': ['num1', 'num2', 'operator'], 'type': 'object'}

工具名称:weather_query
工具描述:查询指定城市天气
输入参数:{'properties': {'city': {'title': 'City', 'type': 'string'}}, 'required': ['city'], 'type': 'object'}
============================================================

客户端调用服务端工具:
1) 计算器结果:[TextContent(type='text', text='计算结果:65.0', annotations=None)]
2) 天气查询结果:[TextContent(type='text', text='晴,25℃,微风', annotations=None)]

客户端调用提示词与资源:
1) 系统提示词:[PromptMessage(role='user', content=TextContent(type='text', text='你是小明的智能助手,必须使用工具回答,禁止编造信息', annotations=None))]
2) 静态资源:[TextResourceContents(uri=AnyUrl('info://bot'), mimeType='text/plain', text='我是基于FastMCP开发的智能助手,支持工具调用与知识库查询')]
[MCP-CLIENT 17:40:59] 客户端已断开

七、工程化对接LLM

  1. 启动服务
    • 本地调用:mcp.run()(stdio模式,传输FC-JSON)
    • 远程调用:mcp.run(transport="sse", port=8000)(SSE模式)
  1. LLM配置
    所有支持Function Calling的LLM(通义千问、GPT-4o、Claude等),直接连接FastMCP服务地址,即可自动识别FC工具列表,完成工具调用。

八、总结

  1. FastMCP底层核心:基于标准Function Calling,所有交互都是FC格式的JSON;
  2. JSON映射关系:FastMCP通过装饰器自动将Python函数转为FC工具定义JSON,无需手写;
  3. 传输方式:stdio(本地高效)、SSE(远程生产),均用于传递FC-JSON数据;
  4. LLM交互:围绕FC协议完成「工具发现→指令下发→执行响应」全流程;
  5. 开发优势:彻底告别繁琐的FC-JSON手写,几行代码搭建兼容所有LLM的工具服务。
Logo

有“AI”的1024 = 2048,欢迎大家加入2048 AI社区

更多推荐