Backend SDK

Server-side client that turns the frontend payload + request context into a logged event during Phase 2, then into a verdict once your workflow is live. Five stacks: Python (default), Node.js, Go, Java — or the plain HTTP API from any language.

Install

Python 3.10+. Published on PyPI as signalgate.

pip install signalgate

The only runtime dependency is requests (2.28+).

Client setup

One Client per process. Thread-safe; reuse the same instance across all your action handlers.

from signalgate import Client

# api_key is positional; every other option is keyword-only (defaults shown)
client = Client(
    "pk_live_...",
    check_timeout_ms=3000,
    log_timeout_ms=1000,
    log_queue_capacity=10000,
    log_max_retries=3,
    log_retry_base_ms=200,
    fail_open=True,
)

Constructor options

OptionTypeDefaultPurpose
api_keystrrequiredYour pk_live_ key. Sent as a Bearer token.
check_timeout_msint3000Hard ceiling for a check() call.
log_timeout_msint1000Per-attempt timeout for the async log worker.
log_queue_capacityint10000In-memory queue size for async log delivery.
log_max_retriesint3Retries for failed log deliveries (up to 4 attempts total).
log_retry_base_msint200Base delay for the log retry backoff (200 → 400 → 800 ms).
fail_openboolTrueOn timeout, network failure or 5xx, check() returns a synthesized allow instead of raising.
fail_open (on by default) covers timeouts, network failures, 5xx and malformed 2xx responses — check() then returns a synthesized allow with failed_open=True. A 4xx always raises ServerError regardless: it signals a bug in the request, so surface it loudly. Advanced seams http and logger exist for custom transports and logging.

Shutdown

close() drains the log queue (within 5× log_timeout_ms) so in-flight events aren't lost. Call it at process shutdown — or use the client as a context manager.

from signalgate import Client

client = Client("pk_live_...")

# At process shutdown: drains the log queue (within 5 x log_timeout_ms)
client.close()

# Or scope the client to a block:
with Client(api_key="pk_live_...") as client:
    ...

Event shape

What you pass to log() and check(). Build the payload from the browser envelope verbatim: EncryptedPayload(**payload_dict).

from signalgate import Event, EncryptedPayload

event = Event(
    user_id="u_123",                        # your user ID
    ip="203.0.113.42",                      # from X-Forwarded-For or request.remote_addr
    method="login",                         # action name: "login", "deposit", ...
    timestamp="2026-04-01T13:08:50+00:00",  # ISO 8601 string
    payload=EncryptedPayload(
        encrypted="<base64 blob from the frontend SDK>",
        timestamp=1748102400000,            # int, unix milliseconds
        nonce="aZ19bCde3fGhI4jK",
        v=2,                                # the frontend SDK emits v=2
    ),
    custom={"plan": "pro"},                 # optional labeling fields
)

# Pass the browser envelope through verbatim:
payload_dict = {"encrypted": "<base64 blob>", "timestamp": 1748102400000,
                "nonce": "aZ19bCde3fGhI4jK", "v": 2}
payload = EncryptedPayload(**payload_dict)
Two different timestamps: Event.timestamp is an ISO-8601 string; payload.timestamp is an integer in unix milliseconds — never unify them. The frontend SDK emits v=2; v and custom are omitted from the wire when None. method names the funnel point: the action handler and the target-action handler each log under their own value (e.g. "login" and "deposit") — the distinct method at the target point is the primary linkage. Each logged event needs its own freshly-collected browser payload; a reused nonce is rejected as a replay.

Methods

check

Synchronous; returns a CheckResult with the verdict. No retries; bounded by check_timeout_ms. Call it at the top of the gated action handler — before the protected action — and gate on action == "block". With fail_open on (the default), degraded-service conditions return a synthesized allow with failed_open=True.

log

Fire-and-forget: enqueues the event and returns immediately; never raises. Bounded in-memory queue (10,000), delivered by a daemon thread with up to 4 attempts (200/400/800 ms backoff); a 4xx is dropped immediately. Call it after the action succeeds — and again at the target action, after it succeeds. Delivery failures are visible via client.metrics.

Examples

Phase 2 — log only

Two handlers: the action handler logs after a successful authenticate(); the target-action handler (the downstream conversion — here a deposit) builds its own event from its own request's fingerprint payload and logs after its own success. Never reuse one browser payload across two funnel points.

from datetime import datetime, timezone

from flask import Flask, request
from signalgate import Client, Event, EncryptedPayload

app = Flask(__name__)
client = Client(api_key="pk_live_...")


def sg_event(body: dict, method: str, custom: dict | None = None) -> Event:
    """Build an Event from THIS request's own fingerprint payload."""
    return Event(
        user_id=body["email"],
        ip=request.headers.get("X-Forwarded-For", request.remote_addr),
        method=method,
        timestamp=datetime.now(timezone.utc).isoformat(),
        payload=EncryptedPayload(**body["signalgate"]),
        custom=custom,
    )


