callm

The production toolkit for LLM calls.

Caching, retries, provider fallback, cost tracking, budgets, PII redaction, prompt-injection detection, structured output validation and telemetry — in one decorator, on top of the SDKs you already use. No proxy. No database server. Zero required dependencies.

Documentation · Quickstart · Features · How it works · CLI

import openai
from pydantic import BaseModel
from callm import callm

class Summary(BaseModel):
    title: str
    bullets: list[str]

client = openai.OpenAI()

@callm(cache=True, retry=3, fallback=["anthropic/claude-sonnet-5"], max_cost=0.25,
       block_pii=True, detect_injection=True, output_schema=Summary)
def summarize(text: str) -> Summary:
    return client.chat.completions.create(
        model="gpt-4o", messages=[{"role": "user", "content": text}]
    )

summarize() still calls OpenAI with your code, your client and your API key — but now every call is cached, retried on rate limits, failed over to Claude if OpenAI is down, refused if it would cost more than 25¢, stripped of emails and phone numbers before it leaves your process, scanned for prompt injection, validated into a Summary object, and recorded in a local cost dashboard.

Why callm

Every team shipping LLM features writes the same production checklist: retry on 429s, cache repeated prompts, fall back when a provider has an outage, track spend, keep PII out of prompts, catch injection attempts, and make the model return valid JSON. That usually means stitching together tenacity, a cache, a PII library, an output-parsing library and a pile of glue code — or deploying a proxy service.

callm is a library: install it, add a decorator, ship.

• No rewrite. Keep calling openai, anthropic or google-genai directly. callm intercepts the SDK call inside decorated functions and hands you back the SDK's own response type.
• No infrastructure. State lives in a local SQLite file (or memory, or Redis if you want a shared cache).
• No required dependencies. The core is standard library only; features that need extra packages are optional extras.
• Sync and async. Same behaviour for def and async def, including concurrent tasks.

Before and after

A simplified version of the glue code callm replaces, for one provider and without fallback, budgets or telemetry:

import hashlib, json, re
from tenacity import (retry, stop_after_attempt,
    wait_random_exponential, retry_if_exception_type)
import openai
from pydantic import ValidationError

EMAIL = re.compile(r"[\w.+-]+@[\w-]+\.[\w.]+")
_cache = {}
PRICES = {"gpt-4o": (2.50, 10.00)}  # $/1M tokens

@retry(
    retry=retry_if_exception_type(
        (openai.RateLimitError, openai.APIConnectionError,
         openai.InternalServerError)),
    wait=wait_random_exponential(max=30),
    stop=stop_after_attempt(4),
)
def _create(**kwargs):
    return client.chat.completions.create(**kwargs)

def summarize(text: str) -> Summary:
    text = EMAIL.sub("[EMAIL]", text)
    messages = [{"role": "user", "content": text}]
    key = hashlib.sha256(json.dumps(messages).encode()).hexdigest()
    if key in _cache:
        return _cache[key]
    for _ in range(3):
        response = _create(model="gpt-4o", messages=messages)
        usage = response.usage
        inp, out = PRICES["gpt-4o"]
        log_cost((usage.prompt_tokens * inp
                  + usage.completion_tokens * out) / 1e6)
        content = response.choices[0].message.content
        try:
            result = Summary.model_validate_json(content)
            _cache[key] = result
            return result
        except ValidationError as exc:
            messages += [
                {"role": "assistant", "content": content},
                {"role": "user", "content": f"Fix: {exc}"},
            ]
    raise RuntimeError("model never returned a valid Summary")

from callm import callm

@callm(cache=True, retry=3, block_pii=True,
       output_schema=Summary)
def summarize(text: str) -> Summary:
    return client.chat.completions.create(
        model="gpt-4o",
        messages=[{"role": "user", "content": text}],
    )

Quickstart

Try it without an API key

examples/offline_demo.py runs the real OpenAI and Anthropic SDKs against a scripted fake server and walks through a rate-limit retry, PII masking, schema validation, a cache hit, a fallback to Claude during an outage, a blocked expensive call and a flagged injection:

pip install "callm-toolkit[openai,anthropic,validation]"
curl -O https://raw.githubusercontent.com/TanbirRamim/callm/main/examples/offline_demo.py
export CALLM_HOME=/tmp/callm-demo      # keep demo data out of ~/.callm
python offline_demo.py
callm stats

(If you cloned the repository, run python examples/offline_demo.py instead.)

With your API key

pip install "callm-toolkit[openai,validation]"   # or callm-toolkit[all]

The package is published as callm-toolkit; you import it as callm and the CLI is callm.

import openai
from callm import callm

client = openai.OpenAI()

@callm()  # zero-config: retries + cost tracking
def ask(question: str) -> str:
    response = client.chat.completions.create(
        model="gpt-4o-mini", messages=[{"role": "user", "content": question}]
    )
    return response.choices[0].message.content

print(ask("What is the capital of France?"))

$ callm stats
Provider   Calls   Tokens    Cost   Cache Hits   Saved   Errors   Latency
──────────────────────────────────────────────────────────────────────────
openai         1       31   $0.0000     0 (0%)   $0.00        0     412ms

Features

Examples

Structured output with automatic repair

from pydantic import BaseModel
from callm import callm

class Invoice(BaseModel):
    vendor: str
    total: float
    currency: str

@callm(output_schema=Invoice, validation_retries=2)
def extract(text: str):
    return anthropic_client.messages.create(
        model="claude-sonnet-5", max_tokens=1024,
        messages=[{"role": "user", "content": f"Extract the invoice as JSON:\n{text}"}],
    )

invoice = extract(raw_text)   # -> Invoice(vendor=..., total=..., currency=...)

Fallback across providers

