replaystack-sdk 1.1.2

ReplayStack Python SDK for backend event capture, exception tracing, and replay debugging.

ReplayStack Python SDK

ReplayStack's Python SDK captures backend API events, exceptions, stack traces, breadcrumbs, and custom events, then ships them to the ReplayStack ingestion API. It's a feature-aligned port of the official @replaystack/sdk for Node.js.

It supports:

• Manual and middleware-driven event capture
• Manual exception capture with parsed stack frames
• Breadcrumb buffer with request-scoped context
• Sensitive data masking (matches the TS default mask list)
• Configurable timeouts, retries, sampling, and payload truncation
• Offline queue + manual flush() / close() lifecycle
• Periodic background flush
• Auto-detection of the request's Authorization mode (bearer / basic / api_key / cookie / other / none) before masking
• Process guards for unhandled exceptions + graceful shutdown flushing
• FastAPI / Starlette, Flask, and Django integrations with middleware options matching TS Express options
• Queue jobs, cron jobs, background workers, and custom events

Default ingestion URL:

https://api.replaystack.co/api/v1/ingest/events

SDK headers sent on every request:

x-tracereplay-api-key: rs_live_xxxxxxxxx
x-replaystack-api-key: rs_live_xxxxxxxxx
x-replaystack-sdk: python
x-replaystack-sdk-version: 1.0.0

x-tracereplay-api-key is the canonical header for the TraceReplay backend; x-replaystack-api-key is kept as a duplicate alias so older receivers keep working.

---

Installation

pip install replaystack-sdk

For framework extras:

pip install replaystack-sdk[fastapi]
pip install replaystack-sdk[flask]
pip install replaystack-sdk[django]
pip install replaystack-sdk[all]

---

Environment Variables

The SDK reads the same variables the TS SDK does:

REPLAYSTACK_API_KEY=rs_live_xxxxxxxxxxxxxxxxx
REPLAYSTACK_ENDPOINT=https://api.replaystack.co
REPLAYSTACK_INGEST_URL=https://api.replaystack.co/api/v1/ingest/events
REPLAYSTACK_SERVICE_NAME=order-service
REPLAYSTACK_ENVIRONMENT=production
REPLAYSTACK_APP_VERSION=1.0.0
REPLAYSTACK_COMMIT_HASH=a7f91c

# Runtime tuning
REPLAYSTACK_ENABLED=true
REPLAYSTACK_TIMEOUT_MS=2500
REPLAYSTACK_RETRIES=1
REPLAYSTACK_SAMPLE_RATE=1.0
REPLAYSTACK_CAPTURE_SUCCESS=true
REPLAYSTACK_MAX_PAYLOAD_SIZE_BYTES=524288
REPLAYSTACK_MAX_BREADCRUMBS=50
REPLAYSTACK_OFFLINE_QUEUE_MAX=100
REPLAYSTACK_FLUSH_INTERVAL_MS=0

REPLAYSTACK_COMMIT_HASH identifies the exact deployed code version that generated the event or exception.

---

Configuration Reference

Every option can be passed as a keyword argument to ReplayStackClient(...). Environment variables (above) act as defaults when the kwarg is omitted.

Argument	Type	Default	Notes
api_key	str	REPLAYSTACK_API_KEY env	Required. Raises ValueError when missing.
endpoint	str	https://api.replaystack.co	Base URL (no path).
ingest_url	str	<endpoint>/api/v1/ingest/events	Override the full ingest URL.
service_name	str	env	Service shown in dashboard.
environment	str	env / development	local / development / staging / production / arbitrary string.
app_version	str	env	Release version (e.g. 1.2.0).
commit_hash	str	env	Git SHA.
enabled	bool	True	Disable the SDK at runtime without removing code.
timeout_seconds	float	2.5	Request timeout per attempt.
max_retries	int	1	Retries on network failure (in addition to the first attempt).
sample_rate	float	1.0	0.1 = capture 10% of events.
capture_success	bool	True	When False, only failed/warning events are sent.
max_payload_bytes	int	524288 (512 KB)	Truncates request/response payloads above this size.
mask_fields	Iterable[str]	[]	Custom field names to mask (case-insensitive).
ignored_paths	Iterable[str]	[]	Path prefixes or prefix* globs to skip.
max_breadcrumbs	int	50	Per-request breadcrumb cap.
offline_queue_max	int	100	Events buffered after the API errors out. Set to 0 to disable.
flush_interval_seconds	float	0	When > 0, a daemon thread calls flush() on this interval.
async_send	bool	True	When True, sends in a background thread. Set False for synchronous workers.
raise_on_error	bool	False	Re-raise the last transport error after retries (useful in tests).
on_error	Callable[[BaseException], None]	None	Internal-error hook (e.g. send to logs/metrics).
on_queue_drop	Callable[[dict], None]	None	Fires when offline queue drops the oldest event.
transport	object	requests	Anything exposing .post(url, json, headers, timeout). Useful for tests.

