Slidespeak

from typing import Any, Optional, Literal
import httpx
import os
import time
import asyncio
import logging
import json
from mcp.server.fastmcp import FastMCP

# --- Configuration & Constants ---

# Initialize FastMCP server
mcp = FastMCP("slidespeak")

# API Configuration
API_BASE = "https://api.slidespeak.co/api/v1"
USER_AGENT = "slidespeak-mcp/0.0.3"
API_KEY = os.environ.get('SLIDESPEAK_API_KEY')

if not API_KEY:
    logging.warning("SLIDESPEAK_API_KEY environment variable not set.")

# Default Timeouts
DEFAULT_TIMEOUT = 30.0
GENERATION_TIMEOUT = 90.0 # Total time allowed for generation + polling
POLLING_INTERVAL = 2.0 # Seconds between status checks
POLLING_TIMEOUT = 10.0 # Timeout for each individual status check request

async def _make_api_request(
    method: Literal["GET", "POST"],
    endpoint: str,
    payload: Optional[dict[str, Any]] = None,
    timeout: float = DEFAULT_TIMEOUT
) -> Optional[dict[str, Any]]:
    """
    Makes an HTTP request to the SlideSpeak API.

    Args:
        method: HTTP method ('GET' or 'POST').
        endpoint: API endpoint path (e.g., '/presentation/templates').
        payload: JSON payload for POST requests. Ignored for GET.
        timeout: Request timeout in seconds.

    Returns:
        The parsed JSON response as a dictionary on success, None on failure.
    """
    if not API_KEY:
        logging.error("API Key is missing. Cannot make API request.")
        return None

    headers = {
        "User-Agent": USER_AGENT,
        "Accept": "application/json",
        "X-API-Key": API_KEY,
    }

    # Construct full URL
    url = f"{API_BASE}{endpoint}"

    async with httpx.AsyncClient() as client:
        try:
            if method == "POST":
                response = await client.post(url, json=payload, headers=headers, timeout=timeout)
            else: # Default to GET
                response = await client.get(url, headers=headers, timeout=timeout)

            response.raise_for_status() # Raise exception for 4xx or 5xx status codes
            return response.json()

        except httpx.HTTPStatusError as e:
            logging.error(f"HTTP error calling {method} {url}: {e.response.status_code} - {e.response.text}")
        except httpx.RequestError as e:
            logging.error(f"Request error calling {method} {url}: {str(e)}")
        except Exception as e:
            logging.error(f"An unexpected error occurred calling {method} {url}: {str(e)}")

        return None

@mcp.tool()
async def get_available_templates() -> str:
    """Get all available presentation templates."""
    templates_endpoint = "/presentation/templates"

    if not API_KEY:
        return "API Key is missing. Cannot process any requests."

    templates_data = await _make_api_request("GET", templates_endpoint)

    if not templates_data:
        return "Unable to fetch templates due to an API error. Check server logs."

    if not isinstance(templates_data, list):
         return f"Unexpected response format received for templates: {type(templates_data).__name__}"

    if not templates_data:
        return "No templates available."

    formatted_templates = "Available templates:\n"
    for template in templates_data:
        # Add more robust checking for expected keys
        name = template.get("name", "default")
        images = template.get("images", {})
        cover = images.get("cover", "No cover image URL")
        content = images.get("content", "No content image URL")
        formatted_templates += f"- {name}\n  Cover: {cover}\n  Content: {content}\n\n"

    return formatted_templates.strip()

@mcp.tool()
async def get_me() -> str:
    """
    Get details about the current API key (user_name and remaining credits).
    """
    if not API_KEY:
        return "API Key is missing. Cannot process any requests."

    result = await _make_api_request("GET", "/me")
    if not result:
        return "Failed to fetch current user details."
    return json.dumps(result) + "\n Note: Generating slides costs 1 credit / slide"

