在上一篇中，我們構建了一個基于內存存儲的食譜助手。說實話，內存存儲雖然簡單，但有個致命問題：程序一重啟，數據就全沒了。
所以這篇我們要解決數據持久化的問題，將食譜助手從內存存儲升級到SQLite數據庫。
項目結構重組
先把項目結構整理得更清晰一些。之前我們所有代碼都在一個文件里，現在按功能分模塊：
mkdir -p recipe-assistant/app recipe-assistant/data
cd recipe-assistant
touch app/__init__.py app/main.py app/models.py
新的目錄結構：
recipe-assistant/app/__init__.pymain.py # 主應用入口（包含所有MCP組件）models.py # 數據模型和數據庫管理data/recipes.db # SQLite數據庫文件bb
重要說明：FastMCP不支持模塊化的方式（如include_router），所有的資源、工具和提示詞組件都必須定義在同一個FastMCP實例上。因此我們將所有MCP組件都放在main.py中。
數據庫設計
創建 app/models.py，定義數據模型和數據庫操作：
# app/models.py
from pydantic import BaseModel
from typing import List, Optional, Dict
import sqlite3
import json
import os# 數據模型定義（與第1篇保持一致）
class Ingredient(BaseModel):name: strquantity: strclass Recipe(BaseModel):id: strname: strcuisine: strdescription: stringredients: List[Ingredient]steps: List[str]difficulty: strclass UserPreference(BaseModel):user_id: strfavorite_cuisines: List[str]dietary_restrictions: List[str]cooking_skill: str# 數據庫管理類
class DatabaseManager:def __init__(self, db_path="data/recipes.db"):self.db_path = db_pathself.conn = Noneself.initialize_db()def get_connection(self):if self.conn is None:# 確保數據目錄存在os.makedirs(os.path.dirname(self.db_path), exist_ok=True)self.conn = sqlite3.connect(self.db_path)self.conn.execute("PRAGMA foreign_keys = ON")self.conn.row_factory = sqlite3.Rowreturn self.conndef initialize_db(self):"""創建數據庫表并導入第1篇的示例數據"""conn = self.get_connection()cursor = conn.cursor()# 創建食譜表cursor.execute('''CREATE TABLE IF NOT EXISTS recipes (id TEXT PRIMARY KEY,name TEXT NOT NULL,cuisine TEXT NOT NULL,description TEXT,ingredients TEXT NOT NULL,steps TEXT NOT NULL,difficulty TEXT NOT NULL)''')# 創建用戶偏好表cursor.execute('''CREATE TABLE IF NOT EXISTS user_preferences (user_id TEXT PRIMARY KEY,favorite_cuisines TEXT NOT NULL,dietary_restrictions TEXT NOT NULL,cooking_skill TEXT NOT NULL)''')conn.commit()# 導入第1篇的示例數據self.import_data()def import_data(self):"""導入第1篇的示例數據"""conn = self.get_connection()cursor = conn.cursor()# 檢查是否已有數據cursor.execute("SELECT COUNT(*) FROM recipes")count = cursor.fetchone()[0]if count == 0:# 第1篇的示例食譜數據recipes = [{"id": "recipe_001","name": "宮保雞丁","cuisine": "川菜","description": "經典川菜，麻辣鮮香","ingredients": [{"name": "雞胸肉", "quantity": "300g"},{"name": "花生米", "quantity": "50g"},{"name": "干辣椒", "quantity": "10個"},{"name": "花椒", "quantity": "1茶匙"},{"name": "蔥", "quantity": "2根"},{"name": "姜", "quantity": "3片"},{"name": "蒜", "quantity": "3瓣"}],"steps": ["雞胸肉切丁，用料酒、生抽、淀粉腌制15分鐘","熱鍋涼油，放入花椒和干辣椒爆香","加入雞丁翻炒至變色","加入蔥姜蒜繼續翻炒","加入調好的宮保汁炒勻","最后加入花生米炒勻即可"],"difficulty": "中等"},{"id": "recipe_002","name": "番茄炒蛋","cuisine": "家常菜","description": "簡單易做的家常菜","ingredients": [{"name": "番茄", "quantity": "2個"},{"name": "雞蛋", "quantity": "3個"},{"name": "蔥", "quantity": "適量"},{"name": "鹽", "quantity": "適量"},{"name": "糖", "quantity": "少許"}],"steps": ["番茄切塊，雞蛋打散","熱鍋倒油，倒入雞蛋炒熟盛出","鍋中再倒少許油，放入番茄翻炒","番茄出汁后加入鹽和糖調味","倒入炒好的雞蛋翻炒均勻","撒上蔥花即可"],"difficulty": "簡單"}]# 插入食譜數據for recipe in recipes:cursor.execute('''INSERT INTO recipes (id, name, cuisine, description, ingredients, steps, difficulty)VALUES (?, ?, ?, ?, ?, ?, ?)''', (recipe["id"],recipe["name"],recipe["cuisine"],recipe["description"],json.dumps(recipe["ingredients"], ensure_ascii=False),json.dumps(recipe["steps"], ensure_ascii=False),recipe["difficulty"]))# 插入示例用戶偏好preferences = {"user_001": {"favorite_cuisines": ["川菜"],"dietary_restrictions": ["少油", "少鹽"],"cooking_skill": "初級"},"user_002": {"favorite_cuisines": ["家常菜"],"dietary_restrictions": ["健康"],"cooking_skill": "初級"}}for user_id, prefs in preferences.items():cursor.execute('''INSERT INTO user_preferences (user_id, favorite_cuisines, dietary_restrictions, cooking_skill)VALUES (?, ?, ?, ?)''', (user_id,json.dumps(prefs["favorite_cuisines"], ensure_ascii=False),json.dumps(prefs["dietary_restrictions"], ensure_ascii=False),prefs["cooking_skill"]))conn.commit()print("示例數據導入完成")# 食譜相關操作def get_all_recipes(self):"""獲取所有食譜"""conn = self.get_connection()cursor = conn.cursor()cursor.execute("SELECT * FROM recipes")rows = cursor.fetchall()recipes = []for row in rows:recipe = {"id": row["id"],"name": row["name"],"cuisine": row["cuisine"],"description": row["description"],"ingredients": json.loads(row["ingredients"]),"steps": json.loads(row["steps"]),"difficulty": row["difficulty"]}recipes.append(recipe)return recipesdef get_recipe_by_id(self, recipe_id: str):"""根據ID獲取食譜"""conn = self.get_connection()cursor = conn.cursor()cursor.execute("SELECT * FROM recipes WHERE id = ?", (recipe_id,))row = cursor.fetchone()if row:return {"id": row["id"],"name": row["name"],"cuisine": row["cuisine"],"description": row["description"],"ingredients": json.loads(row["ingredients"]),"steps": json.loads(row["steps"]),"difficulty": row["difficulty"]}return Nonedef search_recipes_by_ingredient(self, ingredient: str):"""根據食材搜索食譜"""conn = self.get_connection()cursor = conn.cursor()cursor.execute("SELECT * FROM recipes WHERE ingredients LIKE ?", (f'%{ingredient}%',))rows = cursor.fetchall()recipes = []for row in rows:recipe = {"id": row["id"],"name": row["name"],"cuisine": row["cuisine"],"description": row["description"],"ingredients": json.loads(row["ingredients"]),"steps": json.loads(row["steps"]),"difficulty": row["difficulty"]}recipes.append(recipe)return recipes# 用戶偏好相關操作def get_user_preferences(self, user_id: str):"""獲取用戶偏好"""conn = self.get_connection()cursor = conn.cursor()cursor.execute("SELECT * FROM user_preferences WHERE user_id = ?", (user_id,))row = cursor.fetchone()if row:return {"user_id": row["user_id"],"favorite_cuisines": json.loads(row["favorite_cuisines"]),"dietary_restrictions": json.loads(row["dietary_restrictions"]),"cooking_skill": row["cooking_skill"]}return None# 全局數據庫實例
db = DatabaseManager()def _get_all_recipes() -> Dict:"""獲取所有食譜數據"""try:recipes = db.get_all_recipes()return {"success": True,"recipes": recipes,"count": len(recipes)}except Exception as e:return {"success": False,"error": f"獲取食譜數據失敗: {str(e)}"}def _get_recipe_by_id(recipe_id: str) -> Dict:"""根據ID獲取特定食譜"""try:recipe = db.get_recipe_by_id(recipe_id)if recipe:return {"success": True,"recipe": recipe}else:return {"success": False,"error": f"未找到ID為{recipe_id}的食譜"}except Exception as e:return {"success": False,"error": f"獲取食譜失敗: {str(e)}"}def _get_user_preferences(user_id: str) -> Dict:"""獲取用戶偏好數據"""try:preferences = db.get_user_preferences(user_id)if preferences:return {"success": True,"preferences": preferences}else:return {"success": False,"error": f"未找到ID為{user_id}的用戶"}except Exception as e:return {"success": False,"error": f"獲取用戶偏好失敗: {str(e)}"}def _search_recipes_by_ingredient(ingredient: str) -> Dict:"""根據食材查詢食譜"""try:if not ingredient or not ingredient.strip():return {"success": False,"error": "請提供有效的食材名稱"}recipes = db.search_recipes_by_ingredient(ingredient.strip())if recipes:return {"success": True,"message": f"找到了{len(recipes)}個包含{ingredient}的食譜","recipes": recipes}else:return {"success": True,"message": f"抱歉，沒有找到包含{ingredient}的食譜","recipes": []}except Exception as e:return {"success": False,"error": f"查詢食譜時出錯: {str(e)}"}def _recommend_recipes(user_id: str, available_ingredients: List[str] = None) -> Dict:"""根據用戶偏好推薦食譜"""try:# 獲取用戶偏好user_prefs = db.get_user_preferences(user_id)if not user_prefs:return {"success": False,"error": f"未找到ID為{user_id}的用戶偏好"}# 獲取所有食譜all_recipes = db.get_all_recipes()recommended_recipes = []for recipe in all_recipes:# 根據用戶喜好的菜系過濾if recipe["cuisine"] in user_prefs["favorite_cuisines"]:# 如果提供了可用食材，檢查是否匹配if available_ingredients:recipe_ingredients = [ing["name"] for ing in recipe["ingredients"]]if any(avail_ing in recipe_ingredients for avail_ing in available_ingredients):recommended_recipes.append(recipe)else:recommended_recipes.append(recipe)return {"success": True,"message": f"為您推薦了{len(recommended_recipes)}道菜","recipes": recommended_recipes[:5] # 限制返回數量}except Exception as e:return {"success": False,"error": f"推薦食譜時出錯: {str(e)}"}
這個數據庫管理類把之前內存中的操作都移到了SQLite里。
主應用入口
現在創建 app/main.py，將所有MCP組件集中在一個文件中：
# app/main.py
from fastmcp import FastMCP
from typing import Dict, List
from .models import (_get_all_recipes, _get_recipe_by_id, _get_user_preferences,_search_recipes_by_ingredient,_recommend_recipes
)# 創建主應用實例
mcp = FastMCP("RecipeAssistant")# 資源組件
@mcp.resource("recipes://all")
def get_all_recipes() -> Dict:"""獲取所有食譜數據Returns:包含所有食譜的字典"""return _get_all_recipes()@mcp.resource("recipes://{recipe_id}")
def get_recipe_by_id(recipe_id: str) -> Dict:"""根據ID獲取特定食譜Args:recipe_id: 食譜IDReturns:食譜詳細信息"""return _get_recipe_by_id(recipe_id)@mcp.resource("users://{user_id}/preferences")
def get_user_preferences(user_id: str) -> Dict:"""獲取用戶偏好數據Args:user_id: 用戶IDReturns:用戶偏好數據"""return _get_user_preferences(user_id)# 工具組件
@mcp.tool()
def search_recipes_by_ingredient(ingredient: str) -> Dict:"""根據食材查詢食譜Args:ingredient: 食材名稱Returns:包含匹配食譜的字典"""return _search_recipes_by_ingredient(ingredient)@mcp.tool()
def recommend_recipes(user_id: str, available_ingredients: List[str] = None) -> Dict:"""根據用戶偏好和可用食材推薦食譜Args:user_id: 用戶IDavailable_ingredients: 可用食材列表（可選）Returns:包含推薦食譜的字典"""return _recommend_recipes(user_id, available_ingredients)# 提示詞組件
@mcp.prompt()
def generate_recipe_search_response(ingredient: str) -> str:"""生成食譜查詢的回復Args:ingredient: 食材名稱Returns:格式化的回復文本"""search_result = _search_recipes_by_ingredient(ingredient)if not search_result["success"]:return f"抱歉，查詢食譜時出現了問題：{search_result.get('error', '未知錯誤')}"recipes = search_result["recipes"]if not recipes:return f"抱歉，我沒有找到包含{ingredient}的食譜。請嘗試其他食材。"# 生成回復文本response = f"我找到了{len(recipes)}個包含{ingredient}的食譜：\n\n"for i, recipe in enumerate(recipes, 1):response += f"{i}. {recipe['name']} - {recipe['description']}\n"response += f" 難度：{recipe['difficulty']}\n"response += f" 主要食材：{', '.join(ing['name'] for ing in recipe['ingredients'][:3])}\n\n"response += f"想了解某個食譜的詳細做法，請告訴我食譜的編號。"return response@mcp.prompt()
def generate_recipe_details(recipe_id: str) -> str:"""生成食譜詳細信息的回復Args:recipe_id: 食譜IDReturns:格式化的食譜詳情"""recipe_result = _get_recipe_by_id(recipe_id)if not recipe_result["success"]:return f"抱歉，無法獲取食譜詳情：{recipe_result.get('error', '未知錯誤')}"recipe = recipe_result["recipe"]# 生成詳細食譜信息response = f"# {recipe['name']}\n\n"response += f"{recipe['description']}\n\n"response += "## 食材準備\n\n"for ing in recipe["ingredients"]:response += f"- {ing['name']}: {ing['quantity']}\n"response += "\n## 烹飪步驟\n\n"for i, step in enumerate(recipe["steps"], 1):response += f"{i}. {step}\n"response += f"\n難度：{recipe['difficulty']}\n"response += f"菜系：{recipe['cuisine']}\n"return responseif __name__ == "__main__":mcp.run()
測試驗證
現在來測試我們升級后的應用。
使用MCP Inspector測試
首先，啟動我們的應用：
cd recipe-assistant
python -m app.main
在另一個終端啟動MCP Inspector測試，可以按照我們之前的方式進行測試：
npx @modelcontextprotocol/inspector stdio python -m app.main
下一篇預告：我們將通過實際代碼對比，學習如何將一個FastMCP工具改寫成原生SDK版本，直觀理解兩者的差異和使用場景。
