在人工智能領域,模型上下文協議(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 的高級用法和實戰案例。