FastMCP:构建 MCP 服务器和客户端的高效 Python 框架

在人工智能领域,模型上下文协议(Model Context Protocol,简称 MCP)作为一种标准化的协议,为大型语言模型(LLM)提供了丰富的上下文和工具支持。而 FastMCP 作为构建 MCP 服务器和客户端的 Python 框架,以其简洁的 API 设计、高效的开发体验以及强大的扩展能力,正逐渐成为开发者们的首选工具。

一、FastMCP 简介

FastMCP 是一个用于构建 MCP 服务器和客户端的 Python 框架,其目标是让开发者能够以更少的代码、更高的效率构建出功能强大的 MCP 应用。相比官方的 MCP Python SDK,FastMCP 在多个方面进行了优化和改进,提供了更简洁的 API 设计、更高效的开发体验以及更强大的扩展能力。

FastMCP 的核心优势主要体现在以下几个方面:

(一)简洁的 API 设计

FastMCP 采用了 Pythonic 的设计风格,通过装饰器(Decorator)即可轻松定义工具(Tools)、资源(Resources)和提示(Prompts)。这种设计方式极大地减少了开发过程中的样板代码,让开发者能够更加专注于业务逻辑的实现。

(二)高效的开发体验

FastMCP 提供了丰富的内置功能,如服务器组合、远程服务器代理、OpenAPI/FastAPI 集成等,这些功能不仅提高了开发效率,还为开发者提供了更多的灵活性。此外,FastMCP 还支持多种客户端/服务端传输模式,包括 Stdio、SSE 和内存传输,能够满足不同场景下的开发需求。

(三)强大的扩展能力

FastMCP 的设计具有高度的可扩展性,开发者可以根据自己的需求轻松添加新的功能和模块。无论是构建复杂的 AI 应用,还是进行简单的数据处理,FastMCP 都能够提供强大的支持。

二、FastMCP 与官方 SDK 的关系

FastMCP 1.0 的核心概念已经被纳入官方的 MCP Python SDK,而当前的 FastMCP 2.0 则是在此基础上进行的进一步扩展和优化。FastMCP 2.0 不仅继承了 1.0 版本的基础功能,还引入了完整的客户端支持、服务器组合、OpenAPI/FastAPI 集成、远程服务器代理以及内置测试工具等新功能,显著提升了开发效率和应用的灵活性。

三、开发示例

为了更好地理解 FastMCP 的强大功能和简洁的开发方式,我们以一个基于 FastMCP 的数学运算智能问答应用为例,详细介绍其开发过程。

(一)安装 FastMCP

首先,需要安装 FastMCP 框架。通过以下命令即可完成安装:

uv pip install fastmcp

(二)服务端实现

服务端的实现非常简单,只需要定义几个基本的数学运算工具即可。以下是服务端的代码示例:

from fastmcp import FastMCPmcp = FastMCP(name="MyAssistantServer")@mcp.tool()
def add(a: float, b: float) -> float:"""加法运算参数:a: 第一个数字b: 第二个数字返回:两数之和"""return a + b@mcp.tool()
def subtract(a: float, b: float) -> float:"""减法运算参数:a: 第一个数字b: 第二个数字返回:两数之差 (a - b)"""return a - b@mcp.tool()
def multiply(a: float, b: float) -> float:"""乘法运算参数:a: 第一个数字b: 第二个数字返回:两数之积"""return a * b@mcp.tool()
def divide(a: float, b: float) -> float:"""除法运算参数:a: 被除数b: 除数返回:两数之商 (a / b)异常:ValueError: 当除数为零时"""if b == 0:raise ValueError("除数不能为零")return a / bif __name__ == "__main__":mcp.run(transport='sse', host="127.0.0.1", port=8001)

在上述代码中,我们定义了四个基本的数学运算工具:加法、减法、乘法和除法。通过装饰器 @mcp.tool(),这些函数被注册为 MCP 服务器的工具,可供客户端调用。最后,通过调用 mcp.run() 方法启动服务器,并指定使用 SSE 传输模式。

