Create app.py

app.py

@@ -0,0 +1,179 @@
+# MCP-Powered Culinary Voice Assistant
+# Hugging Face Space Implementation
+
+import gradio as gr
+import numpy as np
+from mcp.server.fastmcp import FastMCP
+from agents import Agent, trace
+from agents.mcp import MCPServerSse, MCPServerStdio
+from agents.voice import VoicePipeline, TTSModelSettings, AudioInput
+import sqlite3
+import json
+import requests
+from PIL import Image
+import io
+
+# ------ Custom MCP Cooking Tools Server ------
+mcp = FastMCP("Culinary Tools Server")
+
+@mcp.tool()
+def get_recipe_by_ingredients(ingredients: list) -> dict:
+    """Find recipes based on available ingredients"""
+    print(f"[Culinary Server] Finding recipes with: {', '.join(ingredients)}")
+    # In a real implementation, this would call a recipe API
+    return {
+        "recipes": [
+            {"name": "Vegetable Stir Fry", "time": 20, "difficulty": "Easy"},
+            {"name": "Pasta Primavera", "time": 30, "difficulty": "Medium"}
+        ]
+    }
+
+@mcp.tool()
+def get_recipe_image(recipe_name: str) -> str:
+    """Generate an image of the finished recipe"""
+    print(f"[Culinary Server] Generating image for: {recipe_name}")
+    # This would call DALL-E or Stable Diffusion in production
+    return "https://example.com/recipe-image.jpg"
+
+@mcp.tool()
+def convert_measurements(amount: float, from_unit: str, to_unit: str) -> dict:
+    """Convert cooking measurements between units"""
+    print(f"[Culinary Server] Converting {amount} {from_unit} to {to_unit}")
+    # Simple conversion logic - real implementation would handle more units
+    conversions = {
+        ("tbsp", "tsp"): lambda x: x * 3,
+        ("cups", "ml"): lambda x: x * 240,
+        ("oz", "g"): lambda x: x * 28.35
+    }
+    conversion_key = (from_unit.lower(), to_unit.lower())
+    if conversion_key in conversions:
+        return {"result": conversions[conversion_key](amount), "unit": to_unit}
+    return {"error": "Conversion not supported"}
+
+# ------ Recipe Database (SQLite) ------
+def init_recipe_db():
+    conn = sqlite3.connect('file:recipes.db?mode=memory&cache=shared', uri=True)
+    c = conn.cursor()
+    c.execute('''CREATE TABLE IF NOT EXISTS recipes
+                 (id INTEGER PRIMARY KEY, name TEXT, ingredients TEXT, instructions TEXT, prep_time INT)''')
+    
+    # Sample recipes
+    recipes = [
+        ("Classic Pancakes", "['flour', 'eggs', 'milk', 'baking powder']", 
+         "1. Mix dry ingredients\n2. Add wet ingredients\n3. Cook on griddle", 15),
+        ("Tomato Soup", "['tomatoes', 'onion', 'garlic', 'vegetable stock']", 
+         "1. Sauté onions\n2. Add tomatoes\n3. Simmer and blend", 30)
+    ]
+    
+    c.executemany("INSERT INTO recipes (name, ingredients, instructions, prep_time) VALUES (?,?,?,?)", recipes)
+    conn.commit()
+    return conn
+
+# ------ Voice Assistant Setup ------
+def create_culinary_agent(mcp_servers):
+    """Create the culinary assistant agent"""
+    culinary_agent = Agent(
+        name="ChefAssistant",
+        instructions="""
+        You are a professional chef assistant. Help users with cooking tasks:
+        1. Use get_recipe_by_ingredients when users have specific ingredients
+        2. Use get_recipe_details for known recipes
+        3. Use convert_measurements for unit conversions
+        4. Use get_recipe_image when the user asks to see a dish
+        5. Keep responses concise and practical for kitchen use
+        6. Use a warm, encouraging tone suitable for cooking
+        """,
+        mcp_servers=mcp_servers,
+        model="gpt-4.1-mini",
+    )
+    return culinary_agent
+
+# ------ Gradio Interface ------
+def process_voice_command(audio, state):
+    """Process voice command through the agent system"""
+    sr, audio_data = audio
+    audio_array = (audio_data / np.iinfo(audio_data.dtype).max).astype(np.float32)
+    
+    # Initialize on first run
+    if state is None:
+        init_recipe_db()
+        state = {
+            "mcp_servers": [],
+            "agent": None,
+            "voice_pipeline": VoicePipeline(
+                workflow=None,
+                config=VoicePipelineConfig(
+                    tts_settings=TTSModelSettings(
+                        instructions="Warm, encouraging chef voice"
+                    )
+                )
+            )
+        }
+        
+        # Start MCP servers
+        with MCPServerSse(
+            name="Culinary Tools",
+            params={"url": "http://localhost:8000/sse"},
+            client_session_timeout_seconds=15,
+        ) as culinary_server:
+            with MCPServerStdio(
+                params={"command": "uvx", "args": ["mcp-server-sqlite", "--db-path", "file:recipes.db?mode=memory&cache=shared"]},
+            ) as db_server:
+                state["mcp_servers"] = [culinary_server, db_server]
+                state["agent"] = create_culinary_agent(state["mcp_servers"])
+    
+    # Process audio through agent
+    audio_input = AudioInput(buffer=audio_array, sample_rate=sr)
+    response = state["voice_pipeline"].run(state["agent"], audio_input)
+    
+    # For demo purposes, return mock response
+    return (
+        "https://example.com/response.wav", 
+        "I found 3 recipes for your ingredients! Vegetable Stir Fry (20 mins) and Pasta Primavera (30 mins).",
+        "https://example.com/stir-fry.jpg",
+        state
+    )
+
+# ------ Hugging Face Space UI ------
+with gr.Blocks(title="MCP Culinary Voice Assistant") as demo:
+    state = gr.State(value=None)
+    
+    with gr.Row():
+        gr.Markdown("# 🧑‍🍳 MCP-Powered Culinary Voice Assistant")
+    
+    with gr.Row():
+        audio_input = gr.Audio(source="microphone", type="numpy", label="Speak to Chef Assistant")
+        audio_output = gr.Audio(label="Assistant Response", interactive=False)
+    
+    with gr.Row():
+        text_output = gr.Textbox(label="Transcription", interactive=False)
+        image_output = gr.Image(label="Recipe Image", interactive=False)
+    
+    with gr.Row():
+        submit_btn = gr.Button("Process Command", variant="primary")
+    
+    submit_btn.click(
+        fn=process_voice_command,
+        inputs=[audio_input, state],
+        outputs=[audio_output, text_output, image_output, state]
+    )
+
+    gr.Examples(
+        examples=[
+            ["What can I make with eggs and flour?", "", ""],
+            ["Show me how tomato soup looks", "", ""],
+            ["Convert 2 cups to milliliters", "", ""]
+        ],
+        inputs=[text_output],
+        label="Example Queries"
+    )
+
+if __name__ == "__main__":
+    # Start MCP server in background thread
+    import threading
+    server_thread = threading.Thread(target=mcp.run, kwargs={"transport": "sse"})
+    server_thread.daemon = True
+    server_thread.start()
+    
+    # Launch Gradio interface
+    demo.launch(server_name="0.0.0.0", server_port=7860)