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.
• Off your hot path. Calls are buffered and flushed in batches on a background thread, so your agent's latency never depends on Margin being up. See Batching — read it before you integrate, because the guarantees and their costs are both stated there.
• 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

pip install margin-meter

⚠️ This used to read "published as a git-installable subpath of the Margin repo (no PyPI account needed pre-launch)" against a git+https://github.com/subhsubh24/Margin.ai URL. Both halves were wrong: that repo is private (HTTP 404 for anyone but the owner, so nobody could run the command) and margin-meter is on PyPI. Fixed 2026-08-18, MAR-485.

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)

Auto-instrument (a few lines for a whole repo)

For instrumenting an existing agent you didn't write, you don't want to edit every call site. Point the meter at the installed OpenAI / Anthropic / Gemini SDKs and every model call is timed and metered automatically — tokens are read straight off each provider's response, so there's no set_tokens bookkeeping:

from margin_meter import MarginMeter, instrumented

meter = MarginMeter()  # MARGIN_INGEST_URL + MARGIN_INGEST_KEY

with instrumented(meter, workflow_id="aider-fix"):
    agent.run()        # all OpenAI/Anthropic/Gemini calls inside are metered

meter.record_outcome(workflow_id="aider-fix", passed=tests_green)

instrumented (and instrument_all) patches whichever provider SDKs are installed and skips the rest — a repo that only calls OpenAI doesn't need Anthropic installed. It also covers litellm (the router Aider, SWE-agent, CrewAI and many agents call through) via litellm's callback API. Restoration is automatic on exit. Single-provider entrypoints (instrument_openai / instrument_anthropic / instrument_gemini) return an uninstrument() thunk and raise MarginInstrumentError if that SDK isn't importable.

The litellm hook registers on two points so metering survives an agent that reconfigures litellm mid-run: the classic litellm.success_callback / failure_callback functions and a CustomLogger on litellm.callbacks (the durable hook — some agents reset success_callback for their own analytics, but rarely touch litellm.callbacks). A call delivered to both is de-duped by litellm_call_id and metered exactly once.

Two lower-level primitives when you want tighter control:

# Wrap ONE call (no monkeypatch) — returns a drop-in replacement.
create = meter.wrap(client.chat.completions.create,
                    workflow_id="fit-scoring", provider="openai")
resp = create(model="gpt-4o-mini", messages=[...])   # metered; resp untouched

# Or meter a response you already have.
meter.record_response(workflow_id="fit-scoring", response=resp, provider="openai")

Honesty: tokens are read verbatim from the real response (extract_usage) — cache reads are billed separately, never double-counted as fresh input — and a metering failure is fail-safe (your call still returns / still raises). A failed provider call is still recorded with status="error" so it appears in the supply chain.

Report your evals (pytest)

Your test suite is already an eval harness: each test is a pass/fail graded by code, which is exactly the outcome signal cost-per-outcome divides by. So you do not have to hand-write a record_outcome call — run pytest with a workflow id and every test becomes an outcome:

export MARGIN_INGEST_URL="https://margin-ai-rho.vercel.app"
export MARGIN_INGEST_KEY="mgk_…"
pytest --margin-workflow my-eval-suite

The adapter ships with the package and auto-loads once margin-meter is installed. It does nothing until you pass --margin-workflow (or set MARGIN_PYTEST_WORKFLOW), so installing it never changes how your suite runs, and a Margin outage degrades to a warning — it never fails a test.

Each test maps to one outcome: passed from the result, quality_score 1.0/0.0, quality_method="ground_truth" and grader_kind="code" (a code assertion is a ground-truth code grade), and the test nodeid as the link. Skipped tests report nothing.

To measure value-per-outcome, tag a test with what a pass is worth:

import pytest

@pytest.mark.margin_value(120.0)   # this outcome is worth $120
def test_resolves_the_ticket():
    ...

Absent a marker the test is measured as cost-per-outcome — the value is never defaulted to zero. Add --margin-simulated for a trial run whose rows must not seat real metrics.

Report your evals (promptfoo)

If your evals run under promptfoo, the JSON it already writes is an outcome file. Run your eval, then hand the output to margin-meter:

export MARGIN_INGEST_URL="https://margin-ai-rho.vercel.app"
export MARGIN_INGEST_KEY="mgk_…"
promptfoo eval -o results.json
margin-meter promptfoo --workflow my-eval-suite results.json

Each graded result maps to one outcome: passed from success, quality_score from promptfoo's own score (passed through, not binary-forced), and the grading provenance derived from the assert types — code-graded asserts (equals, contains, is-json, javascript, …) record quality_method="ground_truth" / grader_kind="code"; a model-graded assert anywhere (llm-rubric, factuality, similar, …) records quality_method="llm_judge" / grader_kind="judge"; anything unreadable is unknown. A rubric grade is never reported as a code grade. The link is the test description when present, else the prompt label plus the row index.

To measure value-per-outcome, put margin_value in the test's vars (or its metadata):

tests:
  - description: resolves the refund ticket
    vars:
      margin_value: 120        # this outcome is worth $120
    assert:
      - type: contains
        value: "refund issued"

Absent, the outcome is measured as cost-per-outcome — never defaulted to zero. Add --simulated for a trial run whose rows must not seat real metrics.

Report your evals (DeepEval)

DeepEval writes every run to .deepeval/.latest_run_full.json (and timestamped exports to $DEEPEVAL_RESULTS_FOLDER). Point margin-meter at that file:

export MARGIN_INGEST_URL="https://margin-ai-rho.vercel.app"
export MARGIN_INGEST_KEY="mgk_…"
deepeval test run test_eval.py
margin-meter deepeval --workflow my-eval-suite .deepeval/.latest_run_full.json

Each test case maps to one outcome: passed from the case's success, quality_score from the mean of its metric scores. DeepEval's metrics are mostly LLM-as-judge, so the grading provenance is derived from the metrics — a metric graded by a model records quality_method="llm_judge" / grader_kind="judge" and rides that model as judge_model_id; a metric that names no model records unknown (a code grade is never claimed where none is visible). The link is the case name. Put margin_value in a test case's additionalMetadata to measure value-per-outcome; --simulated marks a trial run.

Report your evals (Braintrust)

If you run evals on Braintrust, serialize a run's per-case results to JSON (a list of results, or the {results: […]} / {events: […]} shapes its SDK and experiment export produce) and point margin-meter at that file:

export MARGIN_INGEST_URL="https://margin-ai-rho.vercel.app"
export MARGIN_INGEST_KEY="mgk_…"
margin-meter braintrust --workflow my-eval-suite --pass-threshold 0.7 results.json

Braintrust reports a named score per scorer and has no intrinsic pass/fail, so the verdict is derived — never fabricated. With --pass-threshold T a case passes when its mean score clears T; without one, a case is recorded only if every scorer is binary (0/1), and a continuous-score case is skipped rather than given a made-up verdict. quality_score is the mean of the case's scores, and the grading provenance is derived from the scorer names — a model-graded autoeval (Factuality, AnswerRelevancy, …) records quality_method="llm_judge" / grader_kind="judge"; a code scorer (ExactMatch, Levenshtein, …) records ground_truth / code; the weaker honest claim always wins. Put an opt-in margin_judge_model in a result's metadata to record the judging model as judge_model_id (a bare model field is the generation model under test, not the judge, so it is not used). Put margin_value in a result's metadata to measure value-per-outcome; --simulated marks a trial run.

Report your evals (Ragas)

If you evaluate RAG with Ragas, serialize the result and point margin-meter at it:

export MARGIN_INGEST_URL="https://margin-ai-rho.vercel.app"
export MARGIN_INGEST_KEY="mgk_…"
python -c "import json; json.dump(result.to_pandas().to_dict(orient='records'), open('ragas.json','w'))"
margin-meter ragas --workflow my-rag --pass-threshold 0.7 ragas.json

Each sample maps to one outcome. The adapter separates Ragas's metric columns (faithfulness, answer_relevancy, context_precision, …) from the dataset columns (question, answer, contexts, ground_truth, …) and sets quality_score to the mean of the metric scores. Ragas has no intrinsic pass/fail and its scores are continuous, so the verdict is derived — with --pass-threshold T a sample passes when its mean clears T; without one, only all-binary samples are recorded and continuous-score samples are skipped rather than given a made-up verdict. Ragas metrics are LLM/embedding judges, so the outcome records quality_method="llm_judge" / grader_kind="judge". Put margin_value on a sample to measure value-per-outcome; --simulated marks a trial run.