(三)客户端实现

客户端的实现同样简单。通过一行代码即可创建一个 MCP 客户端,并连接到服务端。以下是客户端的代码示例:

from fastmcp import Client
import asyncioasync def main():# 测试 mcp 客户端的功能async with Client("http://127.0.0.1:8001/sse") as mcp_client:tools = await mcp_client.list_tools()print(f"Available tools: {tools}")result = await mcp_client.call_tool("add", {"a": 5, "b": 3})print(f"Result: {result[0].text}")if __name__ == "__main__":asyncio.run(main())

在上述代码中,我们通过 Client 类创建了一个 MCP 客户端,并连接到服务端。通过调用 list_tools() 方法,我们可以获取服务端提供的所有工具列表。然后,通过调用 call_tool() 方法,我们可以调用服务端的工具并获取结果。

(四)数学运算智能问答应用

基于 FastMCP,我们还可以构建一个数学运算智能问答应用。该应用通过与大语言模型(LLM)进行交互,根据用户的输入调用相应的工具进行计算,并返回结果。以下是该应用的代码示例:

import asyncio
import json
import logging
import os
from typing import List, Dictfrom fastmcp import Client
from openai import OpenAIclass LLMClient:"""LLM客户端,负责与大语言模型API通信"""def __init__(self, model_name: str, url: str, api_key: str) -> None:self.model_name: str = model_nameself.url: str = urlself.client = OpenAI(api_key=api_key, base_url=url)def get_response(self, messages: List[Dict[str, str]]) -> str:"""发送消息给LLM并获取响应"""response = self.client.chat.completions.create(model=self.model_name,messages=messages,stream=False)return response.choices[0].message.contentclass ChatSession:"""聊天会话,处理用户输入和LLM响应,并与MCP工具交互"""def __init__(self, llm_client: LLMClient, mcp_client: Client) -> None:self.mcp_client: Client = mcp_clientself.llm_client: LLMClient = llm_clientasync def process_llm_response(self, llm_response: str) -> str:"""处理LLM响应,解析工具调用并执行"""try:# 尝试移除可能的markdown格式if llm_response.startswith('```json'):llm_response = llm_response.strip('```json').strip('```').strip()tool_call = json.loads(llm_response)if "tool" in tool_call and "arguments" in tool_call:# 检查工具是否可用tools = await self.mcp_client.list_tools()if any(tool.name == tool_call["tool"] for tool in tools):try:# 执行工具调用result = await self.mcp_client.call_tool(tool_call["tool"], tool_call["arguments"])return f"Tool execution result: {result}"except Exception as e:error_msg = f"Error executing tool: {str(e)}"logging.error(error_msg)return error_msgreturn f"No server found with tool: {tool_call['tool']}"return llm_responseexcept json.JSONDecodeError:# 如果不是JSON格式,直接返回原始响应return llm_responseasync def start(self, system_message: str) -> None:"""启动聊天会话的主循环"""messages = [{"role": "system", "content": system_message}]while True:try:# 获取用户输入user_input = input("用户: ").strip().lower()if user_input in ["quit", "exit", "退出"]:print('AI助手退出')breakmessages.append({"role": "user", "content": user_input})# 获取LLM的初始响应llm_response = self.llm_client.get_response(messages)print("助手: ", llm_response)# 处理可能的工具调用result = await self.process_llm_response(llm_response)# 如果处理结果与原始响应不同,说明执行了工具调用,需要进一步处理while result != llm_response:messages.append({"role": "assistant", "content": llm_response})messages.append({"role": "system", "content": result})# 将工具执行结果发送回LLM获取新响应llm_response = self.llm_client.get_response(messages)result = await self.process_llm_response(llm_response)print("助手: ", llm_response)messages.append({"role": "assistant", "content": llm_response})except KeyboardInterrupt:print('AI助手退出')breakasync def main():async with Client("http://127.0.0.1:8001/sse") as mcp_client:# 初始化LLM客户端,使用通义千问模型llm_client = LLMClient(model_name='qwen-plus-latest', api_key=os.getenv('DASHSCOPE_API_KEY'),url='https://dashscope.aliyuncs.com/compatible-mode/v1')# 获取可用工具列表并格式化为系统提示的一部分tools = await mcp_client.list_tools()dict_list = [tool.__dict__ for tool in tools]tools_description = json.dumps(dict_list, ensure_ascii=False)# 系统提示,指导LLM如何使用工具和返回响应system_message = f'''你是一个智能助手,严格遵循以下协议返回响应:可用工具:{tools_description}响应规则:1、当需要计算时,返回严格符合以下格式的纯净JSON:{{"tool": "tool-name","arguments": {{"argument-name": "value"}}}}2、禁止包含以下内容:- Markdown标记(如```json)- 自然语言解释(如"结果:")- 格式化数值(必须保持原始精度)- 单位符号(如元、kg)校验流程:✓ 参数数量与工具定义一致✓ 数值类型为number✓ JSON格式有效性检查正确示例:用户:单价88.5买235个多少钱?响应:{{"tool":"multiply","arguments":{{"a":88.5,"b":235}}}}错误示例:用户:总金额是多少?错误响应:总价500元 → 含自然语言错误响应:```json{{...}}```→ 含Markdown3、在收到工具的响应后:- 将原始数据转化为自然、对话式的回应- 保持回复简洁但信息丰富- 聚焦于最相关的信息- 使用用户问题中的适当上下文- 避免简单重复使用原始数据'''# 启动聊天会话chat_session = ChatSession(llm_client=llm_client, mcp_client=mcp_client)await chat_session.start(system_message=system_message)if __name__ == "__
__main__":asyncio.run(main())