# --- your existing app logic (stubs for this example) ---
def authenticate(email: str, password: str):
    ...  # your credential check; returns a user or None


def issue_session(user) -> str:
    ...  # your session issuance


def apply_deposit(email: str, amount: float) -> bool:
    ...  # your business logic; True on success


# ── Action handler: the moment you protect ──
@app.post("/api/login")
def login():
    body = request.get_json()

    user = authenticate(body["email"], body["password"])
    if not user:
        return {"error": "invalid credentials"}, 401

    # Log AFTER the action succeeded
    client.log(sg_event(body, method="login"))
    return {"ok": True, "session": issue_session(user)}


# ── Target-action handler: the downstream conversion ──
@app.post("/api/deposit")
def deposit():
    body = request.get_json()

    if not apply_deposit(body["email"], body["amount"]):
        return {"error": "deposit failed"}, 402

    # Its OWN event from THIS request's fresh fingerprint payload
    client.log(sg_event(body, method="deposit", custom={"target_action": True}))
    return {"ok": True}

Once your workflow is live — check + log

Once your workflow is live, add check() at the top of the gated action handler — before the protected action — and gate on "block". The gated request now carries two sealed envelopes from the browser SDK — "signalgate" for the check and "signalgate_log" for the post-success log — because each API call consumes its payload's single-use nonce, so the checked payload can never be reused for the log. The target-action handler is unchanged.

from datetime import datetime, timezone

from flask import Flask, request
from signalgate import Client, Event, EncryptedPayload, ServerError

app = Flask(__name__)
client = Client(api_key="pk_live_...")


def sg_event(body: dict, method: str, envelope: str = "signalgate",
             custom: dict | None = None) -> Event:
    return Event(
        user_id=body["email"],
        ip=request.headers.get("X-Forwarded-For", request.remote_addr),
        method=method,
        timestamp=datetime.now(timezone.utc).isoformat(),
        payload=EncryptedPayload(**body[envelope]),
        custom=custom,
    )


# --- your existing app logic (stubs for this example) ---
def authenticate(email: str, password: str):
    ...  # your credential check; returns a user or None


def issue_session(user) -> str:
    ...  # your session issuance


@app.post("/api/login")
def login():
    # The gated request body carries TWO sealed envelopes from the browser
    # SDK: "signalgate" (for the check) and "signalgate_log" (a second fresh
    # capture, for the post-success log)
    body = request.get_json()

    # 1. check() BEFORE the protected action — gate on "block"
    try:
        verdict = client.check(sg_event(body, method="login"))
        if verdict.action == "block":
            return {"error": "request rejected"}, 403
    except ServerError as exc:
        # With fail_open on (the default) this is a 4xx — a bug in the
        # request, not a fraud signal. Surface it loudly; don't block the user.
        app.logger.error("signalgate check failed: %s", exc)

    # 2. The protected action itself
    user = authenticate(body["email"], body["password"])
    if not user:
        return {"error": "invalid credentials"}, 401

    # 3. log() AFTER the action succeeded — a SECOND event from
    # "signalgate_log": each API call consumes its payload's single-use
    # nonce — never reuse the checked payload for the log
    client.log(sg_event(body, method="login", envelope="signalgate_log"))
    return {"ok": True, "session": issue_session(user)}

# The target-action handler is unchanged from Phase 2.

Verdict shape

The CheckResult dataclass returned by check().

from dataclasses import dataclass

@dataclass
class CheckResult:
    action: str              # "allow" | "admin_alert" | "dry_run_block" | "block" — open set
    score: float             # 0.0 | 0.25 | 0.5 | 1.0 (server-fixed per action)
    request_id: str          # UUID, useful for tracing in your logs
    tenant_id: str
    timestamp: str           # ISO 8601
    processing_time_us: int  # server-side processing time
    failed_open: bool        # True only for a synthesized fail-open allow

Errors

All exceptions inherit from SignalGateError. Catch the specific types your code handles. signalgate.TimeoutError and signalgate.NetworkError shadow Python built-ins — import them from signalgate.

from signalgate import (
    Client,
    Event,
    EncryptedPayload,
    SignalGateError,   # base class
    ConfigError,       # invalid API key shape / constructor argument
    TimeoutError,      # exceeded check_timeout_ms (shadows the builtin!)
    NetworkError,      # DNS, TCP or TLS failure
    ServerError,       # non-2xx API response
)

client = Client(api_key="pk_live_...", fail_open=False)

event = Event(
    user_id="u_123",
    ip="203.0.113.42",
    method="login",
    timestamp="2026-04-01T13:08:50+00:00",
    payload=EncryptedPayload(
        encrypted="<base64 blob from the frontend SDK>",
        timestamp=1748102400000,
        nonce="aZ19bCde3fGhI4jK",
        v=2,
    ),
)

try:
    result = client.check(event)
except ServerError as e:
    # 4xx always raises, even with fail_open=True — a bug in the request
    print(e.status_code, e.code, e.message, e.request_id)