---

Basic Usage

from replaystack_sdk import ReplayStackClient

replaystack = ReplayStackClient(
    api_key="rs_live_xxxxxxxxx",
    service_name="order-service",
    environment="production",
    app_version="1.0.0",
    commit_hash="a7f91c",
)

replaystack.capture_event(
    event_type="custom",
    endpoint="daily-report-job",
    status="success",
    response_payload={"message": "Report generated"},
)

---

Manual Exception Capture

try:
    user = None
    user_id = user["id"]
except Exception as exc:
    replaystack.capture_exception(
        exc,
        event_type="custom",
        endpoint="process-user",
        request_payload={"user": user},
    )
    raise

The SDK captures:

• error name + error message
• formatted stack trace
• parsed stack frames (functionName, fileName, lineNumber)
• active breadcrumbs

---

Breadcrumbs

Breadcrumbs are step-by-step debugging hints attached to the next captured event.

replaystack.add_breadcrumb("Order API started")
replaystack.add_breadcrumb("Validating request payload")
replaystack.add_breadcrumb("Calling payment provider", category="payment")

Inside a framework integration each request gets its own breadcrumb scope automatically (via run_with_replaystack_context). Outside a request, breadcrumbs go into the per-client buffer instead — perfect for cron jobs and background workers.

For custom workers, open a scope manually:

from replaystack_sdk import run_with_replaystack_context

with run_with_replaystack_context():
    replaystack.add_breadcrumb("worker started")
    do_work()

---

Sampling, captureSuccess, and Ignored Paths

replaystack = ReplayStackClient(
    api_key="rs_live_xxx",
    sample_rate=0.1,           # capture 10% of events probabilistically
    capture_success=False,     # only ship failed/warning events
    ignored_paths=["/health", "/metrics", "/internal/*"],
)

Sampling is purely probabilistic — failed events are not auto-promoted. If you want to always capture failures and 10% of successes, combine capture_success=False with your own filtering, or wrap capture_event yourself.

---

Sensitive Data Masking

The default mask list matches the TS SDK and includes authorization, password, token, access_token, refresh_token, apikey, api_key, secret, client_secret, cookie, set-cookie, cardNumber, card_number, cvv, otp, and others. Add more via mask_fields.

replaystack = ReplayStackClient(
    api_key="rs_live_xxx",
    mask_fields=["ssn", "tax_id", "phone_number"],
)

Detection is case-insensitive and treats -, _, and spaces as equivalent.

Authorization headers are inspected before masking so the dashboard can show the auth mode (e.g. bearer, basic, api_key, cookie) without ever storing the raw token.

---

Offline Queue, Flush, and Graceful Shutdown

When the API is unreachable after retries, the prepared payload is buffered in memory (bounded by offline_queue_max). Call flush() or close() to drain it.

replaystack = ReplayStackClient(
    api_key="rs_live_xxx",
    offline_queue_max=200,
    flush_interval_seconds=10,     # periodic best-effort drain
    on_queue_drop=lambda info: print("dropped", info),
)

# Drain manually (e.g. in a /shutdown handler):
replaystack.flush(timeout_seconds=5)

# Stop the periodic timer and drain one last time:
replaystack.close(timeout_seconds=5)

---

Process Guards (recommended for long-running servers)

Register interpreter-level hooks so unhandled exceptions are captured and the offline queue is drained when the process exits.

from replaystack_sdk import ReplayStackClient, install_replaystack_process_guards

client = ReplayStackClient(api_key="rs_live_xxx", flush_interval_seconds=10)
unsubscribe = install_replaystack_process_guards(client)

# Later, e.g. in tests:
unsubscribe()

install_replaystack_process_guards wires up:

• sys.excepthook → captures uncaught exceptions
• sys.unraisablehook → captures unraisable exceptions (3.8+)
• asyncio loop exception handler → captures unhandled task failures
• SIGINT, SIGTERM, and atexit → call client.flush() before exit

Options:

install_replaystack_process_guards(
    client,
    uncaught_exception=True,
    unraisable_exception=True,
    asyncio_unhandled_rejection=True,
    flush_on_shutdown=True,
    shutdown_signals=[signal.SIGINT, signal.SIGTERM],
)