(五)运行验证

运行服务端代码:

python fast_mcp_server.py

运行客户端代码:

python fast_mcp_client.py

在客户端中输入数学问题,例如:

用户: 现在要购买一批货,单价是 1034.32423,数量是 235326。商家后来又说,可以在这个基础上,打95折,折后总价是多少?

客户端会调用服务端的 multiply 工具进行计算,并返回结果:

助手:  {"tool": "multiply","arguments": {"a": 1034.32423,"b": 235326}
}
助手:  {"tool": "multiply","arguments": {"a": 243403383.74898,"b": 0.95}
}
助手:  折后总价是231233214.56。

四、FastMCP 的更多功能

除了上述提到的功能外,FastMCP 还提供了许多其他强大的功能,例如:

(一)服务器组合

通过 mcp.mount()mcp.import_server() 方法,可以将多个 FastMCP 实例组合到一个父服务器中,从而构建出模块化的应用程序。这种方式不仅提高了代码的可维护性,还方便了功能的扩展。

(二)OpenAPI/FastAPI 集成

FastMCP 提供了从现有的 OpenAPI 规范或 FastAPI 应用程序生成 FastMCP 服务器的功能。通过这种方式,开发者可以轻松地将现有的 Web API 集成到 MCP 生态系统中,进一步扩展了 MCP 的应用场景。

(三)代理服务器

通过 FastMCP.as_proxy() 方法,可以创建一个代理服务器,该服务器可以作为本地或远程 MCP 服务器的中间层。这种方式特别适用于桥接不同的传输协议(例如,将远程 SSE 服务器代理到本地 Stdio 客户端)或为不受控制的服务器添加逻辑层。

(四)内置测试工具

FastMCP 提供了强大的内置测试工具,支持通过内存传输直接连接到 FastMCP 服务器实例,从而在测试过程中无需进行进程管理和网络调用。这种方式极大地提高了测试效率,降低了测试成本。

五、总结

FastMCP 作为一个高效、简洁且功能强大的 Python 框架,为构建 MCP 服务器和客户端提供了极大的便利。通过其简洁的 API 设计、高效的开发体验和强大的扩展能力,开发者可以快速构建出功能丰富的 MCP 应用。无论是简单的工具开发,还是复杂的 AI 应用构建,FastMCP 都能够满足开发者的需求。

