added all mcp

This commit is contained in:
liph22
2025-12-19 23:59:54 +01:00
parent edfffd76e1
commit 14bc398915
6240 changed files with 2421602 additions and 0 deletions

View File

@@ -0,0 +1,5 @@
#!/bin/bash
# Get the directory of this script (which will be inside the bundle)
SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
cd "$SCRIPT_DIR"
exec ./venv/bin/python "$SCRIPT_DIR/server.py"

View File

@@ -0,0 +1,16 @@
from .client import MealieClient
from .food import FoodMixin
from .group import GroupMixin
from .mealplan import MealplanMixin
from .recipe import RecipeMixin
from .user import UserMixin
class MealieFetcher(
RecipeMixin,
UserMixin,
GroupMixin,
MealplanMixin,
FoodMixin,
MealieClient,
):
pass

View File

@@ -0,0 +1,15 @@
from .client import MealieClient
from .group import GroupMixin
from .mealplan import MealplanMixin
from .recipe import RecipeMixin
from .user import UserMixin
class MealieFetcher(
RecipeMixin,
UserMixin,
GroupMixin,
MealplanMixin,
MealieClient,
):
pass

View File

@@ -0,0 +1,146 @@
import json
import logging
import traceback
from typing import Any, Dict
import httpx
from httpx import ConnectError, HTTPStatusError, ReadTimeout
logger = logging.getLogger("mealie-mcp")
class MealieApiError(Exception):
"""Custom exception for Mealie API errors with status code and response details."""
def __init__(self, status_code: int, message: str, response_text: str = None):
self.status_code = status_code
self.message = message
self.response_text = response_text
super().__init__(f"{message} (Status Code: {status_code})")
class MealieClient:
def __init__(self, base_url: str, api_key: str):
if not base_url:
raise ValueError("Base URL cannot be empty")
if not api_key:
raise ValueError("API key cannot be empty")
logger.debug({"message": "Initializing MealieClient", "base_url": base_url})
try:
self._client = httpx.Client(
base_url=base_url,
headers={
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json",
},
timeout=30.0, # Set a reasonable timeout for requests
)
# Test connection
logger.debug({"message": "Testing connection to Mealie API"})
self._client.get("/api/app/about")
logger.info({"message": "Successfully connected to Mealie API"})
except ConnectError as e:
error_msg = f"Failed to connect to Mealie API at {base_url}: {str(e)}"
logger.error({"message": error_msg})
logger.debug(
{"message": "Error traceback", "traceback": traceback.format_exc()}
)
raise ConnectionError(error_msg) from e
except Exception as e:
error_msg = f"Error initializing Mealie client: {str(e)}"
logger.error({"message": error_msg})
logger.debug(
{"message": "Error traceback", "traceback": traceback.format_exc()}
)
raise
def _handle_request(self, method: str, url: str, **kwargs) -> Dict[str, Any] | str:
"""Common request handler with error handling for all API calls."""
try:
logger.debug(
{
"message": "Making API request",
"method": method,
"url": url,
"body": kwargs.get("json"),
}
)
if "params" in kwargs:
logger.debug(
{"message": "Request parameters", "params": kwargs["params"]}
)
if "json" in kwargs:
logger.debug({"message": "Request payload", "payload": kwargs["json"]})
response = self._client.request(method, url, **kwargs)
response.raise_for_status() # Raise an exception for 4XX/5XX responses
logger.debug(
{"message": "Request successful", "status_code": response.status_code}
)
# Log the response content at debug level
try:
response_data = response.json()
logger.debug({"message": "Response content", "data": response_data})
return response_data
except json.JSONDecodeError:
logger.debug(
{"message": "Response content (non-JSON)", "content": response.text}
)
return response.text
except HTTPStatusError as e:
status_code = e.response.status_code
error_detail = f"HTTP Error {status_code}"
# Try to parse error details from response
try:
error_detail = e.response.json()
except Exception:
error_detail = e.response.text
error_msg = f"API error for {method} {url}: {error_detail}"
logger.error(
{
"message": "API request failed",
"method": method,
"url": url,
"status_code": status_code,
"error_detail": error_detail,
}
)
logger.debug(
{"message": "Failed Request body", "content": e.request.content}
)
raise MealieApiError(status_code, error_msg, e.response.text) from e
except ReadTimeout:
error_msg = f"Request timeout for {method} {url}"
logger.error({"message": error_msg, "method": method, "url": url})
logger.debug(
{"message": "Error traceback", "traceback": traceback.format_exc()}
)
raise TimeoutError(error_msg)
except ConnectError as e:
error_msg = f"Connection error for {method} {url}: {str(e)}"
logger.error(
{"message": error_msg, "method": method, "url": url, "error": str(e)}
)
logger.debug(
{"message": "Error traceback", "traceback": traceback.format_exc()}
)
raise ConnectionError(error_msg) from e
except Exception as e:
error_msg = f"Unexpected error for {method} {url}: {str(e)}"
logger.error(
{"message": error_msg, "method": method, "url": url, "error": str(e)}
)
logger.debug(
{"message": "Error traceback", "traceback": traceback.format_exc()}
)
raise

