rudow

Rudow — AI that reads your logs and tells you what broke, in plain English.

Send logs to Rudow, then ask questions like "why did checkout errors spike this morning?" from the dashboard, Slack, or any MCP-compatible AI agent (Claude Code, Cursor, Windsurf).

• Read-only by design — Rudow never mutates, rewrites, or acts on your systems. It reads logs. That's it.
• Flat pricing — monthly caps, never usage meters. Limits are caps, not overage charges.
• Free-forever tier — the Hobby plan is free, no credit card, no time limit.
• MCP server — claude mcp add --transport http rudow https://mcp.rudow.ai/mcp

Batches log entries and ships them to the ingest API.

Homepage: https://rudow.ai · Documentation: https://rudow.ai/docs

Install

pip install rudow

Quick Start

from rudow import create_logger

logger = create_logger({
    "api_key": "your-api-key-here",
})

logger.info("Server started", {"port": 8000})
logger.error("Database connection failed", {"error": "ECONNREFUSED"})

Configuration

Options passed to create_logger():

• api_key (required): Your project's API key
• endpoint: Ingest endpoint URL (defaults to https://api.rudow.ai/v1/ingest)
• batch_size: Flush when this many entries are buffered (default: 20)
• flush_interval_ms: Flush at most this often in milliseconds (default: 2000)
• source: Logical source label (default: "backend")
• max_buffer_size: Max entries kept in memory (default: 10000)
• max_retries: Send attempts per flush before re-queueing (default: 4)
• retry_backoff_ms: Base delay for exponential backoff (default: 500)
• gzip_threshold: Compress payloads larger than this (bytes, 0 to disable, default: 1024)
• persist_path: File path for crash-durable buffering (optional)
• flush_on_exit: Auto-flush on SIGTERM/SIGINT (default: True)
• fallback_to_stderr: Print undelivered logs to stderr (default: True)
• enable_metrics_heartbeat: Send periodic metrics (default: True)
• metrics_heartbeat_ms: Metrics interval in milliseconds (default: 15000)

Log Levels

The logger supports these levels:

• debug(message, attributes)
• info(message, attributes)
• warn(message, attributes)
• error(message, attributes)
• fatal(message, attributes)

The generic log(level, message, attributes) method is also available.

Structured Events

The SDK supports rich structured event types:

HTTP Requests

Record completed HTTP requests with timing and status:

from rudow import record_http_request, HttpRequestEvent

record_http_request(logger, HttpRequestEvent(
    method="GET",
    route="/api/users",
    status=200,
    duration_ms=125.5,
    extra={"user_id": "user123"}  # Optional context
))

Exceptions

Record exceptions with stack traces:

from rudow import record_exception, ExceptionEvent

try:
    do_something()
except Exception as e:
    record_exception(logger, ExceptionEvent(
        error=e,
        method="POST",
        route="/api/process",
        status=500,
        extra={"request_id": "req-123"}
    ))

Metrics

Periodically emit process metrics (uptime, memory):

from rudow import start_metrics_heartbeat

stop = start_metrics_heartbeat(logger, interval_ms=15000)

# Later, to stop:
stop.set()

Durability

• Batching: Entries are buffered and sent every 2 seconds or when 20 entries accumulate
• Retries: Failed sends are retried with exponential backoff
• Persistence: With persist_path set, undelivered entries survive process restarts
• Graceful shutdown: Automatic flush on exit; call logger.flush() to ensure delivery

Example with FastAPI

See examples/fastapi-app/ for a complete example application.

from fastapi import FastAPI
from rudow import create_logger, record_http_request, record_exception, HttpRequestEvent
import os
import time

app = FastAPI()
logger = create_logger({"api_key": os.environ.get("RUDOW_API_KEY", "test-key")})

@app.middleware("http")
async def log_requests(request, call_next):
    start = time.time()
    try:
        response = await call_next(request)
        duration_ms = (time.time() - start) * 1000
        
        # Record structured HTTP event
        record_http_request(logger, HttpRequestEvent(
            method=request.method,
            route=request.url.path,
            status=response.status_code,
            duration_ms=duration_ms
        ))
        return response
    except Exception as exc:
        duration_ms = (time.time() - start) * 1000
        record_exception(logger, ExceptionEvent(
            error=exc,
            method=request.method,
            route=request.url.path,
            duration_ms=duration_ms
        ))
        raise

@app.get("/api/users")
async def get_users():
    logger.info("Fetching users", {"endpoint": "/api/users"})
    return {"users": []}

@app.exception_handler(Exception)
async def exception_handler(request, exc):
    record_exception(logger, ExceptionEvent(
        "path": str(request.url),
        "error": str(exc),
    })
    return {"error": str(exc)}