如果你对 MCP 或 FastMCP 感兴趣,不妨尝试使用它来构建自己的应用。相信你一定会被其强大的功能和简洁的设计所吸引。同时,也欢迎关注我的后续文章,我将继续介绍更多关于 MCP 的内容,以及 FastMCP 的高级用法和实战案例。


本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处:http://www.pswp.cn/bicheng/83557.shtml

如若内容造成侵权/违法违规/事实不符,请联系多彩编程网进行投诉反馈email:809451989@qq.com,一经查实,立即删除!

相关文章

动态库导出符号与extern “C“

1. windows下动态库导出符号 根据C/C语法规则,函数声明中的修饰符(如__declspec(dllexport))可以放在返回类型之前或返回类型之后、函数名之前。这两种方式在功能上是等价的,编译器会以相同的方式处理。 __declspec(dllexport) …

Linux(9)——进程(控制篇——下)

目录 三、进程等待 1)进程等待的必要性 2)获取子进程的status 3)进程的等待方法 wait方法 waitpid方法 多进程创建以及等待的代码模型 非阻塞的轮训检测 四、进程程序替换 1)替换原理 2)替换函数 3&…

Datatable和实体集合互转

1.使用已废弃的 JavaScriptSerializer,且反序列化为弱类型 ArrayList。可用但不推荐。 using System; using System.Collections; using System.Collections.Generic; using System.Data; using System.Linq; using System.Reflection; using System.Web; using Sy…

阿里云服务器ECS详解:云服务器是什么,云服务器优势和应用场景及参考

云服务器ECS是阿里云众多云产品中,最受用户关注的产品,阿里云服务器提供多样化的计算能力,支持x86、Arm架构,涵盖CPU、GPU等多种服务器类型,满足各种用户需求。其便捷易用特性包括分钟级交付、通用API和性能监控框架&a…

【Oracle】游标

个人主页:Guiat 归属专栏:Oracle 文章目录 1. 游标基础概述1.1 游标的概念与作用1.2 游标的生命周期1.3 游标的分类 2. 显式游标2.1 显式游标的基本语法2.1.1 声明游标2.1.2 带参数的游标 2.2 游标的基本操作2.2.1 完整的游标操作示例 2.3 游标属性2.3.1…

pikachu靶场通关笔记11 XSS关卡07-XSS之关键字过滤绕过(三种方法渗透)

目录 一、源码分析 1、进入靶场 2、代码审计 3、攻击思路 二、渗透实战 1、探测过滤信息 2、注入Payload1 3、注入Payload2 4、注入Payload3 本系列为通过《pikachu靶场通关笔记》的XSS关卡(共10关)渗透集合,通过对XSS关卡源码的代码审计找到安…

XML 元素:基础、应用与优化

XML 元素:基础、应用与优化 引言 XML(可扩展标记语言)作为一种数据交换的标准格式,广泛应用于互联网数据交换、数据存储等领域。XML 元素是 XML 文档的核心组成部分,本文将深入探讨 XML 元素的概念、特性、应用以及优化方法。 一、XML 元素概述 1.1 XML 元素的定义 X…

【Axure高保真原型】交通事故大屏可视化分析案例

今天和大家分享交通事故大屏可视化分析案例的原型模板,包括饼图分类分析、动态显示发生数、柱状图趋势分析、中部地图展示最新事故发现地点和其他信息、右侧列表记录发生事故的信息…… 通过多种可视化图表展示分析结果,具体效果可以点击下方视频观看或…

HCIP(BGP基础)

一、BGP 基础概念 1. 网络分类与协议定位 IGP(内部网关协议):用于自治系统(AS)内部路由,如 RIP、OSPF、EIGRP,关注选路效率、收敛速度和资源占用。EGP(外部网关协议)&a…

【HarmonyOS 5】 ArkUI-X开发中的常见问题及解决方案

