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://trymargin.io"
export MARGIN_INGEST_KEY="mgk_…"   # mint your own at https://trymargin.io/start

Get your key

Open https://trymargin.io/start, enter the email you want this project attached to, and the page shows you an mgk_… key. It takes about ten seconds and needs nobody's approval.

The raw key is shown once — only its hash is stored. Keep it as MARGIN_INGEST_KEY.

⭐ Sign in at https://console.trymargin.io with that same email to read your own cost-per-outcome. One email is one project, and the sign-in resolves through that address, so the console reads exactly the rows your key writes.

⛔⛔ THIS SECTION USED TO DESCRIBE A DIFFERENT, IMPOSSIBLE PATH, AND IT IS WORTH KNOWING WHY IT SURVIVED. It said the key was "issued by the Margin owner" via python3 scripts/issue_ingest_key.py <your-project-slug> — a script inside the Margin repository, which is private. So the documented way to get started handed a reader a command they could not run, against code they could not read, on a path that needed a human to be awake. Measured 2026-09-21: the self-serve route had existed the whole time and this file never mentioned it once. The lesson is not "the README was stale" — it is that an instruction a reader cannot execute does not fail loudly. It reads as onboarding, and the reader concludes the product is not for them yet.

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://trymargin.io"
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://trymargin.io"
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://trymargin.io"
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://trymargin.io"
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://trymargin.io"
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://trymargin.io"
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.

Report BOTH plugs from Langfuse

Langfuse is the one source that carries both halves of the two-plug shape in a single export, which is why it gets two commands rather than one.

export MARGIN_INGEST_URL="https://trymargin.io"
export MARGIN_INGEST_KEY="mgk_…"

# 1) Your AI in: every generation observation becomes a metered call.
margin-meter langfuse-calls --workflow my-app observations.json

# 2) Your evals in: every score becomes an outcome candidate.
margin-meter langfuse --workflow my-app --pass-threshold 0.7 --min-confidence 0.7 scores.json

⚠️ A numeric score needs --pass-threshold; a categorical needs --pass-values A,B; only a boolean maps to a verdict on its own. Run margin-meter langfuse --help for the current set — a score that arrives without the flag its data type needs is an outcome with no verdict, which will not move your denominator.

Serialise either the SDK's fetch_observations / fetch_scores result or a UI export to JSON. A bare list works, as do {"data": […]} / {"scores": […]} / {"observations": […]} envelopes. No live Langfuse API key is needed here, the same as every other adapter.

⭐ langfuse-calls is worth trying before you instrument anything. A generation observation already carries model, token usage, cost, and the trace and parent ids, so the estate tree falls out of your trace hierarchy without a line of code in your agent. If your generations are in Langfuse, this is the shortest path to a real cost-per-outcome.

⛔ --min-confidence IS A REFUSAL, NOT A FILTER, AND THE DIFFERENCE MATTERS. A score whose metadata or comment embeds a probability is counted only at or above the threshold; below it the row is not an outcome at all — neither a pass nor a fail. Forcing a low-confidence score into a binary "least wrong answer" would fabricate the denominator that cost-per-outcome divides by. A score with no stated probability is unaffected: the floor gates a confidence somebody stated, it never invents one.

⚠️ substrate is left empty, deliberately. substrate names the ROUTE a call took (direct, openrouter, bedrock), which is a different thing from the model, and a Langfuse export carries no route evidence. provider is derived from the model string, overridable per batch with --provider, and left unknown rather than fabricated when neither names it.

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

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
record_response(...)	record_call	meters straight off a provider response — no manual tokens
wrap(fn, ...)	record_call per call	returns an auto-metered drop-in for any LLM call
instrument_all / instrumented	—	monkeypatch the installed OpenAI/Anthropic/Gemini SDKs + hook litellm (two-hook, reset-resilient)
stage(op, session_id=…)	—	context manager; attributes calls inside it to one pipeline stage under a shared run id
delivery_stats()	—	what this meter delivered against what it tried to (see below)
margin-meter doctor / scan_paths(...)	—	static scan for provider call sites your code never meters (see below)

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.

knob	default	env override
records per flush	100	MARGIN_BATCH_MAX_RECORDS
seconds between flushes	10	MARGIN_BATCH_FLUSH_SECONDS
queued records (memory ceiling)	10,000	MARGIN_BATCH_MAX_QUEUE

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