from fastmcp import FastMCP, Context
from contextlib import asynccontextmanager
from collections.abc import AsyncIterator
from dataclasses import dataclass
from dotenv import load_dotenv
import asyncio
import json
import os
from typing import List, Optional
from utils import get_fdc_api_client, FoodDataCentralAPI
load_dotenv()
# Create a dataclass for our application context
@dataclass
class FoodDataContext:
"""Context for the Food Data Central MCP server."""
fdc_client: FoodDataCentralAPI
@asynccontextmanager
async def fdc_lifespan(server: FastMCP) -> AsyncIterator[FoodDataContext]:
"""
Manages the Food Data Central API client lifecycle.
Args:
server: The FastMCP server instance
Yields:
FoodDataContext: The context containing the FDC API client
"""
# Create and return the FDC API client
fdc_client = get_fdc_api_client()
try:
yield FoodDataContext(fdc_client=fdc_client)
finally:
# Clean up the HTTP client
await fdc_client.close()
# Initialize FastMCP server with the Food Data Central API client as context
mcp = FastMCP(
"food-data-central-mcp",
instructions="MCP server for accessing USDA's FoodData Central database",
lifespan=fdc_lifespan,
)
@mcp.tool()
async def search_foods(
ctx: Context,
query: str,
data_type: Optional[List[str]] = None,
page_size: int = 50,
page_number: int = 1,
sort_by: Optional[str] = None,
sort_order: Optional[str] = None,
brand_owner: Optional[str] = None,
start_date: Optional[str] = None,
end_date: Optional[str] = None
) -> str:
"""Search for foods in the USDA Food Data Central database.
This tool searches for foods using keywords and returns a list of matching food items
with their basic information including FDC ID, description, and key nutrients.
Args:
ctx: The MCP server provided context which includes the FDC API client
query: Search terms to find foods (e.g., "cheddar cheese", "apple")
data_type: Optional filter on data type (e.g., ["Branded", "Foundation", "Survey (FNDDS)", "SR Legacy"])
page_size: Maximum number of results to return (default: 50, max: 200)
page_number: Page number to retrieve (default: 1)
sort_by: Sort field (e.g., "dataType.keyword", "lowercaseDescription.keyword", "fdcId", "publishedDate")
sort_order: Sort direction ("asc" or "desc")
brand_owner: Filter by brand owner (only for Branded Foods)
start_date: Filter foods published on or after this date (YYYY-MM-DD)
end_date: Filter foods published on or before this date (YYYY-MM-DD)
"""
try:
fdc_client = ctx.request_context.lifespan_context.fdc_client
result = await fdc_client.search_foods(
query=query,
data_type=data_type,
page_size=page_size,
page_number=page_number,
sort_by=sort_by,
sort_order=sort_order,
brand_owner=brand_owner,
start_date=start_date,
end_date=end_date
)
return json.dumps(result.model_dump(), indent=2)
except Exception as e:
return f"Error searching foods: {str(e)}"
@mcp.tool()
async def get_food_details(
ctx: Context,
fdc_id: int,
format_type: str = "full",
nutrients: Optional[List[int]] = None
) -> str:
"""Get detailed information about a specific food item by its FDC ID.
This tool retrieves comprehensive information about a single food item including
all available nutrients, ingredients, and other details.
Args:
ctx: The MCP server provided context which includes the FDC API client
fdc_id: The Food Data Central ID of the food item
format_type: Format of the response ("full" for all data, "abridged" for basic data)
nutrients: Optional list of specific nutrient IDs to include (max 25)
"""
try:
fdc_client = ctx.request_context.lifespan_context.fdc_client
result = await fdc_client.get_food_details(
fdc_id=fdc_id,
format_type=format_type,
nutrients=nutrients
)
return json.dumps(result.model_dump(), indent=2)
except Exception as e:
return f"Error getting food details: {str(e)}"
@mcp.tool()
async def get_multiple_foods(
ctx: Context,
fdc_ids: List[int],
format_type: str = "full",
nutrients: Optional[List[int]] = None
) -> str:
"""Get detailed information about multiple food items by their FDC IDs.
This tool retrieves comprehensive information about multiple food items at once.
Useful when you need details for several foods simultaneously.
Args:
ctx: The MCP server provided context which includes the FDC API client
fdc_ids: List of Food Data Central IDs (max 20 items)
format_type: Format of the response ("full" for all data, "abridged" for basic data)
nutrients: Optional list of specific nutrient IDs to include (max 25)
"""
try:
if len(fdc_ids) > 20:
return "Error: Maximum 20 FDC IDs allowed per request"
fdc_client = ctx.request_context.lifespan_context.fdc_client
results = await fdc_client.get_multiple_foods(
fdc_ids=fdc_ids,
format_type=format_type,
nutrients=nutrients
)
return json.dumps([result.model_dump() for result in results], indent=2)
except Exception as e:
return f"Error getting multiple foods: {str(e)}"
async def main():
transport = os.getenv("TRANSPORT", "streamable-http")
if transport in ['sse', 'streamable-http']:
# Run the MCP server with HTTP transport (sse is deprecated but still supported)
await mcp.run_http_async(
transport=transport,
host=os.getenv("HOST", "0.0.0.0"),
port=int(os.getenv("PORT", "8050"))
)
else:
# Run the MCP server with stdio transport
await mcp.run_stdio_async()
if __name__ == "__main__":
asyncio.run(main())
