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
繁體地址,請注明出處: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日,山東高密友道化學有限公司的車間爆炸聲,像一把利刃劃破了化工行業的平靜。劇烈的沖擊波將車間夷為平地,黑色蘑菇云騰空而起,刺鼻的化學氣味彌漫…