这是我的第469篇原创文章。

一、引言

Model Context Protocol (MCP) 这个协议简单说就是给大语言模型接入外部数据和工具提供了一套标准化方案。MCP 统一了模型和各种数据源、工具服务之间的交互方式。

如果你有开发经验可以理解为MCP的每一个“能力”其实就是一个 可远程调用的函数(tool)。只要你能和 MCP Server 建立通信(通常通过 WebSocket、Stdio 或 HTTP RPC),你就可以像调用 REST API 一样调用 MCP 的函数。只不过MCP 的「函数」并不是像 Python RPC 那样“自动调用”,它还需要定义 Schema(输入输出结构)+ 元信息(描述),告诉大语言模型 怎么用、什么时候用、怎么传参数

官方 SDK 当然也能实现 MCP 服务器,但 FastMCP 提供的东西更多。可以说是一个生产级框架,内置了企业身份认证、客户端库、测试工具,还能自动生成 API 文档。FastMCP是目前用 Python 构建 MCP 服务器最顺手的框架,把底层那些复杂的协议实现全都封装好了,开发者只需要关注业务逻辑就行

二、实现过程

2.1 准备MCP服务

每个 MCP 服务器都从 FastMCP 类开始初始化。这个类管理着工具、资源和提示模板。我们先写个最简单的加法服务器。

新建 server.py 文件:

from fastmcp import FastMCP
mcp = FastMCP("Demo Server")
@mcp.tool
def add(a: int, b: int) -> int:
    """Add two numbers and return the result"""
    return a + b
if __name__ == "__main__": 
    mcp.run()

就这么多代码,一个功能完整的 MCP 服务器就写好了。客户端调用 add 工具时,服务器会执行加法运算并返回结果。

本地运行很简单

fastmcp run server.py

图片

FastMCP 服务器部署很灵活。测试阶段 fastmcp run 命令足够用。生产环境可以部署到 FastMCP Cloud,它提供现成的 HTTPS 端点和认证服务。

如果想自己托管的话,也可以用 HTTP 或 SSE 传输从自己的服务器提供服务:

 mcp.run(transport="http", host="0.0.0.0", port=8080)

部署完就能和语言模型、Web 客户端或者自动化工作流对接了。

2.2 客户端连接方式

服务器跑起来后,用 fastmcp.Client 就能连接。支持 STDIO、HTTP、SSE 等方式,测试时甚至可以跑内存模式。

新建 client.py文件:

from fastmcp import Client
import asyncio
async def main():
    async with Client("server.py") as client: 
        tools = await client.list_tools() 
        print("Available tools:", tools) 
        result = await client.call_tool("add", {"a": 5, "b": 7}) 
        print("Result:", result.content[0].text)
        
asyncio.run(main())

实际运行效果:

图片

2.3 客户端集成

以 CherryStudio为例,其他类型,添加一个: MCPServer,问答时选中MCP Server即可。

2.4 代码集成

服务器部署好之后,下一步是把它接到大语言模型上。这样 LLM 就能安全地调用服务器函数、读取资源、在对话中执行各种操作。

首先要定义 MCP 配置文件,列出可用的服务器、连接方式、认证需求这些信息。配置完成后 LLM 能自动发现 MCP 工具并按需调用。

假设服务器暴露了 add 或 summarize 工具,模型就能像调用内置功能一样使用它们。聊天场景下,用户说"总结下最新那篇文章",LLM 会自动调 summarize 工具,处理完结果再回复。

如果用 OpenAI Assistants API 或者 LangChain 这类框架开发应用,可以把 MCP 服务器注册成外部工具。LLM 通过 MCP 客户端库和它交互。

大模型调用MCP服务的代码:

import asyncio
from langchain_mcp_adapters.client import MultiServerMCPClient
from langgraph.prebuilt import create_react_agent
from langchain_openai import ChatOpenAI
import os
# 大模型请求地址
BASE_URL = "https://api.deepseek.com/v1"
API_KEY = "sk-xxxxxxxxxxxxxxxx"
MODEL_NAME = "deepseek-chat"
async def main():
    try: # 初始化MCP客户端 stdio 方式
        client = MultiServerMCPClient(
            {
                "Demo Server": 
                {
                "command":"python3",
                "args": [os.path.abspath("server.py")],  # 注意mcp的路径
                "transport":"stdio"
                } # 其它MCP   
            }
            )
        # 获取工具
        tools = await client.get_tools()
        if not tools:
            raise ValueError("未获取到任何工具")
        # 初始化一个 ChatOpenAI 实例,用于与大模型交互
        llm = ChatOpenAI(base_url=BASE_URL, openai_api_key=API_KEY, model=MODEL_NAME, timeout=60.0, max_retries=2)
        # 创建agent
        agent = create_react_agent(llm, tools)
        while True:  
            user_input = input("\n 请输入需求(或输入 exit 退出):\n> ")
            if user_input.strip().lower() == "exit":
                break
            async for chunk in agent.astream({"messages": user_input}):
                print(chunk)
    except Exception as e:
        print(f"程序初始化失败: {e}")
if __name__ == "__main__":
    asyncio.run(main())

代码解析:

    • MultiServerMCPClient 初始化

      • 配置MCP服务,指定启动命令(如:python server.py)、参数和通信方式(这里是 stdio)。

      • 可以同时接入多个MCP服务(如:MySQL、Redis、Web API等等)。

    • 获取工具(get_tools)

      • MCP会返回一个工具清单(类似API描述),大模型就知道能调用什么方法。

    • ChatOpenAI初始化

      • 通过ChatOpenAI封装大模型调用,配置base_url、API Key、模型名称、超时和重试机制。

    • 创建ReAct Agent

      • 使用create_react_agent,让大模型采用Reason + Act的模式:

        • 先思考该做什么(Reason),

        • 再调用MCP工具(Act),

        • 最后输出结果。

实际运行效果:

图片

三、小结

FastMCP 让 AI 和真实世界的数据、API、工具对接变得相当简单。几行 Python 代码就能搭起来功能完整的 MCP 服务器,对接语言模型、自动化工作流,安全地处理实际业务逻辑。

作者简介:

读研期间发表6篇SCI数据挖掘相关论文,现在某研究院从事数据算法相关科研工作,结合自身科研实践经历不定期分享关于Python、机器学习、深度学习、人工智能系列基础知识与应用案例。致力于只做原创,以最简单的方式理解和学习,关注我一起交流成长。需要数据集和源码的小伙伴可以关注底部公众号添加作者微信。

更多推荐