# Run approved prompts from Python

Install the Python Agent SDK, start a resident runtime, render a prompt, wrap your openai or anthropic client, attribute calls explicitly, and instrument LiteLLM with the release's inference settings.

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.

**Coming to PyPI**

The five distributions (`airprompter-agent`, `airprompter-agent-core`, `-sync`, `-runtime`, `-telemetry`) are at version 0.2.13 in the repository and are not on PyPI yet. Until they are, build the wheels from the repository (`python tools/build_all.py` puts every sdist and wheel into one `dist/`) and install from there. The forthcoming public install is `pip install airprompter-agent`.

## 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):

```sh
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

```python
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.

```python
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:

```python
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:

```python
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:

```python
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

```python
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:

```python
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

```python
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

```python
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](https://airprompter.com/docs/integrations/fleet), read what every row that leaves the host contains in [Telemetry](https://airprompter.com/docs/integrations/telemetry), or compare with the [TypeScript](https://airprompter.com/docs/integrations/typescript) integration for a mixed estate.

---

Canonical: https://airprompter.com/docs/integrations/python