import logging from typing import Dict, Any, List from dataclasses import asdict logger = logging.getLogger(__name__) class PodcastTool: """ MCP Tool for podcast generation from documents """ def __init__(self, podcast_generator): """ Initialize Podcast Tool Args: podcast_generator: PodcastGeneratorService instance """ self.podcast_generator = podcast_generator async def generate_podcast( self, document_ids: List[str], style: str = "conversational", duration_minutes: int = 10, host1_voice: str = "Rachel", host2_voice: str = "Adam" ) -> Dict[str, Any]: """ MCP Tool: Generate podcast from documents Args: document_ids: List of document IDs to generate podcast from style: Podcast style (conversational, educational, technical, casual) duration_minutes: Target duration in minutes host1_voice: Voice name for first host host2_voice: Voice name for second host Returns: Dictionary with podcast ID, audio URL, transcript, and metadata """ try: if not document_ids or len(document_ids) == 0: return { "success": False, "error": "No documents provided. Please select at least one document." } logger.info(f"Generating podcast from {len(document_ids)} documents") # Generate podcast using service result = await self.podcast_generator.generate_podcast( document_ids=document_ids, style=style, duration_minutes=duration_minutes, host1_voice=host1_voice, host2_voice=host2_voice ) if result.success: return { "success": True, "podcast_id": result.podcast_id, "audio_file": result.audio_file_path, "audio_url": f"/data/podcasts/{result.podcast_id}.mp3", "transcript": result.transcript, "metadata": asdict(result.metadata) if result.metadata else {}, "generation_time": result.generation_time, "message": f"Podcast generated successfully! Duration: {result.metadata.duration_seconds/60:.1f} minutes" } else: return { "success": False, "error": result.error or "Unknown error during podcast generation" } except Exception as e: logger.error(f"Podcast generation failed: {str(e)}") return { "success": False, "error": str(e) } async def generate_transcript( self, document_ids: List[str], style: str = "conversational", duration_minutes: int = 10 ) -> Dict[str, Any]: """ MCP Tool: Generate podcast transcript ONLY (no audio) Args: document_ids: List of document IDs style: Podcast style duration_minutes: Target duration Returns: Dictionary with transcript and analysis """ try: if not document_ids: return {"success": False, "error": "No documents provided"} logger.info(f"Generating transcript for {len(document_ids)} docs") # 1. Analyze analysis = await self.podcast_generator.analyze_documents(document_ids) # 2. Generate Script script = await self.podcast_generator.generate_script( analysis, style, duration_minutes ) return { "success": True, "transcript": script.to_text(), "word_count": script.word_count, "estimated_duration": script.total_duration_estimate, "key_insights": analysis.key_insights, "topics": analysis.topics } except Exception as e: logger.error(f"Transcript generation failed: {str(e)}") return {"success": False, "error": str(e)} def get_podcast_audio(self, podcast_id: str) -> Dict[str, Any]: """ MCP Tool: Get audio file path for a podcast Args: podcast_id: Podcast ID Returns: Dictionary with audio file path """ try: podcast = self.podcast_generator.get_podcast(podcast_id) if not podcast: return {"success": False, "error": "Podcast not found"} # Construct absolute path (assuming local running) # In a real remote setup, this might return a URL audio_path = f"/data/podcasts/{podcast_id}.mp3" return { "success": True, "podcast_id": podcast_id, "audio_path": audio_path, "exists": True } except Exception as e: logger.error(f"Failed to get audio path: {str(e)}") return {"success": False, "error": str(e)} def list_podcasts(self, limit: int = 10) -> Dict[str, Any]: """ List previously generated podcasts Args: limit: Maximum number of podcasts to return Returns: Dictionary with list of podcast metadata """ try: podcasts = self.podcast_generator.list_podcasts(limit=limit) return { "success": True, "podcasts": [asdict(p) for p in podcasts], "total": len(podcasts) } except Exception as e: logger.error(f"Failed to list podcasts: {str(e)}") return { "success": False, "error": str(e), "podcasts": [] } def get_podcast(self, podcast_id: str) -> Dict[str, Any]: """ Get specific podcast by ID Args: podcast_id: Podcast identifier Returns: Dictionary with podcast metadata """ try: podcast = self.podcast_generator.get_podcast(podcast_id) if podcast: return { "success": True, "podcast": asdict(podcast) } else: return { "success": False, "error": "Podcast not found" } except Exception as e: logger.error(f"Failed to get podcast: {str(e)}") return { "success": False, "error": str(e) }