@mcp.tool()
async def generate_powerpoint(
    plain_text: str,
    length: int,
    template: str,
    document_uuids: Optional[list[str]] = None,
    *,
    language: Optional[str] = "ORIGINAL",
    fetch_images: Optional[bool] = True,
    use_document_images: Optional[bool] = False,
    tone: Optional[Literal['default','casual','professional','funny','educational','sales_pitch']] = 'default',
    verbosity: Optional[Literal['concise','standard','text-heavy']] = 'standard',
    custom_user_instructions: Optional[str] = None,
    include_cover: Optional[bool] = True,
    include_table_of_contents: Optional[bool] = True,
    add_speaker_notes: Optional[bool] = False,
    use_general_knowledge: Optional[bool] = False,
    use_wording_from_document: Optional[bool] = False,
    response_format: Optional[Literal['powerpoint','pdf']] = 'powerpoint',
    use_branding_logo: Optional[bool] = False,
    use_branding_fonts: Optional[bool] = False,
    use_branding_color: Optional[bool] = False,
    branding_logo: Optional[str] = None,
    branding_fonts: Optional[dict[str, str]] = None,
) -> str:
    """
    Generate a PowerPoint or PDF presentation based on text, length, and template.
    Supports optional settings (language, tone, verbosity, images, structure, etc.).
    Waits up to a configured time for the result.

    Parameters:
    Required:
    - plain_text (str): The topic to generate a presentation about
    - length (int): The number of slides
    - template (str): Template name or ID

    Optional:
    - document_uuids (list[str]): UUIDs of uploaded documents to use
    - language (str): Language code (default: 'ORIGINAL')
    - fetch_images (bool): Include stock images (default: True)
    - use_document_images (bool): Include images from documents (default: False)
    - tone (str): Text tone - 'default', 'casual', 'professional', 'funny', 'educational', 'sales_pitch' (default: 'default')
    - verbosity (str): Text length - 'concise', 'standard', 'text-heavy' (default: 'standard')
    - custom_user_instructions (str): Custom generation instructions
    - include_cover (bool): Include cover slide (default: True)
    - include_table_of_contents (bool): Include TOC slides (default: True)
    - add_speaker_notes (bool): Add speaker notes (default: False)
    - use_general_knowledge (bool): Expand with related info (default: False)
    - use_wording_from_document (bool): Use document wording (default: False)
    - response_format (str): 'powerpoint' or 'pdf' (default: 'powerpoint')
    - use_branding_logo (bool): Include brand logo (default: False)
    - use_branding_fonts (bool): Apply brand fonts (default: False)
    - use_branding_color (bool): Apply brand colors (default: False)
    - branding_logo (str): Custom logo URL
    - branding_fonts (dict): The object of brand fonts to be used in the slides
    """
    generation_endpoint = "/presentation/generate"
    status_endpoint_base = "/task_status" # Base path for status checks

    if not API_KEY:
        return "API Key is missing. Cannot process any requests."

    # Prepare the JSON body for the generation request
    # Validate cross-field requirements
    if (use_document_images or use_wording_from_document) and not document_uuids:
        return (
            "When use_document_images or use_wording_from_document is true, you must provide document_uuids."
        )

    payload: dict[str, Any] = {
        "plain_text": plain_text,
        "length": length,
        "template": template,
    }
    if document_uuids:
        payload["document_uuids"] = document_uuids
    if language:
        payload["language"] = language
    if fetch_images:
        payload["fetch_images"] = fetch_images
    if use_document_images:
        payload["use_document_images"] = use_document_images
    if tone:
        payload["tone"] = tone
    if verbosity:
        payload["verbosity"] = verbosity
    if custom_user_instructions is not None and custom_user_instructions.strip():
        payload["custom_user_instructions"] = custom_user_instructions
    if include_cover:
        payload["include_cover"] = include_cover
    if include_table_of_contents:
        payload["include_table_of_contents"] = include_table_of_contents
    if add_speaker_notes:
        payload["add_speaker_notes"] = add_speaker_notes
    if use_general_knowledge:
        payload["use_general_knowledge"] = use_general_knowledge
    if use_wording_from_document:
        payload["use_wording_from_document"] = use_wording_from_document
    if response_format:
        payload["response_format"] = response_format
    if use_branding_logo:
        payload["use_branding_logo"] = use_branding_logo
    if use_branding_fonts:
        payload["use_branding_fonts"] = use_branding_fonts
    if use_branding_color:
        payload["use_branding_color"] = use_branding_color
    if branding_logo:
        payload["branding_logo"] = branding_logo
    if branding_fonts:
        payload["branding_fonts"] = branding_fonts

    # Step 1: Initiate generation (POST request)
    init_result = await _make_api_request("POST", generation_endpoint, payload=payload, timeout=GENERATION_TIMEOUT)

    if not init_result:
        return "Failed to initiate PowerPoint generation due to an API error. Check server logs."

    task_id = init_result.get("task_id")
    if not task_id:
        return f"Failed to initiate PowerPoint generation. API response did not contain a task ID. Response: {init_result}"

    logging.info(f"PowerPoint generation initiated. Task ID: {task_id}")

    # Step 2: Poll for the task status
    status_endpoint = f"{status_endpoint_base}/{task_id}"
    start_time = time.time()
    final_result = None

    while time.time() - start_time < GENERATION_TIMEOUT:
        logging.debug(f"Polling status for task {task_id}...")
        status_result = await _make_api_request("GET", status_endpoint, timeout=POLLING_TIMEOUT)

        if status_result:
            task_status = status_result.get("task_status")
            task_result = status_result.get("task_result") # Assuming result might be here

            if task_status == "SUCCESS":
                logging.info(f"Task {task_id} completed successfully.")
                # Prefer task_result if available, otherwise return the whole status dict as string
                final_result = str(task_result) if task_result else str(status_result)

                final_result = f"Make sure to return the pptx url to the user if available. Here is the result: {final_result}"
                break
            elif task_status == "FAILED": # Use 'FAILED' consistently if possible in API
                logging.error(f"Task {task_id} failed. Status response: {status_result}")
                error_message = task_result.get("error", "Unknown error") if isinstance(task_result, dict) else "Unknown error"
                final_result = f"PowerPoint generation failed for task {task_id}. Reason: {error_message}"
                break
            elif task_status == "PENDING" or task_status == "PROCESSING" or task_status == "SENT": # Add other intermediate states if known
                logging.debug(f"Task {task_id} status: {task_status}. Waiting...")
            else:
                 logging.warning(f"Task {task_id} has unknown status: {task_status}. Response: {status_result}")
                 # Continue polling, but log this unexpected state

        else:
            # Failure during polling
            logging.warning(f"Failed to get status for task {task_id} during polling. Will retry.")
            # Optionally add a counter to break after several consecutive polling failures

        await asyncio.sleep(POLLING_INTERVAL) # Use asyncio.sleep in async functions

    # After loop: check if we got a result or timed out
    if final_result:
        return final_result
    else:
        logging.warning(f"Timeout ({GENERATION_TIMEOUT}s) while waiting for PowerPoint generation task {task_id}.")
        return f"Timeout while waiting for PowerPoint generation (Task ID: {task_id}). The task might still be running."


