#!/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())