except (TimeoutError, NetworkError) as e:
    # only reachable with fail_open=False; the default returns allow instead
    print("degraded:", e)
ServerError carries .status_code, .code, .message and .request_id (.details is None in production). Delivery counters live on client.metrics: client.metrics.get("check_total"), client.metrics.get("log_dropped_total", reason="queue_full"), client.metrics.snapshot().

Where each call goes

The same contract in every stack: two handlers, each with its own event.

check() — before the protected action

Call check() at the top of the gated action handler, before the action itself, and gate on action == "block".

log() — after the action succeeds

Log the action-place event only after the action itself succeeded — not before, and not on failure.

Target-action log() — after the target action succeeds

The downstream conversion is its own funnel point: its handler builds its own event from its own request and logs after its own success. A distinct method value at the target point is the primary linkage between the two.

A fresh payload for every funnel point

Never reuse a browser envelope across funnel points — the target-action log must carry its own freshly-collected payload. A reused nonce is rejected as a replay.

Verdicts

The four actions check can return, and how to handle each. Treat the set as open: new values may appear over time — match on the ones you handle and fall through safely for anything unknown.

allowscore 0.0

No risk detected. Let the request through. The vast majority of legitimate traffic falls here.

Recommended handling: Continue normally.

admin_alertscore 0.25

The request is allowed, but flagged for review — it surfaces in the dashboard events log for your team to inspect.

Recommended handling: Allow; review the flagged activity in the dashboard.

dry_run_blockscore 0.5

A would-be-block in shadow mode — the request is allowed, so you can see what enforcement would catch before turning it on.

Recommended handling: Allow, but treat as a future block.

blockscore 1.0

High-confidence fraud. Reject the action.

Recommended handling: Refuse the action; surface a generic error to the user.

HTTP API reference

The raw REST surface every SDK wraps — and the direct integration path for any other language. Both endpoints share one request-body shape and one response envelope.

Base URL & auth

https://api.signalgate.ai

Authorization: Bearer pk_live_...
Content-Type: application/json

Optional headers: X-Request-Id (fresh per attempt, for tracing) and Idempotency-Key (stable across retries of the same event; accepted, but not yet processed server-side).

POST /v0/check (once your workflow is live)

Synchronous verdict. Returns 200 OK with the verdict in data. Call it before the protected action and gate on data.action == "block".

POST /v0/check
{
  "user_id": "u_123",
  "ip": "203.0.113.42",
  "method": "login",
  "timestamp": "2026-04-01T13:08:50+00:00",
  "payload": {
    "encrypted": "<base64 blob from the frontend SDK>",
    "timestamp": 1748102400000,
    "nonce": "aZ19bCde3fGhI4jK",
    "v": 2
  },
  "custom": { "plan": "pro" }
}

POST /v0/log (Phase 2 — and ongoing after launch)

Fire-and-forget event logging. The request body is identical to /v0/check. Returns 200 OK with an empty envelope — there is no 202.

200 OK
{
  "ok": true,
  "data": null,
  "error": null
}
user_id, ip, method, timestamp and payload are required; custom is optional — omit the key entirely, never send null. The outer timestamp is an ISO-8601 string; payload.timestamp is an int64 in unix milliseconds — two different types, never unify them. payload.v: 2 is required: pass the browser envelope through verbatim.

Error envelope

Every non-2xx response uses the same envelope. details is omitted in production.

401 Unauthorized
{
  "ok": false,
  "data": null,
  "error": {
    "code": "UNAUTHORIZED",
    "message": "invalid or revoked API key",
    "request_id": "550e8400-e29b-41d4-a716-446655440000"
  }
}
CodeStatusMeaning
UNAUTHORIZED401Missing, invalid, or revoked API key.
BAD_REQUEST400Malformed body or missing required field.
INVALID_PAYLOAD400The fingerprint payload could not be read — corrupt or truncated envelope.
REQUEST_REJECTED422Stale or replayed payload (e.g. a reused nonce).

Status codes

StatusMeaning
200 OKSuccess — check returns the verdict in data; log returns an empty envelope.
400 Bad RequestMalformed body (BAD_REQUEST) or corrupt fingerprint payload (INVALID_PAYLOAD).
401 UnauthorizedMissing, invalid, or revoked API key (UNAUTHORIZED).
422 Unprocessable EntityStale or replayed payload (REQUEST_REJECTED).
5xxService degraded — apply the fail-open guidance below.

Fail-open guidance

check — on timeout, network failure or 5xx, synthesize {"action": "allow", "score": 0} locally and let the request through. Never retry a check. A 4xx is a bug in your request — surface it loudly.

log — retry up to 3 times with 200/400/800 ms backoff on timeout, network failure or 5xx; drop the event on any 4xx. A log failure must never break your action.

Continue reading

Back to Overview →

The two-phase integration journey, event model, and SDK version index.

Frontend SDK →

How customers integrate the JS SDK and where the payload originates.