cloud_mcp_server.py

"""
FastMCP Cloud Server for AI Phone Agent Platform

This is a URL/HTTP-based MCP server that can be deployed to Railway, Render, Fly.io, etc.
It calls the python-server-est REST API instead of importing application code directly.

SECURITY: This MCP server is scoped to a specific organization using API key
authentication. For cloud deployment, the API key comes from request headers.
For local development, it can use environment variables as fallback.
All operations are automatically filtered to only access data belonging to
the authenticated organization.
"""

import os
import json
import logging
import sys
from typing import Annotated, Optional
from datetime import datetime
from dotenv import load_dotenv

# Import FastMCP
from mcp.server.fastmcp import FastMCP

# Load environment variables from a local .env file before any module reads os.getenv
load_dotenv(os.path.join(os.path.dirname(os.path.abspath(__file__)), ".env"))

# Import our handlers (these now use HTTP calls instead of direct imports)
import mcp_handlers

# Configure logging
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
    stream=sys.stderr
)
logger = logging.getLogger(__name__)

# Create FastMCP instance
APP_NAME = "Call-Agent-MCP"
mcp = FastMCP(APP_NAME)

# ============================================================================
# AUTHENTICATION SETUP
# ============================================================================

# Organization API key (only for local testing fallback)
ORGANIZATION_API_KEY = os.getenv("ORGANIZATION_API_KEY")
ORGANIZATION_ID_ENV = os.getenv("ORGANIZATION_ID")

# Supabase configuration for API key validation
SUPABASE_URL = os.getenv("SUPABASE_URL")
SUPABASE_SERVICE_KEY = os.getenv("SUPABASE_SERVICE_KEY")

# Cache for organization data
_organization_cache = {}


async def _get_organization_from_api_key(api_key: str) -> dict:
    """
    Validate API key and get organization ID from Supabase.
    
    Args:
        api_key: The API key to validate
        
    Returns:
        Dict with organization_id and organization data
        
    Raises:
        ValueError if API key is invalid
    """
    # Check cache first
    if api_key in _organization_cache:
        return _organization_cache[api_key]
    
    if not SUPABASE_URL or not SUPABASE_SERVICE_KEY:
        raise ValueError("Supabase configuration is missing")
    
    try:
        import httpx
        
        # Query Supabase to validate API key
        headers = {
            "apikey": SUPABASE_SERVICE_KEY,
            "Authorization": f"Bearer {SUPABASE_SERVICE_KEY}",
            "Content-Type": "application/json"
        }
        
        url = f"{SUPABASE_URL}/rest/v1/organization_service_credentials"
        params = {
            "service_name": "eq.master_api_key",
            "credentials->>api_key": f"eq.{api_key}",
            "select": "organization_id,organizations(id,name)"
        }
        
        async with httpx.AsyncClient() as client:
            response = await client.get(url, headers=headers, params=params)
            
            if response.status_code != 200:
                raise ValueError("Failed to validate API key")
            
            data = response.json()
            
            if not data or len(data) == 0:
                raise ValueError("Invalid API key")
            
            cred_data = data[0]
            org_id = cred_data.get("organization_id")
            org_data = cred_data.get("organizations")
            
            if not org_id or not org_data:
                raise ValueError("API key is not associated with a valid organization")
            
            result = {
                "organization_id": org_id,
                "organization": org_data
            }
            
            # Cache the result
            _organization_cache[api_key] = result
            
            logger.info(f"Authenticated organization: {org_data.get('name')} ({org_id})")
            
            return result
            
    except Exception as e:
        logger.error(f"Failed to authenticate: {e}", exc_info=True)
        raise ValueError(f"Authentication failed: {str(e)}")


def _json(value) -> str:
    """Convert value to JSON string."""
    return json.dumps(value, ensure_ascii=False, indent=2, default=str)


