Run approved prompts from Python

The Python SDK is the same protocol, the same conformance vectors, and the same on-disk store as the TypeScript SDK: a store one wrote is a store the other, or the host daemon, opens. Python 3.10 or later; cryptography and httpx are its runtime dependencies. Everything is synchronous and thread-safe; resident mode runs its sync, heartbeat, and update-window timers on daemon threads.

Before you start

You need:

• Python 3.10 or later;
• the agent’s organization id, agent id, and target environment;
• an Agent key for that environment in AIRPROMPTER_AGENT_KEY;
• the AirPrompter root public key for the deployment you talk to, as a JSON Web Key (the queue worker example reads it from AIRPROMPTER_ROOT_JWK); and
• one of the openai, anthropic, or litellm clients with its own credential.

Install (forthcoming on PyPI; today from a wheel you built):

pip install airprompter-agent            # + [openai] [anthropic] [litellm] [kms] [vault] [keyring]
pip install ./dist/airprompter_agent-0.2.13-py3-none-any.whl   # until the PyPI release

The extras install only the provider client or key provider named; the SDK imports each lazily and none is required.

Start a resident runtime

import json
import os

from airprompter_agent import AirPrompterAgent

ap = AirPrompterAgent.start(
    organization_id=os.environ["AIRPROMPTER_ORG_ID"],
    agent_id=os.environ["AIRPROMPTER_AGENT_ID"],
    target="prod",                                      # this app's environment: dev | staging | prod
    api_key=os.environ.get("AIRPROMPTER_AGENT_KEY"),    # absent: serve the last verified release, never call home
    state_dir="/var/lib/acme-support-bot",
    root={"pinned": json.loads(os.environ["AIRPROMPTER_ROOT_JWK"]), "hosted_environment": "prod"},
    sync={"mode": "resident", "poll_seconds": 30},
    models=["claude-sonnet-5", "claude-haiku-4-5"],     # what this process can call; promotion refuses a slot whose model is absent
    apply={"window": "02:00-04:00 Europe/Berlin"},      # staged releases go live on their own at night
    logger=lambda event: print(json.dumps(event)),      # ids, reasons, counts; never prompt text
)

start() reads the store before any network call, tries the other slot and then a vendored_bundle= if you passed one, and only with nothing verified runs one synchronous sync. If that fails it raises AgentStartError("no_verified_release") with the cause in the message, or "store_newer" when a newer package wrote the store. root takes the pinned key plus hosted_environment (the AirPrompter deployment that signed it; the public service is prod), or a full root document.

Expected result: ap.status().generation is the promoted generation and ap.status().apply_state is "active". ap.generation is the same number.

Render a prompt and call your model

1. Wrap your provider client once at start with ap.wrap().
2. Render the prompt by its key in code with the variables it declares; pass subject= when a rollout should be sticky per customer.
3. Send r.text as the system prompt to r.model through the wrapped client.
4. Store r.run_ref beside your own record so you can file feedback later.

from openai import OpenAI

openai = ap.wrap(OpenAI())                     # once: the same client, every call observed

r = ap.prompt("support.triage", subject=customer_id).render(team="Billing", ticket=ticket_text)
reply = openai.chat.completions.create(
    model=r.model,
    messages=[{"role": "system", "content": r.text}, {"role": "user", "content": ticket_text}],
)
ticket_store.save(run_ref=r.run_ref, version_id=r.version_id, arm=r.arm)

render() accepts a mapping or keyword arguments and never touches the network. The Rendered it returns carries text, model, version_id, arm, generation, run_ref, tag, and an optional inference mapping. A variable declared as end-user input is fenced (<ticket>…</ticket>) so the model sees where untrusted input starts and stops; a missing required variable raises MissingVariableError; a value for an undeclared variable raises UnknownVariableError. A freeze from the console or a lapsed lease on a halt environment raises RenderRefusedError with .reason; the queue worker example parks the message and returns the reason.

ap.wrap(client) works for OpenAI / AsyncOpenAI and Anthropic / AsyncAnthropic. It returns a proxy over the same client and observes chat.completions.create / .parse / .stream(), responses.create / .parse / .stream(), messages.create, and messages.stream(); a stream=True call reads usage from the final chunk (pass stream_options={"include_usage": True} on OpenAI). Nothing the wrapper does can fail your call; an unattributed call passes straight through and is logged wrap_unattributed.

