探索高效工具调用的新范式,让本地函数秒变云端服务

今天介绍一个强大的工具调用协议——MCP(Message Call Protocol),作为AI工作者,无论你是想构建本地CLI工具还是云端Web服务,MCP都能提供统一的解决方案。


一、极简入门:安装与基础工具开发

# 安装fastmcp库
pip install fastmcp

只需5行代码,即可创建你的第一个MCP工具:

from fastmcp import FastMCP

mcp = FastMCP('demo.mcp')

@mcp.tool()
def greet(name: str) -> str:
    return f'Hello, {name}'

二、工具自测与调用技巧

开发完成后,立即自测验证功能:

import asyncio
from fastmcp import Client

async def main():
    client = Client(mcp)
    async with client:
        # 查看可用工具
        tools = await client.list_tools()
        print('可用工具:', tools)
        
        # 调用工具(参数名必须匹配)
        result = await client.call_tool('greet', {'name': '技术爱好者'})
        print('调用结果:', result)

asyncio.run(main())

关键点解析

  1. 使用@mcp.tool()装饰器暴露函数
  2. 客户端通过call_tool调用,参数为字典格式
  3. 参数名必须与函数定义完全一致

三、MCP服务器:三种传输模式详解

1、STDIO模式(默认)

  • 适用场景:本地进程间通信
  • 特点:零网络开销,高性能
  • 启动方式:
    mcp.run()  # 默认STDIO模式
    
    STDIO服务器配置示例

2、Streamable HTTP(推荐)

  • 适用场景:现代Web服务
  • 特点:双向实时通信,支持流式传输
  • 启动方式:
    mcp.run(transport="streamable-http", port=8000, path="/mcp")
    
    Streamable HTTP配置

3、SSE模式(兼容旧系统)

  • 适用场景:传统浏览器兼容
  • 特点:单向服务器推送
  • 启动方式:
    mcp.run(transport="sse", port=8000, path="/sse")
    
    SSE服务器配置

智能启动脚本

import os

def run_server():
    mode = os.getenv('MCP_MODE', "").lower()
    if mode == 'sse':
        mcp.run(transport="sse", port=8000, path="/sse")
    elif mode == 'streamable-http':
        mcp.run(transport="streamable-http", port=8000, path="/mcp")
    else:
        mcp.run()  # 默认STDIO

if __name__ == '__main__':
    run_server()

4、总结比较

传输模式 适用场景 性能特性 推荐指数
STDIO 本地进程通信、命令行工具 零延迟,无网络开销 ⭐⭐⭐⭐
Streamable HTTP 实时Web服务、云端部署 支持双向流式传输 ⭐⭐⭐⭐⭐
SSE 兼容旧系统、浏览器 单向推送,存在连接保持开销 ⭐⭐

四、实战:Streamable HTTP客户端调用

import asyncio
from fastmcp.client import Client

async def main():
    # 连接到HTTP服务器
    async with Client('http://localhost:8000/mcp/') as client:
        # 获取工具列表
        tools = await client.list_tools()
        print(f'可用工具: {[t.name for t in tools]}')
        
        # 调用远程工具
        result = await client.call_tool('greet', {'name': 'Python开发者'})
        print(f'返回结果: {result[0].text}')  # 提取TextContent内容

asyncio.run(main())

执行结果

可用工具: ['greet']
返回结果: Hello, Python开发者

五、应用场景及集成

  1. 本地工具链集成
    STDIO模式连接Python脚本和Shell命令

    # 直接调用MCP工具
    echo '{"name": "Terminal用户"}' | python mcp_app.py
    
  2. 微服务架构
    将工具部署为独立HTTP服务,通过API网关调用

  3. Chatbot插件系统
    聊天窗口集成MCP工具

  4. 自动化工作流
    组合多个MCP工具构建复杂流水线


六、MCP服务高级功能

在探索FastMCP的旅程中,许多开发者都有一个认知误区:认为MCP服务等同于MCP工具,但是MCP服务有三大核心支柱:工具(Tool)、提示(Prompt)和资源(Resource)。这三者共同构成了一个完整的AI应用开发生态系统。

1、MCP服务的三维架构

MCP服务
工具 Tools
提示 Prompts
资源 Resources
执行具体操作
指导LLM生成
提供静态/动态数据

核心组件的功能对比

组件类型 核心功能 典型应用场景 装饰器语法
工具 执行具体任务 数据计算、API调用 @mcp.tool()
提示 生成LLM指导消息 标准化提问、响应格式化 @mcp.prompt()
资源 提供可复用数据 配置信息、模板文件 @mcp.resource()

2、工具(Tool):执行引擎

作为最基础的功能,工具负责执行具体操作

@mcp.tool()
def calculate_discount(price: float, discount: float) -> float:
    """计算商品折扣价"""
    return price * (1 - discount/100)

特点

  • 输入输出类型严格验证
  • 支持同步/异步操作
  • 可直接集成现有业务逻辑

3、提示(Prompt):AI指导者

提示是控制LLM行为的智能模板

@mcp.prompt()
def generate_interview_questions(position: str, level: str) -> str:
    """生成职位面试问题"""
    return f"""作为资深{position}面试官,请生成5个适合{level}级候选人的技术问题,
要求:
1. 包含代码题和理论题
2. 难度递增
3. 标注考察点"""

核心优势

  1. 动态参数化:支持变量注入
  2. 输出标准化:确保LLM响应一致性
  3. 集中化管理:统一维护提示模板
  4. 自动验证:严格检查输入参数类型

4、资源(Resource):数据供给站

资源是MCP最被低估的功能,它提供静态或动态的可复用数据

# Basic dynamic resource returning a string
@mcp.resource(uri="resource://greeting/{name}",
    name='greeting',
    description='用于演示的一个资源协议')
def get_greeting(name: str) -> str:
    """Provides a simple greeting message."""
    return f"Hello from {name} Resources!"

# Resource returning JSON data (dict is auto-serialized)
@mcp.resource("resource://config")
def get_config() -> dict:
    """Provides application configuration as JSON."""
    return {
        "theme": "dark",
        "version": "1.2.0",
        "features": ["tools", "resources"],
    }

资源类型

资源类型 刷新机制 适用场景
静态资源 服务启动时加载 配置信息、常量数据
定时刷新资源 固定间隔刷新 市场数据、天气信息
按需加载资源 请求时实时获取 用户个性化数据

七、避坑指南

  1. 参数名不匹配
    call_tool的字典key必须与函数参数名完全一致

  2. 端口冲突
    多服务运行时使用不同端口:

    mcp.run(port=8001)  # 指定端口
    
  3. 异步陷阱
    工具函数需为同步函数,耗时操作应使用线程池:

    @mcp.tool()
    def process_data(data: str):
        # CPU密集型任务
        return heavy_computation(data)
    

八、结语

MCP服务的三位一体架构为AI应用开发提供了完整解决方案:

  • 工具是肌肉 - 执行具体操作
  • 提示是大脑 - 指导智能行为
  • 资源是血液 - 流动数据养分

MCP协议通过统一接口打通了本地与云端工具的界限。无论你构建的是:

  • 🔧 本地开发工具链
  • ☁️ 云端API服务
  • 🤖 智能对话插件

fastmcp都能提供优雅的解决方案。现在就开始你的MCP之旅吧!欢迎在评论区分享你的应用场景和问题~

原创不易,转载请注明来源

Logo

欢迎加入 MCP 技术社区!与志同道合者携手前行,一同解锁 MCP 技术的无限可能!

更多推荐