async def _resolve_authentication() -> tuple[str, str]:
    """Resolve API key and organization ID from request headers or MCP context."""
    context = mcp.get_context()
    provided_api_key: str | None = None

    # DEBUG: Log what we're receiving
    logger.info(f"[AUTH DEBUG] Context type: {type(context)}")
    if context:
        logger.info(f"[AUTH DEBUG] Context attributes: {dir(context)}")
        request_context = getattr(context, "request_context", None)
        if request_context:
            logger.info(f"[AUTH DEBUG] Request context type: {type(request_context)}")
            logger.info(f"[AUTH DEBUG] Request context attributes: {dir(request_context)}")
            
            # Try to get headers from request_context
            if hasattr(request_context, "headers"):
                headers = getattr(request_context, "headers", {})
                logger.info(f"[AUTH DEBUG] Headers found: {headers}")
                if isinstance(headers, dict):
                    provided_api_key = headers.get("X-API-Key") or headers.get("x-api-key")
                    
            # Try raw request if available
            if hasattr(request_context, "request"):
                raw_request = getattr(request_context, "request", None)
                if raw_request:
                    logger.info(f"[AUTH DEBUG] Raw request type: {type(raw_request)}")
                    logger.info(f"[AUTH DEBUG] Raw request attributes: {dir(raw_request)}")
                    if hasattr(raw_request, "headers"):
                        raw_headers = getattr(raw_request, "headers", {})
                        logger.info(f"[AUTH DEBUG] Raw headers: {dict(raw_headers) if hasattr(raw_headers, '__iter__') else raw_headers}")
                        if not provided_api_key:
                            # Try different header formats
                            if hasattr(raw_headers, "get"):
                                provided_api_key = (
                                    raw_headers.get("X-API-Key") or 
                                    raw_headers.get("x-api-key") or
                                    raw_headers.get("X-Api-Key")
                                )
                            elif hasattr(raw_headers, "__getitem__"):
                                try:
                                    provided_api_key = raw_headers["X-API-Key"]
                                except (KeyError, TypeError):
                                    try:
                                        provided_api_key = raw_headers["x-api-key"]
                                    except (KeyError, TypeError):
                                        pass

    logger.info(f"[AUTH DEBUG] Extracted API key: {provided_api_key[:20] if provided_api_key else None}...")

    # For stdio mode or as fallback, check MCP client config
    if not provided_api_key and context:
        request_context = getattr(context, "request_context", None)
        potential_configs = []

        if hasattr(context, "client_config") and context.client_config:
            potential_configs.append(context.client_config)

        if request_context:
            if hasattr(request_context, "client_config") and request_context.client_config:
                potential_configs.append(request_context.client_config)
            if hasattr(request_context, "clientConfig") and request_context.clientConfig:
                potential_configs.append(request_context.clientConfig)
            metadata = getattr(request_context, "metadata", None)
            if isinstance(metadata, dict):
                cfg = metadata.get("client_config") or metadata.get("clientConfig")
                if isinstance(cfg, dict):
                    potential_configs.append(cfg)

        for cfg in potential_configs:
            if isinstance(cfg, dict):
                provided_api_key = cfg.get("api_key") or cfg.get("organization_api_key")
                if provided_api_key:
                    break

    # Final fallback to environment variable (for local testing only)
    if not provided_api_key and ORGANIZATION_API_KEY:
        provided_api_key = ORGANIZATION_API_KEY
        logger.info("[AUTH DEBUG] Using fallback environment variable")

    if not provided_api_key:
        logger.error("[AUTH DEBUG] No API key found anywhere!")
        raise PermissionError("Organization API key must be provided via request headers or MCP client configuration")

    # Resolve organization_id either via environment or Supabase lookup
    if ORGANIZATION_ID_ENV and provided_api_key == ORGANIZATION_API_KEY:
        organization_id = ORGANIZATION_ID_ENV
    else:
        org_data = await _get_organization_from_api_key(provided_api_key)
        organization_id = org_data["organization_id"]

    return organization_id, provided_api_key


# ============================================================================
# MCP TOOLS
# ============================================================================

@mcp.tool()
async def start_campaign_call_scheduled(
    campaign_id: Annotated[str, "UUID of the campaign to use for this call"],
    phone_number: Annotated[str, "Phone number in E.164 format (e.g., +37256011298)"],
    scheduled_time: Annotated[str, "Desired scheduled start time in ISO 8601 format (e.g., 2025-10-01T08:30:00+02:00)"],
    first_name: Annotated[Optional[str], "Client's first name for personalization"] = None,
    last_name: Annotated[Optional[str], "Client's last name"] = None,
    company: Annotated[Optional[str], "Company name from custom data"] = None,
    email: Annotated[Optional[str], "Email from custom data"] = None,
    extrainfo: Annotated[Optional[str], "Additional information from custom data"] = None,
) -> str:
    """
    Schedule a single outbound call for a specific time in a campaign.

    The call will be scheduled for the exact time provided. The scheduled time must be a
    valid ISO 8601 timestamp and will be interpreted according to the campaign's timezone settings.

    Args:
        campaign_id: UUID of the campaign to use for this call
        phone_number: Phone number in E.164 format (e.g., +372560112918)
        first_name: Optional client's first name
        last_name: Optional client's last name
        company: Optional clients company name
        email: Optional clients email
        extrainfo: Optional additional information about the client, that does not fit into other fields

    Returns:
        JSON string with success status and call details including scheduled time and call ID
    """
    # Prepare client payload
    client_payload = {"phone_number": phone_number}
    if first_name:
        client_payload["first_name"] = first_name
    if last_name:
        client_payload["last_name"] = last_name

    custom_data = {}
    if company:
        custom_data["company"] = company
    if email:
        custom_data["email"] = email
    if extrainfo:
        custom_data["extrainfo"] = extrainfo
    if custom_data:
        client_payload["custom_data"] = custom_data

    request_payload = {
        "campaign_id": campaign_id,
        "scheduled_time": scheduled_time,
        "clients": client_payload
    }

    organization_id, provided_api_key = await _resolve_authentication()

    result = await mcp_handlers.handle_start_campaign_call_scheduled(
        payload=request_payload,
        organization_id=organization_id,
        api_key=provided_api_key
    )

    return _json(result)