Attribute a call explicitly

A call is matched to a render by an enclosing ap.attribute block first, then by request text that equals one of the last 256 renders. Use the block when the rendered text is not in the request verbatim; it is a contextvars variable, so it follows await and threads started with a copied context:

import anthropic

client = ap.wrap(anthropic.Anthropic())

with ap.attribute(r):
    message = client.messages.create(
        model=r.model,
        max_tokens=400,
        system=r.text,
        messages=[{"role": "user", "content": ticket_text}],
    )

The explicit helpers remain for teams that prefer a named call: airprompter_agent.integrations.openai.chat_completion(ap, r, client, messages=[...]) and airprompter_agent.integrations.anthropic.messages_create(ap, r, client, messages=[...], max_tokens=256) place the rendered text as the system prompt and call ap.observe for you; both have _async twins. Async applications wrap a call with await ap.observe_async(r, lambda: client.messages.create(...)).

Variables from your system

A prompt author adds {{customer_tier}} to a version; your app knows a customer’s tier. Register once how the value is found, and every version that uses the variable is filled at render time — a version that does not use it never causes the lookup:

ap = AirPrompterAgent.start(
    ...,
    variables={
        "brand": "Acme",                                                  # a literal: always this
        "customer_tier": {                                                # a source: your system, asked at render
            "resolve": lambda ctx: crm.tier_of(ctx.subject),              # None = "I have none"; a plain callable or a coroutine function
            "trust": "operator",                                          # required: whose text this is
            "timeout_seconds": 0.5,                                       # default 2 s; 64 KiB byte bound
        },
    },
)
ap.variables.provide("region", os.environ["REGION"])                      # later is fine

The SDK is synchronous by design, so render() runs a plain-callable source on a daemon thread of its own under its timeout, holding none of the agent’s locks; a coroutine-function source needs await ap.prompt(tag).render_async(...), which awaits it and cancels it at the deadline. The value a render uses is decided in one order: the call site’s value, then a registered source (consulted only for a declared variable that is required or present in the text), then the default the prompt’s settings declare for an optional variable, then nothing — a required variable is MissingVariableError.

A source names its trust, and the render uses the stricter of the prompt’s declaration and the source’s: an end_user source is fenced wherever the prompt declared operator, and a source can never loosen a declaration. A failure is VariableSourceError with .tag, .variable and .reason (threw, timeout, too_large, not_text, empty), logged by name and reason only, and counted on the board as render_missing_variable. A value is never logged or reported — only a variable’s name; in hosted mode the filled value is posted with the run like any other variable.

ap.prompt(tag).needs(**values) names what the call site must still pass; ap.status().variables lists your sources and, per prompt and rollout arm, the required names nothing fills. A workflow’s render_step(step_id, **values) / render_step_async fills a step the same way; ManagedAgent.start(variables=...) fills required and runtime-sourced variables before a hosted run is posted (plain callables only: the managed client is synchronous). The console side — Settings › Prompt settings › Variables, a default for an optional variable your code writes, and a version whose text uses an undeclared name refused from the update drawer — is the same as for TypeScript.

The release’s inference settings

An agent’s update declares how each prompt is called: temperature, top-p, an output cap, stop sequences, and reasoning effort, set beside the model when the update is staged; the release seals those with the text (r.inference, integers on the wire: temperatureMilli 200 is 0.2, topPBps 9000 is 0.9). A wrapped call attributed to that render goes out with the release’s values. A call site that wrote a different value is told once (wrap_inference_overridden names the parameters), never failed. The settings go only on a call to the release’s model: a call naming another model keeps its own parameters (wrap_inference_model_mismatch). A setting the call cannot carry is reported, not sent (wrap_inference_unsupported): a stop sequence on Responses, a reasoning effort on Messages, top-p beside temperature on Anthropic Messages, anything beside a thinking block you set.

Code that calls a provider itself uses apply_inference(kind, params, inference, model=...), exported from airprompter_agent.