Report your evals (LangSmith)

If you evaluate with LangSmith, serialize the evaluate() result and point margin-meter at it:

export MARGIN_INGEST_URL="https://margin-ai-rho.vercel.app"
export MARGIN_INGEST_KEY="mgk_…"
python -c "import json; json.dump(result.to_pandas().to_dict(orient='records'), open('ls.json','w'))"
margin-meter langsmith --workflow my-app --pass-threshold 0.7 ls.json

Each example maps to one outcome. The adapter reads both serializations — the object form (each row's evaluation_results.results) and the pandas form (feedback in feedback.<key> columns) — and sets quality_score to the mean of the feedback scores. LangSmith has no single pass/fail, so the verdict is derived — with --pass-threshold T a row passes when its mean clears T; without one, only rows whose feedback is all binary are recorded and continuous-score rows are skipped rather than given a made-up verdict. The grading provenance is derived from the evaluator keys — LangChain's built-in judges (correctness, criteria, score_string, …) record quality_method="llm_judge" / grader_kind="judge"; string_distance / exact_match record ground_truth / code; a user-named key records unknown rather than an assumed grade. Put margin_value on a row to measure value-per-outcome; --simulated marks a trial run.

Multi-agent pipelines (name each stage)

A crew of agents — a CrewAI pipeline, a planner→coder hand-off — runs in one process under one workflow. Auto-instrumentation binds a single operation at wire-up, so every agent would meter under the same label and the run reads as one blob. stage() splits it: mark the current step around each hand-off and every metered call inside the block is attributed to that stage, under a shared run id.

from margin_meter import MarginMeter, instrument_all, stage

meter = MarginMeter()
instrument_all(meter, workflow_id="crew-pipeline")

run_id = "run-42"
with stage("classify", session_id=run_id):
    classifier.run(ticket)         # its calls meter as operation="classify"
with stage("implement", session_id=run_id):
    implementer.run(plan)          # ... and these as operation="implement"

meter.record_outcome(workflow_id="crew-pipeline", session_id=run_id, passed=ok)

The two stages share session_id but carry distinct operations, so the run is a real multi-stage pipeline: the Money Map decomposes it, and Margin can measure the between-agent spend a botched upstream step drags downstream. stage() is a contextvar, so it stays correct across threads and asyncio; it re-labels the operation only — tokens, cost, and the outcome are untouched.

Say what each call was FOR (task_key)

workflow_id names the pipeline and operation names the step inside it. Neither says what the work actually WAS — a billing question and a password reset look identical on both. task_key is that third axis, in your own vocabulary:

meter.record_call(
    workflow_id="support-triage",
    provider="openai",
    model="gpt-4o-mini",
    input_tokens=812,
    output_tokens=96,
    task_key=ticket.topic,        # "billing" | "password-reset" | anything you use
)

Why it earns a field. Margin measures a cheaper route's parity on a sample and records what that sample was made of. If the distribution of arriving work moves — new ticket topics, a mix that shifts as you grow — parity decays with no model change at all, and nothing that watches for a route getting worse will fire, because nothing got worse. With task_key flowing, Margin compares the measured mix against the arriving one and says steady or moved. Without it the answer is unknown, which is what the console shows and is the honest state, not a bug.

It is never priced and never rolled into cost. Send it on more than 60% of a workflow's calls and the comparison becomes answerable; below that Margin declines to rule rather than describing your instrumented slice as if it were your traffic.

On the auto-instrument path, pass a resolver, not a value. When you wrap a provider with instrument_openai / instrument_all, one wrapper meters a whole run — so task_key is a zero-arg callable read at call time, never a constant (a constant would stamp every call the same and manufacture a single class). Read it from wherever your request context lives — a ContextVar, the current ticket:

from contextvars import ContextVar
from margin_meter import instrument_all

current_task = ContextVar("current_task", default=None)
instrument_all(meter, workflow_id="support-triage", task_key=current_task.get)

# ... per request, before the model calls it drives ...
current_task.set(ticket.topic)   # "billing" | "password-reset" | …

Leave it off and the row's task_key is simply unset (unknown) — never a fabricated bucket.

API

Every method returns an IngestResult(ok, status_code, body, error, reason). ok is True only on HTTP 200; body holds call_id/outcome_id + source. reason is the failure as a stable bucket (no_key, transport, sink_write, sink_corrupt, http_<status>) — read that rather than parsing error, whose wording is prose and will move.

Batching (what it does to your process)

Calls are buffered and flushed in batches — 100 records or 10 seconds, whichever comes first — on a daemon background thread. record_call() appends to a bounded queue and returns; nothing about your agent's latency depends on Margin being reachable.

Why it works this way: at 1M metered calls/day, one call per HTTP request is 30M serverless invocations a month and the same traffic batched at 100 is 300K. That 100x is Margin's cost, not yours, which is exactly why it belongs in the SDK rather than the server — only the client can decide not to make the request.

What you are guaranteed: metering never blocks your call path, never raises into it, and cannot grow without bound in your process. The queue ceiling is ~4-15 MB of your memory at typical row sizes.

What it costs you, stated plainly:

• record_call() returns ok=True, body={"queued": True} — accepted into the buffer, not written to the ledger. delivery_stats() is where you learn what landed.
• A hard kill (SIGKILL, a crash) loses at most 100 records or 10 seconds of them. atexit handles a normal exit; no in-memory buffer survives kill -9.
• If Margin is unreachable long enough to fill the queue, the newest records are dropped, and every drop is counted under queue_full.
• Delivery is at-least-once. Every record carries an event_id and the server is idempotent on it, so a redelivery never double-counts your spend.

meter.flush(timeout=5) blocks until everything queued has been delivered or counted. meter.close() drains and stops the worker. raise_on_error=True skips the buffer entirely and posts at the call site, because strict mode's whole meaning is "raise where I called you".

Full contract, including the arithmetic behind every default: docs/reference/SDK_BATCHING.md.

Knowing what you dropped

Metering is fail-safe: a failed emit returns ok=False and your call carries on. That is the right trade — telemetry should never take your agent down — but it means a lost row looks, from the ledger, exactly like a call that never happened. Cost per outcome then reads cheaper than your bill.

So the meter counts its own losses:

stats = meter.delivery_stats()
print(stats.attempted, stats.delivered, stats.dropped, f"{stats.drop_rate:.1%}")
print(stats.dropped_by_reason)   # {"http_502": 3, "transport": 1}

Two things are kept out of dropped on purpose. A de-duplicated hook is reported as deduped: when instrument_all patches both litellm and the OpenAI SDK, one wire call surfaces at two hooks and the second is correctly suppressed — counting that as a loss would invent a failure rate out of the de-dup working. A retried claim counts both trips to the wire, with one landing.

unmeasured is the remainder made visible. An unreadable sink file lost an unknown number of buffered rows, and there is no honest count to put there, so it is named instead and complete goes False. When that happens, read drop_rate as a floor rather than the answer.

Find uninstrumented call sites (margin-meter doctor)

delivery_stats above counts rows you sent and lost. The quieter shortfall is a call site that never sends anything — a provider call in your code that no instrumenter reaches. It emits no row, no error, no warning; the only symptom is a total that is quietly too small, which from a reconcile looks exactly like a broken meter. On a real Anthropic bill the unmetered call sites carried 100% of the cache-write spend, so the gap read as "Margin's numbers don't add up."

doctor scans your source and names them:

margin-meter doctor path/to/app          # or: python -m margin_meter doctor .
margin-meter doctor --strict .           # exit 1 if any call site is unmetered
margin-meter doctor --json .             # machine-readable report

margin-meter doctor: COVERAGE GAP
  3/6 provider call site(s) metered  (50.0% coverage)  across 41 file(s)

  UNMETERED — these 3 call site(s) emit nothing to the meter:
    api/ios-score.ts:12  [anthropic] raw.messages.create(
    ...

It reads source text — no import of your code, no network — so it runs on a laptop in the first minute. A call site is reported covered only when coverage is provable from the source: a global instrument_all() / instrument_anthropic(), a per-client instrumenter (instrumentAnthropic(client, meter)), or an inline meter.wrap(...). Anything it cannot prove is listed as a gap to check rather than silently passed, and an empty scan reports NO_CALL_SITES — never full coverage. Programmatic entry point: from margin_meter import scan_paths.

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=…).