nlproxy/cli/runserver.py

#!/usr/bin/env python3
"""
ASGI server startup command for NLProxy.

Provides a production-ready FastAPI/uvicorn launcher with environment-aware 
configuration, graceful shutdown, structured logging, and dev/prod mode handling.

Usage
-----
# Production startup (auto-detects CPU cores, disables reload)
$ python -m nlproxy runserver

# Development with auto-reload & debug logging
$ python -m nlproxy runserver --reload --log-level debug

# Bind to specific interface with explicit workers
$ python -m nlproxy runserver --host 0.0.0.0 --port 8080 --workers 4

Configuration
-------------
Environment variables:
    NLPROXY_HOST          Server bind host (default: 0.0.0.0)
    NLPROXY_PORT          Server bind port (default: 8000)
    NLPROXY_LOG_LEVEL     Logging level: debug, info, warning, error, critical
    NLPROXY_RELOAD        Set to "true" to enable dev auto-reload

Author: IntelliDeep Labs Team
License: BSL 1.1
"""
from __future__ import annotations

import argparse
import logging
import multiprocessing
import os
import sys
from typing import Dict, Optional

logger = logging.getLogger(__name__)


# ---------------------------------------------------------------------------
# .env file loader (no external dependency required)
# ---------------------------------------------------------------------------
def _load_dotenv(path: Optional[str] = None) -> Dict[str, str]:
    """
    Parse a .env file and return environment variables as a dict.
    
    Looks for ``.env`` in the current working directory by default.
    Supports ``KEY=VALUE`` syntax with optional quoting, and ``#`` comments.
    Skips empty lines.
    """
    dotenv_path = path or os.path.join(os.getcwd(), ".env")
    result: Dict[str, str] = {}
    if not os.path.isfile(dotenv_path):
        return result
    try:
        with open(dotenv_path, "r") as f:
            for line in f:
                line = line.strip()
                if not line or line.startswith("#") or "=" not in line:
                    continue
                key, _, value = line.partition("=")
                key = key.strip()
                value = value.strip().strip("\"'")
                if key:
                    result[key] = value
    except OSError:
        pass
    return result


# Attempt to load .env variables into the environment so they are available
# for :func:`os.getenv` calls that follow.
_dotenv_vars = _load_dotenv()
for _k, _v in _dotenv_vars.items():
    os.environ.setdefault(_k, _v)
del _dotenv_vars


def setup_logging(level: str = "INFO") -> None:
    """Configure structured console logging for CLI/server processes."""
    numeric_level = getattr(logging, level.upper(), logging.INFO)
    logging.basicConfig(
        level=numeric_level,
        format="%(asctime)s [%(levelname)-8s] %(message)s",
        datefmt="%H:%M:%S",
        stream=sys.stderr,
    )


DEFAULT_MODEL_PER_PROVIDER = {
    "gemini": "gemini-2.0-flash",
    "claude": "claude-3-sonnet-20240229",
    "openai": "gpt-4",
    "deepseek": "deepseek-chat",
    "qwen": "qwen-max",
    "kimi": "kimi",
    "openrouter": "openai/gpt-4",
}