Instrument LiteLLM

A LiteLLM callback sees a call after the fact, so the attribution and the inference settings are handed in as keyword arguments:

import litellm
from airprompter_agent.integrations.litellm import AirPrompterLiteLLMCallback, litellm_inference, litellm_metadata

litellm.callbacks = [AirPrompterLiteLLMCallback(ap)]

r = ap.prompt("support.triage", subject=customer_id).render(team="Billing", ticket=ticket_text)
reply = litellm.completion(
    model=r.model,
    messages=[{"role": "system", "content": r.text}, {"role": "user", "content": ticket_text}],
    metadata=litellm_metadata(r),      # ids and the model only: tag, versionId, arm, model, runRef
    **litellm_inference(r),            # the release's settings as OpenAI chat parameter names; {} when none
)

Without metadata=, the callback still attributes by the messages’ text or an open ap.attribute block.

Run a workflow

flow = ap.workflow("onboarding.flow", subject=customer_id)
previous = ""
for step in flow.steps:                # WorkflowStep: step_id, ordinal, version_id, text, run_ref, model, inference
    with ap.attribute(step):
        reply = openai.chat.completions.create(
            model=flow.model,
            messages=[{"role": "system", "content": step.text}, {"role": "user", "content": previous or intake}],
        )
    previous = reply.choices[0].message.content or ""

Your program executes the steps and whatever work sits between them; the SDK supplies each step’s text, run reference, and inference settings, set per step when the update is staged, for the workflow’s pinned model.

File feedback later

In day-to-day work the signal arrives long after the call, when a person touches the ticket, so the worker keeps run_ref and files against it then:

ap.feedback(r.run_ref, thumbs="up")
ap.feedback(r.run_ref, accepted=True, rating=4, timeToAcceptMs=42_000)

Signals are numbers, booleans, and declared enums from the catalogue; free text is refused by schema, a name outside the catalogue is dropped and logged, and feedback() returns False when nothing was accepted. Any process on the host can file against a run reference minted by another, because the key behind it is the store’s.

Expose health

code, headers, body = ap.healthz_response()   # 200 / 503, {"content-type": ...}, JSON
doc = ap.healthz()                            # ok, status (ok | degraded | failing), reasons, generation, apply_state, ...

Return the tuple from any framework’s route. failing means nothing verified to serve, or a lapsed lease under a halt policy; degraded covers a lapsed lease that still serves, three or more consecutive sync failures, an uploader backing off, and a spool near its budget.

Serverless

sync={"mode": "on_invoke"} and ap.invoke(handler) sync around each invocation; the invocation’s counters are flushed as one segment before invoke() returns (TelemetryOptions(flush="background") is the opt-out). ap.flush_telemetry() does the same by hand.

Stop cleanly

import signal

signal.signal(signal.SIGTERM, lambda *_: ap.stop())

stop() closes the current telemetry window and stops the timers. A restart serves from the store before any network call.

Troubleshooting

AgentStartError: no_verified_release

Read the message. A 404 means nothing is promoted to that environment; a 401 is the key; a 403 names a key bound to another agent or environment. With api_key absent the runtime is offline and needs a store or a vendored_bundle=.

AgentStartError: store_newer

A newer SDK, or the daemon, wrote store.json. The message names the package that wrote it; upgrade this process or roll that writer back.

A LiteLLM call is not counted

Set litellm.callbacks before the call and pass metadata=litellm_metadata(r), or make sure the rendered text is in the messages verbatim.

The release’s settings did not reach LiteLLM

The callback cannot change a call it only observes. Pass **litellm_inference(r) to litellm.completion.

Completion checklist

• The runtime starts against dev with a fresh state_dir and reports the promoted generation.
• The Agent key reaches the process only through the environment.
• Your openai, anthropic, or LiteLLM calls are attributed and the Metrics tab fills a few minutes after upload.
• run_ref is stored beside your own record and feedback is filed against it.
• Your framework serves ap.healthz_response() on a probe route.

Next steps

Keep Agent keys off application hosts with the fleet pattern, read what every row that leaves the host contains in Telemetry, or compare with the TypeScript integration for a mixed estate.