margin-meter 0.1.0

Tiny stdlib-only client that emits LLM call + outcome economics to a Margin ingest API.

margin-meter (Python SDK)

The tiny client a Python project imports to connect to Margin. It wraps your LLM calls and records their outcomes, emitting each one over HTTP to a Margin ingest API (POST /api/ingest/calls / /api/ingest/outcomes), authenticated with a per-project ingest key. This is the customer-shaped path — the same SDK a stranger drops in — not the in-process meter Margin runs on itself.

• Standalone + stdlib-only. No dependency on Margin's server code and no third-party deps. The default transport is urllib.
• Fail-safe. A failed emit returns an IngestResult(ok=False, …) instead of crashing your app. Pass raise_on_error=True for strict/CI behaviour.
• Provenance-honest. is_simulated is carried through untouched; the written source is forced server-side to your key's project — you cannot spoof another project's economics.

Install

Published as a git-installable subpath of the Margin repo (no PyPI account needed pre-launch):

pip install "git+https://github.com/subhsubh24/Margin.ai.git#subdirectory=sdk/python"

Configure

Two environment variables — the deployed API base and your project's key:

export MARGIN_INGEST_URL="https://margin-ai-rho.vercel.app"
export MARGIN_INGEST_KEY="mgk_…"   # issued by the Margin owner (see below)

The Margin owner issues your project a key with the provisioning CLI in the Margin repo:

python3 scripts/issue_ingest_key.py <your-project-slug>

The raw mgk_… key is shown once — only its hash is stored. Give it to your project as MARGIN_INGEST_KEY.

Use

from margin_meter import MarginMeter

meter = MarginMeter()  # reads MARGIN_INGEST_URL + MARGIN_INGEST_KEY

# 1) Wrap the LLM call — latency is timed automatically, cost computed server-side.
with meter.measure(workflow_id="fit-scoring", provider="google",
                   model="gemini-2.5-flash") as m:
    resp = call_the_model(...)
    m.set_tokens(input_tokens=1200, output_tokens=300, cache_read_tokens=800)

# 2) Record the outcome it produced (the unit of productivity).
meter.record_outcome(workflow_id="fit-scoring", passed=True,
                     quality_score=0.94, quality_method="ground_truth")

Or record a call directly (when you already have the token counts):

res = meter.record_call(
    workflow_id="fit-scoring", provider="google", model="gemini-2.5-flash",
    input_tokens=1200, output_tokens=300, cache_read_tokens=800,
)
if not res.ok:
    log.warning("margin ingest failed: %s (%s)", res.error, res.status_code)

API

Method	Emits to	Notes
record_call(...)	POST /api/ingest/calls	cost computed from the pricing table when cost_usd omitted
record_outcome(...)	POST /api/ingest/outcomes	quality_method records HOW the score was graded
measure(...)	record_call on exit	context manager; times latency, status="error" on exception

Every method returns an IngestResult(ok, status_code, body, error). ok is True only on HTTP 200; body holds call_id/outcome_id + source.

Testing against a live app (no network)

The network boundary is a single transport.post(path, json=..., headers=...) protocol, which a FastAPI TestClient satisfies exactly — so you can exercise the real ingest endpoints hermetically by injecting one:

from fastapi.testclient import TestClient
import asgi
from margin_meter import MarginMeter

meter = MarginMeter(api_key=raw_key, transport=TestClient(asgi.app),
                    raise_on_error=True)

Rate + validation bounds

Ingest is auth'd, validated, and rate-bounded server-side. A bad key → 401, a malformed/implausible row → 422, and a full rolling per-project window → 429. In fail-safe mode these come back as IngestResult(ok=False, status_code=…).