@mcp.tool()
async def generate_slide_by_slide(
    template: str,
    slides: list[dict[str, Any]],
    language: Optional[str] = None,
    fetch_images: Optional[bool] = True,
) -> str:
    """
    Generate a PowerPoint presentation using Slide-by-Slide input.

    Parameters
    - template (string): The name of the template or the ID of a custom template. See the custom templates section for more information.
    - language (string, optional): Language code like 'ENGLISH' or 'ORIGINAL'.
    - include_cover (bool, optional): Whether to include a cover slide in addition to the specified slides.
    - include_table_of_contents (bool, optional): Whether to include the ‘table of contents’ slides.
    - slides (list[dict]): A list of slides, each defined as a dictionary with the following keys:
      - title (string): The title of the slide.
      - layout (string): The layout type for the slide. See available layout options below.
      - item_amount (integer): Number of items for the slide (must match the layout constraints).
      - content (string): The content that will be used for the slide.

    Available Layouts
    - items: 1-5 items
    - steps: 3-5 items
    - summary: 1-5 items
    - comparison: exactly 2 items
    - big-number: 1-5 items
    - milestone: 3-5 items
    - pestel: exactly 6 items
    - swot: exactly 4 items
    - pyramid: 1-5 items
    - timeline: 3-5 items
    - funnel: 3-5 items
    - quote: 1 item
    - cycle: 3-5 items
    - thanks: 0 items

    Returns
    - A string containing the final task result (including the PPTX URL when available),
      or an error/timeout message.
    """
    endpoint = "/presentation/generate/slide-by-slide"
    status_endpoint_base = "/task_status"

    if not API_KEY:
        return "API Key is missing. Cannot process any requests."

    # Basic validation
    if not isinstance(slides, list) or len(slides) == 0:
        return "Parameter 'slides' must be a non-empty list of slide objects."

    payload: dict[str, Any] = {
        "template": template,
        "slides": slides,
    }
    if language:
        payload["language"] = language
    if fetch_images is not None:
        payload["fetch_images"] = fetch_images

    # Step 1: Initiate slide-by-slide generation
    init_result = await _make_api_request("POST", endpoint, payload=payload, timeout=GENERATION_TIMEOUT)
    if not init_result:
        return "Failed to initiate slide-by-slide generation due to an API error. Check server logs."

    task_id = init_result.get("task_id")
    if not task_id:
        return f"Failed to initiate slide-by-slide generation. API response did not contain a task ID. Response: {init_result}"

    logging.info(f"Slide-by-slide generation initiated. Task ID: {task_id}")

    # Step 2: Poll for the task status
    status_endpoint = f"{status_endpoint_base}/{task_id}"
    start_time = time.time()
    final_result: Optional[str] = None

    while time.time() - start_time < GENERATION_TIMEOUT:
        logging.debug(f"Polling status for task {task_id} (slide-by-slide)...")
        status_result = await _make_api_request("GET", status_endpoint, timeout=POLLING_TIMEOUT)

        if status_result:
            task_status = status_result.get("task_status")
            task_result = status_result.get("task_result")

            if task_status == "SUCCESS":
                logging.info(f"Task {task_id} completed successfully (slide-by-slide).")
                final_result = str(task_result) if task_result else str(status_result)
                final_result = (
                    f"Make sure to return the pptx url to the user if available. Here is the result: {final_result}"
                )
                break
            elif task_status == "FAILED":
                logging.error(f"Task {task_id} failed (slide-by-slide). Status response: {status_result}")
                error_message = (
                    task_result.get("error", "Unknown error") if isinstance(task_result, dict) else "Unknown error"
                )
                final_result = f"Slide-by-slide generation failed for task {task_id}. Reason: {error_message}"
                break
            elif task_status in ("PENDING", "PROCESSING", "SENT"):
                logging.debug(f"Task {task_id} status: {task_status}. Waiting...")
            else:
                logging.warning(
                    f"Task {task_id} has unknown status: {task_status}. Response: {status_result}"
                )
        else:
            logging.warning(
                f"Failed to get status for task {task_id} during polling (slide-by-slide). Will retry."
            )

        await asyncio.sleep(POLLING_INTERVAL)

    if final_result:
        return final_result
    else:
        logging.warning(
            f"Timeout ({GENERATION_TIMEOUT}s) while waiting for slide-by-slide task {task_id}."
        )
        return (
            f"Timeout while waiting for slide-by-slide generation (Task ID: {task_id}). The task might still be running."
        )

