#!/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("""

đŸŽĨ MCP Video Analysis with Claude AI

Intelligent video content analysis powered by Modal backend and Anthropic Claude

""") 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("

📝 Example Videos to Try:

") 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 )