anvil-sdk (Python)

Trace instrumentation for LLM apps. Spec: plan/sdk.md.

pip install anvil-sdk           # + [langchain] for the callback handler

import anvil_sdk as anvil

anvil.init(api_key="ak_...", base_url="http://localhost:8000")

with anvil.trace("answer_question", input=question) as t:
    with anvil.span("retrieve", type="retrieval") as s:
        s.output = search(question)
    with anvil.span("generate", type="llm") as s:
        s.end(output=answer, tokens_in=120, tokens_out=30)
    t.output = answer

LangChain apps need one line instead:

from anvil_sdk.langchain import AnvilCallbackHandler

chain.invoke(question, config={"callbacks": [AnvilCallbackHandler()]})

The guarantee

This SDK never blocks or crashes your app (ADR-0013). Traces go to a bounded in-memory queue and a background thread ships them. If Anvil is slow, down, or misconfigured: the queue sheds its oldest entries, counts them in anvil.stats(), and your request path never notices. Nothing in this package raises into your code except exceptions your own functions raised.

API

trace.id is a client-generated UUID exposed on the object — hand it to your frontend so a thumbs-down can be attached to that exact trace later.

Span types: llm, retrieval, tool, embedding, chain, custom.

feedback — closing the loop

Round-trip trace.id to your frontend, then send the vote back with it:

@app.post("/ask")
def ask(body: AskRequest):
    handler = AnvilCallbackHandler(trace_name="ask")
    answer = chain.invoke(body.question, config={"callbacks": [handler]})
    return {"answer": answer, "trace_id": handler.trace.id}

@app.post("/feedback")
def feedback(body: FeedbackRequest):
    anvil.feedback(body.trace_id, -1, comment=body.comment, user_ref=body.user_id)
    return {"ok": True}

Two things worth knowing:

• Call it as soon as the user clicks. The vote routinely beats its own trace to Anvil's database; the server holds it and waits for the trace rather than rejecting the race. You never need to sleep first.
• Pass user_ref if you have one. With it, repeat votes upsert — one user toggling 👎 → 👍 replaces their row instead of leaving two contradictory ones. Without it, every call is an independent anonymous vote.

Fire-and-forget like everything else here: feedback() never blocks or raises, and its own counters (feedback_queued/feedback_sent/feedback_dropped) are separate from the trace ones so a busy trace queue can't hide a lost vote.

Downvoted traces show up in the dashboard's failure inbox — thumbs-down traces not yet in any dataset — where one click turns a real failure into a test case.

redact — scrub PII before it leaves your process

anvil.init(api_key=..., redact=scrub)

def scrub(obj):          # called for every trace *and* every span
    obj.input = mask(obj.input)
    return obj           # or None to drop the object entirely

Payloads are usually dicts and lists, not flat strings — walk them recursively. examples/demo-rag-app/demo/app.py has a worked example. A hook that raises causes the object to be dropped rather than shipped un-scrubbed.

Fields over 100 KB are truncated to a preview before sending, mirroring the server-side policy so large payloads degrade instead of 413ing.

Tests

pip install -e .[dev] && python -m pytest -q

Keyless: LLM calls use LangChain's fake models and the transport is mocked with respx. Emitted payloads are validated against packages/shared/trace-schema.json, the same schema the ingest endpoint enforces.