def cmd_runserver(args: argparse.Namespace) -> None:
    setup_logging(args.log_level)
    
    try:
        import uvicorn
    except ImportError:
        logger.error("uvicorn is required for server mode...")
        sys.exit(1)

    workers = args.workers or multiprocessing.cpu_count()
    if args.reload and workers > 1:
        logger.warning(
            "--reload requested but multiple workers configured (%d). "
            "Auto-reload requires a single worker process; forcing workers=1 to enable reload.",
            workers,
        )
        workers = 1

    provider = args.llm_client or os.getenv("NLPROXY_DEFAULT_LLM_PROVIDER") or "openai"
    provider = provider.lower()
    if provider not in DEFAULT_MODEL_PER_PROVIDER:
        logger.error(f"Unsupported provider: {provider}. Valid: {list(DEFAULT_MODEL_PER_PROVIDER.keys())}")
        sys.exit(1)

    if args.model:
        model = args.model
    else:
        model = os.getenv("NLPROXY_DEFAULT_LLM_MODEL") or DEFAULT_MODEL_PER_PROVIDER.get(provider)

    api_key = args.api_key_client
    if not api_key:
        env_key_name = {
            "gemini": "GEMINI_API_KEY",
            "claude": "ANTHROPIC_API_KEY",
            "openai": "OPENAI_API_KEY",
            "deepseek": "DEEPSEEK_API_KEY",
            "qwen": "QWEN_API_KEY",
            "kimi": "KIMI_API_KEY",
            "openrouter": "OPENROUTER_API_KEY",
        }.get(provider)
        if env_key_name:
            api_key = os.getenv(env_key_name)
    if not api_key:
        logger.error(f"No API key found for provider '{provider}'. Provide via --api-key-client or set {env_key_name}")
        sys.exit(1)

    env_map = {
        "gemini": "GEMINI_API_KEY",
        "claude": "ANTHROPIC_API_KEY",
        "openai": "OPENAI_API_KEY",
        "deepseek": "DEEPSEEK_API_KEY",
        "qwen": "QWEN_API_KEY",
        "kimi": "KIMI_API_KEY",
        "openrouter": "OPENROUTER_API_KEY",
    }
    env_var = env_map.get(provider)
    if env_var:
        os.environ[env_var] = api_key
        os.environ[env_var.lower()] = api_key
    os.environ["NLPROXY_DEFAULT_LLM_PROVIDER"] = provider
    os.environ["nlproxy_default_llm_provider"] = provider
    os.environ["NLPROXY_DEFAULT_LLM_MODEL"] = model
    os.environ["nlproxy_default_llm_model"] = model
    os.environ["LLM_CLIENT"] = provider
    os.environ["LLM_API_CLIENT"] = api_key

    logger.info(f"Configured LLM: provider={provider}, model={model}")

    if args.list_models:
        logger.info("Available models per provider:")
        for prov, mdl in DEFAULT_MODEL_PER_PROVIDER.items():
            logger.info(f"  {prov}: {mdl}")
        return

    server_config = {
        "app": "nlproxy.server:app",
        "host": args.host,
        "port": args.port,
        "workers": workers,
        "log_level": args.log_level.lower(),
        "access_log": args.access_log,
        "reload": args.reload,
        "reload_dirs": ["nlproxy"],
        "loop": "auto",
        "http": "httptools",
        "ws": "websockets",
        "timeout_graceful_shutdown": 30,
        "use_colors": True,
    }
    
    logger.info(f"Setting env: NLPROXY_DEFAULT_LLM_PROVIDER={os.environ.get('NLPROXY_DEFAULT_LLM_PROVIDER')}")
    logger.info(f"Setting env: {env_var}={api_key[:10]}...") 

    uvicorn.run(**server_config)


def main(argv: Optional[list[str]] = None) -> int:
    parser = argparse.ArgumentParser(
        prog="nlproxy runserver",
        description="Start the NLProxy Enterprise FastAPI server with production-ready configuration.",
    )
    parser.add_argument("--host", type=str, default=os.getenv("NLPROXY_HOST", "0.0.0.0"))
    parser.add_argument("--port", type=int, default=int(os.getenv("NLPROXY_PORT", "8000")))
    parser.add_argument("--workers", type=int, default=1)
    parser.add_argument("--llm-client", type=str, default=os.getenv("NLPROXY_DEFAULT_LLM_PROVIDER", None),
                        help="LLM provider to use (gemini|claude|openai|deepseek|qwen|kimi|openrouter).")
    parser.add_argument("--model", type=str, default=None,
                        help="Model name (e.g., gpt-4, gemini-2.0-flash). Overrides env NLPROXY_DEFAULT_LLM_MODEL.")
    parser.add_argument("--api-key-client", type=str, default=None,
                        help="API key for the selected LLM provider (overrides provider-specific env var).")
    parser.add_argument("--list-models", action="store_true",
                        help="List available models for each provider and exit.")
    parser.add_argument("--reload", action="store_true", help="Enable auto-reload for development")
    parser.add_argument("--log-level", type=str, default=os.getenv("NLPROXY_LOG_LEVEL", "info"),
                        choices=["debug", "info", "warning", "error", "critical"])
    parser.add_argument("--access-log", action="store_true", default=True, help="Enable HTTP access logging")
    parser.add_argument("-q", "--quiet", action="store_true", help="Suppress non-essential output")

    args = parser.parse_args(argv)
    if args.quiet:
        logging.getLogger("nlproxy").setLevel(logging.WARNING)
        logging.getLogger("uvicorn").setLevel(logging.WARNING)
        
    try:
        cmd_runserver(args)
        return 0
    except KeyboardInterrupt:
        return 0
    except Exception as e:
        logger.error(str(e))
        return 1

if __name__ == "__main__":
    sys.exit(main())