mcp_tools/voice_tool.py

import logging
from typing import Dict, Any, Optional
import asyncio

logger = logging.getLogger(__name__)

class VoiceTool:
    """
    Enhanced MCP Tool for voice-based Q&A using ElevenLabs conversational AI
    
    Improvements:
    - Better error handling and user feedback
    - Support for conversation context
    - Streaming responses support
    - Session management
    """
    
    def __init__(self, elevenlabs_service):
        """
        Initialize Voice Tool
        
        Args:
            elevenlabs_service: ElevenLabs service instance
        """
        self.elevenlabs_service = elevenlabs_service
    
    async def voice_qa(
        self, 
        question: str,
        session_id: Optional[str] = None
    ) -> Dict[str, Any]:
        """
        Ask a question using voice assistant (text-based for web UI)
        
        Args:
            question: User's question
            session_id: Optional session ID for conversation context
            
        Returns:
            Dictionary with answer and metadata
        """
        try:
            # Check if service is available
            if not self.elevenlabs_service or not self.elevenlabs_service.is_available():
                return {
                    "success": False,
                    "error": "Voice assistant not configured. Please set ELEVENLABS_API_KEY in your .env file.",
                    "help": "Get your API key from: https://elevenlabs.io/app/settings/api-keys"
                }
            
            if not question or not question.strip():
                return {
                    "success": False,
                    "error": "Please enter a question"
                }
            
            logger.info(f"Voice QA (session: {session_id}): {question}")
            
            # Send message through ElevenLabs service
            result = await self.elevenlabs_service.send_text_message(
                message=question,
                session_id=session_id or "default"
            )
            
            if result.get("success"):
                return {
                    "success": True,
                    "question": question,
                    "answer": result["answer"],
                    "session_id": session_id,
                    "mode": "text"
                }
            else:
                return {
                    "success": False,
                    "error": result.get("error", "Unknown error"),
                    "question": question
                }
            
        except Exception as e:
            logger.error(f"Voice QA failed: {str(e)}", exc_info=True)
            return {
                "success": False,
                "error": f"An error occurred: {str(e)}",
                "question": question
            }
    
    async def start_session(self, session_id: str) -> Dict[str, Any]:
        """
        Start a new voice assistant session
        
        Args:
            session_id: Unique session identifier
            
        Returns:
            Session start status
        """
        try:
            result = await self.elevenlabs_service.start_conversation(session_id)
            return result
        except Exception as e:
            logger.error(f"Failed to start session: {str(e)}")
            return {
                "success": False,
                "error": str(e)
            }
    
    async def end_session(self, session_id: str) -> Dict[str, Any]:
        """
        End a voice assistant session
        
        Args:
            session_id: Session identifier
            
        Returns:
            Session end status
        """
        try:
            success = await self.elevenlabs_service.end_conversation(session_id)
            return {
                "success": success,
                "message": "Session ended" if success else "Session not found"
            }
        except Exception as e:
            logger.error(f"Failed to end session: {str(e)}")
            return {
                "success": False,
                "error": str(e)
            }
    
    def get_conversation_history(self, session_id: str) -> Dict[str, Any]:
        """
        Get conversation history for a session
        
        Args:
            session_id: Session identifier
            
        Returns:
            Dictionary with conversation history
        """
        try:
            history = self.elevenlabs_service.get_conversation_history(session_id)
            return {
                "success": True,
                "history": history,
                "message_count": len(history)
            }
        except Exception as e:
            logger.error(f"Failed to get history: {str(e)}")
            return {
                "success": False,
                "error": str(e),
                "history": []
            }
    
    async def test_connection(self) -> Dict[str, Any]:
        """
        Test voice assistant connection
        
        Returns:
            Connection test results
        """
        try:
            if not self.elevenlabs_service:
                return {
                    "success": False,
                    "message": "Service not initialized"
                }
            
            result = await self.elevenlabs_service.test_connection()
            return {
                "success": result["status"] == "success",
                **result
            }
        except Exception as e:
            logger.error(f"Connection test failed: {str(e)}")
            return {
                "success": False,
                "message": str(e)
            }