Loki Logging with byteforge-loki-logging

This skill helps you integrate Grafana Loki logging using the byteforge-loki-logging library, which handles structured JSON logging and asynchronous Loki shipping with graceful fallback.

When to Use This Skill

Use this skill when:

• Starting a new Python/Flask application
• You want centralized logging with Loki
• You need structured JSON logs for production
• You want easy local development with console logs

What This Skill Creates

1. requirements.txt entry - Adds byteforge-loki-logging dependency (public GitHub library)
2. Logging initialization - Adds configure_logging() call to main application file
3. CA certificate configuration - LOKI_CA_BUNDLE_PATH, plus a Docker Compose volume mount if your Loki endpoint uses a private CA
4. Environment variable documentation - All required Loki configuration

Step 1: Gather Project Information

IMPORTANT: Before making changes, ask the user these questions:

1. "What is your application tag/name?" (e.g., "myapp", "another-service")
   • This identifies your service in Loki logs
   • This becomes the application label in Loki — all services across the stack (Python and TypeScript) must use application as the label name, never app or service
2. "What is your main application file?" (e.g., "app.py", "server.py", "myapp.py")
   • Where to add the logging configuration
3. "Does your Loki endpoint use a private CA or a publicly trusted certificate?"
   • Private CA: also ask for the certificate filename (e.g., "loki-ca.pem", "my-org-ca.pem")
   • Publicly trusted: no certificate file is needed, but LOKI_CA_BUNDLE_PATH must still point at the system CA bundle (see Step 4)

Step 2: Add byteforge-loki-logging to requirements.txt

Add this line to requirements.txt:

# Logging configuration with Loki support (public GitHub library)
byteforge-loki-logging @ git+https://github.com/jmazzahacks/byteforge-loki-logging.git

Install the dependency:

pip install -r requirements.txt

Step 3: Configure Logging in Application

For MCP servers, use the MCP integration below instead of this module-level setup. Select the transport before configuring logging: stdio requires stderr-only logging because stdout carries JSON-RPC.

For Flask apps, use the Flask + Gunicorn pattern instead, which configures logging inside create_app(). The module-level setup below is for plain scripts and other single-process programs.

Add to the top of your main script (e.g., {app_file}.py):

import os
from byteforge_loki_logging import configure_logging

# Configure logging with byteforge-loki-logging
# Use debug_local=True for local development, False for production with Loki
debug_mode = os.environ.get('DEBUG_LOCAL', 'true').lower() == 'true'
log_level = os.environ.get('LOG_LEVEL', 'INFO')
configure_logging(
    application_tag='{application_tag}',
    debug_local=debug_mode,
    local_level=log_level
)

CRITICAL: Replace:

• {app_file} -> Your main script filename (e.g., "myapp")
• {application_tag} -> Your service name (e.g., "myapp")

Place this before any other initialization.

Structured JSON Logging

JSON formatting is enabled by default (json_format=True). Log records are formatted as:

{"logger": "myapp", "level": "INFO", "message": "Request processed", "user_id": "123", "latency_ms": 42}

Query in Grafana: {application="my-service"} | json | user_id="123"

Graceful Fallback

If the Loki connection test fails at startup, logging automatically falls back to stdout with a warning on stderr. Your application never crashes due to logging issues.

Step 4: Configure CA Certificate

LOKI_CA_BUNDLE_PATH is required whenever DEBUG_LOCAL=false, so every production deployment sets it. Only the value depends on your certificate.

Private CA: mount the certificate into the container via Docker Compose as a read-only volume. Do not bake the certificate into the Dockerfile with COPY.

In docker-compose.yaml:

services:
  {app_name}:
    volumes:
      - /path/to/{ca_cert_filename}:/app/certs/loki-ca.pem:ro
    environment:
      - LOKI_CA_BUNDLE_PATH=/app/certs/loki-ca.pem

CRITICAL: Replace:

• {app_name} -> Your service name in docker-compose
• {ca_cert_filename} -> Your actual CA certificate filename from Step 1

The certificate will be available at /app/certs/loki-ca.pem inside the container.

Publicly trusted certificate: skip the volume mount and point LOKI_CA_BUNDLE_PATH at the system CA bundle. Leaving it unset stops the app at startup. On Debian, Ubuntu and Alpine images, including python:*-slim and python:*-alpine:

services:
  {app_name}:
    environment:
      - LOKI_CA_BUNDLE_PATH=/etc/ssl/certs/ca-certificates.crt

Step 5: Document Environment Variables

Add to README.md or .env.example:

Environment Variables

Logging Configuration (Local Development):

• DEBUG_LOCAL - Set to 'true' for local development (console logs), 'false' for production (Loki)
  • Default: 'true'
  • Production: 'false'
• LOG_LEVEL - Logging level: DEBUG, INFO, WARNING, ERROR, CRITICAL
  • Default: 'INFO'

Loki Configuration (Production Only):

ALL FOUR variables below are hard-required when DEBUG_LOCAL=false, regardless of whether your Loki endpoint uses a private or public CA. _validate_loki_env_vars() raises RuntimeError if any is unset or empty. Under gunicorn this crash-loops the worker visibly at boot; under celery the failure is silent unless you wired the _configure_or_die pattern from the Celery Workers section (see the "Celery worker running but no task logs in Loki, container is green" troubleshooting entry).

