FastMCP:為大語言模型構建強大的上下文和工具服務
在人工智能快速發展的今天,大語言模型(LLM)已經成為許多應用的核心。然而,如何讓這些模型更好地與外部世界交互,獲取實時信息,執行特定任務,一直是開發者面臨的挑戰。模型上下文協議(Model Context Protocol,簡稱MCP)應運而生,它為LLM提供了一種標準化的方式來訪問外部數據和功能。而FastMCP則是一個高效、簡潔的Python庫,讓開發者能夠輕松構建符合MCP規范的服務器和客戶端。
本文將深入介紹FastMCP的核心功能、架構設計和實際應用,幫助開發者快速掌握這一強大工具,為大語言模型構建更智能、更實用的應用。
1. FastMCP簡介
1.1 什么是MCP協議
模型上下文協議(Model Context Protocol)是一種專為大語言模型設計的標準化協議,它允許LLM以安全、一致的方式與外部系統交互。MCP協議常被描述為"AI的USB-C接口",提供了一種統一的方式連接LLM與它們可以使用的資源。
MCP協議的核心功能包括:
- 資源(Resources):類似于GET端點,用于將信息加載到LLM的上下文中
- 工具(Tools):類似于POST端點,用于執行代碼或產生副作用
- 提示(Prompts):可重用的LLM交互模板
- 上下文(Context):提供額外的交互功能,如日志記錄和進度報告
1.2 FastMCP的優勢
FastMCP是MCP協議的高級Python實現,它大大簡化了創建MCP服務器和客戶端的過程。與直接使用底層MCP SDK相比,FastMCP具有以下優勢:
- 高效開發:高級接口意味著更少的代碼和更快的開發速度
- 簡潔設計:最小化樣板代碼,專注于業務邏輯
- Python風格:API設計符合Python開發者的習慣
- 完整實現:提供MCP規范的全面實現
1.3 FastMCP 2.0的新特性
FastMCP 2.0是該項目的活躍開發版本,相比于貢獻給官方MCP SDK的1.0版本,它增加了許多強大的新功能:
- 強大的客戶端功能
- 服務器代理和組合
- OpenAPI/FastAPI集成
- 更多高級特性
2. Servers模塊:構建MCP服務器
Servers模塊是FastMCP的核心,提供了創建和管理MCP服務器的基礎設施。
2.1 FastMCP服務器基礎
創建一個基本的FastMCP服務器非常簡單:
from fastmcp import FastMCP# 創建一個FastMCP服務器實例
mcp = FastMCP(name="MyServer")# 啟動服務器
if __name__ == "__main__":mcp.run()
您還可以添加更多元數據來描述您的服務器:
from fastmcp import FastMCP# 創建帶有元數據的服務器
mcp = FastMCP(name="WeatherServer",description="提供天氣預報和歷史天氣數據",version="1.0.0",contact={"name": "開發團隊", "email": "dev@example.com"}
)if __name__ == "__main__":mcp.run()
2.2 Tools:為LLM提供功能
Tools(工具)是FastMCP中最常用的功能之一,它允許LLM執行特定的操作或函數。使用@mcp.tool()
裝飾器可以輕松將Python函數轉換為LLM可調用的工具:
from fastmcp import FastMCPmcp = FastMCP(name="CalculatorServer")@mcp.tool()
def add(a: float, b: float) -> float:"""將兩個數字相加。Args:a: 第一個數字b: 第二個數字Returns:兩個數字的和"""return a + b@mcp.tool()
def multiply(a: float, b: float) -> float:"""將兩個數字相乘。Args:a: 第一個數字b: 第二個數字Returns:兩個數字的乘積"""return a * bif __name__ == "__main__":mcp.run()
FastMCP支持異步工具函數,這對于需要I/O操作的任務特別有用:
import asyncio
from fastmcp import FastMCPmcp = FastMCP(name="AsyncServer")@mcp.tool()
async def fetch_data(url: str) -> dict:"""從指定URL異步獲取數據。Args:url: 要獲取數據的URLReturns:獲取的數據字典"""# 模擬異步網絡請求await asyncio.sleep(1)return {"url": url, "status": "success", "data": "示例數據"}
工具函數還可以使用上下文對象來訪問額外功能:
from fastmcp import FastMCP
from fastmcp.server import Contextmcp = FastMCP(name="ContextServer")@mcp.tool()
async def log_message(ctx: Context, message: str) -> bool:"""記錄消息并返回成功狀態。Args:ctx: FastMCP上下文對象message: 要記錄的消息Returns:操作是否成功"""# 使用上下文發送日志ctx.info(f"收到消息: {message}")return True
2.3 Resources:提供上下文數據
Resources(資源)用于向LLM提供數據和上下文信息。與工具不同,資源主要用于讀取數據而非執行操作:
from fastmcp import FastMCPmcp = FastMCP(name="DocumentServer")@mcp.resource()
def company_info() -> str:"""提供公司基本信息。"""return """公司名稱: 示例科技有限公司成立時間: 2020年主營業務: 人工智能解決方案員工人數: 100+"""@mcp.resource()
def pricing_policy() -> dict:"""返回產品定價策略。"""return {"basic_plan": {"price": 99, "features": ["基礎功能", "電子郵件支持"]},"pro_plan": {"price": 199, "features": ["所有基礎功能", "高級功能", "優先支持"]},"enterprise": {"price": "聯系銷售", "features": ["定制解決方案", "專屬支持團隊"]}}
資源也可以接受參數,這使得它們更加靈活:
@mcp.resource()
def user_profile(user_id: int) -> dict:"""獲取指定用戶的個人資料。Args:user_id: 用戶IDReturns:用戶個人資料"""# 在實際應用中,這里會從數據庫獲取用戶信息users = {1: {"name": "張三", "email": "zhang@example.com", "role": "管理員"},2: {"name": "李四", "email": "li@example.com", "role": "用戶"},}if user_id in users:return users[user_id]else:return {"error": "用戶不存在"}
2.4 Prompts:創建提示模板
Prompts(提示)允許您創建可重用的提示模板,這些模板可以被參數化并用于標準化LLM交互:
from fastmcp import FastMCP
from fastmcp.types import Messagemcp = FastMCP(name="CustomerServiceBot")@mcp.prompt()
def greeting(customer_name: str) -> list[Message]:"""創建一個個性化的問候提示。Args:customer_name: 客戶名稱Returns:包含問候消息的提示"""return [{"role": "system", "content": "你是一位專業的客戶服務代表,語氣友好且樂于助人。"},{"role": "user", "content": f"你好,我是{customer_name},我需要一些幫助。"}]@mcp.prompt()
def product_inquiry(product_name: str, specific_question: str = None) -> list[Message]:"""創建產品咨詢提示。Args:product_name: 產品名稱specific_question: 具體問題(可選)Returns:包含產品咨詢的提示"""content = f"我想了解關于{product_name}的信息。"if specific_question:content += f" 具體來說,{specific_question}"return [{"role": "system", "content": "你是產品專家,提供準確、簡潔的產品信息。"},{"role": "user", "content": content}]
2.5 Context:訪問服務器功能
Context(上下文)對象提供了在工具和資源函數內部訪問FastMCP服務器功能的方法:
from fastmcp import FastMCP
from fastmcp.server import Contextmcp = FastMCP(name="ContextDemoServer")@mcp.tool()
async def process_task(ctx: Context, task_name: str, complexity: int) -> dict:"""處理一個任務并報告進度。Args:ctx: FastMCP上下文對象task_name: 任務名稱complexity: 任務復雜度(1-10)Returns:任務處理結果"""# 發送信息日志ctx.info(f"開始處理任務: {task_name}")# 報告進度total_steps = complexityfor step in range(1, total_steps + 1):# 更新進度ctx.progress(step / total_steps, f"完成步驟 {step}/{total_steps}")# 如果任務復雜,發送詳細日志if complexity > 5:ctx.debug(f"步驟 {step} 詳細信息: 正在處理中...")# 任務完成,發送成功日志ctx.info(f"任務 {task_name} 已完成")return {"task": task_name,"status": "completed","complexity_level": complexity}
上下文對象提供的主要功能包括:
- 日志記錄(debug、info、warning、error)
- 進度報告
- 資源訪問
- 請求元數據訪問
2.6 Proxy Servers:代理遠程服務器
Proxy Servers(代理服務器)允許一個FastMCP服務器實例作為另一個MCP服務器的前端:
from fastmcp import FastMCP, Client# 假設我們有一個遠程MCP服務器運行在這個URL
remote_url = "https://api.example.com/mcp"# 創建一個客戶端連接到遠程服務器
client = Client(remote_url)# 從客戶端創建代理服務器
proxy_server = FastMCP.from_client(client=client,name="LocalProxy",description="本地代理到遠程MCP服務"
)if __name__ == "__main__":# 在本地運行代理服務器proxy_server.run()
代理服務器的主要用途包括:
- 為遠程MCP服務器提供不同的傳輸方式
- 集成不同實現的MCP服務器
- 創建服務器網關
2.7 Composition:組合多個服務器
Composition(組合)功能允許您將多個FastMCP服務器組合成一個更大的應用:
from fastmcp import FastMCP# 創建主服務器
main_server = FastMCP(name="MainApplication")# 創建工具服務器
tools_server = FastMCP(name="ToolsServer")@tools_server.tool()
def utility_function(text: str) -> str:return f"處理結果: {text.upper()}"# 創建數據服務器
data_server = FastMCP(name="DataServer")@data_server.resource()
def get_config() -> dict:return {"api_version": "v2", "timeout": 30}# 將工具服務器和數據服務器導入到主服務器
main_server.import_server(tools_server, prefix="tools")
main_server.import_server(data_server, prefix="data")if __name__ == "__main__":main_server.run()
FastMCP支持兩種組合方式:
- 靜態導入:使用
import_server
方法一次性復制組件 - 動態掛載:使用
mount
方法創建實時委托鏈接
# 動態掛載服務器
main_server.mount(auth_server, prefix="auth")
main_server.mount(content_server, prefix="content")
3. Deployment模塊:部署MCP服務器
Deployment模塊提供了多種方式來運行和部署FastMCP服務器。
3.1 Running the Server:運行服務器
最基本的運行方式是使用run()
方法:
from fastmcp import FastMCPmcp = FastMCP(name="SimpleServer")@mcp.tool()
def hello(name: str) -> str:return f"你好,{name}!"if __name__ == "__main__":# 使用默認設置運行服務器mcp.run()
您可以配置多種運行選項:
if __name__ == "__main__":# 使用HTTP傳輸并指定端口mcp.run(transport="http",host="0.0.0.0",port=8080,log_level="info")
FastMCP支持多種傳輸協議:
- HTTP:標準HTTP傳輸
- SSE:服務器發送事件傳輸
- STDIO:標準輸入/輸出傳輸(用于命令行工具)
3.2 ASGI Integration:與ASGI框架集成
FastMCP可以輕松集成到現有的ASGI應用中,如Starlette和FastAPI:
from fastmcp import FastMCP
from starlette.applications import Starlette
from starlette.routing import Route, Mount
from starlette.responses import JSONResponse# 創建FastMCP服務器
mcp = FastMCP(name="APIServer")@mcp.tool()
def greet(name: str) -> str:return f"你好,{name}!"# 創建Starlette路由
async def homepage(request):return JSONResponse({"message": "歡迎訪問API服務器"})# 獲取MCP的ASGI應用
mcp_app = mcp.http_app()# 創建Starlette應用并掛載MCP
app = Starlette(routes=[Route("/", homepage),Mount("/mcp", app=mcp_app)
])# 使用uvicorn運行
# uvicorn myapp:app
與FastAPI集成:
from fastmcp import FastMCP
from fastapi import FastAPI# 創建FastMCP服務器
mcp = FastMCP(name="FastAPIIntegration")@mcp.tool()
def analyze_text(text: str) -> dict:word_count = len(text.split())char_count = len(text)return {"word_count": word_count,"character_count": char_count,"summary": f"文本包含{word_count}個單詞,{char_count}個字符"}# 創建FastAPI應用
app = FastAPI(title="集成示例")@app.get("/")
def read_root():return {"message": "歡迎訪問API"}# 掛載MCP到FastAPI
app.mount("/mcp", mcp.http_app())
ASGI集成的主要用途包括:
- 向現有網站或API添加MCP功能
- 在特定URL路徑下掛載MCP服務器
- 結合多種服務在單一應用中
3.3 Authentication:服務器認證
FastMCP利用底層MCP SDK的OAuth 2.0支持提供認證功能:
from fastmcp import FastMCP# 創建帶有認證的服務器
mcp = FastMCP(name="SecureServer",# 配置認證(這里使用簡化示例,實際使用時需參考MCP認證文檔)auth={"type": "oauth2","flows": {"clientCredentials": {"tokenUrl": "https://auth.example.com/token","scopes": {"read": "讀取權限","write": "寫入權限"}}}}
)@mcp.tool()
def secure_operation(data: str) -> str:return f"安全處理: {data}"
3.4 CLI:命令行界面
FastMCP提供了一個命令行界面,使您可以輕松運行和管理服務器:
# 直接運行服務器
fastmcp run server.py# 以開發模式運行(帶MCP檢查器)
fastmcp dev server.py# 安裝到Claude桌面應用
fastmcp install server.py# 查看版本信息
fastmcp version
CLI還支持依賴管理:
# 使用指定依賴運行
fastmcp run server.py --with pandas,numpy# 使用可編輯依賴運行
fastmcp run server.py --with-editable ./my_package
4. Clients模塊:與MCP服務器交互
Clients模塊提供了與MCP服務器交互的客戶端功能。
4.1 Client Overview:客戶端概述
基本的客戶端使用方式:
import asyncio
from fastmcp import Clientasync def main():# 連接到MCP服務器async with Client("http://localhost:8080") as client:# 調用工具result = await client.tools.call("add", a=5, b=3)print(f"計算結果: {result}")# 讀取資源weather = await client.resources.read("current_weather", city="北京")print(f"天氣信息: {weather}")if __name__ == "__main__":asyncio.run(main())
客戶端還支持回調處理:
import asyncio
from fastmcp import Clientasync def progress_callback(progress: float, message: str):print(f"進度: {progress*100:.1f}% - {message}")async def main():async with Client("http://localhost:8080") as client:# 注冊回調client.register_progress_callback(progress_callback)# 調用可能產生進度更新的工具result = await client.tools.call("process_large_file", filename="data.csv")print(f"處理結果: {result}")
4.2 Transports:客戶端傳輸
FastMCP客戶端支持多種傳輸方式:
import asyncio
from fastmcp import Client
from fastmcp.client.transports import StreamableHttpTransportasync def main():# 創建HTTP傳輸transport = StreamableHttpTransport(base_url="http://localhost:8080")# 使用指定傳輸創建客戶端async with Client(transport=transport) as client:result = await client.tools.call("hello", name="世界")print(result)
對于測試,可以使用內存傳輸:
import asyncio
from fastmcp import FastMCP, Client
from fastmcp.client.transports import FastMCPTransport# 創建服務器
mcp = FastMCP(name="TestServer")@mcp.tool()
def echo(message: str) -> str:return f"回聲: {message}"async def main():# 創建內存傳輸transport = FastMCPTransport(mcp=mcp)# 使用內存傳輸創建客戶端async with Client(transport=transport) as client:result = await client.tools.call("echo", message="測試消息")print(result) # 輸出: 回聲: 測試消息
FastMCP支持的主要傳輸類型包括:
- StreamableHttpTransport:用于HTTP連接(推薦用于遠程服務器)
- SSETransport:用于SSE連接(舊版選項)
- FastMCPTransport:用于內存中測試
- UvxStdioTransport/NpxStdioTransport:用于本地服務器
5. Patterns模塊:高級使用模式
Patterns模塊提供了一系列高級使用模式和集成方案。
5.1 Decorating Methods:裝飾方法模式
在類中使用FastMCP裝飾器需要特別注意:
from fastmcp import FastMCPmcp = FastMCP(name="ClassMethodsDemo")class Calculator:def __init__(self, precision: int = 2):self.precision = precision# 不要直接裝飾實例方法def add_wrong(self, a: float, b: float) -> float:"""這種方式會導致錯誤,因為裝飾器在實例存在前捕獲方法"""return round(a + b, self.precision)# 正確方式:使用靜態方法或類方法@staticmethod@mcp.tool()def add(a: float, b: float, precision: int = 2) -> float:"""正確方式:使用靜態方法并將實例屬性作為參數傳遞"""return round(a + b, precision)# 另一種方式:工廠函數@classmethoddef register_methods(cls, server: FastMCP):"""使用工廠方法注冊實例方法"""calculator = cls()@server.tool()def multiply(a: float, b: float) -> float:return round(a * b, calculator.precision)return calculator# 注冊方法
calc = Calculator.register_methods(mcp)
5.2 HTTP Requests:訪問HTTP請求信息
在FastMCP服務器中訪問HTTP請求信息:
from fastmcp import FastMCP
from fastmcp.server.dependencies import get_http_request
from starlette.requests import Requestmcp = FastMCP(name="HTTPRequestDemo")@mcp.tool()
async def user_agent_info(request: Request = get_http_request()) -> dict:"""返回有關客戶端的信息。Args:request: HTTP請求對象(自動注入)Returns:客戶端信息"""return {"user_agent": request.headers.get("user-agent", "未知"),"client_ip": request.client.host if request.client else "未知","request_path": request.url.path,"query_params": dict(request.query_params)}
5.3 OpenAPI:從OpenAPI生成服務器
FastMCP可以從OpenAPI規范自動生成MCP服務器:
import httpx
from fastmcp import FastMCP# 創建API客戶端
api_client = httpx.AsyncClient(base_url="https://api.example.com")# 加載OpenAPI規范
openapi_spec = {"openapi": "3.0.0","info": {"title": "示例API", "version": "1.0.0"},"paths": {"/users": {"get": {"summary": "獲取用戶列表","operationId": "getUsers","responses": {"200": {"description": "成功響應","content": {"application/json": {}}}}}},"/users/{userId}": {"get": {"summary": "獲取用戶詳情","operationId": "getUserById","parameters": [{"name": "userId","in": "path","required": True,"schema": {"type": "integer"}}],"responses": {"200": {"description": "成功響應","content": {"application/json": {}}}}}}}
}# 從OpenAPI規范創建MCP服務器
mcp = FastMCP.from_openapi(openapi_spec=openapi_spec,client=api_client
)
5.4 FastAPI:從FastAPI應用生成服務器
FastMCP可以從FastAPI應用自動生成MCP服務器:
from fastapi import FastAPI, Path
from fastmcp import FastMCP# 創建FastAPI應用
app = FastAPI(title="用戶API")@app.get("/users")
async def get_users():"""獲取所有用戶列表"""return [{"id": 1, "name": "張三"},{"id": 2, "name": "李四"}]@app.get("/users/{user_id}")
async def get_user(user_id: int = Path(..., description="用戶ID")):"""獲取特定用戶信息"""users = {1: {"id": 1, "name": "張三", "email": "zhang@example.com"},2: {"id": 2, "name": "李四", "email": "li@example.com"}}if user_id in users:return users[user_id]return {"error": "用戶不存在"}# 從FastAPI應用創建MCP服務器
mcp = FastMCP.from_fastapi(app)
5.5 Contrib Modules:貢獻模塊
FastMCP包含一個contrib包,其中包含社區貢獻的擴展模塊:
from fastmcp import FastMCP
from fastmcp.contrib import my_module # 假設存在這樣的貢獻模塊mcp = FastMCP(name="ContribDemo")# 使用貢獻模塊的功能
my_module.setup(mcp)
5.6 Testing:測試MCP服務器
FastMCP提供了測試MCP服務器的工具和模式:
import pytest
from fastmcp import FastMCP, Client
from fastmcp.client.transports import FastMCPTransport# 創建要測試的服務器
@pytest.fixture
def mcp_server():mcp = FastMCP(name="TestServer")@mcp.tool()def add(a: float, b: float) -> float:return a + breturn mcp# 創建客戶端測試夾具
@pytest.fixture
async def client(mcp_server):transport = FastMCPTransport(mcp=mcp_server)async with Client(transport=transport) as client:yield client# 編寫測試
async def test_add_tool(client):result = await client.tools.call("add", a=2, b=3)assert result == 5
總結
FastMCP是一個強大而靈活的Python庫,為開發者提供了構建符合MCP規范的服務器和客戶端的簡便方法。通過其簡潔的API和豐富的功能,FastMCP使得為大語言模型創建上下文和工具服務變得前所未有的簡單。
無論您是想為LLM提供數據訪問能力,還是想讓它們能夠執行特定操作,FastMCP都能滿足您的需求。隨著AI應用的不斷發展,FastMCP將成為連接大語言模型與外部世界的重要橋梁。
通過本文的介紹,希望您已經對FastMCP有了全面的了解,并能夠開始使用它來構建自己的MCP服務。隨著FastMCP 2.0的持續發展,我們可以期待更多強大功能的加入,進一步擴展大語言模型的能力邊界。