Spaces:
Sleeping
Sleeping
| #!/usr/bin/env python3 | |
| """ | |
| MCP Video Analysis Client with Anthropic Integration | |
| This application serves as an MCP (Model Context Protocol) client that: | |
| 1. Connects to video analysis tools via MCP | |
| 2. Integrates with Anthropic's Claude for intelligent video understanding | |
| 3. Provides a Gradio interface for user interaction | |
| """ | |
| import os | |
| import json | |
| import asyncio | |
| import logging | |
| from typing import Dict, Any, List, Optional | |
| import gradio as gr | |
| import httpx | |
| from anthropic import Anthropic | |
| # Configure logging | |
| logging.basicConfig(level=logging.INFO) | |
| logger = logging.getLogger(__name__) | |
| class MCPVideoAnalysisClient: | |
| """MCP Client for video analysis with Anthropic integration.""" | |
| def __init__(self): | |
| # Initialize Anthropic client | |
| self.anthropic_api_key = os.getenv("ANTHROPIC_API_KEY") | |
| if not self.anthropic_api_key: | |
| raise ValueError("ANTHROPIC_API_KEY environment variable is required") | |
| self.anthropic_client = Anthropic(api_key=self.anthropic_api_key) | |
| # Modal backend endpoint | |
| self.modal_endpoint = os.getenv( | |
| "MODAL_VIDEO_ANALYSIS_ENDPOINT_URL", | |
| "https://jomasego--video-analysis-gradio-pipeline-process-video-analysis.modal.run" | |
| ) | |
| logger.info(f"Initialized MCP Video Analysis Client with Modal endpoint: {self.modal_endpoint}") | |
| async def analyze_video_with_modal(self, video_url: str) -> Dict[str, Any]: | |
| """Call the Modal backend for comprehensive video analysis.""" | |
| try: | |
| async with httpx.AsyncClient(timeout=300.0) as client: | |
| logger.info(f"Calling Modal backend for video analysis: {video_url}") | |
| response = await client.post( | |
| self.modal_endpoint, | |
| json={"video_url": video_url}, | |
| headers={"Content-Type": "application/json"} | |
| ) | |
| response.raise_for_status() | |
| return response.json() | |
| except Exception as e: | |
| logger.error(f"Error calling Modal backend: {e}") | |
| return {"error": f"Modal backend error: {str(e)}"} | |
| def enhance_analysis_with_claude(self, video_analysis: Dict[str, Any], user_query: str = None) -> str: | |
| """Use Claude to provide intelligent insights about the video analysis.""" | |
| # Prepare the analysis data for Claude | |
| analysis_summary = self._format_analysis_for_claude(video_analysis) | |
| # Create the prompt for Claude | |
| system_prompt = """You are an expert video analyst with deep knowledge of multimedia content, storytelling, and visual communication. You excel at interpreting video analysis data and providing meaningful insights. | |
| Your task is to analyze the provided video analysis data and give intelligent, actionable insights. Focus on: | |
| 1. Content understanding and themes | |
| 2. Visual storytelling elements | |
| 3. Technical quality assessment | |
| 4. Audience engagement potential | |
| 5. Key moments and highlights | |
| 6. Contextual relevance | |
| Be concise but thorough, and tailor your response to be useful for content creators, marketers, or researchers.""" | |
| if user_query: | |
| user_prompt = f"""Here is the video analysis data: | |
| {analysis_summary} | |
| User's specific question: {user_query} | |
| Please provide a comprehensive analysis addressing the user's question while incorporating insights from all the available data.""" | |
| else: | |
| user_prompt = f"""Here is the video analysis data: | |
| {analysis_summary} | |
| Please provide a comprehensive analysis of this video, highlighting the most important insights and potential applications.""" | |
| try: | |
| response = self.anthropic_client.messages.create( | |
| model="claude-3-5-sonnet-20241022", | |
| max_tokens=2000, | |
| temperature=0.3, | |
| system=system_prompt, | |
| messages=[{"role": "user", "content": user_prompt}] | |
| ) | |
| return response.content[0].text | |
| except Exception as e: | |
| logger.error(f"Error calling Anthropic API: {e}") | |
| return f"Error generating Claude analysis: {str(e)}" | |
| def _format_analysis_for_claude(self, analysis: Dict[str, Any]) -> str: | |
| """Format the video analysis data for Claude consumption.""" | |
| formatted = [] | |
| # Handle transcription | |
| if "transcription" in analysis: | |
| transcription = analysis["transcription"] | |
| if isinstance(transcription, str) and not transcription.startswith("Error"): | |
| formatted.append(f"**TRANSCRIPTION:**\n{transcription}\n") | |
| else: | |
| formatted.append(f"**TRANSCRIPTION:** {transcription}\n") | |
| # Handle caption | |
| if "caption" in analysis: | |
| caption = analysis["caption"] | |
| if isinstance(caption, str) and not caption.startswith("Error"): | |
| formatted.append(f"**VIDEO CAPTION:**\n{caption}\n") | |
| else: | |
| formatted.append(f"**VIDEO CAPTION:** {caption}\n") | |
| # Handle actions | |
| if "actions" in analysis: | |
| actions = analysis["actions"] | |
| if isinstance(actions, list) and actions: | |
| action_text = [] | |
| for action in actions: | |
| if isinstance(action, dict): | |
| if "error" in action: | |
| action_text.append(f"Error: {action['error']}") | |
| else: | |
| # Format action detection results | |
| action_text.append(str(action)) | |
| else: | |
| action_text.append(str(action)) | |
| formatted.append(f"**ACTION RECOGNITION:**\n{'; '.join(action_text)}\n") | |
| else: | |
| formatted.append(f"**ACTION RECOGNITION:** {actions}\n") | |
| # Handle objects | |
| if "objects" in analysis: | |
| objects = analysis["objects"] | |
| if isinstance(objects, list) and objects: | |
| object_text = [] | |
| for obj in objects: | |
| if isinstance(obj, dict): | |
| if "error" in obj: | |
| object_text.append(f"Error: {obj['error']}") | |
| else: | |
| # Format object detection results | |
| object_text.append(str(obj)) | |
| else: | |
| object_text.append(str(obj)) | |
| formatted.append(f"**OBJECT DETECTION:**\n{'; '.join(object_text)}\n") | |
| else: | |
| formatted.append(f"**OBJECT DETECTION:** {objects}\n") | |
| # Handle any errors | |
| if "error" in analysis: | |
| formatted.append(f"**ANALYSIS ERROR:**\n{analysis['error']}\n") | |
| return "\n".join(formatted) if formatted else "No analysis data available." | |
| async def process_video_request(self, video_url: str, user_query: str = None) -> tuple[str, str]: | |
| """Process a complete video analysis request with Claude enhancement.""" | |
| if not video_url or not video_url.strip(): | |
| return "Please provide a valid video URL.", "" | |
| try: | |
| # Step 1: Get video analysis from Modal backend | |
| logger.info(f"Starting video analysis for: {video_url}") | |
| video_analysis = await self.analyze_video_with_modal(video_url.strip()) | |
| # Step 2: Format the raw analysis for display | |
| raw_analysis = json.dumps(video_analysis, indent=2) | |
| # Step 3: Enhance with Claude insights | |
| logger.info("Generating Claude insights...") | |
| claude_insights = self.enhance_analysis_with_claude(video_analysis, user_query) | |
| return claude_insights, raw_analysis | |
| except Exception as e: | |
| error_msg = f"Error processing video request: {str(e)}" | |
| logger.error(error_msg) | |
| return error_msg, "" | |
| # Initialize the MCP client | |
| try: | |
| mcp_client = MCPVideoAnalysisClient() | |
| logger.info("MCP Video Analysis Client initialized successfully") | |
| except Exception as e: | |
| logger.error(f"Failed to initialize MCP client: {e}") | |
| mcp_client = None | |
| # Gradio Interface Functions | |
| async def analyze_video_interface(video_url: str, user_query: str = None) -> tuple[str, str]: | |
| """Gradio interface function for video analysis.""" | |
| if not mcp_client: | |
| return "MCP Client not initialized. Please check your environment variables.", "" | |
| return await mcp_client.process_video_request(video_url, user_query) | |
| def create_gradio_interface(): | |
| """Create and configure the Gradio interface.""" | |
| with gr.Blocks( | |
| title="MCP Video Analysis with Claude", | |
| theme=gr.themes.Soft(), | |
| css=""" | |
| .gradio-container { | |
| max-width: 1200px !important; | |
| } | |
| .main-header { | |
| text-align: center; | |
| margin-bottom: 30px; | |
| } | |
| .analysis-output { | |
| max-height: 600px; | |
| overflow-y: auto; | |
| } | |
| """ | |
| ) as interface: | |
| gr.HTML(""" | |
| <div class="main-header"> | |
| <h1>🎥 MCP Video Analysis with Claude AI</h1> | |
| <p>Intelligent video content analysis powered by Modal backend and Anthropic Claude</p> | |
| </div> | |
| """) | |
| with gr.Tab("🔍 Video Analysis"): | |
| with gr.Row(): | |
| with gr.Column(scale=1): | |
| video_url_input = gr.Textbox( | |
| label="Video URL", | |
| placeholder="Enter YouTube URL or direct video link...", | |
| lines=2 | |
| ) | |
| user_query_input = gr.Textbox( | |
| label="Specific Question (Optional)", | |
| placeholder="Ask a specific question about the video...", | |
| lines=2 | |
| ) | |
| with gr.Row(): | |
| analyze_btn = gr.Button("🚀 Analyze Video", variant="primary", size="lg") | |
| clear_btn = gr.Button("🗑️ Clear", variant="secondary") | |
| with gr.Column(scale=2): | |
| claude_output = gr.Textbox( | |
| label="🤖 Claude AI Insights", | |
| lines=20, | |
| elem_classes=["analysis-output"], | |
| interactive=False | |
| ) | |
| with gr.Row(): | |
| raw_analysis_output = gr.JSON( | |
| label="📊 Raw Analysis Data", | |
| elem_classes=["analysis-output"] | |
| ) | |
| # Example videos | |
| gr.HTML("<h3>📝 Example Videos to Try:</h3>") | |
| with gr.Row(): | |
| example_urls = [ | |
| "https://www.youtube.com/watch?v=dQw4w9WgXcQ", | |
| "https://www.youtube.com/watch?v=jNQXAC9IVRw", | |
| "https://www.youtube.com/watch?v=9bZkp7q19f0" | |
| ] | |
| for i, url in enumerate(example_urls, 1): | |
| gr.Button(f"Example {i}", size="sm").click( | |
| lambda url=url: url, outputs=video_url_input | |
| ) | |
| with gr.Tab("ℹ️ About"): | |
| gr.Markdown(""" | |
| ## About MCP Video Analysis | |
| This application combines multiple AI technologies to provide comprehensive video analysis: | |
| ### 🔧 Technology Stack | |
| - **Modal Backend**: Scalable cloud compute for video processing | |
| - **Whisper**: Speech-to-text transcription | |
| - **Computer Vision Models**: Object detection, action recognition, and captioning | |
| - **Anthropic Claude**: Advanced AI for intelligent content analysis | |
| - **MCP Protocol**: Model Context Protocol for seamless integration | |
| ### 🎯 Features | |
| - **Transcription**: Extract spoken content from videos | |
| - **Visual Analysis**: Identify objects, actions, and scenes | |
| - **Content Understanding**: AI-powered insights and summaries | |
| - **Custom Queries**: Ask specific questions about video content | |
| ### 🚀 Usage | |
| 1. Enter a video URL (YouTube or direct link) | |
| 2. Optionally ask a specific question | |
| 3. Click "Analyze Video" to get comprehensive insights | |
| 4. Review both Claude's intelligent analysis and raw data | |
| ### 🔒 Privacy & Security | |
| - Video processing is handled securely in the cloud | |
| - No video data is stored permanently | |
| - API keys are handled securely via environment variables | |
| """) | |
| # Event handlers | |
| def clear_all(): | |
| return "", "", "", "" | |
| analyze_btn.click( | |
| fn=analyze_video_interface, | |
| inputs=[video_url_input, user_query_input], | |
| outputs=[claude_output, raw_analysis_output], | |
| show_progress=True | |
| ) | |
| clear_btn.click( | |
| fn=clear_all, | |
| outputs=[video_url_input, user_query_input, claude_output, raw_analysis_output] | |
| ) | |
| return interface | |
| # Create and launch the interface | |
| if __name__ == "__main__": | |
| interface = create_gradio_interface() | |
| interface.launch( | |
| server_name="0.0.0.0", | |
| server_port=7860, | |
| share=False, | |
| show_error=True | |
| ) | |