dispatch-kit

A tiny, pure, dependency-free library for gating expensive remote/external work — the same machinery for a cloud GPU job (a Cloud Run L4 reached over a tailnet) and a paid LLM/SDK API call (Gemini, Claude, Rowan). It answers three questions, fail-closed:

• Can we afford it? — a hard, reserve-on-approval budget cap (per-run + per-month).
• Where should it run? — a pure router: LOCAL → LAN → CLOUD → SDK, SDK opt-in only.
• Is the external call safe? — opt-in, audited API egress with reference-only secrets.

It owns the policy (afford / route / approve / egress); your app keeps its job entity, persistence, and executor. The transport auth (who may talk) is a separate concern — pair this with tailnet-guard. Stdlib only; every check is fail-closed (default budget 0 = paid work off; SDK never auto-selected; a missing key refuses).

Use

from decimal import Decimal
from dispatch_kit import (
    BudgetCap, BudgetState, CostRates, admits, estimate_cost,   # the hard $ cap
    select_backend, BackendKind, ToolRequirements,              # the where
    SecretRef, ExternalEndpoint, log_egress,                    # opt-in API egress
    Approval,                                                   # the approval audit fact
)

# 1. Reserve-on-approval: refuse a job that would push past the cap (both windows).
rates = CostRates(gpu_usd_per_s=Decimal("0.0008"), vcpu_usd_per_s=Decimal("0.00001"),
                  gib_usd_per_s=Decimal("0.000002"), idle_tail_s=Decimal(600))
cost = estimate_cost(rates, max_runtime_s=3600, vcpus=8, memory_gib=32)   # an UPPER bound
decision = admits(cost, run_state, month_state, BudgetCap(run_usd=Decimal(50), month_usd=Decimal(500)))
if not decision.admitted:
    raise OverBudget(decision.reason)        # default cap is $0 — paid work is off until you set one

# 2. Pick where it runs — LOCAL first, SDK only if explicitly allowed.
backend = select_backend(my_backends, ToolRequirements(tool_id="cofold", min_vram_gb=24.0))

# 3. An LLM/SDK key is a REFERENCE (env var name), resolved at call time, never logged.
gemini = ExternalEndpoint("gemini", "https://generativelanguage.googleapis.com",
                          SecretRef("GEMINI_API_KEY"))
log_egress(gemini, detail="summarize")       # audit that data left the boundary
headers = {"Authorization": gemini.bearer()} # raises if the key is unset (never an unauth call)

What's in the box

Notes

• The budget cap lives in your dispatch service, never the UI — an agent hitting the API directly is still gated. Default cap $0; if spend can't be computed, refuse.
• Reserve on approval, reconcile on completion — approving reserves the estimate immediately so a burst counts against the cap; the worker's true runtime reconciles reserved → spent.
• SDK / external egress is the one deliberate exception — never the default (allow_sdk / opt-in), always logged, the key sourced from a secret at call time and never written to a log.