# server URL: /pounce/api/server/ Section: api Description: Server — orchestrates the full pounce lifecycle. Manages the state machine: CONFIG → DETECT → BIND → LIFESPAN → SERVE → SHUTDOWN When ``workers == 1`` the server runs a single-worker fast path with no supervisor overhead. When ``workers > 1`` the supervisor spawns and monitors worker threads (nogil) or processes (GIL). Signal handling: SIGINT/SIGTERM trigger graceful shutdown. --- > For a complete page index, fetch /pounce/llms.txt. Open LLM text (/pounce/api/server/index.txt) Share with AI Ask Claude (https://claude.ai/new?q=Please%20help%20me%20understand%20this%20documentation%3A%20%2Fpounce%2Fapi%2Fserver%2Findex.txt) Ask ChatGPT (https://chatgpt.com/?q=Please%20help%20me%20understand%20this%20documentation%3A%20%2Fpounce%2Fapi%2Fserver%2Findex.txt) Ask Gemini (https://gemini.google.com/app?q=Please%20help%20me%20understand%20this%20documentation%3A%20%2Fpounce%2Fapi%2Fserver%2Findex.txt) Ask Copilot (https://copilot.microsoft.com/?q=Please%20help%20me%20understand%20this%20documentation%3A%20%2Fpounce%2Fapi%2Fserver%2Findex.txt) Module # `server` Server — orchestrates the full pounce lifecycle. Manages the state machine: ``` CONFIG → DETECT → BIND → LIFESPAN → SERVE → SHUTDOWN ``` When`workers == 1`the server runs a single-worker fast path with no supervisor overhead. When`workers > 1`the supervisor spawns and monitors worker threads (nogil) or processes (GIL). Signal handling: SIGINT/SIGTERM trigger graceful shutdown. 1Class2Functions ## Classes `Server` 27 ▼ Top-level server that orchestrates the full lifecycle. Creates the socket(s), runs lifespan events… Top-level server that orchestrates the full lifecycle. Creates the socket(s), runs lifespan events, starts workers (via the supervisor when multi-worker), and handles shutdown signals. #### Methods `bound_addr` 0 `tuple[str, int] | None` ▼ The actual (host, port) the server bound to, or ``None`` if not yet started. property `def bound_addr(self) -> tuple[str, int] | None` ##### Returns `tuple[str, int] | None` `run` 0 ▼ Start the server (blocking). Lifecycle: 1. Configure logging 2. Resolve effect… `def run(self) -> None` Start the server (blocking). Lifecycle: - Configure logging - Resolve effective worker count and detect worker mode - Create TLS context (if configured) - Print startup banner - Bind socket(s) - Run ASGI lifespan startup (once, in the main thread) - Start worker(s) — single-worker fast path or supervisor - Wait for shutdown signal - Run ASGI lifespan shutdown - Close socket(s) `shutdown` 0 ▼ Request graceful shutdown. Thread-safe and idempotent. Can be called from any … `def shutdown(self) -> None` Request graceful shutdown. Thread-safe and idempotent. Can be called from any thread to stop a running server. In single-worker mode this wakes the asyncio event loop via `call_soon_threadsafe`. In multi-worker mode this delegates to the supervisor's shutdown. Safe to call before`run()`— the server will exit immediately on startup. Safe to call multiple times. Internal Methods 24 ▼ `__init__` 6 ▼ `def __init__(self, config: ServerConfig, app: ASGIApp, *, app_path: str | None = None, lifecycle_collector: LifecycleCollector | None = None, sync_app: SyncApp | None = None, cli_display: CliDisplayOverrides | None = None) -> None` ##### Parameters Name Type Description `config` `—` `app` `—` `app_path` `—` Default:`None` `lifecycle_collector` `—` Default:`None` `sync_app` `—` Default:`None` `cli_display` `—` Default:`None` `_run_single` 0 ▼ Run with a single worker — no supervisor, minimal overhead. `def _run_single(self) -> None` `_reload_watch_dirs` 0 `list[Path]` ▼ Directories the file watcher should scan in ``--reload`` mode. Always includes… `def _reload_watch_dirs(self) -> list[Path]` Directories the file watcher should scan in`--reload`mode. Always includes the current working directory and any explicit `reload_dirs`. Also includes configured static-mount directories (`static_files` is `{url_path: directory}`) so that editing content/assets served from outside cwd still triggers a reload. Duplicates and missing directories are tolerated by the watcher. ##### Returns `list[Path]` `_run_single_with_reload` 0 ▼ Single-worker mode with auto-reload on source changes. Runs the worker in a lo… `def _run_single_with_reload(self) -> None` Single-worker mode with auto-reload on source changes. Runs the worker in a loop. When the file watcher detects changes it sets`_reload_requested`and signals shutdown, causing the asyncio loop to exit. The outer loop then restarts the worker with a fresh socket and event loop. `_run_single_async` 2 ▼ Async entry point for single-worker mode. async `async def _run_single_async(self, sock: socket.socket, udp_sock: socket.socket | None = None) -> None` ##### Parameters Name Type Description `sock` `—` `udp_sock` `—` Default:`None` `_run_multi` 2 ▼ Run with multiple workers managed by the supervisor. Lifespan runs once in the… `def _run_multi(self, effective_workers: int, mode: WorkerMode) -> None` Run with multiple workers managed by the supervisor. Lifespan runs once in the main process/thread before workers are spawned. Workers do not run lifespan. When`--reload`is active, a file watcher thread runs alongside the supervisor and triggers `restart_workers()` (/pounce/api/supervisor/#Supervisor) on changes. ##### Parameters Name Type Description `effective_workers` `—` `mode` `—` `_run_lifespan_then_supervise` 3 ▼ Run lifespan in the main thread, then hand off to supervisor. Installs asyncio… async `async def _run_lifespan_then_supervise(self, supervisor: Supervisor, sockets: list[socket.socket], udp_sockets: list[socket.socket] | None = None) -> None` Run lifespan in the main thread, then hand off to supervisor. Installs asyncio signal handlers so SIGINT/SIGTERM trigger a coordinated graceful shutdown through the supervisor instead of a raw`KeyboardInterrupt`. On the first signal the supervisor drains workers; a second signal removes the handler so the default`KeyboardInterrupt`forces an immediate exit. ##### Parameters Name Type Description `supervisor` `—` `sockets` `—` `udp_sockets` `—` Default:`None` `_run_single_h3` 2 ▼ Run HTTP/3 datagram endpoint in single-worker mode. async `async def _run_single_h3(self, udp_sock: socket.socket, lifespan_state: dict[str, Any]) -> None` ##### Parameters Name Type Description `udp_sock` `—` `lifespan_state` `—` `_effective_worker_mode` 0 `WorkerMode (/pounce/api/_runtime/#WorkerMode)` ▼ Return the worker spawning strategy that ``run()`` will use. Mirrors the resol… `def _effective_worker_mode(self) -> WorkerMode` Return the worker spawning strategy that`run()`will use. Mirrors the resolution in`run`(): an explicit `worker_mode="subinterpreter"`wins, otherwise the GIL state decides between thread (3.14t) and process (GIL build) mode. Used to reason about whether in-process backpressure state is shared (thread mode) or copied per worker (process/subinterpreter mode). ##### Returns `WorkerMode (/pounce/api/_runtime/#WorkerMode)` `_log_worker_mode_notice` 1 ▼ Emit a compatibility reminder for explicit subinterpreter mode. The lifecycle … `def _log_worker_mode_notice(self, mode: WorkerMode) -> None` Emit a compatibility reminder for explicit subinterpreter mode. The lifecycle contract is stable, but applications and C extensions still need to support isolated interpreters. ##### Parameters Name Type Description `mode` `—` `_apply_integrations` 0 ▼ Configure optional integrations and wrap the app. Wrapping order (outermost fi… `def _apply_integrations(self) -> None` Configure optional integrations and wrap the app. Wrapping order (outermost first): - Sentry (catches exceptions from all inner layers) - Request queue (load shedding) - Rate limiter (per-IP throttling) - User middleware - Static files - Metrics endpoint (intercepts /metrics) - App (innermost) `_apply_otel` 0 ▼ Configure OpenTelemetry tracing if endpoint is set. `def _apply_otel(self) -> None` `_apply_lifecycle_logging` 0 ▼ Configure lifecycle event logging if enabled. `def _apply_lifecycle_logging(self) -> None` `_apply_metrics` 0 ▼ Configure Prometheus metrics endpoint if enabled. `def _apply_metrics(self) -> None` `_apply_static_files` 0 ▼ Wrap app with configured static file mounts. `def _apply_static_files(self) -> None` `_apply_middleware` 0 ▼ Wrap app with middleware stack if configured. `def _apply_middleware(self) -> None` `_apply_rate_limiter` 0 ▼ Configure per-IP rate limiting if enabled. `def _apply_rate_limiter(self) -> None` `_apply_request_queue` 0 ▼ Configure request queueing (load shedding) if enabled. Each worker thread gets… `def _apply_request_queue(self) -> None` Configure request queueing (load shedding) if enabled. Each worker thread gets its own RequestQueue/QueueMetrics on first use — asyncio.Lock and asyncio.Semaphore are single-event-loop primitives, so a shared instance would break under free-threading where multiple workers run independent loops. `_apply_sentry` 0 ▼ Configure Sentry error tracking if DSN is set. `def _apply_sentry(self) -> None` `_warn_if_introspection_public` 0 ▼ Warn at startup if the introspection endpoint is publicly reachable. The ``/_p… `def _warn_if_introspection_public(self) -> None` Warn at startup if the introspection endpoint is publicly reachable. The`/_pounce/info`endpoint exposes a redacted view of the running config. The redaction allowlist is fail-closed, but a misconfigured deploy that binds pounce to a public interface still hands strangers a free runtime probe. Surface the risk loudly with a code agents can catalog-link to. `_create_udp_listener_if_h3` 1 `socket.socket | None` ▼ Create a single UDP listener for HTTP/3 if configured and available. `def _create_udp_listener_if_h3(self, actual_addr: tuple[str, int]) -> socket.socket | None` ##### Parameters Name Type Description `actual_addr` `—` ##### Returns `socket.socket | None` `_create_udp_listeners_if_h3` 2 `list[socket.socket]` ▼ Create multiple UDP listeners for HTTP/3 if configured and available. `def _create_udp_listeners_if_h3(self, actual_addr: tuple[str, int], count: int) -> list[socket.socket]` ##### Parameters Name Type Description `actual_addr` `—` `count` `—` ##### Returns `list[socket.socket]` `_print_banner` 2 ▼ Print the startup banner to stderr. In JSON mode, emits a single structured JS… `def _print_banner(self, effective_workers: int, mode: WorkerMode) -> None` Print the startup banner to stderr. In JSON mode, emits a single structured JSON log line. In text/pretty mode, prints a human-friendly ASCII banner. ##### Parameters Name Type Description `effective_workers` `—` `mode` `—` `_close_sockets` 1 ▼ Close all sockets, deduplicating shared-fd sockets. On macOS (no SO_REUSEPORT)… staticmethod `def _close_sockets(sockets: list[socket.socket]) -> None` Close all sockets, deduplicating shared-fd sockets. On macOS (no SO_REUSEPORT) all workers share the same socket fd. Deduplicate by fd and guard against already-closed fds to avoid `ValueError: Invalid file descriptor: -1`on shutdown. ##### Parameters Name Type Description `sockets` `—` ## Functions `_is_loopback_bind` 1 `bool` ▼ Return True if *host* refers to a loopback address. Recognises ``localhost``, … `def _is_loopback_bind(host: str) -> bool` Return True if host refers to a loopback address. Recognises`localhost`, IPv4 `127.0.0.0/8`, and IPv6 `::1`. An unparseable host (a name we can't resolve here) is treated as non-loopback so the public-bind warning errs on the loud side. ##### Parameters Name Type Description `host` `str` ##### Returns `bool` `_get_version` 0 `str` ▼ Get the pounce version string. `def _get_version() -> str` ##### Returns `str`