View File

@@ -0,0 +1,49 @@
import logging
from typing import Any, Dict, Optional
logger = logging.getLogger("mealie-mcp")
class FoodMixin:
"""Mixin for Mealie food and unit API operations."""
def get_foods(
self,
search: Optional[str] = None,
page: Optional[int] = None,
per_page: Optional[int] = None,
) -> Dict[str, Any]:
"""Get list of foods."""
params = {}
if search:
params["search"] = search
if page:
params["page"] = page
if per_page:
params["perPage"] = per_page
logger.info({"message": "Retrieving foods", "parameters": params})
return self._handle_request("GET", "/api/foods", params=params)
def get_units(self) -> Dict[str, Any]:
"""Get list of all units."""
logger.info({"message": "Retrieving units"})
return self._handle_request("GET", "/api/units")
def create_food(self, name: str, description: Optional[str] = None) -> Dict[str, Any]:
"""Create a new food item."""
data = {"name": name}
if description:
data["description"] = description
logger.info({"message": "Creating food", "name": name})
return self._handle_request("POST", "/api/foods", json=data)
def create_unit(self, name: str, abbreviation: Optional[str] = None) -> Dict[str, Any]:
"""Create a new unit."""
data = {"name": name}
if abbreviation:
data["abbreviation"] = abbreviation
logger.info({"message": "Creating unit", "name": name})
return self._handle_request("POST", "/api/units", json=data)

View File

@@ -0,0 +1,17 @@
import logging
from typing import Any, Dict
logger = logging.getLogger("mealie-mcp")
class GroupMixin:
"""Mixin class for group-related API endpoints"""
def get_current_group(self) -> Dict[str, Any]:
"""Get information about the current user's group.
Returns:
Dictionary containing group details such as id, name, slug, and other group information.
"""
logger.info({"message": "Retrieving current group information"})
return self._handle_request("GET", "/api/groups/self")

View File

@@ -0,0 +1,105 @@
import logging
from typing import Any, Dict, List, Optional
from utils import format_api_params
logger = logging.getLogger("mealie-mcp")
class MealplanMixin:
"""Mixin class for mealplan-related API endpoints"""
def get_mealplans(
self,
start_date: Optional[str] = None,
end_date: Optional[str] = None,
page: Optional[int] = None,
per_page: Optional[int] = None,
) -> Dict[str, Any]:
"""Get all mealplans for the current household with pagination.
Args:
start_date: Start date for filtering meal plans (ISO format YYYY-MM-DD)
end_date: End date for filtering meal plans (ISO format YYYY-MM-DD)
page: Page number to retrieve
per_page: Number of items per page
Returns:
JSON response containing mealplan items and pagination information
Raises:
MealieApiError: If the API request fails
"""
param_dict = {
"startDate": start_date,
"endDate": end_date,
"page": page,
"perPage": per_page,
}
params = format_api_params(param_dict)
logger.info({"message": "Retrieving mealplans", "parameters": params})
response = self._handle_request(
"GET", "/api/households/mealplans", params=params
)
return response
def create_mealplan(
self,
date: str,
recipe_id: Optional[str] = None,
title: Optional[str] = None,
entry_type: str = "breakfast",
) -> Dict[str, Any]:
"""Create a new mealplan entry.
Args:
date: Date for the mealplan in ISO format (YYYY-MM-DD)
recipe_id: UUID of the recipe to add to the mealplan (optional)
title: Title for the mealplan entry if not using a recipe (optional)
entry_type: Type of mealplan entry (breakfast, lunch, dinner, etc.)
Returns:
JSON response containing the created mealplan entry
Raises:
ValueError: If neither recipe_id nor title is provided
MealieApiError: If the API request fails
"""
if not recipe_id and not title:
raise ValueError("Either recipe_id or title must be provided")
if not date:
raise ValueError("Date cannot be empty")
# Build the request payload
payload = {
"date": date,
"entryType": entry_type,
}
if recipe_id:
payload["recipeId"] = recipe_id
if title:
payload["title"] = title
logger.info(
{
"message": "Creating mealplan entry",
"date": date,
"entry_type": entry_type,
}
)
return self._handle_request("POST", "/api/households/mealplans", json=payload)
def get_todays_mealplan(self) -> List[Dict[str, Any]]:
"""Get the mealplan entries for today.
Returns:
List of today's mealplan entries
Raises:
MealieApiError: If the API request fails
"""
logger.info({"message": "Retrieving today's mealplan"})
return self._handle_request("GET", "/api/households/mealplans/today")

View File

