ladybug11's picture
add logging
dee4b1a
raw
history blame
30.3 kB
import gradio as gr
import os
import requests
import random
import tempfile
from openai import OpenAI
from smolagents import CodeAgent, MCPClient, tool
from huggingface_hub import InferenceClient
from moviepy.editor import VideoFileClip, ImageClip, CompositeVideoClip, AudioFileClip
from PIL import Image, ImageDraw, ImageFont
import textwrap
import numpy as np
from elevenlabs import ElevenLabs, VoiceSettings
# Initialize clients
openai_client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))
PEXELS_API_KEY = os.getenv("PEXELS_API_KEY")
elevenlabs_client = ElevenLabs(api_key=os.getenv("ELEVENLABS_API_KEY"))
# Initialize MCP Client (connecting to existing MCP server)
try:
mcp_client = MCPClient("https://abidlabs-mcp-tools.hf.space")
mcp_enabled = True
except Exception as e:
print(f"MCP initialization warning: {e}")
mcp_enabled = False
# Define custom tools for the MCP agent
@tool
def generate_quote_tool(niche: str, style: str) -> str:
"""
Generate a powerful inspirational quote using OpenAI.
Args:
niche: The category of quote (Motivation, Business, Fitness, etc.)
style: The visual style (Cinematic, Nature, Urban, Minimal, Abstract)
Returns:
A powerful quote string
"""
prompt = f"""Generate a powerful {niche} quote suitable for an Instagram/TikTok video.
Style: {style}
Requirements:
- 2-4 sentences (can be longer)
- Inspirational and impactful
- Deep and meaningful
- Should resonate deeply with viewers
Return ONLY the quote text, nothing else."""
try:
response = openai_client.chat.completions.create(
model="gpt-4o-mini",
messages=[
{"role": "system", "content": "You are a quote generator for social media content."},
{"role": "user", "content": prompt}
],
max_tokens=150,
temperature=0.8
)
quote = response.choices[0].message.content.strip()
quote = quote.strip('"').strip("'")
return quote
except Exception as e:
return f"Error generating quote: {str(e)}"
@tool
def search_pexels_video_tool(style: str, niche: str) -> dict:
"""
Search and fetch a matching video from Pexels based on style and niche.
Args:
style: Visual style (Cinematic, Nature, Urban, Minimal, Abstract)
niche: Content niche (Motivation, Business, Fitness, etc.)
Returns:
Dictionary with video_url, search_query, and pexels_url
"""
# Intelligent search strategy mapping
search_strategies = {
"Motivation": {
"Cinematic": ["person climbing mountain", "running sunrise", "achievement success"],
"Nature": ["sunrise mountain peak", "ocean waves powerful", "forest light"],
"Urban": ["city skyline dawn", "person running city", "urban success"],
"Minimal": ["minimal motivation", "single person silhouette", "clean inspiring"],
"Abstract": ["light rays hope", "particles rising", "abstract energy"]
},
"Business/Entrepreneurship": {
"Cinematic": ["business cityscape", "office modern", "handshake deal"],
"Nature": ["growth plant", "river flowing", "sunrise new beginning"],
"Urban": ["city business", "office skyline", "modern workspace"],
"Minimal": ["desk minimal", "workspace clean", "simple office"],
"Abstract": ["network connections", "growth chart", "abstract progress"]
},
"Fitness": {
"Cinematic": ["athlete training", "gym workout", "running outdoor"],
"Nature": ["outdoor workout", "mountain hiking", "beach running"],
"Urban": ["city running", "urban fitness", "street workout"],
"Minimal": ["gym minimal", "simple workout", "clean fitness"],
"Abstract": ["energy motion", "strength power", "dynamic movement"]
},
"Mindfulness": {
"Cinematic": ["meditation sunset", "peaceful landscape", "calm water"],
"Nature": ["forest peaceful", "calm lake", "zen garden"],
"Urban": ["city peaceful morning", "quiet street", "urban calm"],
"Minimal": ["minimal zen", "simple meditation", "clean peaceful"],
"Abstract": ["calm waves", "gentle motion", "soft particles"]
},
"Stoicism": {
"Cinematic": ["ancient architecture", "statue philosopher", "timeless landscape"],
"Nature": ["mountain strong", "oak tree", "stone nature"],
"Urban": ["classical building", "statue city", "ancient modern"],
"Minimal": ["stone minimal", "simple strong", "pillar minimal"],
"Abstract": ["marble texture", "stone abstract", "timeless pattern"]
},
"Leadership": {
"Cinematic": ["team meeting", "leader speaking", "group collaboration"],
"Nature": ["eagle flying", "lion pride", "mountain top"],
"Urban": ["office leadership", "boardroom", "city leadership"],
"Minimal": ["chess pieces", "simple leadership", "clean professional"],
"Abstract": ["network leader", "connection points", "guiding light"]
},
"Love & Relationships": {
"Cinematic": ["couple sunset", "romance beautiful", "love cinematic"],
"Nature": ["couple nature", "romantic sunset", "peaceful together"],
"Urban": ["couple city", "romance urban", "love city lights"],
"Minimal": ["hands holding", "simple love", "minimal romance"],
"Abstract": ["hearts flowing", "love particles", "connection abstract"]
}
}
# Get queries for this niche + style combination
queries = search_strategies.get(niche, {}).get(style, ["aesthetic nature"])
try:
headers = {"Authorization": PEXELS_API_KEY}
# Pick a random query for variety
query = random.choice(queries)
url = f"https://api.pexels.com/videos/search?query={query}&per_page=15&orientation=portrait"
response = requests.get(url, headers=headers)
data = response.json()
if "videos" in data and len(data["videos"]) > 0:
# Pick a random video from results
video = random.choice(data["videos"][:10])
video_files = video.get("video_files", [])
# Find portrait/vertical video
portrait_videos = [vf for vf in video_files if vf.get("width", 0) < vf.get("height", 0)]
if portrait_videos:
selected = random.choice(portrait_videos)
return {
"video_url": selected.get("link"),
"search_query": query,
"pexels_url": video.get("url"),
"success": True
}
# Fallback to any HD video
if video_files:
return {
"video_url": video_files[0].get("link"),
"search_query": query,
"pexels_url": video.get("url"),
"success": True
}
return {
"video_url": None,
"search_query": query,
"pexels_url": None,
"success": False,
"error": "No suitable videos found"
}
except Exception as e:
return {
"video_url": None,
"search_query": "",
"pexels_url": None,
"success": False,
"error": str(e)
}
@tool
def generate_voice_narration_tool(quote_text: str, output_path: str) -> dict:
"""
Generate voice narration for the quote using ElevenLabs.
Args:
quote_text: The quote text to narrate
output_path: Path where to save the audio file
Returns:
Dictionary with success status and output path
"""
try:
# Generate audio using ElevenLabs
audio = elevenlabs_client.text_to_speech.convert(
text=quote_text,
voice_id="pNInz6obpgDQGcFmaJgB", # Adam voice - clear and motivational
model_id="eleven_multilingual_v2",
voice_settings=VoiceSettings(
stability=0.5,
similarity_boost=0.75,
style=0.5,
use_speaker_boost=True
)
)
# Save audio to file
with open(output_path, 'wb') as f:
for chunk in audio:
f.write(chunk)
return {
"success": True,
"output_path": output_path,
"message": "Voice narration created successfully!"
}
except Exception as e:
return {
"success": False,
"output_path": None,
"message": f"Error creating voice: {str(e)}"
}
@tool
def create_quote_video_tool(video_url: str, quote_text: str, output_path: str, audio_path: str = None) -> dict:
"""
Create a final quote video by overlaying text on the background video.
Uses Modal for fast processing (4-8x faster) with local fallback.
Optionally adds voice narration audio.
Args:
video_url: URL of the background video from Pexels
quote_text: The quote text to overlay
output_path: Path where to save the final video
audio_path: Optional path to audio file for voice narration
Returns:
Dictionary with success status and output path
"""
# Check if Modal is configured
modal_endpoint = os.getenv("MODAL_ENDPOINT_URL")
if modal_endpoint:
try:
import requests
import base64
print("πŸš€ Processing on Modal (fast!)...")
# For now, skip audio in Modal (would need to upload to cloud storage)
# We'll process without audio for speed
audio_url = None
# Call Modal endpoint
response = requests.post(
modal_endpoint,
json={
"video_url": video_url,
"quote_text": quote_text,
"audio_url": audio_url
},
timeout=120
)
if response.status_code == 200:
result = response.json()
if result.get("success"):
# Decode video bytes
video_b64 = result["video"]
video_bytes = base64.b64decode(video_b64)
# Save to output path
with open(output_path, 'wb') as f:
f.write(video_bytes)
print(f"βœ… Modal processing complete! {result['size_mb']:.2f}MB")
return {
"success": True,
"output_path": output_path,
"message": f"Video created via Modal in ~20s ({result['size_mb']:.2f}MB)"
}
# If Modal failed, fall through to local processing
print("⚠️ Modal failed, falling back to local processing...")
except Exception as e:
print(f"⚠️ Modal error: {e}, falling back to local processing...")
# LOCAL PROCESSING (Fallback or if Modal not configured)
print("πŸ”§ Processing locally...")
try:
# Step 1: Download the video
response = requests.get(video_url, stream=True, timeout=30)
response.raise_for_status()
# Create temporary file for downloaded video
temp_video = tempfile.NamedTemporaryFile(delete=False, suffix='.mp4')
with open(temp_video.name, 'wb') as f:
for chunk in response.iter_content(chunk_size=8192):
f.write(chunk)
# Step 2: Load video with MoviePy
video = VideoFileClip(temp_video.name)
# Get video dimensions
w, h = video.size
# Step 3: Create text overlay using PIL
def make_text_frame(t):
"""Generate a text frame using PIL"""
# Create transparent image
img = Image.new('RGBA', (w, h), (0, 0, 0, 0))
draw = ImageDraw.Draw(img)
# Calculate font size (2.5% of video height - smaller for better aesthetic)
font_size = int(h * 0.025)
# Try to load a font, fall back to default if needed
try:
# Try common fonts available on Linux
font = ImageFont.truetype("/usr/share/fonts/truetype/dejavu/DejaVuSans-Bold.ttf", font_size)
except:
try:
font = ImageFont.truetype("/usr/share/fonts/truetype/liberation/LiberationSans-Bold.ttf", font_size)
except:
# Fall back to default font
font = ImageFont.load_default()
# Wrap text to fit width (60% of video width for better proportions)
max_width = int(w * 0.6)
# Manual text wrapping with better line length
words = quote_text.split()
lines = []
current_line = []
for word in words:
test_line = ' '.join(current_line + [word])
# Get text bbox to check width
bbox = draw.textbbox((0, 0), test_line, font=font)
text_width = bbox[2] - bbox[0]
if text_width <= max_width:
current_line.append(word)
else:
if current_line:
lines.append(' '.join(current_line))
current_line = [word]
else:
lines.append(word)
if current_line:
lines.append(' '.join(current_line))
# Calculate total text height with better line spacing
line_spacing = int(font_size * 0.4)
text_block_height = len(lines) * (font_size + line_spacing)
# Start y position (centered vertically)
y = (h - text_block_height) // 2
# Draw each line centered
for line in lines:
# Get text size
bbox = draw.textbbox((0, 0), line, font=font)
text_width = bbox[2] - bbox[0]
# Center horizontally
x = (w - text_width) // 2
# Draw black outline (stroke) - thinner for smaller text
outline_width = max(2, int(font_size * 0.08))
for adj_x in range(-outline_width, outline_width + 1):
for adj_y in range(-outline_width, outline_width + 1):
draw.text((x + adj_x, y + adj_y), line, font=font, fill='black')
# Draw white text on top
draw.text((x, y), line, font=font, fill='white')
y += font_size + line_spacing
return np.array(img)
# Step 4: Create text clip from function
text_clip = ImageClip(make_text_frame(0), duration=video.duration)
# Step 5: Composite video with text
final_video = CompositeVideoClip([video, text_clip])
# Step 5.5: Add voice narration if provided
if audio_path and os.path.exists(audio_path):
try:
audio_clip = AudioFileClip(audio_path)
# Use the shorter duration between video and audio
audio_duration = min(audio_clip.duration, final_video.duration)
audio_clip = audio_clip.subclip(0, audio_duration)
final_video = final_video.set_audio(audio_clip)
except Exception as audio_error:
print(f"Warning: Could not add audio: {audio_error}")
# Step 6: Export final video
final_video.write_videofile(
output_path,
codec='libx264',
audio_codec='aac',
temp_audiofile='temp-audio.m4a',
remove_temp=True,
fps=24,
preset='ultrafast', # Faster encoding
threads=4
)
# Cleanup
video.close()
final_video.close()
os.unlink(temp_video.name)
return {
"success": True,
"output_path": output_path,
"message": "Video created successfully!"
}
except Exception as e:
return {
"success": False,
"output_path": None,
"message": f"Error creating video: {str(e)}"
}
# Initialize the MCP-powered agent
def initialize_agent():
"""Initialize the CodeAgent with MCP capabilities"""
try:
# Use Hugging Face Inference API for the agent's LLM
model = InferenceClient(token=os.getenv("HF_TOKEN"))
# Create agent with custom tools
agent = CodeAgent(
tools=[generate_quote_tool, search_pexels_video_tool, generate_voice_narration_tool, create_quote_video_tool],
model=model,
additional_authorized_imports=["requests", "openai", "random", "tempfile", "os"],
max_steps=15
)
# Add MCP client if available
if mcp_enabled:
agent.mcp_clients = [mcp_client]
return agent, None
except Exception as e:
return None, f"Agent initialization error: {str(e)}"
# Initialize agent
agent, agent_error = initialize_agent()
def mcp_agent_pipeline(niche, style, num_variations=3, add_voice=True):
"""
MCP-POWERED AUTONOMOUS AGENT PIPELINE
Uses smolagents with proper MCP server integration
Generates multiple video variations with optional voice narration
"""
status_log = []
status_log.append("πŸ€– **MCP AGENT STARTING**\n")
if agent_error:
status_log.append(f"❌ Agent initialization failed: {agent_error}")
status_log.append("\nπŸ”„ Falling back to direct tool execution...\n")
return fallback_pipeline(niche, style, num_variations, add_voice)
try:
# STEP 1: Agent receives task
status_log.append("πŸ“‹ **TASK RECEIVED:**")
status_log.append(f" β†’ Generate {niche} quote with {style} aesthetic")
status_log.append(f" β†’ Create {num_variations} video variations")
if add_voice:
status_log.append(f" β†’ Add voice narration with ElevenLabs")
status_log.append("")
# STEP 2: Agent executes quote generation
status_log.append("🧠 **MCP TOOL: generate_quote_tool**")
quote = generate_quote_tool(niche, style)
if "Error" in quote:
return "\n".join(status_log) + f"\n❌ Failed: {quote}", None, []
status_log.append(f" βœ… Generated: \"{quote[:100]}...\"" if len(quote) > 100 else f" βœ… Generated: \"{quote}\"\n")
# STEP 3: Generate voice narration if requested
audio_path = None
if add_voice:
status_log.append("🎀 **MCP TOOL: generate_voice_narration_tool**")
status_log.append(" ⏳ Creating AI voice narration...")
audio_dir = "/tmp/quote_audio"
os.makedirs(audio_dir, exist_ok=True)
import time
audio_filename = f"narration_{int(time.time())}.mp3"
audio_path = os.path.join(audio_dir, audio_filename)
voice_result = generate_voice_narration_tool(quote, audio_path)
if voice_result["success"]:
status_log.append(f" βœ… Voice narration created!\n")
else:
status_log.append(f" ⚠️ Voice creation failed, continuing without audio\n")
audio_path = None
# STEP 4: Search for multiple videos
status_log.append(f"πŸ” **MCP TOOL: search_pexels_video_tool (x{num_variations})**")
status_log.append(f" ⏳ Finding {num_variations} different videos...")
video_results = []
for i in range(num_variations):
video_result = search_pexels_video_tool(style, niche)
if video_result["success"]:
video_results.append(video_result)
status_log.append(f" βœ… Video {i+1}: {video_result['search_query']}")
if not video_results:
return "\n".join(status_log) + "\n❌ No videos found", None, []
status_log.append("")
# STEP 5: Create multiple video variations
status_log.append(f"🎬 **MCP TOOL: create_quote_video_tool (x{len(video_results)})**")
status_log.append(f" ⏳ Creating {len(video_results)} video variations...")
output_dir = "/tmp/quote_videos"
os.makedirs(output_dir, exist_ok=True)
created_videos = []
import time
timestamp = int(time.time())
for i, video_result in enumerate(video_results):
output_filename = f"quote_video_v{i+1}_{timestamp}.mp4"
output_path = os.path.join(output_dir, output_filename)
creation_result = create_quote_video_tool(
video_result["video_url"],
quote,
output_path,
audio_path if add_voice else None
)
if creation_result["success"]:
created_videos.append(creation_result["output_path"])
status_log.append(f" βœ… Variation {i+1} created!")
else:
error_msg = creation_result.get("message", "Unknown error")
status_log.append(f" ⚠️ Variation {i+1} failed: {error_msg}")
if not created_videos:
status_log.append("\n❌ All video creations failed")
return "\n".join(status_log), video_results[0]["video_url"] if video_results else None, []
status_log.append("")
# STEP 6: MCP Server integration status
status_log.append("πŸ”— **MCP SERVER STATUS:**")
if mcp_enabled:
status_log.append(" βœ… Connected to: abidlabs-mcp-tools.hf.space")
else:
status_log.append(" ⚠️ MCP server connection pending")
status_log.append("")
# STEP 7: Success!
status_log.append("✨ **PIPELINE COMPLETE!**")
status_log.append(f" 🎬 Created {len(created_videos)} video variations")
if add_voice:
status_log.append(f" 🎀 With AI voice narration")
status_log.append(f" πŸ“₯ Choose your favorite and download!")
final_status = "\n".join(status_log)
return final_status, video_results[0]["video_url"] if video_results else None, created_videos
except Exception as e:
status_log.append(f"\n❌ Pipeline error: {str(e)}")
return "\n".join(status_log), None, []
def fallback_pipeline(niche, style, num_variations=3, add_voice=True):
"""Fallback pipeline if MCP agent fails"""
status_log = []
status_log.append("πŸ”„ **FALLBACK MODE (Direct Tool Execution)**\n")
# Generate quote
status_log.append("🧠 Generating quote...")
quote = generate_quote_tool(niche, style)
if "Error" in quote:
return "\n".join(status_log) + f"\n❌ {quote}", None, []
status_log.append(f" βœ… Quote generated\n")
# Generate voice if requested
audio_path = None
if add_voice:
status_log.append("🎀 Creating voice narration...")
audio_dir = "/tmp/quote_audio"
os.makedirs(audio_dir, exist_ok=True)
import time
audio_filename = f"narration_{int(time.time())}.mp3"
audio_path = os.path.join(audio_dir, audio_filename)
voice_result = generate_voice_narration_tool(quote, audio_path)
if voice_result["success"]:
status_log.append(f" βœ… Voice created\n")
else:
audio_path = None
status_log.append(f" ⚠️ Voice failed\n")
# Search videos
status_log.append(f"πŸ” Searching for {num_variations} videos...")
video_results = []
for i in range(num_variations):
video_result = search_pexels_video_tool(style, niche)
if video_result["success"]:
video_results.append(video_result)
if not video_results:
return "\n".join(status_log) + "\n❌ No videos found", None, []
status_log.append(f" βœ… Found {len(video_results)} videos\n")
# Create videos
status_log.append("🎬 Creating videos...")
output_dir = "/tmp/quote_videos"
os.makedirs(output_dir, exist_ok=True)
import time
timestamp = int(time.time())
created_videos = []
for i, video_result in enumerate(video_results):
output_filename = f"quote_video_v{i+1}_{timestamp}.mp4"
output_path = os.path.join(output_dir, output_filename)
creation_result = create_quote_video_tool(
video_result["video_url"],
quote,
output_path,
audio_path if add_voice else None
)
if creation_result["success"]:
created_videos.append(creation_result["output_path"])
else:
error_msg = creation_result.get("message", "Unknown error")
status_log.append(f" ❌ Video {i+1} error: {error_msg}")
if not created_videos:
return "\n".join(status_log) + "\n❌ Video creation failed", video_results[0]["video_url"] if video_results else None, []
status_log.append(f" βœ… Created {len(created_videos)} videos!\n")
status_log.append("🎬 **COMPLETE!**")
return "\n".join(status_log), video_results[0]["video_url"] if video_results else None, created_videos
# Gradio Interface
with gr.Blocks(title="AIQuoteClipGenerator - MCP Edition", theme=gr.themes.Soft()) as demo:
gr.Markdown("""
# 🎬 AIQuoteClipGenerator
### MCP-Powered Autonomous AI Agent with Voice Narration
**MCP Integration Features:**
- πŸ”— **MCP Server:** Connected to smolagents framework
- πŸ› οΈ **4 Custom MCP Tools:** Quote generation + Video search + Voice narration + Video creation
- πŸ€– **Agent Reasoning:** Autonomous task execution
- ⚑ **Tool Orchestration:** Intelligent pipeline management
- 🎀 **ElevenLabs Voice:** AI narration for videos
- 🎨 **Multiple Variations:** Get 3 different video styles
""")
with gr.Row():
with gr.Column():
gr.Markdown("### 🎯 Input")
niche = gr.Dropdown(
choices=[
"Motivation",
"Business/Entrepreneurship",
"Fitness",
"Mindfulness",
"Stoicism",
"Leadership",
"Love & Relationships"
],
label="πŸ“‚ Select Niche",
value="Motivation"
)
style = gr.Dropdown(
choices=[
"Cinematic",
"Nature",
"Urban",
"Minimal",
"Abstract"
],
label="🎨 Visual Style",
value="Cinematic"
)
num_variations = gr.Slider(
minimum=1,
maximum=5,
value=3,
step=1,
label="🎬 Number of Video Variations",
info="Generate multiple versions to choose from"
)
add_voice = gr.Checkbox(
value=False,
label="🎀 Add Voice Narration (ElevenLabs)",
info="AI voice will read the quote (optional)"
)
generate_btn = gr.Button("πŸ€– Run MCP Agent", variant="primary", size="lg")
with gr.Column():
gr.Markdown("### πŸ“Š MCP Agent Activity Log")
output = gr.Textbox(label="Agent Status", lines=20, show_label=False)
with gr.Row():
with gr.Column():
gr.Markdown("### πŸŽ₯ Background Video Preview")
preview_video = gr.Video(label="Original Pexels Video")
with gr.Row():
gr.Markdown("### ✨ Your Quote Videos (Pick Your Favorite!)")
with gr.Row():
video_gallery = gr.Gallery(
label="Video Variations",
show_label=False,
elem_id="gallery",
columns=3,
rows=2,
height="auto",
object_fit="contain"
)
gr.Markdown("""
---
### ✨ NEW FEATURES!
- 🎀 **ElevenLabs Voice Narration** - AI voice reads your quotes
- 🎨 **Multiple Variations** - Get 3-5 different videos to choose from
- βœ… **4 MCP Tools** - Quote, Video Search, Voice, Video Creation
### ✨ MCP Implementation
- βœ… **smolagents Framework** - Proper MCP integration
- βœ… **Custom MCP Tools** - 4 tools working autonomously
- βœ… **CodeAgent** - Autonomous reasoning and execution
- βœ… **MCP Client** - Connected to external MCP servers
- βœ… **MoviePy + PIL** - Professional text overlay
- βœ… **ElevenLabs** - AI voice narration
### πŸ† Hackathon: MCP 1st Birthday
**Track:** Track 2 - MCP in Action
**Category:** Productivity Tools
**Built with:** Gradio + smolagents + OpenAI + Pexels + ElevenLabs + MoviePy + MCP
""")
generate_btn.click(
mcp_agent_pipeline,
inputs=[niche, style, num_variations, add_voice],
outputs=[output, preview_video, video_gallery]
)
if __name__ == "__main__":
demo.launch()