内容
# 🍽️ MealPlanner MCP 多智能体
使用 [FastMCP](https://gofastmcp.com/)(模型上下文协议)构建的**多模态餐饮规划智能体**。它演示了如何使用 **图像分析**(通过谷歌 Gemini Vision)、**工具**、**资源**和**提示**构建 MCP 服务器。
## 🏗️ 架构
```
┌──────────────────────────────────────────────────────────┐
│ MCP 客户端(client.py) │
│ 连接到服务器,调用工具,读取资源 │
└─────────────────────────┬────────────────────────────────┘
│ MCP 协议(stdio)
┌─────────────────────────▼────────────────────────────────┐
│ MCP 服务器(server.py) │
│ │
│ 🔧 工具: │
│ • identify_ingredients(图像)→ Gemini Vision API │
│ • suggest_recipes(食材,饮食) │
│ • get_nutrition(菜谱名称) │
│ • get_meal_plan(食材,餐食,饮食) │
│ │
│ 📦 资源: │
│ • recipes://all → 完整菜谱数据库 │
│ • recipes://{id} → 单个菜谱详情 │
│ • dietary://profiles → 饮食限制 │
│ │
│ 💬 提示: │
│ • analyze_fridge → 图像分析工作流 │
│ • weekly_meal_prep → 餐食准备规划 │
│ │
│ 🧠 外部 API: │
│ • Google Gemini 2.0 Flash(Vision/Multimodal) │
└──────────────────────────────────────────────────────────┘
```
---
## 🔄 端到端工作流
### 高级序列
```
┌────────┐ ┌────────────┐ ┌────────────┐ ┌────────────┐
│ 用户 │ │ MCP 客户端 │ │ MCP 服务器 │ │ Gemini │
│ │ │ (client.py)│ │ (server.py)│ │ Vision │
└───┬────┘ └─────┬──────┘ └─────┬──────┘ └─────┬──────┘
│ │ │ │
│ 1. 运行客户端 │ │ │
│ 带有 --image │ │ │
│──────────────────>│ │ │
│ │ │ │
│ │ 2. 通过 stdio 传输 │
│ │ 连接 │ │
│ │───────────────────>│ │
│ │ │ │
│ │ 3. list_tools() │ │
│ │ list_resources()│ │
│ │ list_prompts() │ │
│ │───────────────────>│ │
│ │<──────────────────-│ │
│ │ (能力) │ │
│ │ │ │
│ │ 4. read_resource │ │
│ │ ("recipes://all") │ │
│ │───────────────────>│ │
│ │<──────────────────-│ │
│ │ (8 个菜谱) │ │
│ │ │ │
│ │ 5. call_tool │ │
│ │ identify_ │ │
│ │ ingredients(图像) │ │
│ │───────────────────>│ 6. 发送图像 │
│ │ │ + 提示 │
│ │ │────────────────────>│
│ │ │<────────────────────│
│ │ │ 7. 食材 JSON │
│ │<──────────────────-│ │
│ │ (食材列表) │ │
│ │ │ │
│ │ 8. call_tool │ │
│ │ suggest_recipes() │ │
│ │───────────────────>│ │
│ │<──────────────────-│ │
│ │ (匹配的菜谱) │ │
│ │ │ │
│ │ 9. call_tool │ │
│ │ get_nutrition() │ │
│ │───────────────────>│ │
│ │<──────────────────-│ │
│ │ (营养信息) │ │
│ │ │ │
│ │ 10. call_tool │ │
│ │ get_meal_plan() │ │
│ │───────────────────>│ │
│ │<──────────────────-│ │
│ │ (日常计划) │ │
│ │ │ │
│ 11. 打印结果│ │ │
│<──────────────────│ │ │
│ (格式化输出) │ │
```
### 逐步数据流程
---
#### 步骤 1:客户端连接到服务器
```python
# client.py — 客户端通过 stdio 将服务器作为子进程启动
client = Client("server.py")
async with client:
# 使用 stdio 传输的 MCP 协议建立连接
# FastMCP 自动处理:握手、能力协商、生命周期
```
**内部发生的事情:**
- `Client("server.py")` 告诉 FastMCP 以 **子进程** 的方式启动 `server.py`
- 通过 **stdin/stdout** (stdio 传输)进行通信
- 发生 MCP 协议握手
- 客户端发现服务器名称:`"MealPlanner"`
---
#### 步骤 2:发现服务器能力
```python
# client.py — 向服务器询问其能力
tools = await client.list_tools() # → 4 个工具
resources = await client.list_resources() # → 2 个静态资源
templates = await client.list_resource_templates() # → 1 个模板
prompts = await client.list_prompts() # → 2 个提示
```
**服务器响应:**
| 类型 | 名称 | 描述 |
|------|------|-------------|
| 🔧 工具 | `identify_ingredients` | 分析冰箱照片 → 食材(多模态) |
| 🔧 工具 | `suggest_recipes` | 匹配食材 → 菜谱 |
| 🔧 工具 | `get_nutrition` | 菜谱 → 营养信息 |
| 🔧 工具 | `get_meal_plan` | 食材 → 每日餐饮计划 |
| 📦 资源 | `recipes://all` | 完整菜谱数据库(8 个菜谱) |
| 📦 资源 | `dietary://profiles` | 4 个饮食配置文件 |
| 📄 模板 | `recipes://{recipe_id}` | 按 ID 查询单个菜谱 |
| 💬 提示 | `analyze_fridge` | 引导式冰箱分析工作流 |
| 💬 提示 | `weekly_meal_prep` | 每周计划模板 |
---
#### 步骤 3:读取资源(菜谱数据库和饮食配置文件)
```python
# client.py — 读取菜谱数据库
recipes = await client.read_resource("recipes://all")
```
**服务器处理(server.py):**
```python
@mcp.resource("recipes://all")
def get_all_recipes() -> str:
# 遍历 RECIPE_DATABASE 字典 → 返回 8 个菜谱的 JSON 摘要
# 每个菜谱包含:id、名称、美食、准备时间、难度、食材
```
**返回的数据:**
```json
[
{"id": "pasta_primavera", "name": "意大利春蔬意大利面", "cuisine": "意大利", ...},
{"id": "chicken_stir_fry", "name": "鸡肉炒菜", "cuisine": "亚洲", ...},
{"id": "vegetable_omelette", "name": "蔬菜煎蛋", ...},
// ... 共 8 个菜谱
]
```
---
#### 步骤 4:多模态 — 从冰箱照片识别食材 📸
这是 **核心多模态步骤**,其中分析图像。
```python
# client.py — 发送冰箱图像进行分析
result = await client.call_tool(
"identify_ingredients",
{"image_path": "/path/to/fridge_photo.jpg"}
)
```
**服务器处理(server.py)— 4 个子步骤:
```
┌─────────────────────────────────────────────────────────────────┐
│ identify_ingredients(图像路径) │
│ │
│ 步骤 A:从磁盘读取图像文件 │
│ image_bytes = Path(image_path).read_bytes() │
│ │
│ 步骤 B:编码为 base64 │
│ image_base64 = base64.b64encode(image_bytes) │
│ 检测 MIME 类型:.jpg → "image/jpeg" │
│ │
│ 步骤 C:发送到 Google Gemini Vision API │
│ gemini_client.models.generate_content( │
│ model="gemini-2.0-flash", │
│ contents=[text_prompt + inline_image_data] │
│ ) │
│ │
│ 步骤 D:解析并返回结构化 JSON │
│ 从响应中清理 Markdown 代码块 │
│ 以 JSON 字符串返回食材列表 │
└─────────────────────────────────────────────────────────────────┘
```
**发送到 Gemini 的提示:**
```
分析这张冰箱或储藏室的图像。识别所有可见的食品食材。
返回一个 JSON 对象,其结构如下:
{
"ingredients": [
{"name": "...", "category": "produce|dairy|protein|...", "quantity": "..."}
],
"total_count": <number>,
"freshness_notes": "..."
}
```
**Gemini 的示例响应:**
```json
{
"ingredients": [
{"name": "eggs", "category": "protein", "quantity": "6 eggs"},
{"name": "milk", "category": "dairy", "quantity": "1 carton"},
{"name": "tomato", "category": "produce", "quantity": "3 pieces"},
{"name": "cheese", "category": "dairy", "quantity": "1 block"},
{"name": "butter", "category": "dairy", "quantity": "1 stick"}
],
"total_count": 5,
"freshness_notes": "所有物品看起来都很新鲜且存储良好"
}
```
---
#### 步骤 5:根据食材建议菜谱
```python
# client.py — 将识别的食材传递给菜谱匹配器
result = await client.call_tool(
"suggest_recipes",
{"ingredients": ["eggs", "milk", "tomato", "cheese", "butter"],
"dietary_preference": "none"}
)
```
**服务器处理(server.py):**
```
┌─────────────────────────────────────────────────────────────────┐
│ suggest_recipes(食材,饮食偏好) │
│ │
│ 1. 规范化食材 → 小写 │
│ │
│ 2. 加载饮食配置文件 │
│ "none" → 没有排除的食材 │
│ "vegan" → 排除鸡蛋、奶酪、牛奶、黄油等 │
│ │
│ 3. 对于 RECIPE_DATABASE 中的每个菜谱(8 个菜谱): │
│ a. 跳过如果菜谱有排除的食材 │
│ b. 计算可用食材的数量 │
│ c. 计算匹配百分比 = 可用 / 总数 × 100 │
│ d. 如果匹配 ≥ 40%,则包含 │
│ │
│ 4. 按匹配百分比排序(最高优先) │
│ │
│ 5. 返回:菜谱名称、匹配百分比、可用和缺失的食材 │
└─────────────────────────────────────────────────────────────────┘
```
**输出示例:**
```json
{
"dietary_preference": "无限制",
"total_matches": 4,
"recipes": [
{
"name": "蔬菜煎蛋",
"match_percentage": 66.7,
"available_ingredients": ["eggs", "cheese", "butter"],
"missing_ingredients": ["bell pepper", "onion", "mushroom"]
},
{
"name": "烤奶酪三明治",
"match_percentage": 66.7,
"available_ingredients": ["cheese", "butter"],
"missing_ingredients": ["bread"]
}
]
}
```
---
#### 步骤 6:获取顶部菜谱的营养信息
```python
# client.py — 获取最佳匹配的营养信息
result = await client.call_tool(
"get_nutrition",
{"recipe_name": "vegetable_omelette"}
)
```
**服务器处理(server.py):**
```
┌─────────────────────────────────────────────────────────────────┐
│ get_nutrition(菜谱名称) │
│ │
│ 1. 按 ID 或名称(不区分大小写)在 RECIPE_DATABASE 中搜索 │
│ 2. 如果未找到 → 模糊匹配(部分字符串匹配) │
│ 3. 返回完整的菜谱详情: │
│ - 食材列表 │
│ - 分步烹饪说明 │
│ - 营养:卡路里、蛋白质、碳水化合物、脂肪、纤维 │
│ - 根据数值自动生成的健康提示 │
└─────────────────────────────────────────────────────────────────┘
```
**输出示例:**
```json
{
"recipe": "蔬菜煎蛋",
"nutrition": {
"calories": 320,
"protein": "22g",
"carbs": "8g",
"fat": "24g",
"fiber": "2g"
},
"instructions": [
"将鸡蛋与少量牛奶搅拌。",
"在中火上融化黄油至非粘锅中。",
"炒碎蔬菜 2-3 分钟。",
"倒入鸡蛋混合物并煮至边缘凝固。",
"添加奶酪,折叠并上菜。"
],
"health_tips": [
"💪 良好的蛋白质含量 — 有利于肌肉恢复!",
"💧 记得在一天中保持水分充足!"
]
}
```
---
#### 步骤 7:生成每日餐饮计划
```python
# client.py — 创建全天餐饮计划
result = await client.call_tool(
"get_meal_plan",
{"ingredients": ingredient_names, "meals_per_day": 3, "dietary_preference": "none"}
)
```
**服务器处理(server.py):**
```
┌─────────────────────────────────────────────────────────────────┐
│ get_meal_plan(食材、每日餐数、饮食) │
│ │
│ 1. 内部调用 suggest_recipes() 获取匹配的菜谱 │
│ 2. 将菜谱分配到餐位: │
│ - 餐 1 → "早餐"(最佳匹配菜谱) │
│ - 餐 2 → "午餐"(第二最佳匹配) │
│ - 餐 3 → "晚餐"(第三最佳匹配) │
│ 3. 计算估计的总热量 │
│ 4. 返回结构化的餐饮计划 │
└─────────────────────────────────────────────────────────────────┘
```
**输出示例:**
```json
{
"meal_plan": [
{"meal": "早餐", "recipe": "蔬菜煎蛋", "match": "66.7%"},
{"meal": "午餐", "recipe": "烤奶酪三明治", "match": "66.7%"},
{"meal": "晚餐", "recipe": "经典番茄汤", "match": "50.0%"}
],
"estimated_total_calories": 920,
"note": "这是根据可用食材建议的调整。按需调整份量!"
}
```
#### 步骤 8:客户端打印格式化结果
客户端将所有 JSON 响应格式化为 **用户友好的终端输出**,使用表情符号、颜色和清晰的章节。
---
### 完整数据流程图
```
📸 冰箱照片
│
▼
┌──────────────┐ base64 + 提示 ┌──────────────┐
│ MCP 服务器 │ ─────────────────────── │ Google Gemini │
│ 工具: │ │ Vision API │
│ identify_ │ <─── JSON 食材 ── │ (2.0 Flash) │
│ ingredients │ └──────────────┘
└──────┬───────┘
│ ["eggs", "tomato", "cheese", ...]
▼
┌──────────────┐
│ MCP 服务器 │ 对比 RECIPE_DATABASE(8 个菜谱)
│ 工具: │ ──▶ RECIPE_DATABASE(8 个菜谱)
│ suggest_ │ ──▶ DIETARY_PROFILES(4 个配置文件)
│ recipes │
└──────┬───────┘
│ [{name: "Omelette", match: 83%}, ...]
▼
┌──────────────┐
│ MCP 服务器 │ 查询菜谱详情
│ 工具: │ ──▶ 完整说明
│ get_ │ ──▶ 营养数据
│ nutrition │ ──▶ 健康提示(自动生成)
└──────┬───────┘
│ {calories: 320, protein: "22g", ...}
▼
┌──────────────┐
│ MCP 服务器 │ 结合顶级菜谱生成
│ 工具: │ ──▶ 早餐 / 午餐 / 晚餐
│ get_ │ ──▶ 计算总热量
│ meal_plan │
└──────┬───────┘
│
▼
📋 完整餐饮计划
营养信息
```
### Demo 模式 vs 图片模式
| 方面 | Demo 模式 (`python3 client.py`) | 图片模式 (`python3 client.py --image photo.jpg`) |
|--------|------|------|
| 图片分析 | ❌ 跳过 | ✅ 调用 Gemini Vision API |
| 原料 | 硬编码的示例列表(10 个项目) | 由 AI 从图片中提取 |
| 需要 API 密钥 | ❌ 不需要(服务器加载但工具未调用) | ✅ 需要(`.env` 中的 GOOGLE_API_KEY) |
| 食谱匹配 | ✅ 工作正常 | ✅ 工作正常 |
| 营养信息 | ✅ 工作正常 | ✅ 工作正常 |
| 餐计划 | ✅ 工作正常 | ✅ 工作正常 |
---
## 📁 项目结构
```
mealplanner_mcp_multi_agent/
├── server.py # MCP 服务器 — 工具、资源、提示
├── client.py # MCP 客户端 — 示例脚本(通过 HTTP 连接)
├── pyproject.toml # 项目配置和依赖项(uv)
├── uv.lock # 锁定依赖项版本
├── .env.example # 环境变量模板
├── .python-version # Python 版本锁定
└── README.md # 本文件
```
## 🚀 设置
### 1. 安装 uv(如果尚未安装)
```bash
curl -LsSf https://astral.sh/uv/install.sh | sh
```
### 2. 安装依赖项
```bash
uv sync
```
这会创建一个 `.venv` 虚拟环境,并从 `pyproject.toml` 安装所有依赖项。
### 3. 获取 Google Gemini API 密钥
- 访问 [Google AI Studio](https://aistudio.google.com/apikey)
- 创建一个免费 API 密钥
- 复制 `.env.example` 到 `.env` 并粘贴您的密钥:
```bash
cp .env.example .env
# 编辑 .env 并添加您的 GOOGLE_API_KEY
```
### 4. 启动 MCP 服务器(终端 1)
```bash
uv run python3 server.py
```
服务器在 `http://127.0.0.1:8000/mcp` 启动,使用 **Streamable HTTP** 传输。
### 5. 运行客户端(终端 2)
```bash
# Demo 模式(使用示例原料)
uv run python3 client.py
# 使用真实的冰箱照片(多模态!)
uv run python3 client.py --image path/to/fridge_photo.jpg
# 连接到自定义服务器 URL
uv run python3 client.py --url http://your-server:8000/mcp
```
### 6. 使用 MCP Inspector 测试
```bash
uv run fastmcp dev server.py
```
这会打开一个网页 UI,您可以在其中交互式地测试所有工具、资源和提示。
## 🔧 MCP 概念演示
| 概念 | 功能 | 示例 |
|---------|-------------|------------------------|
| **工具** | LLM 可以调用的函数 | `identify_ingredients()` — 发送图片到 Gemini |
| **资源** | LLM 的只读数据 | `recipes://all` — 浏览食谱数据库 |
| **资源模板** | 带有参数的动态资源 | `recipes://{recipe_id}` — 获取特定食谱 |
| **提示** | 可重用的提示模板 | `analyze_fridge` — 引导式冰箱分析工作流 |
## 🖼️ 多模态功能
`identify_ingredients` 工具是 **多模态** 组件:
1. 接受 **图片文件路径**(JPG、PNG、WebP)
2. 将图片编码为 **base64**
3. 将其发送到 **Google Gemini 2.0 Flash**,并附带结构化提示
4. 返回 **JSON 列表**,包含已识别的原料及其类别
这演示了 MCP 工具如何处理非文本输入(图片)以及文本。
## 📝 示例输出
```
============================================================
🍽️ 餐计划 MCP 客户端
============================================================
📋 发现服务器功能...
🔧 可用工具:
• identify_ingredients: 分析冰箱或储藏室的照片...
• suggest_recipes: 根据可用原料建议食谱...
• get_nutrition: 获取详细的营养信息...
• get_meal_plan: 生成简单的日常餐计划...
📦 可用资源:
• recipes://all: 食谱数据库
• dietary://profiles: 膳食配置文件
👨🍳 建议食谱:
🟢 蔬菜煎蛋(83.3% 匹配)
🟢 卡普里沙拉(80.0% 匹配)
🟡 经典番茄汤(66.7% 匹配)
🟡 意大利青酱意大利面(57.1% 匹配)
🟡 烤奶酪三明治(66.7% 匹配)
```
## 🔑 关键学习点
1. **FastMCP 简化了 MCP 服务器创建** — 只需使用 `@mcp.tool()`、`@mcp.resource()`、`@mcp.prompt()` 装饰器
2. **多模态 = 图片 + 文本** — 服务器处理图片编码并发送到视觉 API
3. **工具 vs 资源** — 工具执行操作(带有副作用);资源提供只读数据
4. **提示** — 可重用的模板,引导 LLM 完成多步骤工作流
5. **客户端通过 stdio 连接** — FastMCP 自动处理所有 MCP 协议细节
MCP 配置
以下是此 MCP Server 的配置信息,可直接复制到 Cursor 或其他 MCP 客户端中使用。
mcp.json
连接信息
猜你喜欢
暂无推荐服务器