@@ -0,0 +1,107 @@
import logging
from typing import Any, Dict, List, Optional
from utils import format_api_params
logger = logging.getLogger("mealie-mcp")
class RecipeMixin:
"""Mixin class for recipe-related API endpoints"""
def get_recipes(
self,
search: Optional[str] = None,
order_by: Optional[str] = None,
order_by_null_position: Optional[str] = None,
order_direction: Optional[str] = "desc",
query_filter: Optional[str] = None,
pagination_seed: Optional[str] = None,
page: Optional[int] = None,
per_page: Optional[int] = None,
categories: Optional[List[str]] = None,
tags: Optional[List[str]] = None,
tools: Optional[List[str]] = None,
) -> Dict[str, Any]:
"""Provides paginated list of recipes
Args:
search: Search term to filter recipes by name, description, etc.
order_by: Field to order results by
order_by_null_position: How to handle nulls in ordering ('first' or 'last')
order_direction: Direction to order results ('asc' or 'desc')
query_filter: Advanced query filter
pagination_seed: Seed for consistent pagination
page: Page number to retrieve
per_page: Number of items per page
categories: List of category slugs to filter by
tags: List of tag slugs to filter by
tools: List of tool slugs to filter by
Returns:
JSON response containing recipe items and pagination information
"""
param_dict = {
"search": search,
"orderBy": order_by,
"orderByNullPosition": order_by_null_position,
"orderDirection": order_direction,
"queryFilter": query_filter,
"paginationSeed": pagination_seed,
"page": page,
"perPage": per_page,
"categories": categories,
"tags": tags,
"tools": tools,
}
params = format_api_params(param_dict)
logger.info({"message": "Retrieving recipes", "parameters": params})
return self._handle_request("GET", "/api/recipes", params=params)
def get_recipe(self, slug: str) -> Dict[str, Any]:
"""Retrieve a specific recipe by its slug
Args:
slug: The slug identifier of the recipe to retrieve
Returns:
JSON response containing all recipe details
"""
if not slug:
raise ValueError("Recipe slug cannot be empty")
logger.info({"message": "Retrieving recipe", "slug": slug})
return self._handle_request("GET", f"/api/recipes/{slug}")
def update_recipe(self, slug: str, recipe_data: Dict[str, Any]) -> Dict[str, Any]:
"""Update a specific recipe by its slug
Args:
slug: The slug identifier of the recipe to update
recipe_data: Dictionary containing the recipe properties to update
Returns:
JSON response containing the updated recipe details
"""
if not slug:
raise ValueError("Recipe slug cannot be empty")
if not recipe_data:
raise ValueError("Recipe data cannot be empty")
logger.info({"message": "Updating recipe", "slug": slug})
return self._handle_request("PUT", f"/api/recipes/{slug}", json=recipe_data)
def create_recipe(self, name: str) -> str:
"""Create a new recipe
Args:
name: The name of the new recipe
Returns:
Slug of the newly created recipe
"""
logger.info({"message": "Creating new recipe", "name": name})
return self._handle_request("POST", "/api/recipes", json={"name": name})

View File

@@ -0,0 +1,17 @@
import logging
from typing import Any, Dict
logger = logging.getLogger("mealie-mcp")
class UserMixin:
"""Mixin class for user-related API endpoints"""
def get_current_user(self) -> Dict[str, Any]:
"""Get information about the currently logged in user.
Returns:
Dictionary containing user details such as id, username, email, and other profile information.
"""
logger.info({"message": "Retrieving current user information"})
return self._handle_request("GET", "/api/users/self")

View File

@@ -0,0 +1,6 @@
2025-10-03 19:25:30,936 - httpx - INFO - HTTP Request: GET https://mealie.liphlink.xyz/api/app/about "HTTP/1.1 200 OK"
2025-10-03 19:25:30,937 - mealie-mcp - INFO - {'message': 'Successfully connected to Mealie API'}
2025-10-03 19:25:30,968 - mealie-mcp - INFO - {'message': 'Starting Mealie MCP Server'}
2025-10-03 19:44:01,040 - httpx - INFO - HTTP Request: GET https://mealie.liphlink.xyz/api/app/about "HTTP/1.1 200 OK"
2025-10-03 19:44:01,041 - mealie-mcp - INFO - {'message': 'Successfully connected to Mealie API'}
2025-10-03 19:44:01,071 - mealie-mcp - INFO - {'message': 'Starting Mealie MCP Server'}

View File

@@ -0,0 +1,10 @@
from typing import Optional
from pydantic import BaseModel
class MealPlanEntry(BaseModel):
date: str
recipe_id: Optional[str] = None
title: Optional[str] = None
entry_type: str = "breakfast"

View File