@mcp.tool()
async def list_voice_agents() -> str:
    """List all voice agents (campaigns) for the authenticated organization."""
    organization_id, provided_api_key = await _resolve_authentication()

    result = await mcp_handlers.handle_list_voice_agents(
        organization_id=organization_id,
        api_key=provided_api_key,
    )
    return _json(result)


@mcp.tool()
async def get_call_data_by_id(
    call_id: Annotated[str, "Supabase call identifier"]
) -> str:
    """Fetch the call record identified by call_id."""
    organization_id, provided_api_key = await _resolve_authentication()

    result = await mcp_handlers.handle_get_call_data_by_id(
        call_id=call_id,
        organization_id=organization_id,
        api_key=provided_api_key,
    )
    return _json(result)


@mcp.tool()
async def list_recent_calls(
    limit: int | None = None
) -> str:
    """
    Return the most recent calls for the authenticated organization.
    
    Args:
        limit: Maximum number of call records to return
    """
    organization_id, provided_api_key = await _resolve_authentication()

    # Explicitly convert limit to int if it's provided (handles potential float/number type issues)
    if limit is not None:
        limit = int(limit)

    result = await mcp_handlers.handle_list_recent_calls(
        limit=limit,
        organization_id=organization_id,
        api_key=provided_api_key,
    )
    return _json(result)


@mcp.tool()
async def get_twilio_number() -> str:
    """Return the Twilio phone number configured for the organization."""
    organization_id, provided_api_key = await _resolve_authentication()

    result = await mcp_handlers.handle_get_twilio_number(
        organization_id=organization_id,
        api_key=provided_api_key,
    )
    return _json(result)


@mcp.tool()
async def list_inbound_clients(
    limit: int | None = None
) -> str:
    """
    List inbound clients for the authenticated organization.
    
    Args:
        limit: Maximum number of inbound clients to return
    """
    organization_id, provided_api_key = await _resolve_authentication()

    # Explicitly convert limit to int if it's provided (handles potential float/number type issues)
    if limit is not None:
        limit = int(limit)

    result = await mcp_handlers.handle_list_inbound_clients(
        limit=limit,
        organization_id=organization_id,
        api_key=provided_api_key,
    )
    return _json(result)


@mcp.tool()
async def get_webhook_url() -> str:
    """Return the webhook URL configured for the organization."""
    organization_id, provided_api_key = await _resolve_authentication()

    result = await mcp_handlers.handle_get_webhook_url(
        organization_id=organization_id,
        api_key=provided_api_key,
    )
    return _json(result)


@mcp.tool()
async def get_organisation_members() -> str:
    """Return organization member name(s)"""
    organization_id, provided_api_key = await _resolve_authentication()

    result = await mcp_handlers.handle_get_organisation_members(
        organization_id=organization_id,
        api_key=provided_api_key,
    )
    return _json(result)


# ============================================================================
# HEALTH CHECK (for Railway/Render/Fly.io)
# ============================================================================

# Add a simple health check endpoint
@mcp.custom_route("/health", methods=["GET"])
async def health_check(request):
    """Health check endpoint for deployment platforms."""
    return {"status": "healthy", "service": "est-call-agent-mcp"}


# ============================================================================
# SERVER RUNNER
# ============================================================================

def main():
    """Main entry point for running the FastMCP server."""
    logger.info("=" * 60)
    logger.info("EST Call Agent Platform - FastMCP Cloud Server")
    logger.info("=" * 60)
    logger.info("Configuration:")
    logger.info(f"  - App Name: {APP_NAME}")
    logger.info(f"  - API Base URL: {os.getenv('API_BASE_URL', 'Not set')}")
    logger.info(f"  - Supabase URL: {SUPABASE_URL[:50] if SUPABASE_URL else 'Not set'}...")
    logger.info(f"  - API Key configured (env): {bool(ORGANIZATION_API_KEY)}")
    logger.info("=" * 60)
    
    # Get transport configuration
    # Default to HTTP for MCP client configuration
    transport = os.getenv("MCP_TRANSPORT", "http")

    # Map transport names to FastMCP transport modes
    transport_map = {
        "stdio": "stdio",
        "http": "streamable-http",
        "sse": "sse"
    }

    mcp_transport = transport_map.get(transport, "streamable-http")

    # Start the MCP server
    logger.info("🚀 Starting FastMCP server")
    logger.info(f"Transport: {transport}")

    if mcp_transport == "stdio":
        # Run in stdio mode
        mcp.run(transport="stdio")
    else:
        # Run in HTTP mode using uvicorn
        import uvicorn

        # Get the ASGI app based on transport mode
        if mcp_transport == "sse":
            app = mcp.sse_app()
        else:  # streamable-http
            app = mcp.streamable_http_app()

        # Run with uvicorn
        port = int(os.getenv("PORT", "8000"))
        host = os.getenv("HOST", "localhost")

        logger.info(f"Starting {mcp_transport} server on {host}:{port}")
        uvicorn.run(app, host=host, port=port)


if __name__ == "__main__":
    main()