@mcp.tool()
async def get_task_status(task_id: str) -> str:
    """
    Get the current task status and result by task_id.
    """
    if not API_KEY:
        return "API Key is missing. Cannot process any requests."
    status = await _make_api_request("GET", f"/task_status/{task_id}", timeout=POLLING_TIMEOUT)
    if not status:
        return f"Failed to fetch status for task {task_id}."
    return json.dumps(status)

@mcp.tool()
async def upload_document(file_path: str) -> str:
    """
    Upload a document file and return the task_id for processing.
    Supported file types: .pptx, .ppt, .docx, .doc, .xlsx, .pdf
    """
    if not API_KEY:
        return "API Key is missing. Cannot process any requests."

    url = f"{API_BASE}/document/upload"
    headers = {
        "User-Agent": USER_AGENT,
        "X-API-Key": API_KEY,
    }

    # Validate path
    if not os.path.isfile(file_path):
        return f"File not found: {file_path}"

    try:
        async with httpx.AsyncClient(timeout=DEFAULT_TIMEOUT) as client:
            with open(file_path, "rb") as f:
                files = {"file": (os.path.basename(file_path), f)}
                response = await client.post(url, headers=headers, files=files)
                response.raise_for_status()
                data = response.json()
                return json.dumps(data)
    except httpx.HTTPStatusError as e:
        logging.error(f"HTTP error uploading document: {e.response.status_code} - {e.response.text}")
        return f"Upload failed: {e.response.status_code} {e.response.text}"
    except Exception as e:
        logging.error(f"Unexpected error uploading document: {str(e)}")
        return f"Upload failed: {str(e)}"


if __name__ == "__main__":
    # Configure logging (optional but recommended)
    logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')

    # Check for API Key at startup
    if not API_KEY:
       logging.critical("SLIDESPEAK_API_KEY is not set. The server cannot communicate with the backend API.")
       # Optionally exit here if the API key is absolutely required
       # import sys
       # sys.exit("API Key missing. Exiting.")

    # Initialize and run the server
    mcp.run(transport='stdio')