@@ -0,0 +1,84 @@
from typing import Any, Dict, List, Optional
from pydantic import BaseModel, Field
class RecipeIngredient(BaseModel):
quantity: Optional[float] = None
unit: Optional[str] = None
food: Optional[str] = None
note: str
isFood: Optional[bool] = True
disableAmount: Optional[bool] = False
display: Optional[str] = None
title: Optional[str] = None
originalText: Optional[str] = None
referenceId: Optional[str] = None
class RecipeInstruction(BaseModel):
id: Optional[str] = None
title: Optional[str] = None
summary: Optional[str] = None
text: str
ingredientReferences: List[str] = Field(default_factory=list)
class RecipeNutrition(BaseModel):
calories: Optional[str] = None
carbohydrateContent: Optional[str] = None
cholesterolContent: Optional[str] = None
fatContent: Optional[str] = None
fiberContent: Optional[str] = None
proteinContent: Optional[str] = None
saturatedFatContent: Optional[str] = None
sodiumContent: Optional[str] = None
sugarContent: Optional[str] = None
transFatContent: Optional[str] = None
unsaturatedFatContent: Optional[str] = None
class RecipeSettings(BaseModel):
public: bool = False
showNutrition: bool = False
showAssets: bool = False
landscapeView: bool = False
disableComments: bool = False
disableAmount: bool = False
locked: bool = False
class Recipe(BaseModel):
id: str
userId: str
householdId: str
groupId: str
name: str
slug: str
image: Optional[str] = None
recipeServings: Optional[int] = None
recipeYieldQuantity: Optional[int] = 0
recipeYield: Optional[str] = None
totalTime: Optional[int] = None
prepTime: Optional[int] = None
cookTime: Optional[int] = None
performTime: Optional[int] = None
description: Optional[str] = None
recipeCategory: List[str] = Field(default_factory=list)
tags: List[str] = Field(default_factory=list)
tools: List[str] = Field(default_factory=list)
rating: Optional[float] = None
orgURL: Optional[str] = None
dateAdded: str
dateUpdated: str
createdAt: str
updatedAt: str
lastMade: Optional[str] = None
recipeIngredient: List[RecipeIngredient] = Field(default_factory=list)
recipeInstructions: List[RecipeInstruction] = Field(default_factory=list)
nutrition: RecipeNutrition = Field(default_factory=RecipeNutrition)
settings: RecipeSettings = Field(default_factory=RecipeSettings)
assets: List[Any] = Field(default_factory=list)
notes: List[Any] = Field(default_factory=list)
extras: Dict[str, Any] = Field(default_factory=dict)
comments: List[Any] = Field(default_factory=list)

View File

@@ -0,0 +1,60 @@
from mcp.server.fastmcp import FastMCP
from mcp.server.fastmcp.prompts.base import AssistantMessage, Message, UserMessage
def register_prompts(mcp: FastMCP) -> None:
"""Register all prompt-related tools with the MCP server."""
@mcp.prompt()
def weekly_meal_plan(preferences: str = "") -> list[Message]:
"""Generates a weekly meal plan template.
Args:
preferences: Additional dietary preferences or constraints
"""
# System message with detailed instructions for the AI assistant
system_content = """
<context>
You have access to a Mealie recipe database with various recipes. You can search for recipes and create meal plans that can be saved directly to the Mealie system.
## Tool Usage Guidelines
### Recipe Tools
- get_recipes: Search and list recipes (always set per_page=50. Use null if empty values)
- get_recipe_concise: Get basic recipe details (use by default)
- get_recipe_detailed: Get full recipe information (do not use unless user asks for it)
### Meal Plan Tools
- get_all_mealplans: View existing meal plans
- create_mealplan_bulk: Add multiple recipes to a meal plan at once (requires a list of entries with date in YYYY-MM-DD format, recipe_id if available, and entry_type)
- get_todays_mealplan: View today's planned meals
</context>
<instructions>
# Meal Planning Guidelines
- Include breakfast, lunch, and dinner for all 7 days
- Create a variety of meals using different proteins, grains, and vegetables
- Consider seasonal ingredients and balance nutrition throughout the week
- Use recipes from the Mealie database when available
- Plan for leftovers where appropriate and suggest how they can be repurposed
# User Interaction
- Present the meal plan in table format
- Ask for feedback about meal swaps, leftover utilization, and dietary needs
- When suggesting recipes, include the recipe ID or slug
- Before saving to Mealie, display the complete meal plan in concise summary for final user confirmation
- After confirmation, save the plan to Mealie without showing an additional summary
</instructions>
"""
# User message that initiates the conversation
user_content = "I need help creating a balanced meal plan for the next week that includes breakfast, lunch, and dinner."
if preferences:
user_content += f" My preferences are: {preferences}"
# Create and return a list of Message objects
return [
AssistantMessage(system_content),
UserMessage(user_content),
]

View File

@@ -0,0 +1,29 @@
[project]
name = "mealie-mcp-server"
version = "0.1.0"
description = "Add your description here"
readme = "README.md"
requires-python = ">=3.12"
dependencies = [
"httpx>=0.28.1",
"mcp[cli]>=1.12.0",
"pydantic>=2.11.3",
"python-dotenv>=1.1.0",
]
[tool.black]
line-length = 88
target-version = ["py312"]
include = '\.pyi?$'
[tool.isort]
profile = "black"
line_length = 88
multi_line_output = 3
include_trailing_comma = true
force_grid_wrap = 0
use_parentheses = true
ensure_newline_before_comments = true
[project.optional-dependencies]
dev = ["black>=24.1.0", "isort>=5.13.0"]

16
mealie-mcp-bundle/server/run.py Executable file
View File

