|
|
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 |
|
|
|
|
|
|
|
|
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")) |
|
|
|
|
|
|
|
|
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 |
|
|
|
|
|
|
|
|
@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 |
|
|
""" |
|
|
|
|
|
|
|
|
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"] |
|
|
} |
|
|
} |
|
|
|
|
|
|
|
|
queries = search_strategies.get(niche, {}).get(style, ["aesthetic nature"]) |
|
|
|
|
|
try: |
|
|
headers = {"Authorization": PEXELS_API_KEY} |
|
|
|
|
|
|
|
|
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: |
|
|
|
|
|
video = random.choice(data["videos"][:10]) |
|
|
video_files = video.get("video_files", []) |
|
|
|
|
|
|
|
|
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 |
|
|
} |
|
|
|
|
|
|
|
|
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: |
|
|
|
|
|
audio = elevenlabs_client.text_to_speech.convert( |
|
|
text=quote_text, |
|
|
voice_id="pNInz6obpgDQGcFmaJgB", |
|
|
model_id="eleven_multilingual_v2", |
|
|
voice_settings=VoiceSettings( |
|
|
stability=0.5, |
|
|
similarity_boost=0.75, |
|
|
style=0.5, |
|
|
use_speaker_boost=True |
|
|
) |
|
|
) |
|
|
|
|
|
|
|
|
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 |
|
|
""" |
|
|
|
|
|
|
|
|
modal_endpoint = os.getenv("MODAL_ENDPOINT_URL") |
|
|
|
|
|
if modal_endpoint: |
|
|
try: |
|
|
import requests |
|
|
import base64 |
|
|
|
|
|
print("π Processing on Modal (fast!)...") |
|
|
|
|
|
|
|
|
|
|
|
audio_url = None |
|
|
|
|
|
|
|
|
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"): |
|
|
|
|
|
video_b64 = result["video"] |
|
|
video_bytes = base64.b64decode(video_b64) |
|
|
|
|
|
|
|
|
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)" |
|
|
} |
|
|
|
|
|
|
|
|
print("β οΈ Modal failed, falling back to local processing...") |
|
|
|
|
|
except Exception as e: |
|
|
print(f"β οΈ Modal error: {e}, falling back to local processing...") |
|
|
|
|
|
|
|
|
print("π§ Processing locally...") |
|
|
|
|
|
try: |
|
|
|
|
|
response = requests.get(video_url, stream=True, timeout=30) |
|
|
response.raise_for_status() |
|
|
|
|
|
|
|
|
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) |
|
|
|
|
|
|
|
|
video = VideoFileClip(temp_video.name) |
|
|
|
|
|
|
|
|
w, h = video.size |
|
|
|
|
|
|
|
|
def make_text_frame(t): |
|
|
"""Generate a text frame using PIL""" |
|
|
|
|
|
img = Image.new('RGBA', (w, h), (0, 0, 0, 0)) |
|
|
draw = ImageDraw.Draw(img) |
|
|
|
|
|
|
|
|
font_size = int(h * 0.025) |
|
|
|
|
|
|
|
|
try: |
|
|
|
|
|
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: |
|
|
|
|
|
font = ImageFont.load_default() |
|
|
|
|
|
|
|
|
max_width = int(w * 0.6) |
|
|
|
|
|
|
|
|
words = quote_text.split() |
|
|
lines = [] |
|
|
current_line = [] |
|
|
|
|
|
for word in words: |
|
|
test_line = ' '.join(current_line + [word]) |
|
|
|
|
|
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)) |
|
|
|
|
|
|
|
|
line_spacing = int(font_size * 0.4) |
|
|
text_block_height = len(lines) * (font_size + line_spacing) |
|
|
|
|
|
|
|
|
y = (h - text_block_height) // 2 |
|
|
|
|
|
|
|
|
for line in lines: |
|
|
|
|
|
bbox = draw.textbbox((0, 0), line, font=font) |
|
|
text_width = bbox[2] - bbox[0] |
|
|
|
|
|
|
|
|
x = (w - text_width) // 2 |
|
|
|
|
|
|
|
|
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.text((x, y), line, font=font, fill='white') |
|
|
|
|
|
y += font_size + line_spacing |
|
|
|
|
|
return np.array(img) |
|
|
|
|
|
|
|
|
text_clip = ImageClip(make_text_frame(0), duration=video.duration) |
|
|
|
|
|
|
|
|
final_video = CompositeVideoClip([video, text_clip]) |
|
|
|
|
|
|
|
|
if audio_path and os.path.exists(audio_path): |
|
|
try: |
|
|
audio_clip = AudioFileClip(audio_path) |
|
|
|
|
|
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}") |
|
|
|
|
|
|
|
|
final_video.write_videofile( |
|
|
output_path, |
|
|
codec='libx264', |
|
|
audio_codec='aac', |
|
|
temp_audiofile='temp-audio.m4a', |
|
|
remove_temp=True, |
|
|
fps=24, |
|
|
preset='ultrafast', |
|
|
threads=4 |
|
|
) |
|
|
|
|
|
|
|
|
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)}" |
|
|
} |
|
|
|
|
|
|
|
|
def initialize_agent(): |
|
|
"""Initialize the CodeAgent with MCP capabilities""" |
|
|
try: |
|
|
|
|
|
model = InferenceClient(token=os.getenv("HF_TOKEN")) |
|
|
|
|
|
|
|
|
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 |
|
|
) |
|
|
|
|
|
|
|
|
if mcp_enabled: |
|
|
agent.mcp_clients = [mcp_client] |
|
|
|
|
|
return agent, None |
|
|
except Exception as e: |
|
|
return None, f"Agent initialization error: {str(e)}" |
|
|
|
|
|
|
|
|
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: |
|
|
|
|
|
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("") |
|
|
|
|
|
|
|
|
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") |
|
|
|
|
|
|
|
|
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 |
|
|
|
|
|
|
|
|
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("") |
|
|
|
|
|
|
|
|
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("") |
|
|
|
|
|
|
|
|
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("") |
|
|
|
|
|
|
|
|
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") |
|
|
|
|
|
|
|
|
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") |
|
|
|
|
|
|
|
|
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") |
|
|
|
|
|
|
|
|
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") |
|
|
|
|
|
|
|
|
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 |
|
|
|
|
|
|
|
|
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() |
|
|
|