Update app.py

app.py

@@ -1,26 +1,38 @@
-# MCP-Powered Culinary Voice Assistant
+# MCP-Powered Voice Assistant with Open-Source Tools
 # 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
+import time
 
-# ------ Custom MCP Cooking Tools Server ------
-mcp = FastMCP("Culinary Tools Server")
+# ------ Mock MCP Server Implementation ------
+class MockMCPServer:
+    def __init__(self):
+        self.tools = {}
+        
+    def register_tool(self, name, func, description):
+        self.tools[name] = {
+            "function": func,
+            "description": description
+        }
+    
+    def call_tool(self, tool_name, params):
+        if tool_name in self.tools:
+            return self.tools[tool_name]["function"](**params)
+        return {"error": f"Tool {tool_name} not found"}
 
-@mcp.tool()
-def get_recipe_by_ingredients(ingredients: list) -> dict:
+# ------ Create Mock MCP Server ------
+mcp_server = MockMCPServer()
+
+# ------ Tool Implementations ------
+def get_recipe_by_ingredients(ingredients):
     """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
+    # In a real implementation, this would call an API
     return {
         "recipes": [
             {"name": "Vegetable Stir Fry", "time": 20, "difficulty": "Easy"},
@@ -28,18 +40,16 @@ def get_recipe_by_ingredients(ingredients: list) -> dict:
         ]
     }
 
-@mcp.tool()
-def get_recipe_image(recipe_name: str) -> str:
+def get_recipe_image(recipe_name):
     """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"
+    # In production, this would call a model like Stable Diffusion
+    return {
+        "image_url": "https://example.com/recipe-image.jpg",
+        "alt_text": f"Image of {recipe_name}"
+    }
 