@@ -0,0 +1,16 @@
#!/usr/bin/env python3
import os
import sys
import subprocess
# Get the directory where this script is located
script_dir = os.path.dirname(os.path.abspath(__file__))
# Path to the venv python
venv_python = os.path.join(script_dir, 'venv', 'bin', 'python')
# Path to the actual server
server_py = os.path.join(script_dir, 'server.py')
# Execute the server with the venv python
os.execv(venv_python, [venv_python, server_py])

View File

@@ -0,0 +1,62 @@
import logging
import os
import traceback
from tools.food_tools import register_food_tools
from dotenv import load_dotenv
from mcp.server.fastmcp import FastMCP
from mealie import MealieFetcher
from prompts import register_prompts
from tools import register_all_tools
# Load environment variables first
load_dotenv()
# Get log level from environment variable with INFO as default
log_level_name = os.getenv("LOG_LEVEL", "INFO")
log_level = getattr(logging, log_level_name.upper(), logging.INFO)
# Configure logging
logging.basicConfig(
level=log_level,
format="%(asctime)s - %(name)s - %(levelname)s - %(message)s",
handlers=[logging.StreamHandler(), logging.FileHandler("mealie_mcp_server.log")],
)
logger = logging.getLogger("mealie-mcp")
mcp = FastMCP("mealie")
MEALIE_BASE_URL = os.getenv("MEALIE_BASE_URL")
MEALIE_API_KEY = os.getenv("MEALIE_API_KEY")
if not MEALIE_BASE_URL or not MEALIE_API_KEY:
raise ValueError(
"MEALIE_BASE_URL and MEALIE_API_KEY must be set in environment variables."
)
try:
mealie = MealieFetcher(
base_url=MEALIE_BASE_URL,
api_key=MEALIE_API_KEY,
)
except Exception as e:
logger.error({"message": "Failed to initialize Mealie client", "error": str(e)})
logger.debug({"message": "Error traceback", "traceback": traceback.format_exc()})
raise
register_prompts(mcp)
register_all_tools(mcp, mealie)
register_food_tools(mcp, mealie)
if __name__ == "__main__":
try:
logger.info({"message": "Starting Mealie MCP Server"})
mcp.run(transport="stdio")
except Exception as e:
logger.critical(
{"message": "Fatal error in Mealie MCP Server", "error": str(e)}
)
logger.debug(
{"message": "Error traceback", "traceback": traceback.format_exc()}
)
raise

View File

@@ -0,0 +1,3 @@
#!/bin/bash
cd "$(dirname "$0")"
exec ./venv/bin/python ./server.py

View File

@@ -0,0 +1,19 @@
#!/usr/bin/env python3
import os
import sys
# Find the bundle path (2 levels up from this script)
script_path = os.path.abspath(__file__)
server_dir = os.path.dirname(script_path)
bundle_dir = os.path.dirname(server_dir)
# Add the venv site-packages to Python path
venv_site_packages = os.path.join(server_dir, 'venv', 'lib', 'python3.13', 'site-packages')
sys.path.insert(0, venv_site_packages)
# Change to the bundle directory
os.chdir(bundle_dir)
# Import and run the server module
sys.path.insert(0, server_dir)
import server

View File

@@ -0,0 +1,15 @@
from .mealplan_tools import register_mealplan_tools
from .recipe_tools import register_recipe_tools
def register_all_tools(mcp, mealie):
"""Register all tools with the MCP server."""
register_recipe_tools(mcp, mealie)
register_mealplan_tools(mcp, mealie)
__all__ = [
"register_all_tools",
"register_recipe_tools",
"register_mealplan_tools",
]

View File

@@ -0,0 +1,99 @@
import json
import logging
import traceback
from typing import Optional
from mcp.server.fastmcp import FastMCP
from mealie import MealieFetcher
from utils import format_error_response
logger = logging.getLogger("mealie-mcp")
def register_food_tools(mcp: FastMCP, mealie: MealieFetcher) -> None:
"""Register all food and unit-related tools with the MCP server."""
@mcp.tool()
def get_foods(
search: Optional[str] = None,
page: Optional[int] = None,
per_page: Optional[int] = None,
) -> str:
"""Get a list of foods from your Mealie database.
Args:
search: Search term to filter foods
page: Page number for pagination
per_page: Number of items per page
Returns:
str: JSON list of foods
"""
try:
logger.info({"message": "Fetching foods", "search": search})
result = mealie.get_foods(search=search, page=page, per_page=per_page)
return json.dumps(result, indent=2)
except Exception as e:
error_msg = f"Error fetching foods: {str(e)}"
logger.error({"message": error_msg})
logger.debug({"message": "Error traceback", "traceback": traceback.format_exc()})
return format_error_response(error_msg)
@mcp.tool()
def get_units() -> str:
"""Get a list of all units from your Mealie database.
Returns:
str: JSON list of units
"""
try:
logger.info({"message": "Fetching units"})
result = mealie.get_units()
return json.dumps(result, indent=2)
except Exception as e:
error_msg = f"Error fetching units: {str(e)}"
logger.error({"message": error_msg})
logger.debug({"message": "Error traceback", "traceback": traceback.format_exc()})
return format_error_response(error_msg)
@mcp.tool()
def create_food(name: str, description: Optional[str] = None) -> str:
"""Create a new food item in Mealie.
Args:
name: Name of the food
description: Optional description
Returns:
str: Confirmation with created food details
"""
try:
logger.info({"message": "Creating food", "name": name})
result = mealie.create_food(name=name, description=description)
return json.dumps(result, indent=2)
except Exception as e:
error_msg = f"Error creating food '{name}': {str(e)}"
logger.error({"message": error_msg})
logger.debug({"message": "Error traceback", "traceback": traceback.format_exc()})
return format_error_response(error_msg)
@mcp.tool()
def create_unit(name: str, abbreviation: Optional[str] = None) -> str:
"""Create a new unit in Mealie.
Args:
name: Name of the unit (e.g., "Teaspoon", "Gram")
abbreviation: Optional abbreviation (e.g., "tsp", "g")
Returns:
str: Confirmation with created unit details
"""
try:
logger.info({"message": "Creating unit", "name": name})
result = mealie.create_unit(name=name, abbreviation=abbreviation)
return json.dumps(result, indent=2)
except Exception as e:
error_msg = f"Error creating unit '{name}': {str(e)}"
logger.error({"message": error_msg})
logger.debug({"message": "Error traceback", "traceback": traceback.format_exc()})
return format_error_response(error_msg)

