SDK Protocol

Send events from any language to brakit's ingest endpoint.

Python SDK

The Python SDK is available on PyPI: pip install brakit

Supports FastAPI and Flask with zero config. Add import brakit before your framework import. Queries, HTTP calls, and errors are captured automatically and forwarded to the Node.js dashboard.

import brakit  # must be before framework import
from fastapi import FastAPI

app = FastAPI()

# That's it. Brakit auto-detects FastAPI, instruments
# SQLAlchemy/asyncpg queries, httpx/aiohttp calls,
# and forwards everything to the dashboard.

Custom SDK Protocol

For other languages (Go, Ruby), any language can POST events to the ingest endpoint. The stores, analysis engine, and dashboard are completely reused.

Endpoint

POST http://localhost:<BRAKIT_PORT>/__brakit/api/ingest

Returns 204 No Content on success, 400 on invalid payload.

Payload Format

{
  "_brakit": true,
  "version": 1,
  "sdk": "brakit-python/0.1.0",
  "events": [
    {
      "type": "db.query",
      "requestId": "uuid-from-x-brakit-request-id-header",
      "timestamp": 1708000000000,
      "data": {
        "operation": "SELECT",
        "table": "users",
        "duration": 45,
        "source": "django-orm",
        "sql": "SELECT * FROM users WHERE id = %s",
        "rowCount": 1
      }
    }
  ]
}

Event Types

Type	Required fields	Optional fields
db.query	operation, table, duration	sql, source, rowCount, model
fetch	url, method, statusCode, duration	-
log	level, message	-
error	name, message	stack
auth.check	provider	result

duration is in milliseconds. level is one of log, warn, error, info, debug. operation is one of SELECT, INSERT, UPDATE, DELETE, OTHER.

Request Correlation

For request correlation, your SDK should generate a unique ID per incoming request and include it in all events. Your SDK should:

Generate or read a unique ID for each incoming HTTP request.
Store it in a request-scoped context (thread-local, context var, etc.).
Include it as requestId in every event emitted during that request.

This is how database queries, fetch calls, and logs are correlated back to the HTTP request that triggered them.

Batching

Buffer events and flush periodically (every 50–100ms) or when the buffer reaches a threshold (20 events). Use fire-and-forget HTTP. Don't block on the response. If brakit isn't running, silently drop events.

The SDK should never interfere with your application's behavior. If brakit isn't running, everything should work exactly the same. events are just dropped.

Minimal SDK Structure

brakit-<language>/
  src/
    transport     # Batched HTTP POST to ingest endpoint
    context       # Request-scoped storage for requestId
    middleware    # HTTP middleware to extract x-brakit-request-id
    adapters/
      django      # Patches Django ORM query execution
      sqlalchemy  # Patches SQLAlchemy engine.execute

The SDK should auto-detect installed libraries and patch only those found. The same pattern as the Node.js adapter registry.

Example: Minimal Transport

If you're building a custom SDK, here's the minimal transport pattern. The Python SDK implements this fully with auto-detection, adapters, and safe wrapping.

import urllib.request
import json
import threading
from collections import deque

class BrakitTransport:
    def __init__(self, port):
        self.url = f"http://localhost:{port}/__brakit/api/ingest"
        self.buffer = deque(maxlen=10_000)

    def emit(self, event):
        self.buffer.append(event)
        if len(self.buffer) >= 20:
            self.flush()

    def flush(self):
        events = list(self.buffer)
        self.buffer.clear()
        if not events:
            return
        try:
            payload = json.dumps({
                "_brakit": True,
                "version": 1,
                "sdk": "my-sdk/0.1.0",
                "events": events,
            }).encode()
            req = urllib.request.Request(
                self.url, data=payload,
                headers={"Content-Type": "application/json"},
            )
            urllib.request.urlopen(req, timeout=1)
        except Exception:
            pass  # Fire and forget

Previous← Adapters NextCLI →