-@mcp.tool()
-def convert_measurements(amount: float, from_unit: str, to_unit: str) -> dict:
+def convert_measurements(amount, from_unit, to_unit):
     """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,
@@ -50,18 +60,17 @@ def convert_measurements(amount: float, from_unit: str, to_unit: str) -> dict:
         return {"result": conversions[conversion_key](amount), "unit": to_unit}
     return {"error": "Conversion not supported"}
 
-# ------ Recipe Database (SQLite) ------
+# ------ Recipe Database ------
 def init_recipe_db():
-    conn = sqlite3.connect('file:recipes.db?mode=memory&cache=shared', uri=True)
+    conn = sqlite3.connect(':memory:')
     c = conn.cursor()
-    c.execute('''CREATE TABLE IF NOT EXISTS recipes
+    c.execute('''CREATE TABLE recipes
                  (id INTEGER PRIMARY KEY, name TEXT, ingredients TEXT, instructions TEXT, prep_time INT)''')
     
-    # Sample recipes
     recipes = [
-        ("Classic Pancakes", "['flour', 'eggs', 'milk', 'baking powder']", 
+        ("Classic Pancakes", json.dumps(["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']", 
+        ("Tomato Soup", json.dumps(["tomatoes", "onion", "garlic", "vegetable stock"]), 
          "1. Sauté onions\n2. Add tomatoes\n3. Simmer and blend", 30)
     ]
     
@@ -69,77 +78,106 @@ def init_recipe_db():
     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
+# ------ Voice Processing Functions ------
+def text_to_speech(text):
+    """Mock TTS function - in real use, replace with actual TTS"""
+    print(f"[TTS]: {text}")
+    # Return dummy audio data
+    return np.zeros(16000, dtype=np.float32), 16000
+
+def speech_to_text(audio):
+    """Mock STT function - in real use, replace with actual STT"""
+    # Return dummy text
+    return "Show me pancake recipes"
+
+# ------ Agent Logic ------
+def process_query(query, db_conn):
+    """Process user query using the available tools"""
+    # Simple intent recognition
+    if "recipe" in query.lower() or "make" in query.lower():
+        # Extract ingredients
+        ingredients = ["flour", "eggs"]  # Simplified extraction
+        return mcp_server.call_tool(
+            "get_recipe_by_ingredients",
+            {"ingredients": ingredients}
+        )
+    elif "image" in query.lower() or "show" in query.lower():
+        recipe_name = "Classic Pancakes"  # Simplified extraction
+        return mcp_server.call_tool(
+            "get_recipe_image",
+            {"recipe_name": recipe_name}
+        )
+    elif "convert" in query.lower():
+        # Simplified extraction
+        return mcp_server.call_tool(
+            "convert_measurements",
+            {"amount": 2, "from_unit": "cups", "to_unit": "ml"}
+        )
+    else:
+        # Fallback to database search
+        c = db_conn.cursor()
+        c.execute("SELECT * FROM recipes WHERE name LIKE ?", (f"%{query}%",))
+        return c.fetchall()
+
+# ------ Register Tools with MCP Server ------
+mcp_server.register_tool(
+    "get_recipe_by_ingredients",
+    get_recipe_by_ingredients,
+    "Find recipes based on available ingredients"
+)
+mcp_server.register_tool(
+    "get_recipe_image",
+    get_recipe_image,
+    "Generate an image of the finished recipe"
+)
+mcp_server.register_tool(
+    "convert_measurements",
+    convert_measurements,
+    "Convert cooking measurements between units"
+)
+
+# ------ Initialize System ------
+db_conn = init_recipe_db()
 
 # ------ Gradio Interface ------
-def process_voice_command(audio, state):
+def process_voice_command(audio):
     """Process voice command through the agent system"""
-    sr, audio_data = audio
-    audio_array = (audio_data / np.iinfo(audio_data.dtype).max).astype(np.float32)
+    # Convert audio to text
+    query = speech_to_text(audio)
     
-    # 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 query using agent logic
+    result = process_query(query, db_conn)
     
-    # Process audio through agent
-    audio_input = AudioInput(buffer=audio_array, sample_rate=sr)
-    response = state["voice_pipeline"].run(state["agent"], audio_input)
+    # Generate response text
+    if isinstance(result, list) and result:
+        response_text = f"Found {len(result)} recipes:\n"
+        for item in result:
+            response_text += f"- {item[1]} ({item[4]} mins)\n"
+    elif "recipes" in result:
+        response_text = f"Found {len(result['recipes'])} recipes:\n"
+        for recipe in result["recipes"]:
+            response_text += f"- {recipe['name']} ({recipe['time']} mins)\n"
+    elif "image_url" in result:
+        response_text = f"Here's an image of {result.get('alt_text', 'the recipe')}"
+        image = Image.new('RGB', (300, 200), color=(73, 109, 137))
+    else:
+        response_text = str(result)
+        image = None
     
-    # For demo purposes, return mock response
+    # Convert response to audio
+    audio_data, sr = text_to_speech(response_text)
+    
+    # Return results
     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
+        (sr, audio_data),
+        response_text,
+        image if 'image' in locals() else None
     )
 
 # ------ 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")
+    gr.Markdown("# 🧑‍🍳 MCP-Powered Culinary Voice Assistant (Open-Source)")
+    gr.Markdown("Speak to your cooking assistant about recipes, conversions, and more!")
     
     with gr.Row():
         audio_input = gr.Audio(source="microphone", type="numpy", label="Speak to Chef Assistant")
@@ -154,26 +192,19 @@ with gr.Blocks(title="MCP Culinary Voice Assistant") as demo:
     
     submit_btn.click(
         fn=process_voice_command,
-        inputs=[audio_input, state],
-        outputs=[audio_output, text_output, image_output, state]
+        inputs=[audio_input],
+        outputs=[audio_output, text_output, image_output]
     )
 
     gr.Examples(
         examples=[
-            ["What can I make with eggs and flour?", "", ""],
-            ["Show me how tomato soup looks", "", ""],
-            ["Convert 2 cups to milliliters", "", ""]
+            ["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)
+    demo.launch()