View File

@@ -0,0 +1,133 @@
import logging
import traceback
from typing import Any, Dict, List, Optional
from mcp.server.fastmcp import FastMCP
from mealie import MealieFetcher
from models.mealplan import MealPlanEntry
from utils import format_error_response
logger = logging.getLogger("mealie-mcp")
def register_mealplan_tools(mcp: FastMCP, mealie: MealieFetcher) -> None:
"""Register all mealplan-related tools with the MCP server."""
@mcp.tool()
def get_all_mealplans(
start_date: Optional[str] = None,
end_date: Optional[str] = None,
page: Optional[int] = None,
per_page: Optional[int] = None,
) -> Dict[str, Any]:
"""Get all meal plans for the current household with pagination.
Args:
start_date: Start date for filtering meal plans (ISO format YYYY-MM-DD)
end_date: End date for filtering meal plans (ISO format YYYY-MM-DD)
page: Page number to retrieve
per_page: Number of items per page
Returns:
Dict[str, Any]: JSON response containing mealplan items and pagination information
"""
try:
logger.info(
{
"message": "Fetching mealplans",
"start_date": start_date,
"end_date": end_date,
"page": page,
"per_page": per_page,
}
)
return mealie.get_mealplans(
start_date=start_date,
end_date=end_date,
page=page,
per_page=per_page,
)
except Exception as e:
error_msg = f"Error fetching mealplans: {str(e)}"
logger.error({"message": error_msg})
logger.debug(
{"message": "Error traceback", "traceback": traceback.format_exc()}
)
return format_error_response(error_msg)
@mcp.tool()
def create_mealplan(
entry: MealPlanEntry,
) -> Dict[str, Any]:
"""Create a new meal plan entry.
Args:
entry: MealPlanEntry object containing date, recipe_id, title, and entry_type
Returns:
Dict[str, Any]: JSON response containing the created mealplan entry
"""
try:
logger.info(
{
"message": "Creating mealplan entry",
"entry": entry.model_dump(),
}
)
return mealie.create_mealplan(**entry.model_dump())
except Exception as e:
error_msg = f"Error creating mealplan entry: {str(e)}"
logger.error({"message": error_msg})
logger.debug(
{"message": "Error traceback", "traceback": traceback.format_exc()}
)
return format_error_response(error_msg)
@mcp.tool()
def create_mealplan_bulk(
entries: List[MealPlanEntry],
) -> Dict[str, Any]:
"""Create multiple meal plan entries in bulk.
Args:
entries: List of MealPlanEntry objects
containing date, recipe_id, title, and entry_type
Returns:
Dict[str, Any]: JSON response containing the created mealplan entries
"""
try:
logger.info(
{
"message": "Creating bulk mealplan entries",
"entries_count": len(entries),
}
)
for entry in entries:
mealie.create_mealplan(**entry.model_dump())
return {"message": "Bulk mealplan entries created successfully"}
except Exception as e:
error_msg = f"Error creating bulk mealplan entries: {str(e)}"
logger.error({"message": error_msg})
logger.debug(
{"message": "Error traceback", "traceback": traceback.format_exc()}
)
return format_error_response(error_msg)
@mcp.tool()
def get_todays_mealplan() -> List[Dict[str, Any]]:
"""Get the mealplan entries for today.
Returns:
List[Dict[str, Any]]: List of today's mealplan entries
"""
try:
logger.info({"message": "Fetching today's mealplan"})
return mealie.get_todays_mealplan()
except Exception as e:
error_msg = f"Error fetching today's mealplan: {str(e)}"
logger.error({"message": error_msg})
logger.debug(
{"message": "Error traceback", "traceback": traceback.format_exc()}
)
return format_error_response(error_msg)

