引言
在現代分布式系統和AI應用中,模型上下文協議(MCP)扮演著重要角色,它負責協調客戶端與服務器之間的通信,尤其是在需要頻繁交互的場景中。本文將介紹如何使用FastMCP庫快速實現客戶端與服務器之間的通信,并通過實戰案例展示其核心用法。
項目概述
本項目使用FastMCP庫實現了一個簡單的客戶端-服務器通信示例,包含基本的工具注冊與調用功能。通過本示例,你將了解FastMCP的核心概念、使用方法以及常見問題的解決方案。
項目結構
├── .venv/ # Python虛擬環境
├── client.py # 客戶端代碼
├── server.py # 服務器代碼
└── README.md # 項目說明文檔
實現步驟
步驟1: 創建虛擬環境并安裝依賴
首先,我們需要創建一個獨立的Python虛擬環境并安裝必要的依賴:
-
創建Python虛擬環境:
python -m venv .venv
-
激活虛擬環境:
# Linux/Mac系統 source .venv/bin/activate# Windows系統 .venv\Scripts\activate
-
安裝FastMCP庫:
pip install fastmcp
步驟2: 編寫服務器代碼
創建server.py
文件,實現一個簡單的MCP服務器,該服務器將提供一個hello
工具供客戶端調用:
# 導入FastMCP類,用于創建MCP服務器
from fastmcp import FastMCP# 創建FastMCP服務器實例,命名為"Demo"
mcp = FastMCP("Demo")# 使用裝飾器注冊一個名為"hello"的工具函數
@mcp.tool
def hello(name: str) -> str:"""向指定名稱的用戶發送問候參數:name: 字符串,要問候的用戶名稱返回:字符串,包含問候語的響應"""return f"Hello, {name}!"# 當腳本作為主程序運行時
if __name__ == "__main__":# 啟動MCP服務器# 默認使用STDIO傳輸方式,通過標準輸入輸出與客戶端通信mcp.run()
代碼解析:
- 我們通過
FastMCP
類創建了一個服務器實例,并命名為"Demo" - 使用
@mcp.tool
裝飾器注冊了一個名為hello
的工具函數 - 該工具函數接收一個
name
參數,并返回相應的問候語 - 服務器默認使用STDIO(標準輸入輸出)作為傳輸方式
- 最后通過
mcp.run()
啟動服務器
步驟3: 編寫客戶端代碼
創建client.py
文件,實現一個連接到服務器的客戶端:
# 導入asyncio模塊,用于支持異步操作
import asyncio
# 導入Client類,用于創建MCP客戶端
from fastmcp import Client# 創建Client實例,連接到名為"server.py"的服務器腳本
# 客戶端會自動推斷使用STDIO傳輸方式與服務器通信
client = Client("server.py")# 定義異步函數call_tool,用于調用服務器上的工具
async def call_tool(name: str):"""異步調用服務器上的hello工具參數:name: 字符串,要傳遞給hello工具的名稱參數返回:無返回值,但會打印服務器的響應結果"""# 使用異步上下文管理器連接到服務器async with client:# 調用服務器上的hello工具,傳遞name參數result = await client.call_tool("hello", {"name": name})# 打印服務器返回的結果print(result)# 運行異步函數call_tool,傳入默認名稱"Ford"
asyncio.run(call_tool("Ford"))
代碼解析:
- 我們創建了一個
Client
實例,指定要連接的服務器腳本 - 定義了異步函數
call_tool
用于調用服務器上的工具 - 使用異步上下文管理器
async with client
來管理與服務器的連接 - 通過
await client.call_tool()
方法調用服務器上的hello
工具 - 最后使用
asyncio.run()
運行異步函數,啟動客戶端
步驟4: 解決通信問題
在實現過程中,我們遇到了幾個典型問題并成功解決:
-
傳輸方式識別問題
- 問題:最初客戶端無法自動識別"stdio"傳輸方式
- 解決:通過查看源碼了解到FastMCP使用
PythonStdioTransport
類處理標準輸入輸出通信,客戶端會根據服務器類型自動選擇合適的傳輸方式
-
連接管理問題
- 問題:不清楚如何正確管理客戶端與服務器的連接狀態
- 解決:FastMCP Client采用異步上下文管理器模式進行連接管理,替代了顯式的
connect()
和disconnect()
方法,使用async with client
即可自動管理連接的建立和關閉
-
工具調用參數問題
- 問題:工具調用時參數傳遞不正確
- 解決:
call_tool
方法需要使用name
參數指定工具名稱,第二個參數為工具所需的參數字典
-
工具名稱匹配問題
- 問題:客戶端調用工具時提示找不到工具
- 解決:確保客戶端調用的工具名稱與服務器注冊的工具名稱完全匹配,包括大小寫
步驟5: 運行和測試
完成代碼編寫后,我們可以按照以下步驟運行和測試:
- 首先啟動服務器,在終端中執行:
python server.py
-
打開另一個終端,激活相同的虛擬環境,運行客戶端:
python client.py
-
客戶端將輸出服務器的響應結果:
CallToolResult(content=[TextContent(type='text', text='Hello, Ford!', annotations=None, meta=None)], structured_content={'result': 'Hello, Ford!'}, data='Hello, Ford!', is_error=False)
步驟6: trae中集成MCP
- 在trae ide中點擊手動添加MCP
- 把這段json輸入進去
{"mcpServers": {"local-mcp-server": {"command": "bash","args": ["-c","source /mcp-test/.venv/bin/activate && python /mcp-test/server.py"],"cwd": "/mcp-test"}}
}
3. 看到這個小綠對鉤就是成功了
4. 在對話中輸入“張三”,可以看到正常調用我們自定義的MCP服務
技術要點總結
-
FastMCP庫核心概念
- FastMCP提供了簡單易用的API,簡化了MCP客戶端-服務器通信的實現
- 服務器通過裝飾器注冊工具,客戶端通過名稱調用這些工具
-
STDIO傳輸方式
- 默認使用標準輸入輸出進行進程間通信
- 無需網絡配置,適合本地進程間的簡單通信場景
-
異步編程模型
- 客戶端基于asyncio實現異步操作
- 使用異步上下文管理器管理連接生命周期
- 通過
await
關鍵字處理異步工具調用
-
工具注冊與調用
- 服務器使用
@mcp.tool
裝飾器注冊工具函數 - 客戶端使用
call_tool
方法調用服務器上的工具 - 工具參數以字典形式傳遞,支持復雜數據結構
- 服務器使用
擴展與優化建議
-
對于生產環境,可以考慮:
- 使用更可靠的傳輸方式(如TCP)
- 添加身份驗證和數據加密
- 實現錯誤重試機制
- 添加日志記錄和監控
-
功能擴展:
- 注冊多個工具函數,實現更復雜的業務邏輯
- 添加工具調用的超時處理
- 支持更復雜的參數類型和返回值
- 實現批量工具調用功能
結論
通過本文的示例,我們展示了如何使用FastMCP庫快速實現客戶端與服務器之間的通信。FastMCP提供了簡潔的API和靈活的擴展能力,使開發者能夠專注于業務邏輯的實現,而無需過多關注底層通信細節。無論是在AI模型交互、分布式系統協調還是普通的進程間通信場景,FastMCP都是一個值得嘗試的優秀選擇。
希望本文能幫助你快速入門FastMCP的使用,如果有任何問題或建議,歡迎在評論區留言討論!