Argus

The read path for observability. Query, correlate, and stream Prometheus, Loki & Tempo - for your agents and your UI.

Argus is a Python library (plus optional FastAPI router, MCP server, and CLI) that sits in front of your observability stack and handles everything a custom dashboard, internal tool, or AI agent would otherwise have to re-implement:

• Step math - automatic "nice" step selection, range alignment for cacheable responses, and $__interval / $__rate substitution with safe rate windows (the logic Grafana applies internally).
• Log pagination - opaque cursors over Loki's cursorless query API, with nanosecond-boundary dedup.
• Live tailing - polling-based tail (no fragile /tail WebSocket) as an async iterator or an SSE endpoint, with subscriber fan-out.
• Query guardrails - range/point/limit/concurrency caps so an over-eager agent or an auto-refreshing dashboard can never take the stack down. Rejections explain how to fix the query.
• Cross-signal correlation - one trace id in; the trace, its logs, and RED metrics for every involved service out. Gaps are reported honestly in notes, never silently dropped.
• Anomaly flagging - robust z-score (MAD) over any PromQL result.
• Multi-tenancy - X-Scope-OrgID header injection (authoritative) and optional label-matcher injection (best-effort).
• Normalized models - Pydantic models everywhere; NaN/Inf serialize to null, base64 trace ids become hex, string floats become floats.

Install

pip install -e .              # library only
pip install -e '.[fastapi]'   # + FastAPI router / SSE server
pip install -e '.[mcp]'       # + MCP server for agents
pip install -e '.[dev]'       # everything + pytest

Library

from argus import Telemetry

t = Telemetry(
    prometheus="http://localhost:9090",
    loki="http://localhost:3100",
    tempo="http://localhost:3200",
)

series = await t.metrics.range('sum(rate(http_requests_total[$__rate]))', last="1h")
page = await t.logs.search('{service_name="gateway"} |= "error"', last="1h")
next_page = await page.next()
trace = await t.traces.get("4bf92f3577b34da6a3ce929d0e0e4736")
inv = await t.correlate("4bf92f3577b34da6a3ce929d0e0e4736")

async for entry in t.logs.tail('{service_name="api"}', max_seconds=60):
    print(entry.line)

Or load argus.yaml (copy argus.example.yaml to get started - env vars like ${PROM_TOKEN} are expanded):

t = Telemetry.from_config("argus.yaml")

HTTP server (SSE included)

argus-o11y --config argus.yaml serve --port 8000

Endpoints: /health, /services, /metrics/range|instant|anomalies|labels, /logs/search|tail|labels, /traces/search, /traces/{id}, /correlate/{id}. /logs/tail streams SSE. Or mount into an existing app:

from argus.server import make_router
app.include_router(make_router(t), prefix="/telemetry")

MCP server (for agents)

argus-o11y --config argus.yaml mcp

Exposes every capability as a tool (query_metrics_range, search_logs, get_trace, correlate, ...) over stdio, guarded by the same query limits.

CLI

argus-o11y --prometheus http://localhost:9090 metrics range 'sum(rate(http_requests_total[$__rate]))'
argus-o11y --loki http://localhost:3100 logs search '{service_name="api"} |= "error"' --limit 50
argus-o11y --tempo http://localhost:3200 correlate 4bf92f3577b34da6a3ce929d0e0e4736
argus-o11y --config argus.yaml logs tail '{service_name="api"}'   # NDJSON

Every command prints one JSON document, so output pipes cleanly into jq. Exit codes: 0 ok, 1 Argus error, 130 interrupted.

Tests

pip install -e '.[dev]'
pytest

The suite runs entirely in-process against fake backends and recorded HTTP responses - no Prometheus, Loki, or Tempo required.

License

Apache-2.0