一、跨平台编译与适配问题 1. 平台特定API不兼容 ‌问题现象‌:使用Router模块的replaceUrl或startAbility等鸿蒙专属API时,编译跨平台工程报错cant support crossplatform application。 ‌解决方案‌: 改用ohos.router的跨平台封装API&a…

Matlab2018a---安装教程

目录 壹 | 引 言 贰 | 安装环境 叁 | 安 装 肆 | 结 语 壹 | 引 言 大家好,我是子正。 最近想学习一下DSP数字信号处理有关的知识,要用到Matlab进行数据处理,于是又重新把Matlab捡了回来; 记得上学那会儿用的还是Matlab2012a&#xff…

分布式流处理与消息传递——Kafka ISR(In-Sync Replicas)算法深度解析

Java Kafka ISR(In-Sync Replicas)算法深度解析 一、ISR核心原理 #mermaid-svg-OQtnaUGNQ9PMgbW0 {font-family:"trebuchet ms",verdana,arial,sans-serif;font-size:16px;fill:#333;}#mermaid-svg-OQtnaUGNQ9PMgbW0 .error-icon{fill:#55222…

ARM GIC V3概述

中断类型 locality- specific peripheral interrupt(LPI):LPI是一个有针对性的外设中断,通过affinity路由到特定的PE。 为非安全group1中断边沿触发可以通过its进行路由没有active状态,所以不需要明确的停用操作LPI总…

蓝桥杯国赛训练 day1

目录 k倍区间 舞狮 交换瓶子 k倍区间 取模后算组合数就行 import java.util.HashMap; import java.util.Map; import java.util.Scanner;public class Main {static Scanner sc new Scanner(System.in);public static void main(String[] args) {solve();}public static vo…

安装和配置 Nginx 和 Mysql —— 一步一步配置 Ubuntu Server 的 NodeJS 服务器详细实录6

前言 昨天更新了四篇博客,我们顺利的 安装了 ubuntu server 服务器,并且配置好了 ssh 免密登录服务器,安装好了 服务器常用软件安装, 配置好了 zsh 和 vim 以及 通过 NVM 安装好Nodejs,还有PNPM包管理工具 。 作为服务器的运行…

鸿蒙版Taro 搭建开发环境

鸿蒙版Taro 搭建开发环境 一、配置鸿蒙环境 下载安装 DevEco 建议使用最新版本的 IDE,当前为 5.0.5Release 版本。 二、创建鸿蒙项目 打开 DevEco,点击右上角的 Create Project,在 Application 处选择 Empty Ability,点击 Ne…

Could not get unknown property ‘mUser‘ for Credentials [username: null]

最近遇到jekins打包报错: Could not get unknown property mUser for Credentials [username: null] of type org.gradle.internal.credentials.DefaultPasswordCredentials_Decorated。 项目使用的是gradle,通过pipeline打docker包;因为ma…

Spring Boot + MyBatis-Plus 读写分离与多 Slave 负载均衡示例

Spring Boot + MyBatis-Plus 读写分离与多 Slave 负载均衡示例 一、项目结构 src/main/java/com/example/demo/ ├── config/ │ ├── DataSourceConfig.java # 数据源配置 │ ├── MyBatisPlusConfig.java # MyBatis-Plus配置 ├── constant/ │…

android binder(1)基本原理

一、IPC 进程间通信(IPC,Inter-Process Communication)机制,用于解决不同进程间的数据交互问题。 不同进程之间用户地址空间的变量和函数是不能相互访问的,但是不同进程的内核地址空间是相同和共享的,我们可…

高密爆炸警钟长鸣:AI为化工安全戴上“智能护盾”

一、高密爆炸:一声巨响,撕开化工安全“伤疤” 2025年5月27日,山东高密友道化学有限公司的车间爆炸声,像一把利刃划破了化工行业的平静。剧烈的冲击波将车间夷为平地,黑色蘑菇云腾空而起,刺鼻的化学气味弥漫…