uptime-agent

Ship the errors your Python app already logs to UptimeAgent, so an incident arrives with your own error lines attached to the outside-in probe evidence.

Status: 0.x. Minor versions may break until 1.0.0. The wire protocol is frozen and executable (docs/wire-protocol-v1.md, conformance/), and this package passes every fixture in it.

pip install uptime-agent

The distribution is uptime-agent; the import is uptimeagent.

import uptimeagent

agent = uptimeagent.uptime_agent(
    service="checkout-api",  # required: how the Agents screen groups instances
    serves=["checkout.acme.com"],  # the hostnames exactly as they appear in Monitors
)
# token comes from UPTIMEAGENT_TOKEN

The boundary: evidence, never detection

A quiet package is never read as downtime, and a package reporting healthy is never read as uptime. A process that has stopped answering cannot vote on its own health, so detection belongs entirely to UptimeAgent's external probe regions. Nothing this package sends, or fails to send, can open, close or influence an incident.

What it never does

• Never raises into your code after uptime_agent() returns. Every public method and every hook is wrapped; a failure drops the event and bumps a counter you can read with agent.stats().
• Never blocks your request path. emit is a redaction walk with a hard budget and a buffer append. Network I/O happens on one daemon thread that holds nothing open.
• Never grows without bound. The buffer is hard-capped (lines and bytes), drop-oldest, with the in-flight batch still counted against the cap.
• Never opens a listening socket, never installs a signal handler, never patches print or sys.stdout. There is no inbound control path.
• Never reads an environment variable outside the documented list: the UPTIMEAGENT_* options, the deprecated UA_TOKEN, and uptimeagent.PLATFORM_ENV_VARS (platform detection, proxies, and the three variables the standard library's TLS setup reads: SSL_CERT_FILE, SSL_CERT_DIR, SSLKEYLOGFILE). A test wraps os.environ and fails on any read outside that list.
• Zero runtime dependencies. pip show uptimeagent prints an empty Requires: line. The sdist on PyPI is the exact source that runs in your process.

What it collects

Only what you point it at. Lines at or above min_level (default warn) from the hooks you attach, plus three synthesised kinds: an uncaught exception (source=exception), a failing HTTP request (source=http: method, route, status, duration), and process start and exit (source=process). Every line is redacted in-process before it is buffered.

Redaction

Runs before anything leaves the process, with a fixed node, depth and byte budget so a large or cyclic object cannot slow a request. Two categories, two markers:

• Keys: [redacted:keys]. A key is split into words and every contiguous run of words is matched against the rules, over the whole flattened path, so body.card.number matches cardnumber. Defaults: password passwd pwd secret token apikey accesskey authorization auth cookie session credential privatekey clientsecret otp cvv cardnumber ssn.
• Patterns: [redacted:patterns] over the message and every string value, in a fixed order: bearer tokens, URL credentials, JWTs, AWS access key ids, private key blocks, email addresses.

Your rules extend the defaults; they never replace them.

import re

agent = uptimeagent.uptime_agent(
    service="checkout-api",
    redact={
        "keys": ["customer_ref"],
        "patterns": [
            re.compile(r"\bACC-\d{8}\b"),
            {"name": "iban", "regex": r"\b[A-Z]{2}\d{2}[A-Z0-9]{11,30}\b", "flags": ""},
        ],
    },
)

Default patterns use ASCII \b and \s (re.ASCII), matching the ingest backstop. Your own compiled patterns are used exactly as you compiled them.

Wiring

stdlib logging

On by default: uptime_agent() attaches a handler to the root logger at min_level, so WARNING and above ship and INFO does not. Narrow it with capture_logging={"level": "WARNING", "logger": "myapp"}, or turn it off with capture_logging=False (or UPTIMEAGENT_CAPTURE_LOGGING=0) and attach the handler yourself:

import logging

logging.getLogger("myapp").addHandler(agent.logging_handler())

Levels map by name (CRITICAL/ERROR → error, WARNING → warn, everything else → info), extra= fields become attrs, exc_info becomes the traceback plus exception.name and exception.message, and the logger's name rides in attrs.logger. Records ship as source="logging". structlog and loguru both bridge to stdlib logging; wire the handler there.

WSGI and ASGI

# Flask
app.wsgi_app = agent.wsgi_middleware(app.wsgi_app)
# Django: wrap get_wsgi_application() in wsgi.py
application = agent.wsgi_middleware(get_wsgi_application())
# Starlette / FastAPI, or any ASGI app
app = agent.asgi_middleware(app, route=route_template)

route= is called when the response starts, after routing. For Starlette:

def route_template(scope):
    endpoint = scope.get("endpoint")
    for r in getattr(scope.get("router"), "routes", []):
        if getattr(r, "endpoint", None) is endpoint:
            return r.path  # "/orders/{order_id}"
    return None

A request with status ≥ min_status (default 500) records a source=http line with http.method, http.path, http.status and http.duration_ms. An exception out of the app is recorded and re-raised. The request context is ambient (contextvars), so an error logged three calls down carries the path that produced it. Pass route= to map a request to a low-cardinality template; the default is the raw path without the query string.

with agent.context({"reqId": rid}):
    ...  # every line inside carries reqId

Threads you spawn do not inherit the context; run them with contextvars.copy_context().run.

AWS Lambda

handler = agent.wrap_handler(handler)

On a detected serverless platform (Lambda, Vercel, GCP Functions, Azure Functions) no background thread is started. The buffer is drained at handler end, inside a budget derived from context.get_remaining_time_in_millis(), and the heartbeat rides along.

What it does to your process on a crash

• Uncaught exception on the main thread: sys.excepthook is chained. The line is captured, a synchronous flush runs for at most fatal_flush_timeout_ms (default 2 s), then the previous hook runs. Python prints its own traceback and picks its own exit code; nothing is reproduced or suppressed.
• Uncaught exception in a thread: threading.excepthook is chained. Captured, no flush; the process survives exactly as it would have.
• Exit: an atexit handler records process exiting and flushes with the same deadline. process.exit_code is null unless an uncaught exception was observed (then 1): atexit cannot see the status passed to sys.exit().
• Not captured: os._exit, SIGKILL/OOM, signals whose default action terminates (add your own SIGTERM handler and call agent.close()), KeyboardInterrupt (passed straight through, not recorded), and an app that replaces sys.excepthook after init.

capture_exceptions=False and capture_process_events=False turn these off.

Configuration

Every option has an environment variable. Explicit kwargs win over the environment.

option	env	default
service	UPTIMEAGENT_SERVICE	required
serves / site	UPTIMEAGENT_SERVES (comma-separated) / UPTIMEAGENT_SITE	absent
token	UPTIMEAGENT_TOKEN (UA_TOKEN deprecated)	none: installed and inert
environment	UPTIMEAGENT_ENVIRONMENT	production (a display label; never affects routing)
instance	UPTIMEAGENT_INSTANCE	platform id → hostname → generated
endpoint	UPTIMEAGENT_ENDPOINT	https://ingest.uptimeagent.com
mode	UPTIMEAGENT_MODE; UPTIMEAGENT_DISABLED=1 forces off	send
min_level	UPTIMEAGENT_MIN_LEVEL	warn
redact		{}
capture_exceptions	UPTIMEAGENT_CAPTURE_EXCEPTIONS	True
capture_logging	UPTIMEAGENT_CAPTURE_LOGGING	True
capture_process_events	UPTIMEAGENT_CAPTURE_PROCESS	True
flush_interval_ms	UPTIMEAGENT_FLUSH_INTERVAL_MS	2000
heartbeat_interval_ms	UPTIMEAGENT_HEARTBEAT_INTERVAL_MS	30000
request_timeout_ms	UPTIMEAGENT_REQUEST_TIMEOUT_MS	10000
fatal_flush_timeout_ms	UPTIMEAGENT_FATAL_FLUSH_TIMEOUT_MS	2000
max_buffer_lines / max_buffer_bytes	UPTIMEAGENT_MAX_BUFFER_LINES / _BYTES	10000 / 8 MiB
max_batch_lines / max_batch_bytes	UPTIMEAGENT_MAX_BATCH_LINES / _BYTES	500 / 256 KiB
max_lines_per_second / max_bytes_per_second	UPTIMEAGENT_MAX_LINES_PER_SECOND / _MAX_BYTES_PER_SECOND	500 / 1 MiB
serverless	UPTIMEAGENT_SERVERLESS	auto
debug	UPTIMEAGENT_DEBUG	False
on_error(err, ctx)		none
on_config_error		throw (or disable)

uptime_agent() raises ConfigError only for malformed configuration: a missing service, an unrecognised mode, or a non-HTTPS endpoint (allowed only for localhost, 127.0.0.1, ::1). It happens at startup, before any socket opens. Pass on_config_error="disable" for a warning and an inert agent instead. A missing token never raises: the agent warns once and stays inert.

mode="log": the development path

Renders each batch to stdout as NDJSON and never opens a socket. Needs no backend at all. It is explicit only, never inferred from environment.

Endpoint and environment

The endpoint is never derived from environment. Your staging belongs on the production backend, labelled environment="staging". Proxies: HTTPS_PROXY, HTTP_PROXY and NO_PROXY are honoured (CONNECT tunnelling via urllib); macOS system proxy settings are not consulted. HTTP redirects are not followed.

Failure mode

Everything the SDK cannot do is counted, never raised:

>>> agent.stats().as_dict()
{'service': 'checkout-api', 'instance': 'ip-10-2-3-41', ..., 'enabled': True,
 'disabled_reason': None, 'buffered': {'lines': 0, 'bytes': 0},
 'sent': {'lines': 1204, 'batches': 17},
 'dropped': {'buffer': 0, 'throttle': 0, 'oversize': 0, 'transport': 2, 'internal': 0, 'filtered': 184}}

filtered is lines below min_level. A 401/403 from ingest disables the transport for the life of the process and calls on_error once; after that every line counts as dropped.transport. tests/test_never_hurt_the_app.py is the suite that holds these claims up.

Verification

command	what it proves
make test	~260 tests, including the 89 protocol conformance fixtures and subprocess crash scenarios
make test-matrix	the same on CPython 3.9 through 3.13
make integration	the built wheel under real Flask, Starlette/uvicorn and gunicorn prefork, against a local ingest stub, with a byte scan for planted secrets
make verify	lint, types, tests, package checks: what deploy.sh gates on

Where it runs

CPython 3.9 to 3.13. PyPy is best-effort. gunicorn and uwsgi prefork are handled with os.register_at_fork: a forked worker starts its own sender thread lazily and gets a #<pid> instance suffix. asyncio apps use the ASGI middleware; flush() and close() are synchronous and should not be awaited on an event loop (await asyncio.to_thread(agent.flush)).

Differences from the Node SDK: docs/differences-from-node.md. Publishing: docs/publishing.md. Internal staging notes: docs/staging.md.

MIT. Zero dependencies. Python 3.9+.