poster_server.py

from mcp.server.fastmcp import FastMCP
from mcp import ClientSession
from mcp.client.sse import sse_client
import httpx
import os
import random
import sys
import json
import asyncio

# Initialize FastMCP server
mcp = FastMCP("PosterAgent - Modal Flux AI")

# Your Modal Flux MCP endpoint with SSE
FLUX_MCP_SSE_URL = "https://abedalswaity7--flux-mcp-app-ui.modal.run/gradio_api/mcp/sse"


def log(msg: str):
    """Log to stderr to avoid interfering with MCP protocol on stdout"""
    print(msg, file=sys.stderr, flush=True)


def estimate_trip_price(destination: str, budget: str, travelers: int, duration_days: int = 7) -> str:
    """
    Estimate trip price based on destination, budget level, and duration.
    Returns formatted price string like "From $1,299" or "$2,500 Per Person"
    """
    # Base prices per person per day by budget level
    budget_multipliers = {
        "budget": (80, 150),      # $80-150/day
        "moderate": (200, 350),   # $200-350/day  
        "luxury": (500, 1000)     # $500-1000/day
    }
    
    # Destination cost factors (relative to average)
    destination_factors = {
        "paris": 1.2,
        "tokyo": 1.3,
        "dubai": 1.5,
        "bali": 0.7,
        "rome": 1.1,
        "turkey": 0.8,
        "india": 0.6,
        "sri lanka": 0.65,
        "maldives": 1.8,
        "santorini": 1.3,
        "new york": 1.4,
        "london": 1.3,
        "thailand": 0.6,
        "vietnam": 0.5,
        "morocco": 0.7,
        "egypt": 0.6,
        "mexico": 0.75,
        "spain": 1.0,
        "greece": 1.0,
        "switzerland": 1.6,
    }
    
    # Get budget range
    budget_lower = budget.lower() if budget else "moderate"
    min_rate, max_rate = budget_multipliers.get(budget_lower, (200, 350))
    
    # Get destination factor
    dest_lower = destination.lower()
    dest_factor = 1.0
    for key, factor in destination_factors.items():
        if key in dest_lower:
            dest_factor = factor
            break
    
    # Calculate price range
    base_price = ((min_rate + max_rate) / 2) * duration_days * dest_factor
    
    # Add flight estimate based on destination
    flight_estimates = {
        "budget": 400,
        "moderate": 800,
        "luxury": 1500
    }
    flight_cost = flight_estimates.get(budget_lower, 800) * dest_factor
    
    total_price = int(base_price + flight_cost)
    
    # Round to nice numbers
    if total_price < 1000:
        total_price = round(total_price / 50) * 50
    else:
        total_price = round(total_price / 100) * 100
    
    # Format based on budget level
    if budget_lower == "luxury":
        return f"${total_price:,} Per Person"
    elif budget_lower == "budget":
        return f"From ${total_price:,}"
    else:
        return f"From ${total_price:,} Per Person"