---

FastAPI / Starlette

from fastapi import FastAPI
from replaystack_sdk import ReplayStackClient
from replaystack_sdk.integrations.fastapi import FastAPIMiddlewareOptions, setup_fastapi

app = FastAPI()
client = ReplayStackClient(api_key="rs_live_xxx", service_name="fastapi-service")

setup_fastapi(
    app,
    client,
    FastAPIMiddlewareOptions(
        capture_request_body=True,
        capture_response_body=True,
        capture_headers=True,
        ignored_paths=["/internal/*"],
        get_trace_id=lambda req: req.headers.get("x-request-id"),
        should_capture=lambda info: info["statusCode"] != 204,
    ),
)

@app.post("/orders")
async def create_order(payload: dict):
    client.add_breadcrumb("Creating order")
    return {"success": True, "order": payload}

The middleware emits an absolute requestUrl, sets x-trace-id on the response, and adds started / finished / exception breadcrumbs automatically. Streaming responses are not body-captured (a marker payload is sent instead).

---

Flask

from flask import Flask, jsonify, request
from replaystack_sdk import ReplayStackClient
from replaystack_sdk.integrations.flask import FlaskMiddlewareOptions, setup_flask

app = Flask(__name__)
client = ReplayStackClient(api_key="rs_live_xxx", service_name="flask-service")

setup_flask(
    app,
    client,
    FlaskMiddlewareOptions(
        capture_request_body=True,
        capture_response_body=True,
        capture_headers=True,
        ignored_paths=["/internal/*"],
        get_trace_id=lambda req: req.headers.get("x-request-id"),
        should_capture=lambda info: info["statusCode"] != 204,
    ),
)

@app.post("/orders")
def create_order():
    client.add_breadcrumb("Creating order")
    return jsonify({"success": True, "order": request.json})

/health, /metrics, and /favicon.ico are ignored by default. The middleware sets x-trace-id on the response.

---

Django

In settings.py:

from replaystack_sdk import ReplayStackClient
from replaystack_sdk.integrations.django import DjangoMiddlewareOptions

REPLAYSTACK_CLIENT = ReplayStackClient(
    api_key="rs_live_xxx",
    service_name="django-service",
    environment="production",
)

REPLAYSTACK_OPTIONS = DjangoMiddlewareOptions(
    capture_request_body=True,
    capture_response_body=True,
    capture_headers=True,
    ignored_paths=["/internal/*"],
    get_trace_id=lambda req: req.headers.get("x-request-id"),
    should_capture=lambda info: info["statusCode"] != 204,
)

MIDDLEWARE = [
    # ... your security/common middleware ...
    "replaystack_sdk.integrations.django.ReplayStackDjangoMiddleware",
    # ... your custom middleware ...
]

---

Direct API Contract

POST /api/v1/ingest/events
Content-Type: application/json
x-tracereplay-api-key: rs_live_xxxxxxxxx
x-replaystack-api-key: rs_live_xxxxxxxxx

Example payload:

{
  "traceId": "f4b1d4e3-…-…",
  "eventType": "api",
  "method": "POST",
  "endpoint": "/orders",
  "requestUrl": "https://api.example.com/orders?ref=email",
  "authMode": "bearer",
  "authScheme": "Bearer",
  "status": "failed",
  "statusCode": 500,
  "executionTimeMs": 120,
  "errorName": "TypeError",
  "errorMessage": "'NoneType' object is not subscriptable",
  "stackTrace": "Traceback...",
  "stackFrames": [
    {
      "functionName": "create_order",
      "fileName": "/app/main.py",
      "lineNumber": 18
    }
  ],
  "serviceName": "order-service",
  "environment": "production",
  "appVersion": "1.0.0",
  "commitHash": "a7f91c"
}

---

Production Notes

• The SDK should never crash the user application. Internal errors are silently swallowed unless raise_on_error=True is set.
• Failed network calls re-enter the offline queue and drain on the next flush() / close() / periodic tick.
• Use ignored_paths for /health, /metrics, /favicon.ico (already ignored by integrations) and any other noisy endpoints.
• Use sample_rate for high-volume successful events; use capture_success=False when you only want errors.
• Sensitive fields are masked after auth-mode detection, so authMode is always reliable even when the raw header is masked in the captured payload.
• Always call client.close() (or wire up install_replaystack_process_guards) in long-running workers to avoid losing buffered events on shutdown.