• LOKI_ENDPOINT - Loki push API URL (e.g., https://loki.example.com/loki/api/v1/push)
• LOKI_USER - Loki username for HTTP Basic Auth
• LOKI_PASSWORD - Loki password for HTTP Basic Auth
• LOKI_CA_BUNDLE_PATH - How the client validates Loki's TLS certificate. Three valid shapes:
  • Private CA (Loki behind an internal PKI): path to a .pem file containing your CA cert, e.g. /app/certs/loki-ca.pem. Mount it into the container via docker-compose (see Step 4).
  • Public CA (Grafana Cloud, any Loki fronted by a widely-trusted issuer): path to the system CA bundle, e.g. /etc/ssl/certs/ca-certificates.crt on Debian/Ubuntu/Alpine images, /etc/pki/tls/certs/ca-bundle.crt on RHEL. This is the safe choice — TLS is still verified against the system trust store.
  • "false" (literal string) — disables SSL verification entirely. This is an escape hatch for dev/test/internal-network Loki where you accept the MITM risk. Do NOT reach for it as a shortcut for public-CA endpoints; the system-bundle path above is what you want there.

Logging Behavior

Local Development (DEBUG_LOCAL=true):

• Logs output to console with human-readable formatting
• Easy to read during development
• No Loki connection required
• No need to set LOKI_* variables

Production (DEBUG_LOCAL=false):

• Logs output as structured JSON to Loki asynchronously (1-second batching)
• All LOKI_* variables must be set
• Queryable in Grafana
• Automatic fallback to stdout if Loki is unreachable at startup

Step 6: Usage Examples

Local Development

# In .env or shell
export DEBUG_LOCAL=true
export LOG_LEVEL=DEBUG

pip install -r requirements.txt
python {app_file}.py

Production Deployment

Docker Compose example:

services:
  {app_name}:
    build:
      context: .
    # Private CA shown. For a publicly trusted certificate, remove volumes and
    # use the system CA bundle for LOKI_CA_BUNDLE_PATH (see Step 4).
    volumes:
      - /path/to/{ca_cert_filename}:/app/certs/loki-ca.pem:ro
    environment:
      - DEBUG_LOCAL=false
      - LOG_LEVEL=INFO
      - LOKI_ENDPOINT=${LOKI_ENDPOINT}
      - LOKI_USER=${LOKI_USER}
      - LOKI_PASSWORD=${LOKI_PASSWORD}
      - LOKI_CA_BUNDLE_PATH=/app/certs/loki-ca.pem

NOTE: Set these in your .env file:

LOKI_ENDPOINT=https://loki.example.com/loki/api/v1/push
LOKI_USER=your_loki_user
LOKI_PASSWORD=your_loki_password

How It Works

The byteforge-loki-logging library provides:

1. Automatic mode detection - Console logs for local dev, Loki for production
2. Structured JSON logging - Consistent JSON format for Loki, enabled by default
3. Async shipping - Background thread with 1-second batching for minimal performance impact
4. Secure connection - Uses CA certificate for encrypted Loki communication
5. Graceful fallback - Falls back to stdout if Loki is unreachable, never crashes your app
6. Application tagging - Identifies your service in centralized logs via the application label

You don't need to:

• Write JSON formatters
• Configure logging handlers
• Manage Loki client setup
• Handle certificate validation
• Worry about logging crashing your application

Use the integration below for your server or worker type; MCP stdio uses stderr-only logging instead of configure_logging().

Integration with Other Skills

Flask API Server

If using flask-smorest-api skill, add logging as the first thing inside create_app(), before creating the Flask app. The Flask + Gunicorn section has the full pattern, including the dependency-logger loop.

import os
from flask import Flask
from byteforge_loki_logging import configure_logging

def create_app() -> Flask:
    # Configure logging FIRST
    debug_mode = os.environ.get('DEBUG_LOCAL', 'true').lower() == 'true'
    log_level = os.environ.get('LOG_LEVEL', 'INFO')
    configure_logging(
        application_tag='my-api',
        debug_local=debug_mode,
        local_level=log_level,
    )

    # Then create Flask app
    app = Flask(__name__)

    # IMPORTANT: Propagate Flask's logger to the root logger so unhandled
    # exceptions in route handlers reach Loki. Without this, Flask catches
    # exceptions internally and logs them via werkzeug to stdout/stderr,
    # bypassing the root logger that configure_logging() set up.
    #
    # DO NOT set app.logger's level here. Leaving it at NOTSET lets
    # isEnabledFor()'s ancestor walk reach root and inherit LOG_LEVEL.
    # Pinning to DEBUG would stop the walk at app.logger, so the level
    # check passes, the record propagates to root (whose handler is
    # NOTSET), and DEBUG lines reach Loki regardless of LOG_LEVEL.
    # See the Troubleshooting entry for the full trace.
    app.logger.handlers.clear()
    app.logger.propagate = True

    # ... rest of setup
    return app

Why this matters: Flask catches exceptions in route handlers and returns a 500 response, but by default it logs the traceback through its own app.logger using werkzeug's error handling — not through Python's root logger. Since configure_logging() configures the root logger, those tracebacks never reach Loki unless you clear Flask's default handlers and set propagate = True.

Flask + Gunicorn: Propagate All Dependency Loggers

When running Flask under gunicorn, werkzeug and gunicorn create their own loggers (werkzeug, gunicorn.error, gunicorn.access) with their own StreamHandler instances and set propagate=False. This means log messages from those loggers never reach the root logger — which is where configure_logging() attaches the Loki handler. All application logs go to stdout/stderr instead of Loki.

Everything must be done inside create_app() — not at module level — for two reasons:

1. One place for all logging wiring: gunicorn attaches its own handlers to gunicorn.error / gunicorn.access in the master process before it imports your app, so by the time create_app() runs those handlers exist and can be cleared. Keeping the clearing right next to configure_logging() means the whole root-logger setup lives in one function that runs once per worker.
2. Gunicorn fork/SSL: configure_logging() creates a Loki handler with a requests.Session and SSL context. If called at module level, this runs in gunicorn's master process before fork(). Before byteforge-loki-logging 0.1.11, that state doesn't survive the fork into worker processes, causing SSL errors or lost logs. Moving it into create_app() ensures it is created in the worker process where it will be used. From 0.1.11 the library resets this state after a fork, and you can send master logs to Loki too.

import os
import logging
from flask import Flask
from byteforge_loki_logging import configure_logging

def create_app() -> Flask:
    # Configure logging inside create_app() so it runs post-fork in the
    # gunicorn worker process. Module-level init causes SSL context issues
    # with the Loki handler because the SSL session doesn't survive fork().
    debug_mode = os.environ.get('DEBUG_LOCAL', 'true').lower() == 'true'
    log_level = os.environ.get('LOG_LEVEL', 'INFO')
    configure_logging(
        application_tag='my-api',
        debug_local=debug_mode,
        local_level=log_level,
    )

    app = Flask(__name__)

    # Force Flask, werkzeug, and gunicorn loggers to propagate to root.
    # These loggers create their own StreamHandlers with propagate=False,
    # which bypasses the root logger's Loki handler.
    #
    # Do NOT setLevel() on ANY of these — not app.logger, not the deps.
    # Leaving them at NOTSET lets isEnabledFor()'s ancestor walk reach
    # root and inherit LOG_LEVEL. Pinning one to DEBUG stops the walk
    # there; the record propagates to root (whose handler is NOTSET)
    # and DEBUG lines reach Loki regardless of LOG_LEVEL. Same rule
    # applies to any propagate-to-root wiring elsewhere. See the
    # Troubleshooting entry for the full trace.
    app.logger.handlers.clear()
    app.logger.propagate = True

    for name in ('werkzeug', 'gunicorn', 'gunicorn.error', 'gunicorn.access'):
        dep_logger = logging.getLogger(name)
        dep_logger.handlers.clear()
        dep_logger.propagate = True

    # ... register blueprints, configure Api, etc. — before or after the
    # loop above; the order does not matter (see the note below the
    # Flask + Gunicorn example).

    return app

Start it with gunicorn "{module}:create_app()", which calls the factory in each worker process.

If your entry point must be {module}:app, add app = create_app() at the bottom of the module. Leave that line out if the factory fails fast on missing config, such as a required database password: it would run on every import, so importing the module under pytest crashes test collection. Avoid --preload with either form, because it runs the factory in the master process before fork(). Before 0.1.11 that breaks worker logging. From 0.1.11 logging still works, but the dependency-logger loop clears the master's stdout handlers, so bind failures and worker timeouts disappear from docker logs.

CRITICAL: configure_logging() must be the first thing inside create_app(), before any code that logs. Do NOT call it at module level.

The dependency-logger loop can go anywhere inside create_app() — before or after Flask(__name__), Api(app), or blueprint registration. (Earlier versions of this skill claimed it had to follow Api initialization; that was wrong, and contradicted the examples.) Verified against gunicorn 26.2, Flask 3.1, Werkzeug 3.1, and flask-smorest 0.47 with an A/B test — loop before vs. after Api(app) — which produced identical results: gunicorn.access and worker-side gunicorn.error records reached root in both. Why no ordering applies:

• gunicorn creates the gunicorn.error / gunicorn.access handlers (and sets propagate=False) in the master process before importing your app. Workers do not repeat that setup, so nothing re-creates the handlers after you clear them.
• Werkzeug and Flask attach their default StreamHandler only when no handler exists anywhere up the logger tree. configure_logging() has already put one on root, so they add nothing.
• flask-smorest's Api does not configure logging at all.

The only ordering that matters is the CRITICAL rule above: configure_logging() first.

Verify after a deploy (in Grafana/Loki):

• {application="<your-tag>"} | json | logger="gunicorn.error" — should show Worker exiting (pid: N) records around a container rotation.
• {application="<your-tag>"} | json | logger="gunicorn.access" — per-request lines, but only if an access log is configured (e.g. --access-logfile -). Without one, gunicorn emits no access records at all, so their absence proves nothing.

Gunicorn master-process records don't reach Loki by default. configure_logging() runs post-fork inside the worker, and the master never imports your app, so it never gets the Loki handler. Master records stay on stdout: Starting gunicorn, Listening at, Booting worker with pid, Handling signal, Shutting down: Master — and, more importantly, port-bind failures, workers that die before booting, and WORKER TIMEOUT kills. A service that looks silent in Loki may be healthy-and-idle or failing to start; check docker logs <container> to tell which, or send these records to Loki as described next.

Sending Gunicorn master records to Loki

Requires byteforge-loki-logging 0.1.11 or later. Check with pip show byteforge-loki-logging, and upgrade first if it's older. In a local test on 0.1.10, Booting worker records were lost, the master's startup records were sent more than once, and workers that didn't call configure_logging() themselves sent nothing. If you can't upgrade, leave master records on stdout, and don't move configure_logging() to module level.

Add gunicorn.conf.py and keep create_app() unchanged. Replace {application_tag} as in Step 3:

import logging
import os
from byteforge_loki_logging import configure_logging
from gunicorn.glogging import Logger


class LokiGunicornLogger(Logger):
    def setup(self, cfg) -> None:
        super().setup(cfg)
        configure_logging(
            application_tag='{application_tag}',
            debug_local=os.environ.get('DEBUG_LOCAL', 'true').lower() == 'true',
            local_level=os.environ.get('LOG_LEVEL', 'INFO'),
        )
        # Gunicorn's named loggers set propagate=False, so route them to the
        # root logger's Loki handler. Keep their stdout handlers here: master
        # records then stay in `docker logs` too, for when Loki is down.
        for name in ('gunicorn.error', 'gunicorn.access'):
            logging.getLogger(name).propagate = True


logger_class = LokiGunicornLogger

Start it with gunicorn -c gunicorn.conf.py "{module}:create_app()".

• Workers: each worker inherits this setup, and the configure_logging() call in create_app() reconfigures it. Worker records go to Loki only, as before, because the dependency-logger loop clears the stdout handlers in the worker. The master keeps its handlers.
• Startup failures: a missing LOKI_* variable now stops gunicorn before it binds, and the error shows in docker logs. If Loki is unreachable, logging falls back to stdout and the app still starts.
• Duplicate lines: with DEBUG_LOCAL=true or the stdout fallback, master records appear twice on stdout, once from each handler. Loki is unaffected.
• Forced exits: SIGKILL can lose records that are still queued, so keep checking docker logs when a container dies abruptly.

Multiprocessing / Forked Child Processes

For processes using Loki, call configure_logging() in each child that logs. This is required before byteforge-loki-logging 0.1.11: the Loki handler's background QueueListener thread and requests.Session don't survive fork(), so a child inherits a dead handler and its logs go into a queue that nobody reads. No errors are raised because the queue accepts writes silently. From 0.1.11 a child forked with Python's os.fork() can log through the inherited setup, but calling configure_logging() is still supported and is how a child gets different settings or an application tag. MCP stdio servers and children sharing their stdout must use stderr-only logging instead; see the MCP integration below.

This applies to:

• multiprocessing.Process target functions
• os.fork() child processes
• Any code that spawns child processes that emit logs

Fix: Call configure_logging() at the top of the child process entry point, before any logging:

import os
import logging
import multiprocessing
from byteforge_loki_logging import configure_logging

def run_job(job_id: str) -> None:
    """Target function for multiprocessing.Process — runs in a forked child."""
    # CRITICAL before byteforge-loki-logging 0.1.11: the parent's Loki handler
    # (QueueListener thread + requests.Session) does not survive fork().
    # Re-initialize logging in the child process so it gets its own handler,
    # thread, and SSL context.
    debug_mode = os.environ.get('DEBUG_LOCAL', 'true').lower() == 'true'
    log_level = os.environ.get('LOG_LEVEL', 'INFO')
    configure_logging(
        application_tag='my-worker',
        debug_local=debug_mode,
        local_level=log_level,
    )
    logger = logging.getLogger(__name__)

    logger.info(f"Starting job {job_id}")
    # ... do work ...
    logger.info(f"Finished job {job_id}")


# In the parent process:
process = multiprocessing.Process(target=run_job, args=(job_id,))
process.start()

Summary of where configure_logging() must be called:

Celery Workers

Celery is a fork-based worker pool, so the general multiprocessing rule applies — each prefork child needs its own configure_logging() after fork. Two Celery-specific traps make the naive "just call configure_logging in a signal receiver" recipe silently or catastrophically wrong; both must be handled together.

Trap 1 — the signal-swallow blackout. Celery's Signal.send wraps each receiver in try/except Exception and swallows the error, and setup_logging_subsystem still counts a failed setup_logging receiver as "logging was configured." So a Loki misconfig (DEBUG_LOCAL=false with any of the four LOKI_* vars missing → RuntimeError from _validate_loki_env_vars) leaves the worker running with zero root handlers attached. Tasks keep acking, nothing reaches Loki, and the container's health checks stay green — a silent blackout on the write path. Fix: raise SystemExit(1) from the receiver. SystemExit is a BaseException, not an Exception, so it escapes the swallow and kills the process loudly — the same fail-fast behavior your Flask container gets from create_app().

Trap 2 — the fork-window timeout race. worker_process_init receivers run inside billiard's child-startup window. Celery SIGKILLs any child not fully up within worker_proc_alive_timeout (default 4.0 seconds). But configure_logging()'s synchronous /ready probe against Loki is 3s connect + 3s read, and DNS resolution is unbounded by either — so when Loki degrades, every prefork child blows the 4s window and gets SIGKILL'd, and the pool enters a kill/respawn loop precisely during the outage you need logs from. Fix: raise worker_proc_alive_timeout in the Celery config (30.0s in the reference implementation) whenever a worker_process_init receiver calls configure_logging().

Fix: complete Celery app wired for both traps:

"""Celery application with Loki-routed logging."""

import logging
import os
import sys
from celery import Celery
from celery.signals import setup_logging, worker_process_init
from byteforge_loki_logging import configure_logging

redis_url = os.environ.get('MY_APP_REDIS_URL', 'redis://localhost:6379/0')

celery = Celery('my_app', broker=redis_url, include=['tasks'])

celery.conf.update(
    task_serializer='json',
    accept_content=['json'],
    # The worker_process_init hook below runs configure_logging() in each
    # prefork child, which probes the Loki endpoint synchronously (3s
    # connect + 3s read, and a hung DNS resolver isn't bounded by either).
    # Celery SIGKILLs any child that isn't up within this timeout
    # (default 4.0s) — leaving the default puts every child spawn in a
    # kill/respawn loop exactly when Loki is degraded.
    worker_proc_alive_timeout=30.0,
    # ... your other celery config (task_acks_late, beat_schedule, etc.) ...
)


def _configure_loki_logging() -> None:
    debug_mode = os.environ.get('DEBUG_LOCAL', 'true').lower() == 'true'
    log_level = os.environ.get('LOG_LEVEL', 'INFO')
    configure_logging(
        application_tag='my-app',
        debug_local=debug_mode,
        local_level=log_level,
    )

    # Celery's loggers attach their own StreamHandlers with propagate=False,
    # bypassing the root logger's Loki handler. Clear and propagate; do NOT
    # setLevel() on any of them — see the Flask + Gunicorn section for why
    # pinning a child logger's level breaks LOG_LEVEL inheritance.
    # `celery.redirected` matters if celery's --without-mingle / worker CLI
    # enabled redirect_stdouts (which reroutes print()/sys.stdout writes
    # through logging); `celery.pool` carries billiard pool lifecycle.
    for name in ('celery', 'celery.worker', 'celery.task',
                 'celery.app.trace', 'celery.beat', 'celery.redirected',
                 'celery.pool'):
        dep_logger = logging.getLogger(name)
        dep_logger.handlers.clear()
        dep_logger.propagate = True


def _configure_or_die() -> None:
    # celery's Signal.send catches and swallows Exception from receivers —
    # AND still counts a failed receiver as "logging was configured", so a
    # misconfig (e.g. DEBUG_LOCAL=false with a LOKI_* var missing, which
    # raises RuntimeError) would leave the worker running with ZERO root
    # handlers: a silent logging blackout on the write path. SystemExit is
    # a BaseException, so it escapes the swallow and kills the process
    # loudly — the same crash-on-misconfig behavior the Flask container
    # gets from create_app().
    try:
        _configure_loki_logging()
    except Exception as exc:
        print(f"FATAL: logging configuration failed: {exc}", file=sys.stderr)
        raise SystemExit(1) from exc


@setup_logging.connect
def _on_setup_logging(**kwargs) -> None:
    # Connecting to setup_logging fully suppresses celery's default
    # logging config (including worker_hijack_root_logger), so
    # MainProcess events (broker connect, "ready", beat schedule ticks)
    # reach our Loki handler instead of celery's stdout-only StreamHandler.
    _configure_or_die()


@worker_process_init.connect
def _on_worker_process_init(**kwargs) -> None:
    # Each prefork pool child needs its own configure_logging(): the
    # parent's QueueListener thread and requests.Session SSL context do
    # not survive fork(). Without this, child logs go to a dead queue.
    _configure_or_die()

Verification checklist:

• Start the worker with a missing LOKI_* env var. Expect FATAL: logging configuration failed: ... on stderr followed by process exit, NOT a running worker that swallows tasks silently. If the worker starts, _configure_or_die isn't wired or setup_logging/worker_process_init isn't connected. (No broker needed — setup_logging fires from on_init_blueprint before any broker connection attempt.)
• With Loki up and LOG_LEVEL=INFO, run a task that calls logger.debug(...) from a logging.getLogger(__name__) at module scope. It should NOT appear in Loki. If it does, one of the celery.* loggers in the propagation loop has been setLevel()'d somewhere.
• Simulate Loki degradation by pointing LOKI_ENDPOINT at a black-hole endpoint (e.g. https://198.51.100.1/loki/api/v1/push). Workers should still start within 30s (not 4s), then log fallback warnings and continue serving tasks. If workers kill/respawn-loop, worker_proc_alive_timeout isn't set high enough.

MCP Server (FastMCP / uvicorn)

Use this section when integrating with an MCP server built on FastMCP (the high-level wrapper in the mcp Python SDK). Verified end-to-end against mcp==1.27.x / FastMCP streamable-http and sse transports on 2026-05-18.

mcp package version — bound to <2.0. Every import and constructor in this section (from mcp.server.fastmcp import FastMCP, FastMCP(...), mcp.streamable_http_app(), Context from mcp.server.fastmcp) is 1.x API. The 2.0.0 release (2026-07-28) renamed FastMCP → MCPServer and moved mcp.server.fastmcp → mcp.server.mcpserver, so this example silently breaks on the first fresh install that resolves to 2.x. Bound the requirement in your project's requirements.txt:

mcp>=1.8,<2.0

The 1.8 minimum supplies Streamable HTTP, stateless_http, and streamable_http_app() used below. Update existing dependency locks and sync the environment as well as changing the declaration.

See the mcp-docker-deployment skill's requirements.txt subsection for the full rationale and the deliberate-migration path to 2.x.

The problem

FastMCP's mcp.run(transport="streamable-http") internally constructs a uvicorn.Config without passing log_config=. uvicorn falls back to its default LOGGING_CONFIG, which attaches its own StreamHandler instances to the uvicorn, uvicorn.access, and uvicorn.error loggers with propagate=False. Those records bypass Python's root logger entirely, which means the configure_logging() handler we installed on root never sees them — and HTTP access logs (POST /mcp ... 200) never reach Loki. The container looks alive (you can curl the MCP successfully), but Loki shows nothing under application=<your-tag>.

This is identical in shape to the Flask + Gunicorn case for the same reason (third-party loggers with their own handlers + propagate=False), but the fix is different because FastMCP doesn't expose a hook analogous to create_app().

The fix

FastMCP.run() does not accept a log_config kwarg. Bypass mcp.run() for HTTP transports and drive uvicorn directly, reusing the Starlette app that FastMCP builds:

import asyncio
import logging
import os
import sys

import uvicorn
from byteforge_loki_logging import configure_logging
from mcp.server.fastmcp import FastMCP

mcp = FastMCP("my-mcp-server", host="0.0.0.0", port=8000, stateless_http=True)

# ... register tools with @mcp.tool() ...


def main() -> None:
    transport = os.environ.get("MCP_TRANSPORT", "stdio")
    if transport not in {"stdio", "streamable-http", "sse"}:
        raise ValueError(f"Unsupported MCP_TRANSPORT: {transport!r}")
    log_level = os.environ.get("LOG_LEVEL", "INFO").upper()

    if transport == "stdio":
        # stdout belongs exclusively to MCP JSON-RPC. configure_logging()
        # uses stdout in debug mode AND when Loki is unreachable, so skip it.
        # force=True replaces existing root handlers, including stdout ones.
        logging.basicConfig(level=log_level, stream=sys.stderr, force=True)
        mcp.run(transport="stdio")
        return

    # configure_logging() in main() runs BEFORE uvicorn boots. uvicorn doesn't
    # fork by default — unlike gunicorn — so the QueueListener + SSL session
    # the Loki handler creates are inherited safely by uvicorn's own coroutines.
    # No post-fork concerns here; this is the right place.
    debug_mode = os.environ.get("DEBUG_LOCAL", "true").lower() == "true"
    configure_logging(
        application_tag="my-mcp-server",
        debug_local=debug_mode,
        local_level=log_level,
    )

    # For streamable-http / sse, drive uvicorn directly so we can pass a
    # log_config that routes uvicorn.* loggers through the root logger
    # (where byteforge-loki-logging's handler ships to Loki).
    starlette_app = (
        mcp.streamable_http_app() if transport == "streamable-http"
        else mcp.sse_app()
    )
    uv_config = uvicorn.Config(
        starlette_app,
        host=os.environ.get("FASTMCP_HOST", "0.0.0.0"),
        port=int(os.environ.get("FASTMCP_PORT", "8000")),
        log_level=log_level.lower(),
        log_config=_uvicorn_log_config_propagating_to_root(),
    )
    asyncio.run(uvicorn.Server(uv_config).serve())


def _uvicorn_log_config_propagating_to_root() -> dict:
    """uvicorn log_config that routes uvicorn.* loggers through Python root.

    `disable_existing_loggers: False` is CRITICAL — without it, dictConfig
    silently wipes the root configuration that configure_logging() just
    installed. Setting `handlers: []` + `propagate: True` on each uvicorn
    logger means records bubble up to root where byteforge-loki-logging's
    handler picks them up.
    """
    return {
        "version": 1,
        "disable_existing_loggers": False,
        "loggers": {
            "uvicorn": {"handlers": [], "level": "INFO", "propagate": True},
            "uvicorn.access": {"handlers": [], "level": "INFO", "propagate": True},
            "uvicorn.error": {"handlers": [], "level": "INFO", "propagate": True},
        },
    }


if __name__ == "__main__":
    main()

Why this works (and why it's not a hack)

FastMCP.streamable_http_app() and FastMCP.sse_app() are documented public methods that return a Starlette app. The only thing mcp.run() does on top of uvicorn.run(starlette_app, ...) is wire host/port from FastMCP's settings — which we read directly from the environment anyway. So we're not bypassing internals; we're using FastMCP's documented building blocks plus uvicorn's documented log_config parameter.

If FastMCP ever adds a log_config kwarg to mcp.run(), this whole section becomes a one-line mcp.run(transport=..., log_config=_uvicorn_log_config_propagating_to_root()).

Tool-invocation logging

configure_logging() only sets up the transport — application-level log lines still need to come from somewhere. FastMCP tools are decorated with @mcp.tool() and may take an optional Context parameter (which FastMCP strips from the LLM-visible tool schema). Use it to read request headers and emit per-call audit logs:

import time
from mcp.server.fastmcp import Context

logger = logging.getLogger(__name__)


def _caller_from(ctx: Context) -> tuple[str | None, str | None]:
    """Return (X-Client-ID, X-Client-Name) if present on the current request.

    For HTTP transports (streamable-http, sse), nginx — or whatever gateway
    sits in front — typically forwards client identity via custom headers
    after upstream auth. For stdio, there's no HTTP request; this returns
    (None, None) and callers must handle that gracefully.
    """
    try:
        request = ctx.request_context.request
    except (ValueError, AttributeError):
        # No active request (stdio) or Context outside request scope
        return None, None
    if request is None or not hasattr(request, "headers"):
        return None, None
    return (
        request.headers.get("x-client-id"),
        request.headers.get("x-client-name"),
    )


class _LogToolCall:
    """Async context manager: time a tool call and emit a structured log line
    on exit (success or error). Wrap each tool body with `async with`."""

    def __init__(self, tool: str, ctx: Context) -> None:
        self._tool = tool
        self._ctx = ctx
        self._start = 0.0
        self._error: str | None = None

    async def __aenter__(self) -> "_LogToolCall":
        self._start = time.monotonic()
        return self

    async def __aexit__(self, exc_type: object, _exc: object, _tb: object) -> None:
        if exc_type is not None and isinstance(exc_type, type):
            self._error = exc_type.__name__
        caller_id, caller_name = _caller_from(self._ctx)
        duration_ms = round((time.monotonic() - self._start) * 1000, 2)
        logger.info(
            "tool_invoked",
            extra={
                "tool": self._tool,
                "caller_id": caller_id,
                "caller_name": caller_name,
                "duration_ms": duration_ms,
                "error": self._error,
            },
        )


@mcp.tool()
async def my_read_tool(some_arg: str, ctx: Context) -> dict:
    """..."""
    async with _LogToolCall("my_read_tool", ctx):
        return await do_the_thing(some_arg)

Under byteforge-loki-logging's JSON formatter, each extra= field becomes a top-level key in Loki, so you can query with {application="my-mcp-server"} | json | tool="my_read_tool" or ... | duration_ms > 100.

Verification

Both the uvicorn-through-root wiring and the tool instrumentation should be smoke-tested locally before deploying to a Loki-backed environment:

DEBUG_LOCAL=true \
LOG_LEVEL=INFO \
MCP_TRANSPORT=streamable-http \
FASTMCP_HOST=127.0.0.1 FASTMCP_PORT=8000 \
python -m my_mcp_package.server > /tmp/srv.log 2>&1 &
SRV=$!
sleep 2

# Initialize, then invoke a tool with custom client-identity headers
curl -sS -X POST http://127.0.0.1:8000/mcp \
    -H "Content-Type: application/json" \
    -H "Accept: application/json, text/event-stream" \
    -H "X-Client-ID: smoke-test" -H "X-Client-Name: SmokeTest" \
    -d '{"jsonrpc":"2.0","method":"initialize","id":1,"params":{"protocolVersion":"2024-11-05","capabilities":{},"clientInfo":{"name":"smoke","version":"0.1"}}}' \
    -o /dev/null

kill $SRV; wait $SRV 2>/dev/null

# You should see every line formatted by byteforge-loki-logging, including:
#   <ts> - uvicorn.error - INFO - Started server process [...]
#   <ts> - uvicorn.error - INFO - Uvicorn running on http://127.0.0.1:8000
#   <ts> - uvicorn.access - INFO - 127.0.0.1:... - "POST /mcp HTTP/1.1" 200
#   <ts> - __main__ - INFO - tool_invoked
# If `uvicorn.access` lines are formatted by uvicorn instead (look for
# "INFO:     " with the column-aligned padding), then log_config didn't take
# effect — re-check that you're driving uvicorn yourself, not via mcp.run().
grep -E "(uvicorn|tool_invoked)" /tmp/srv.log

Stdio transport

For stdio (e.g. local use with Claude Code), use the early-return branch above: configure Python logging to stderr, then call mcp.run(transport="stdio"). The MCP transport specification reserves stdout for MCP messages.

Do not call configure_logging() before this branch, at module import, or from a stdio tool. In byteforge-loki-logging 0.1.8, both debug_local=True and the unreachable-Loki fallback install a root StreamHandler(sys.stdout). Setting DEBUG_LOCAL=false alone cannot protect the protocol stream. This stdio branch skips Loki regardless of DEBUG_LOCAL, needs no LOKI_* variables, and sends application logs to stderr rather than Loki.

force=True replaces existing root handlers. Any explicitly attached handlers on named loggers must also use stderr or propagate to this root handler; remove stdout handlers and route diagnostic print() calls to stderr. Keep startup and imported modules free of stdout diagnostics as well.

Verify with a stdio client: initialize, list tools, then invoke a tool that emits a log record. Capture stdout and stderr separately; every stdout line must be a valid MCP JSON-RPC message, with the tool's diagnostic log on stderr. Repeat with DEBUG_LOCAL=true and false and no Loki variables to cover both modes.

Common pitfalls

• Calling configure_logging() before selecting the MCP transport. Its local and fallback stdout handlers corrupt stdio JSON-RPC; take the stderr-only stdio branch first.
• mcp.run(transport="streamable-http", log_config=...) — does NOT exist. FastMCP's run() accepts only transport and mount_path as of 2026-05-18. Passing log_config= raises TypeError. You must construct uvicorn.Config yourself.
• disable_existing_loggers: True (the default!) silently wipes byteforge-loki-logging's root setup. Always set it to False in your uvicorn log_config dict.
• Forgetting to set handlers: [] on uvicorn loggers. If you leave handlers at their default (["default"]), uvicorn's own StreamHandler stays attached AND records propagate to root — you get duplicate output. Empty list + propagate=True is the right combo.
• Not adding ctx: Context to tool signatures. Without it, _caller_from() can't read the request headers and your audit logs are missing per-caller identity. FastMCP strips Context from the LLM-visible tool schema, so the surface to clients is unchanged.

Troubleshooting

Local stdio MCP fails to connect / JSON parse errors after adding Loki logging:

• configure_logging() installs a stdout handler in local mode or when Loki is unreachable, mixing log lines with MCP JSON-RPC.
• Fix: use the MCP section's stderr-only stdio branch before any configure_logging() call. Check imported startup code, named logger handlers, and print() calls for stdout diagnostics too. Turning off DEBUG_LOCAL alone is insufficient.

Logs not appearing in Loki (production):

• Verify DEBUG_LOCAL=false is set
• Check all LOKI_* variables are correct
• Test CA certificate path is accessible in container
• Verify Loki endpoint is reachable from container
• Check stderr for fallback warnings — if you see "Falling back to stdout", the Loki connection failed at startup

Missing CA certificate error:

• Private CA: ensure the volume mount is correct in docker-compose.yaml, LOKI_CA_BUNDLE_PATH points to /app/certs/loki-ca.pem, and the source file exists on the host at the mounted path
• Publicly trusted certificate: verify LOKI_CA_BUNDLE_PATH points at a system CA bundle that exists in the image (see Step 4)

Runtime error: Missing required environment variables:

• Only occurs when DEBUG_LOCAL=false. _validate_loki_env_vars() hard-requires ALL FOUR LOKI_* vars (endpoint, user, password, ca_bundle_path) to be non-empty — regardless of whether the endpoint uses a private or public CA.
• Fix: set every one. For LOKI_CA_BUNDLE_PATH on a public-CA endpoint (Grafana Cloud etc.), point at the system trust store, e.g. /etc/ssl/certs/ca-certificates.crt on Debian/Ubuntu/Alpine — NOT the string "false" unless you accept disabling TLS verification. See Step 5's LOKI_CA_BUNDLE_PATH block.
• Check spelling (LOKI_, not MZ_LOKI_ or MATERIA_LOKI_).
• Where you see it depends on the container: under gunicorn the RuntimeError crashes the worker at boot and gunicorn respawn-loops (visible in docker logs). Under celery WITHOUT the _configure_or_die wrapper from the Celery Workers section, celery's signal machinery swallows the exception AND the worker keeps running with zero root handlers — silent blackout. See "Celery worker running but no task logs in Loki, container is green" below.

Flask route exceptions not appearing in Loki:

• Flask catches exceptions in route handlers and logs them via werkzeug to stdout/stderr, bypassing the root logger
• Fix: clear Flask's default handlers and propagate to root logger (see Flask integration section above)
• Symptoms: 500 errors appear in nginx/container logs but not in Grafana/Loki

app.logger.debug(...) shipping to Loki even when LOG_LEVEL=INFO:

• An app.logger.setLevel(logging.DEBUG) call somewhere in create_app() pinned Flask's own logger below root's effective level. CPython's logging filters at two points: (a) at origination, isEnabledFor(level) walks the ancestor chain via getEffectiveLevel() and returns the first non-NOTSET level it finds; (b) during propagation, each ancestor's handlers are called, but the ancestors' own levels are NOT re-checked. So if app.logger is NOTSET, the walk-up reaches root, finds LOG_LEVEL (say INFO), and DEBUG is suppressed at origination. But if app.logger is pinned to DEBUG, the walk-up stops there and never reaches root — the record is created, propagates to root's handler (which is itself NOTSET), and reaches Loki. Ordinary module loggers (logging.getLogger(__name__) in blueprints) are unaffected because they remain NOTSET and correctly inherit LOG_LEVEL via the walk-up.
• Fix: delete the app.logger.setLevel(...) call. Leave it at NOTSET so it inherits root's level (which configure_logging() set from LOG_LEVEL). Keep the handlers.clear() + propagate = True lines — those are still load-bearing for the propagation half of the story.
• Symptoms: with LOG_LEVEL=INFO, blueprint DEBUG lines are suppressed but app.logger.debug(...) reaches Loki anyway. Under gunicorn the split is wider than it looks because app.logger is logging.getLogger('<module_name>') — the same object your app module itself logs through.

Loki SSL errors on first few startup log messages (gunicorn, byteforge-loki-logging before 0.1.11):

• configure_logging() was called at module level, which runs in gunicorn's master process before fork(). The Loki handler's requests.Session and its SSL context were initialized pre-fork, then broke in the child worker because SSL contexts don't survive fork().
• Fix: move configure_logging() inside create_app() so it runs post-fork in the worker process (see Flask + Gunicorn section above), or upgrade to 0.1.11 or later
• Symptoms: SSL errors only on the first few startup log messages, then everything works fine after the session recovers

Gunicorn/werkzeug logs going to stdout instead of Loki:

• Werkzeug and gunicorn create their own loggers with propagate=False and their own StreamHandler instances
• Fix: clear handlers and set propagate=True on werkzeug, gunicorn, gunicorn.error, and gunicorn.access loggers inside create_app() (see Flask + Gunicorn section above)
• Symptoms: application logs visible in docker logs or container stdout but missing from Grafana/Loki

Gunicorn startup, bind, and worker-timeout messages missing from Loki (app logs arrive fine):

• Expected by default, not a propagation bug. These come from gunicorn's master process, which never imports the app and so never runs configure_logging().
• Fix: read them with docker logs <container>, or on byteforge-loki-logging 0.1.11 or later send them to Loki. On older versions, do not move configure_logging() to module level (it breaks the Loki handler across fork()).
• Symptoms: logger="gunicorn.error" shows Worker exiting records but never Starting gunicorn / Listening at / Booting worker; a container that fails to bind or boot looks silent in Loki.

Logs from multiprocessing.Process child processes not appearing in Loki:

• Before byteforge-loki-logging 0.1.11, the parent's QueueListener thread and requests.Session do not survive fork(). The child inherits a dead handler — logs go into a queue that nobody reads. No errors are raised.
• Fix: call configure_logging() at the top of the child process target function, before any logging (see Multiprocessing section above), or upgrade to 0.1.11 or later
• Symptoms: parent process logs appear in Loki, but all child process logs vanish silently

Celery worker running but no task logs in Loki, container is green:

• The setup_logging / worker_process_init receiver raised an exception during configure_logging() (typically a RuntimeError from _validate_loki_env_vars when a LOKI_* env var is missing under DEBUG_LOCAL=false). Celery's Signal.send swallowed the exception AND counted the failed receiver as "logging was configured", so the worker ran on with zero root handlers. Tasks keep acking; nothing writes to Loki.
• Fix: wrap the receiver body in try/except Exception → raise SystemExit(1). SystemExit is a BaseException, so it escapes celery's swallow and kills the worker loudly on misconfig. See the Celery Workers section for the full pattern.
• Symptoms: celery worker is running, health checks pass, tasks return SUCCESS, but application=<your-tag> is completely absent from Loki. Stderr may show the RuntimeError traceback exactly once (from the swallow's default logger before it dies) but no further errors.

Celery workers kill/respawn-looping during a Loki outage:

• configure_logging()'s synchronous /ready probe (3s connect + 3s read + unbounded DNS) is called from worker_process_init, which runs inside billiard's child-startup window. Celery SIGKILLs any child not up within worker_proc_alive_timeout (default 4.0s). When Loki is slow or DNS-degraded, every prefork child blows the window and gets killed before it can log a fallback warning.
• Fix: set worker_proc_alive_timeout=30.0 (or higher) in the Celery config alongside the worker_process_init receiver. See the Celery Workers section.
• Symptoms: docker logs <worker> shows a stream of celery.worker.consumer.MainProcess ... Process 'ForkPoolWorker-N' pid:X exited with 'signal 9 (SIGKILL)' messages during a broader Loki incident, and the worker never gets to the "ready" state.

MCP server logs / uvicorn access logs not appearing in Loki:

• mcp.run(transport="streamable-http") internally calls uvicorn.Config(...) without a log_config kwarg, so uvicorn falls back to its default config that attaches StreamHandlers to uvicorn, uvicorn.access, and uvicorn.error with propagate=False. Those records bypass root and never reach the byteforge-loki-logging handler.
• Fix: drive uvicorn directly via mcp.streamable_http_app() + uvicorn.Config(..., log_config={...}) (see the "MCP Server (FastMCP / uvicorn)" section above).
• Symptoms: container is alive (MCP initialize handshake succeeds via curl), application=<your-tag> is completely absent from Loki, docker logs <container> shows uvicorn's native INFO: 127.0.0.1:... - "POST /mcp HTTP/1.1" 200 OK format with the column-aligned padding (rather than byteforge-loki-logging's <ts> - <logger> - <level> - <message> format).

Import error for byteforge_loki_logging:

• Run pip install -r requirements.txt
• Verify installed: pip list | grep byteforge-loki-logging

Example Implementation

# myapp.py
import os
import logging
from flask import Flask
from byteforge_loki_logging import configure_logging


def create_app() -> Flask:
    debug_mode = os.environ.get('DEBUG_LOCAL', 'true').lower() == 'true'
    log_level = os.environ.get('LOG_LEVEL', 'INFO')
    configure_logging(
        application_tag='myapp',
        debug_local=debug_mode,
        local_level=log_level,
    )

    app = Flask(__name__)

    # Leave app.logger and the dep loggers below at NOTSET so LOG_LEVEL
    # (applied to root by configure_logging) is honored via the ancestor
    # walk-up. See the Flask + Gunicorn section for the full rationale.
    app.logger.handlers.clear()
    app.logger.propagate = True

    for name in ('werkzeug', 'gunicorn', 'gunicorn.error', 'gunicorn.access'):
        dep_logger = logging.getLogger(name)
        dep_logger.handlers.clear()
        dep_logger.propagate = True

    # ... register blueprints, configure Api, etc. — before or after the
    # loop above; the order does not matter (see the note below the
    # Flask + Gunicorn example).

    return app

Start it with gunicorn "myapp:create_app()".

# docker-compose.yaml
services:
  myapp:
    # Private CA shown; see Step 4 for a publicly trusted certificate.
    volumes:
      - ./certs/loki-ca.pem:/app/certs/loki-ca.pem:ro
    environment:
      - DEBUG_LOCAL=false
      - LOG_LEVEL=INFO
      - LOKI_ENDPOINT=${LOKI_ENDPOINT}
      - LOKI_USER=${LOKI_USER}
      - LOKI_PASSWORD=${LOKI_PASSWORD}
      - LOKI_CA_BUNDLE_PATH=/app/certs/loki-ca.pem

This provides structured logging locally during development and automatic Loki shipping in production with secure encrypted connections.

Reporting problems with this skill

If you find a bug or stale instruction while using this skill:

• If HiveMake is connected and byteforge-skills-maintainer-agent is available, file a ticket with that agent. Use discover_agents to find its project ID.
• Otherwise, open an issue at GitHub Issues. If you cannot submit a report, give the human the details and this link.

Include the skill name, plugin version if known, the instruction followed, expected and actual results, relevant errors/runtime versions, and any fix or workaround that worked. Omit credentials and private data.

Address the user's immediate problem first; report the skill defect afterward.