@callm(retry=2, fallback=["anthropic/claude-sonnet-5", "google/gemini-2.5-flash", "ollama/llama3.1"])
def answer(question: str):
    return openai_client.chat.completions.create(
        model="gpt-4o", messages=[{"role": "user", "content": question}]
    )

If OpenAI keeps returning 429/5xx, callm retries, then sends the same conversation to Claude, then Gemini, then a local model — and answer() still returns an OpenAI ChatCompletion. Requests that use provider-specific features (tools, images, response formats) only fall back to models of the same provider, so a fallback never silently changes what you asked for.

Budgets per user

import callm

@callm.callm(max_cost=0.05)
def chat(messages): ...

def handle(user_id: str, messages):
    with callm.budget_for(f"user:{user_id}", limit=2.00):
        return chat(messages)   # raises callm.BudgetExceeded once the user spent $2

Security without a decorator

from callm import shield

with shield(block_pii=True, detect_injection=True) as s:
    # Any supported SDK call inside the block is protected...
    openai_client.chat.completions.create(model="gpt-4o-mini", messages=user_messages)
    # ...and you can make provider-neutral calls directly.
    response = s.complete(provider="anthropic", model="claude-sonnet-5", messages=user_messages)
    print(response.text, response.cost)

Direct, provider-neutral calls

response = callm.complete("gemini/gemini-2.5-flash", "Summarize: ...", max_tokens=200, cache=True)
response.text, response.usage.total_tokens, response.cost, response.raw

Want a complete program to run? examples/ticket_triage.py triages support tickets with validation, caching, PII masking and a cost report in ~60 lines. More in the cookbook: a support chatbot, RAG answers with citations and data extraction.

How it works

your function                         callm middleware stack
─────────────                         ──────────────────────
@callm(...)                    ┌────────────────────────────────────┐
def summarize():   ─────────►  │ 1. Telemetry     cost, latency, tokens
    client.chat.completions    │ 2. Security      injection scan, PII masking
        .create(...)           │ 3. Cache         exact / semantic lookup
                               │ 4. Validator     parse + re-ask on invalid output
                               │ 5. Fallback      next provider when one keeps failing
                               │ 6. Cost guard    max_cost and budgets, before sending
                               │ 7. Retry         backoff honouring retry-after
                               │ 8. Transport ──► the real SDK call
                               └────────────────────────────────────┘

1. @callm sets a scope (a ContextVar) while your function runs.
2. The official SDK methods (chat.completions.create and .parse, OpenAI's responses.create, messages.create, models.generate_content, sync and async) are instrumented on first use. Outside a callm scope they call straight through, so importing callm never changes other code.
3. Inside a scope, the SDK call is converted into a provider-neutral request and sent through the middleware chain. Each layer is independent and only enabled when configured.
4. The response is converted back into the SDK's native type before it is returned to you.

The middleware is written once as generators and executed by a sync or an async driver, so def and async def functions behave identically.

Configuration

import callm

callm.configure(
    home="~/.callm",          # SQLite database and price overrides
    storage="sqlite",         # "sqlite" | "memory" | a storage object
    telemetry=True,
    otel=False,               # emit OpenTelemetry spans
    default_retries=2,
    on_call=[print],          # called with every CallRecord
)

Every option of @callm is documented in the API reference.

The callm CLI

$ callm stats --since 7d                 # cost, tokens, cache hits and savings by provider
$ callm stats --by function --json       # machine-readable, per function
$ callm calls --limit 20                 # recent calls with retries, fallbacks, flags
$ callm cache stats | clear
$ callm pricing show claude-sonnet-5     # USD per 1M tokens
$ callm pricing update                   # refresh prices from the LiteLLM price list
$ callm info                             # environment and installed extras

Installation extras

How callm compares

callm is not the only tool in this space. An honest summary of where each one fits:

The trade-off: callm covers fewer providers than LiteLLM and fewer validators than Guardrails, in exchange for zero code changes, zero required dependencies and no infrastructure.

Benchmarks

Measured offline with the real OpenAI SDK against a fake server (method and caveats, reproduce with python benchmarks/run.py):

• Overhead: +0.06 ms per call with default settings, +0.10 ms with PII masking, injection detection and budgets enabled.
• Cache: 91% lower spend on support/FAQ-style traffic; 2% on prompts that rarely repeat.
• Reliability: with 20% random provider failures, success rises from 79.9% (plain SDK) to 99.4% with retry=2 and 100% with a fallback deployment.

Design decisions and limits

• The cache is exact-match unless you opt into semantic matching. A semantic cache with a similarity threshold would happily return the answer for "Summarize https://a.example" when asked about "https://b.example". Use cache="semantic" (or CacheConfig(semantic=True)) for FAQ-style traffic; semantic matches never cross different system prompts, histories, parameters or schemas.
• Injection detection flags by default. Heuristics have false positives, so the default logs a warning and records the score; use InjectionConfig(action="block") to refuse. No detector catches every attack — keep treating model output as untrusted.
• Budgets use estimates before a call and actual cost after it. Set max_tokens for a tight worst-case estimate; without it the cost guard assumes 1,024 output tokens.
• Streaming calls get security, retries on connection setup, budgets and telemetry, but are not cached or validated.
• Threads: the callm scope follows asyncio tasks automatically. Work handed to a thread pool needs contextvars.copy_context().run(...), or decorate the function running in the thread.
• Prices are a bundled snapshot. Run callm pricing update or callm.set_price(...) for current or negotiated rates.

Contributing

Contributions are welcome. The good first issues are small and scoped, and every pull request gets a review. See CONTRIBUTING.md. The test suite runs entirely offline against the real provider SDKs with mocked HTTP transports:

uv sync
uv run pytest

If callm is useful to you, a star helps other people find it.

License

MIT