def create_professional_travel_poster_prompt(
    destination: str,
    origin: str = "",
    dates: str = "",
    duration: str = "",
    price: str = "",
    travelers: int = 2,
    budget: str = "moderate",
    interests: list = None,
    company_name: str = "",
    company_phone: str = "",
    company_email: str = "",
    company_website: str = "",
    inclusions: list = None,
    tagline: str = ""
) -> str:
    """
    Create a professional travel agency poster prompt inspired by real tourism marketing.
    
    Design elements from reference posters:
    - Bold destination name with creative typography
    - Polaroid-style photo collages with red pins
    - Traveler silhouettes looking at horizon
    - Trip details (dates, duration, price)
    - Company branding section
    - Icons for inclusions
    - Gradient sky backgrounds (blue to orange sunset)
    """
    
    # Destination-specific landmarks and visual themes
    destination_visuals = {
        "paris": {
            "landmarks": "Eiffel Tower, Arc de Triomphe, Seine River with bridges, Louvre pyramid",
            "colors": "romantic sunset pink and orange, Parisian blue sky",
            "vibe": "romantic, elegant, city of lights",
            "hero_image": "Eiffel Tower at golden hour with city panorama"
        },
        "tokyo": {
            "landmarks": "Mount Fuji with snow cap, Tokyo Tower, Shibuya crossing, cherry blossoms, traditional temples",
            "colors": "cherry blossom pink, sunset orange, zen white, neon accents",
            "vibe": "blend of ancient tradition and futuristic innovation",
            "hero_image": "Mount Fuji with cherry blossoms in foreground"
        },
        "dubai": {
            "landmarks": "Burj Khalifa, Palm Jumeirah, Dubai Marina yachts, desert dunes, Burj Al Arab",
            "colors": "golden sand, luxury gold accents, Arabian blue sky",
            "vibe": "ultra-luxury, futuristic architecture, desert mystique",
            "hero_image": "Burj Khalifa piercing dramatic sunset sky"
        },
        "bali": {
            "landmarks": "Tanah Lot temple, rice terraces, beach with palm trees, traditional gates, waterfalls",
            "colors": "tropical green, turquoise ocean, sunset coral",
            "vibe": "tropical paradise, spiritual serenity, natural beauty",
            "hero_image": "iconic Bali temple silhouette at sunset over ocean"
        },
        "rome": {
            "landmarks": "Colosseum, Trevi Fountain, St. Peter's Basilica dome, Roman Forum ruins",
            "colors": "warm terracotta, marble white, Mediterranean blue",
            "vibe": "ancient grandeur, timeless history, la dolce vita",
            "hero_image": "majestic Colosseum with dramatic golden hour lighting"
        },
        "turkey": {
            "landmarks": "Blue Mosque Istanbul, Cappadocia hot air balloons, Bodrum harbor, Hagia Sophia",
            "colors": "turquoise blue, Ottoman gold, sunset orange",
            "vibe": "crossroads of civilizations, exotic bazaars, coastal beauty",
            "hero_image": "Blue Mosque with minarets against sunset sky"
        },
        "india": {
            "landmarks": "Taj Mahal, Humayun's Tomb, Kerala backwaters, mountain peaks, colorful markets",
            "colors": "marigold orange, royal purple, peacock blue",
            "vibe": "incredible diversity, ancient wonders, spiritual journey",
            "hero_image": "Taj Mahal at sunrise with reflection pool"
        },
        "sri lanka": {
            "landmarks": "Sigiriya rock fortress, tea plantations, elephants, golden beaches, ancient temples",
            "colors": "lush green, ocean blue, golden sand",
            "vibe": "pearl of Indian Ocean, wildlife, ancient ruins",
            "hero_image": "Sigiriya Lion Rock rising from misty jungle"
        },
        "maldives": {
            "landmarks": "overwater bungalows, crystal lagoon, white sand beach, coral reefs",
            "colors": "turquoise paradise blue, pure white, sunset gold",
            "vibe": "ultimate luxury escape, pristine beaches, underwater paradise",
            "hero_image": "overwater villas on crystal clear turquoise lagoon"
        },
        "santorini": {
            "landmarks": "white-washed buildings, blue domes, caldera view, sunset at Oia",
            "colors": "iconic Greek blue, pristine white, sunset pink",
            "vibe": "romantic Greek island, breathtaking views, Mediterranean charm",
            "hero_image": "famous blue domes overlooking Aegean Sea at sunset"
        },
        "new york": {
            "landmarks": "Statue of Liberty, Empire State Building, Times Square, Brooklyn Bridge, Central Park",
            "colors": "metropolitan silver, taxi yellow, skyline blue",
            "vibe": "city that never sleeps, urban excitement, iconic skyline",
            "hero_image": "Manhattan skyline with Statue of Liberty"
        }
    }
    
    # Find matching destination style
    dest_lower = destination.lower()
    visuals = None
    for key in destination_visuals:
        if key in dest_lower:
            visuals = destination_visuals[key]
            break
    
    if not visuals:
        visuals = {
            "landmarks": f"iconic landmarks and stunning scenery of {destination}",
            "colors": "vibrant travel colors, sunset gradient sky",
            "vibe": "adventure and discovery, wanderlust inspiration",
            "hero_image": f"breathtaking panoramic view of {destination}"
        }
    
    # Build professional travel agency poster prompt
    # Inspired by the reference images with polaroid collages, travelers, bold typography
    
    dest_name = destination.split(",")[0].strip().upper()
    
    # Tagline generation
    if not tagline:
        taglines = [
            f"Explore Your {dest_name} Journey",
            f"Discover {dest_name}",
            f"Experience {dest_name}",
            f"Your {dest_name} Adventure Awaits"
        ]
        tagline = random.choice(taglines)
    
    # Build the professional prompt
    prompt = f"""Professional travel agency advertisement poster for {destination} vacation package.

LAYOUT DESIGN (inspired by luxury travel marketing):
- Clean gradient background transitioning from bright blue sky at top to warm sunset orange/gold at bottom
- Large bold destination name "{dest_name}" displayed prominently in creative 3D typography with shadow effects, positioned in upper third
- Stylized tagline "{tagline}" in elegant script font above the destination name
- Main hero image: {visuals['hero_image']} as the dominant visual centerpiece

PHOTO COLLAGE ELEMENTS:
- 3 polaroid-style photo frames with white borders, slightly tilted at different angles
- Photos showing: {visuals['landmarks']}
- Red circular push-pins attached to each polaroid photo
- Photos arranged in artistic scattered layout

TRAVELERS ELEMENT:
- Silhouette of {travelers} travelers with backpacks and luggage standing on a road/path
- Looking towards the destination horizon, creating sense of adventure and anticipation
- Positioned in lower portion of poster

TRIP INFORMATION SECTION:"""

    # Add trip details if provided
    if dates:
        prompt += f"\n- Travel dates: \"{dates}\" displayed in bold banner style"
    if duration:
        prompt += f"\n- Duration badge showing \"{duration}\" in rounded rectangle"
    
    # Auto-calculate price if not provided
    display_price = price
    if not display_price:
        # Extract duration days for price calculation
        duration_days = 7  # default
        if duration:
            try:
                # Parse "6N 7D" format
                parts = duration.upper().split()
                for p in parts:
                    if 'N' in p:
                        duration_days = int(p.replace('N', '')) + 1
                        break
                    elif 'D' in p:
                        duration_days = int(p.replace('D', ''))
                        break
            except:
                pass
        display_price = estimate_trip_price(destination, budget, travelers, duration_days)
    
    prompt += f"\n- Price display: \"{display_price}\" with 'Per Person' text in highlighted box"
    
    # Add inclusions icons if provided
    if inclusions:
        icons_text = ", ".join(inclusions)
        prompt += f"\n- Row of circular icons with labels for inclusions: {icons_text}"
    else:
        prompt += "\n- Row of circular icons for: Flights, Hotels, Tours & Transfers, Meals"

    # Add company branding if provided
    if company_name or company_phone or company_website:
        prompt += "\n\nCOMPANY BRANDING SECTION (bottom of poster):"
        if company_name:
            prompt += f"\n- Company logo/name: \"{company_name}\" in professional styling"
        if company_phone:
            prompt += f"\n- Contact phone: \"{company_phone}\""
        if company_email:
            prompt += f"\n- Email: \"{company_email}\""
        if company_website:
            prompt += f"\n- Website: \"{company_website}\""
        prompt += "\n- 'Book Now' call-to-action button"

    prompt += f"""

VISUAL STYLE:
- Professional travel agency marketing design
- Color palette: {visuals['colors']}
- Mood: {visuals['vibe']}, inspiring wanderlust
- High-end glossy magazine advertisement quality
- Clean modern typography with excellent readability
- Balanced composition with clear visual hierarchy

TECHNICAL QUALITY:
- Ultra high resolution 4K quality
- Professional graphic design execution
- Photorealistic landmark images
- Clean vector-style graphics for icons and text
- Print-ready commercial quality"""

    return prompt