View File

@@ -0,0 +1,305 @@
import json
import logging
import traceback
from typing import List, Optional
from mcp.server.fastmcp import FastMCP
from mealie import MealieFetcher
from models.recipe import Recipe, RecipeIngredient, RecipeInstruction, RecipeNutrition
from utils import format_error_response
logger = logging.getLogger("mealie-mcp")
def register_recipe_tools(mcp: FastMCP, mealie: MealieFetcher) -> None:
"""Register all recipe-related tools with the MCP server."""
@mcp.tool()
def get_recipes(
search: Optional[str] = None,
page: Optional[int] = None,
per_page: Optional[int] = None,
categories: Optional[List[str]] = None,
tags: Optional[List[str]] = None,
) -> str:
"""Provides a paginated list of recipes with optional filtering.
Args:
search: Filters recipes by name or description.
page: Page number for pagination.
per_page: Number of items per page.
categories: Filter by specific recipe categories.
tags: Filter by specific recipe tags.
Returns:
str: Recipe summaries with details like ID, name, description, and image information.
"""
try:
logger.info(
{
"message": "Fetching recipes",
"search": search,
"page": page,
"per_page": per_page,
"categories": categories,
"tags": tags,
}
)
result = mealie.get_recipes(
search=search,
page=page,
per_page=per_page,
categories=categories,
tags=tags,
)
return json.dumps(result, indent=2)
except Exception as e:
error_msg = f"Error fetching recipes: {str(e)}"
logger.error({"message": error_msg})
logger.debug(
{"message": "Error traceback", "traceback": traceback.format_exc()}
)
return format_error_response(error_msg)
@mcp.tool()
def get_recipe_detailed(slug: str) -> str:
"""Retrieve a specific recipe by its slug identifier. Use this when to get full recipe
details for tasks like updating or displaying the recipe.
Args:
slug: The unique text identifier for the recipe, typically found in recipe URLs
or from get_recipes results.
Returns:
str: Comprehensive recipe details including ingredients, instructions,
nutrition information, notes, and associated metadata.
"""
try:
logger.info({"message": "Fetching recipe", "slug": slug})
result = mealie.get_recipe(slug)
return json.dumps(result, indent=2)
except Exception as e:
error_msg = f"Error fetching recipe with slug '{slug}': {str(e)}"
logger.error({"message": error_msg})
logger.debug(
{"message": "Error traceback", "traceback": traceback.format_exc()}
)
return format_error_response(error_msg)
@mcp.tool()
def get_recipe_concise(slug: str) -> str:
"""Retrieve a concise version of a specific recipe by its slug identifier. Use this when you only
need a summary of the recipe, such as for when mealplaning.
Args:
slug: The unique text identifier for the recipe, typically found in recipe URLs
or from get_recipes results.
"""
try:
logger.info({"message": "Fetching recipe", "slug": slug})
recipe_json = mealie.get_recipe(slug)
recipe = Recipe.model_validate(recipe_json)
return json.dumps(recipe.model_dump(
include={
"name",
"slug",
"recipeServings",
"recipeYieldQuantity",
"recipeYield",
"totalTime",
"rating",
"recipeIngredient",
"lastMade",
},
exclude_none=True,
), indent=2)
except Exception as e:
error_msg = f"Error fetching recipe with slug '{slug}': {str(e)}"
logger.error({"message": error_msg})
logger.debug(
{"message": "Error traceback", "traceback": traceback.format_exc()}
)
return format_error_response(error_msg)
@mcp.tool()
def create_recipe(
name: str,
ingredients: list[str],
instructions: list[str],
prep_time: Optional[str] = None,
cook_time: Optional[str] = None,
total_time: Optional[str] = None,
recipe_yield: Optional[str] = None,
servings: Optional[int] = None,
description: Optional[str] = None,
calories: Optional[str] = None,
protein: Optional[str] = None,
carbohydrates: Optional[str] = None,
fat: Optional[str] = None,
fiber: Optional[str] = None,
sugar: Optional[str] = None,
sodium: Optional[str] = None,
) -> str:
"""Create a new recipe
Args:
name: The name of the new recipe to be created.
ingredients: A list of ingredients for the recipe include quantities and units.
instructions: A list of instructions for preparing the recipe.
prep_time: Preparation time (e.g., "30min", "1h", "1h 30min")
cook_time: Cooking time (e.g., "45min", "1h 15min")
total_time: Total time (e.g., "90min", "2h")
recipe_yield: What the recipe yields (e.g., "4 servings", "12 cookies")
servings: Number of servings as integer
description: A brief description of the recipe
calories: Calories per serving (e.g., "350")
protein: Protein content (e.g., "25g")
carbohydrates: Carbohydrate content (e.g., "40g")
fat: Fat content (e.g., "15g")
fiber: Fiber content (e.g., "5g")
sugar: Sugar content (e.g., "8g")
sodium: Sodium content (e.g., "600mg")
Returns:
str: Confirmation message or details about the created recipe.
"""
try:
logger.info({"message": "Creating recipe", "name": name})
slug = mealie.create_recipe(name)
recipe_json = mealie.get_recipe(slug)
recipe = Recipe.model_validate(recipe_json)
recipe.recipeIngredient = [RecipeIngredient(note=i) for i in ingredients]
recipe.recipeInstructions = [
RecipeInstruction(text=i) for i in instructions
]
# Set time information if provided
if prep_time:
recipe.prepTime = prep_time
if cook_time:
recipe.cookTime = cook_time
if total_time:
recipe.totalTime = total_time
if recipe_yield:
recipe.recipeYield = recipe_yield
if servings:
recipe.recipeServings = servings
if description:
recipe.description = description
# Set nutrition information if provided
if any([calories, protein, carbohydrates, fat, fiber, sugar, sodium]):
nutrition = RecipeNutrition()
if calories:
nutrition.calories = calories
if protein:
nutrition.proteinContent = protein
if carbohydrates:
nutrition.carbohydrateContent = carbohydrates
if fat:
nutrition.fatContent = fat
if fiber:
nutrition.fiberContent = fiber
if sugar:
nutrition.sugarContent = sugar
if sodium:
nutrition.sodiumContent = sodium
recipe.nutrition = nutrition
result = mealie.update_recipe(slug, recipe.model_dump(exclude_none=True))
return json.dumps(result, indent=2)
except Exception as e:
error_msg = f"Error creating recipe '{name}': {str(e)}"
logger.error({"message": error_msg})
logger.debug(
{"message": "Error traceback", "traceback": traceback.format_exc()}
)
return format_error_response(error_msg)
@mcp.tool()
def update_recipe(
slug: str,
ingredients: Optional[list[str]] = None,
instructions: Optional[list[str]] = None,
prep_time: Optional[str] = None,
cook_time: Optional[str] = None,
total_time: Optional[str] = None,
recipe_yield: Optional[str] = None,
servings: Optional[int] = None,
description: Optional[str] = None,
calories: Optional[str] = None,
protein: Optional[str] = None,
carbohydrates: Optional[str] = None,
fat: Optional[str] = None,
fiber: Optional[str] = None,
sugar: Optional[str] = None,
sodium: Optional[str] = None,
) -> str:
"""Update an existing recipe with new information.
Args:
slug: The unique text identifier for the recipe to be updated.
ingredients: A list of ingredients for the recipe include quantities and units.
instructions: A list of instructions for preparing the recipe.
prep_time: Preparation time (e.g., "PT15M" for 15 minutes in ISO 8601 format)
cook_time: Cooking time (e.g., "PT30M" for 30 minutes)
total_time: Total time (e.g., "PT45M" for 45 minutes)
recipe_yield: What the recipe yields (e.g., "4 servings")
servings: Number of servings as integer
description: A brief description of the recipe
Returns:
str: Confirmation message or details about the updated recipe.
"""
try:
logger.info({"message": "Updating recipe", "slug": slug})
recipe_json = mealie.get_recipe(slug)
recipe = Recipe.model_validate(recipe_json)
if ingredients:
recipe.recipeIngredient = [RecipeIngredient(note=i) for i in ingredients]
if instructions:
recipe.recipeInstructions = [
RecipeInstruction(text=i) for i in instructions
]
if prep_time:
recipe.prepTime = prep_time
if cook_time:
recipe.cookTime = cook_time
if total_time:
recipe.totalTime = total_time
if recipe_yield:
recipe.recipeYield = recipe_yield
if servings:
recipe.recipeServings = servings
if description:
recipe.description = description
if any([calories, protein, carbohydrates, fat, fiber, sugar, sodium]):
if not recipe.nutrition:
recipe.nutrition = RecipeNutrition()
if calories:
recipe.nutrition.calories = calories
if protein:
recipe.nutrition.proteinContent = protein
if carbohydrates:
recipe.nutrition.carbohydrateContent = carbohydrates
if fat:
recipe.nutrition.fatContent = fat
if fiber:
recipe.nutrition.fiberContent = fiber
if sugar:
recipe.nutrition.sugarContent = sugar
if sodium:
recipe.nutrition.sodiumContent = sodium
result = mealie.update_recipe(slug, recipe.model_dump(exclude_none=True))
return json.dumps(result, indent=2)
except Exception as e:
error_msg = f"Error updating recipe '{slug}': {str(e)}"
logger.error({"message": error_msg})
logger.debug(
{"message": "Error traceback", "traceback": traceback.format_exc()}
)
return format_error_response(error_msg)

View File

@@ -0,0 +1,20 @@
import json
def format_error_response(error_message: str) -> str:
"""Format error responses consistently as JSON strings."""
error_response = {"success": False, "error": error_message}
return json.dumps(error_response)
def format_api_params(params: dict) -> dict:
"""Formats list and None values in a dictionary for API parameters."""
output = {}
for k, v in params.items():
if v is None:
continue
if isinstance(v, list):
output[k] = ",".join(v)
else:
output[k] = v
return output