在人工智能领域,模型上下文协议(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
繁体地址,请注明出处:http://hk.pswp.cn/bicheng/83557.shtml
英文地址,请注明出处:http://en.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日,山东高密友道化学有限公司的车间爆炸声,像一把利刃划破了化工行业的平静。剧烈的冲击波将车间夷为平地,黑色蘑菇云腾空而起,刺鼻的化学气味弥漫…