@mcp.tool()
async def generate_poster_image(
    destination: str,
    origin: str = "",
    dates: str = "",
    duration: str = "",
    price: str = "",
    travelers: int = 2,
    budget: str = "moderate",
    interests: str = "",
    company_name: str = "",
    company_phone: str = "",
    company_email: str = "",
    company_website: str = "",
    inclusions: str = "",
    tagline: str = ""
) -> str:
    """
    Generate a professional travel agency poster using Modal Flux AI.
    Creates marketing-style posters with trip details, pricing, and company branding.
    
    Args:
        destination: The travel destination (e.g., "Paris, France", "Tokyo, Japan").
        origin: Origin city for departure info (e.g., "New York").
        dates: Travel dates to display (e.g., "Dec 28, 2025 - Jan 3, 2026").
        duration: Trip duration (e.g., "6N 7D" for 6 nights 7 days).
        price: Price to display (e.g., "$1,700" or "From $999").
        travelers: Number of travelers (affects silhouette in poster).
        budget: Budget level (budget/moderate/luxury) - affects styling.
        interests: Comma-separated interests (e.g., "culture, food, adventure").
        company_name: Travel company name for branding (e.g., "Wanderlust Travel").
        company_phone: Contact phone number (e.g., "+1 800 555 1234").
        company_email: Contact email (e.g., "[email protected]").
        company_website: Company website (e.g., "www.travel.com").
        inclusions: Comma-separated inclusions (e.g., "Visa, Flights, Hotels, Meals, Tours").
        tagline: Custom tagline (e.g., "Discover Your Journey").
    
    Returns:
        Path to the generated poster image.
    """
    # Parse interests and inclusions
    interests_list = [i.strip() for i in interests.split(",")] if interests else []
    inclusions_list = [i.strip() for i in inclusions.split(",")] if inclusions else None
    
    # Create the professional prompt
    prompt = create_professional_travel_poster_prompt(
        destination=destination,
        origin=origin,
        dates=dates,
        duration=duration,
        price=price,
        travelers=travelers,
        budget=budget,
        interests=interests_list,
        company_name=company_name,
        company_phone=company_phone,
        company_email=company_email,
        company_website=company_website,
        inclusions=inclusions_list,
        tagline=tagline
    )
    
    safe_name = destination.lower().replace(' ', '_').replace(',', '').replace('.', '')
    output_filename = f"poster_{safe_name}_{random.randint(1000, 9999)}.jpg"
    full_path = os.path.abspath(output_filename)
    
    log(f"🎨 Generating PROFESSIONAL travel agency poster for: '{destination}'")
    log(f"   Trip: {dates} | {duration} | {price}")
    log(f"   Company: {company_name or 'Generic'}")
    log(f"   MCP SSE: {FLUX_MCP_SSE_URL}")
    
    # Use 9:16 portrait for social media / print poster style
    aspect_ratio = "9:16 Portrait (768×1360)"
    
    try:
        # Connect to Flux MCP server via SSE
        async with sse_client(FLUX_MCP_SSE_URL) as (read_stream, write_stream):
            async with ClientSession(read_stream, write_stream) as session:
                await session.initialize()
                
                log("📡 Connected to Flux MCP server")
                
                # List available tools
                tools = await session.list_tools()
                tool_name = "generate_flux_image"
                for tool in tools.tools:
                    if "generate" in tool.name.lower():
                        tool_name = tool.name
                        break
                
                log(f"🔧 Calling: {tool_name}")
                log(f"📝 Prompt length: {len(prompt)} chars")
                
                # Call with optimized parameters for quality
                result = await session.call_tool(
                    tool_name,
                    arguments={
                        "prompt": prompt,
                        "aspect_ratio": aspect_ratio,
                        "quality_preset": "✨ Quality (35 steps)",
                        "guidance": 4.5,  # Higher guidance for better prompt following
                        "seed": random.randint(1, 999999)
                    }
                )
                
                log(f"📦 Result received")
                
                # Process the result
                if result.content:
                    for item in result.content:
                        log(f"📄 Processing item type: {type(item).__name__}")
                        
                        # Handle TextContent
                        if hasattr(item, 'text'):
                            text = item.text
                            log(f"📄 Text: {text[:200]}...")
                            
                            if text.startswith("http"):
                                async with httpx.AsyncClient(timeout=60.0, follow_redirects=True) as client:
                                    img_resp = await client.get(text)
                                    with open(full_path, "wb") as f:
                                        f.write(img_resp.content)
                                    log(f"✅ Professional poster saved: {full_path}")
                                    return full_path
                            
                            if "http" in text:
                                import re
                                urls = re.findall(r'https?://[^\s<>"{}|\\^`\[\]]+', text)
                                if urls:
                                    async with httpx.AsyncClient(timeout=60.0, follow_redirects=True) as client:
                                        img_resp = await client.get(urls[0])
                                        with open(full_path, "wb") as f:
                                            f.write(img_resp.content)
                                        log(f"✅ Professional poster saved: {full_path}")
                                        return full_path
                            
                            if text.startswith("/") and os.path.exists(text):
                                return text
                            
                            return text
                        
                        # Handle ImageContent
                        elif hasattr(item, 'data'):
                            data = item.data
                            log(f"📷 Image data type: {type(data)}")
                            
                            if isinstance(data, str):
                                if data.startswith("http"):
                                    async with httpx.AsyncClient(timeout=60.0, follow_redirects=True) as client:
                                        img_resp = await client.get(data)
                                        with open(full_path, "wb") as f:
                                            f.write(img_resp.content)
                                        log(f"✅ Professional poster saved: {full_path}")
                                        return full_path
                                elif data.startswith("data:image"):
                                    import base64
                                    base64_data = data.split(",")[1] if "," in data else data
                                    img_bytes = base64.b64decode(base64_data)
                                    with open(full_path, "wb") as f:
                                        f.write(img_bytes)
                                    log(f"✅ Professional poster saved: {full_path}")
                                    return full_path
                                else:
                                    try:
                                        import base64
                                        img_bytes = base64.b64decode(data)
                                        with open(full_path, "wb") as f:
                                            f.write(img_bytes)
                                        log(f"✅ Professional poster saved: {full_path}")
                                        return full_path
                                    except:
                                        return data
                            else:
                                with open(full_path, "wb") as f:
                                    f.write(data)
                                log(f"✅ Professional poster saved: {full_path}")
                                return full_path
                
                return "Error: No image data in response"
                
    except Exception as e:
        log(f"❌ Error: {e}")
        import traceback
        log(traceback.format_exc())
        return f"Error: {str(e)}"


@mcp.tool()
def get_poster_status() -> str:
    """Check if poster generation service is available."""
    return f"Professional Travel Poster service ready. Creates marketing-style posters with trip details, pricing, and company branding. Using Modal Flux MCP at: {FLUX_MCP_SSE_URL}"


if __name__ == "__main__":
    mcp.run()