---
title: Python SDK
description: Configure AsyncAMIClient, build state and tag context, invoke named decision points, ingest events, trace decisions, and handle Memrail actions in Python.
eyebrow: SDK REFERENCE
keywords: Memrail Python SDK, AsyncAMIClient, Python AI agent control
last_updated: 2026-09-10
---

# Python SDK

The `memrail` package provides the asynchronous client, typed ATOM builders, event ingestion, tracing, and the CLI used to manage EMUs and tools.

## Install

Requires Python 3.9 or newer.

```bash
python3 -m venv .venv
. .venv/bin/activate
python -m pip install --upgrade memrail
```

Verify the import without printing credentials:

```bash
python -c "import memrail; from importlib.metadata import version; print(version('memrail'))"
memrail --help
```

## Configure

Environment configuration keeps scope consistent across calls:

```bash
export AMI_API_KEY='your-api-key'
export AMI_ORG='your-org-slug'
export AMI_TEAM='default'          # optional
export AMI_WORKSPACE='development'
export AMI_PROJECT='support-agent'
```

The client reads `AMI_BASE_URL` automatically. An explicit `base_url` overrides it; otherwise the default is `https://api.memrail.com`. Set `use_env=False` to disable environment fallbacks.

```python
from memrail import AsyncAMIClient
from memrail.atoms import state

async with AsyncAMIClient() as client:
    response = await client.decide(decision_point="support.before_response", context=[state("customer.tier", "enterprise")])
```

For applications that choose scope at runtime, configure explicitly:

```python
import os
from memrail import AsyncAMIClient

async with AsyncAMIClient(
    api_key=os.environ["AMI_API_KEY"],
    org="acme",
    team="platform",
    workspace="production",
    project="support-agent",
) as client:
    ...
```

Use the async context manager so underlying resources are closed cleanly.

## Build state and tags

```python
from memrail.atoms import state, tag, atoms_from_dict

context = [
    state("customer.id", "cust_123"),
    state("customer.tier", "enterprise"),
    state("ticket.priority", "critical"),
    tag("intent", "escalation"),
]

customer_context = atoms_from_dict(
    {
        "id": "cust_123",
        "tier": "enterprise",
        "account": {"region": "us-east", "active": True},
    },
    prefix="customer",
)
```

State values should be strings, numbers, or booleans. `atoms_from_dict` flattens nested dictionaries into dot notation. Keep keys lowercase and give them at least two segments after the `state` namespace.

Validate model-derived tags at runtime, not only with type annotations:

```python
from typing import Literal
from pydantic import TypeAdapter
from memrail.atoms import tag

Intent = Literal["billing", "cancellation", "technical", "other", "unknown"]
intent = TypeAdapter(Intent).validate_python(model_classification)
intent_atom = tag("intent", intent, source="ml")
```

The SDK preserves state/tag `source` in decision requests. Label model-derived values with `source="ml"`, and ensure they cannot impersonate trusted application facts. Provenance labels complement application authorization; they do not replace it.

## Decide

```python
from memrail import AsyncAMIClient
from memrail.atoms import state, tag
from memrail.models import InvokeOptions

async def steer_response(customer, ticket, validated_intent, *, evaluation_only=True):
    async with AsyncAMIClient() as client:
        response = await client.decide(
            decision_point="support.before_response",
            context=[
                state("customer.id", customer.id),
                state("customer.tier", customer.tier),
                state("ticket.priority", ticket.priority),
                tag("intent", validated_intent, source="ml"),
            ],
            options=InvokeOptions(dry_run=evaluation_only),
        )

        for item in response.selected:
            await handle_selected_action(
                item,
                control_point="support.before_response",
                evaluation_only=evaluation_only,
            )

        return response
```

Use `context`; `atoms` is also a supported alias in this version. Provide exactly one. `context_atoms` is a wire-body field, not a `decide()` keyword.

Bind the policy to `support.before_response` and pass that exact name on every call. See the [integration invariants](/concepts/#integration-invariants).

When an application serves multiple scopes, a call may override the client defaults with `workspace="staging"` and `project="support-agent"`.

## Dry run and trace

```python
from memrail.models import InvokeOptions, TraceOptions

response = await client.decide(
    context=context,
    decision_point="support.before_response",
    options=InvokeOptions(
        dry_run=True,
        top_k=5,
    ),
    trace=TraceOptions(enable=True),
)
```

Use the [evaluation and execution contract](/concepts/#evaluation-and-execution); the combined SDK helper keeps dry-run results out of handlers and ACK.

Tracing is enabled by default. Use `TraceOptions(enable=True)` and inspect `response.trace["candidates"]` for eligibility and suppression details. Shadow EMUs appear in traces but are excluded from `selected`.

## Handle selected actions

Keep dispatch explicit and fail closed. `your_app.execution_contract`, `your_app.prompt_builder`, `your_app.review_queue`, `your_app.route_registry`, and `your_app.execute_authorized_tool` below are application-owned code, not SDK methods:

```python
async def handle_selected_action(item, *, control_point, evaluation_only=True):
    if evaluation_only:
        return None  # Never dispatch evaluation results.

    action = item.action

    if not action:
        return None

    if not your_app.execution_contract.permits(item, control_point):
        raise PermissionError("Selected action is not authorized for execution")

    if action["type"] == "context_directive":
        return your_app.prompt_builder.add_directive(action["directive"])
    if action["type"] == "decision_prompt":
        return your_app.review_queue.create(action["message"], action["options"])
    if action["type"] == "route":
        return your_app.route_registry.dispatch(action["destination"], action.get("metadata", {}))
    if action["type"] == "tool_call":
        return await your_app.execute_authorized_tool(action["tool"])

    raise ValueError(f"Unsupported Memrail action type: {action['type']}")
```

Selections expose `policy`, `lifecycle_state`, `emu_version`, and `action`. The custom contract above receives the full selection and must check those fields, the allowed control point, tool version, arguments, caller permissions, and duplicate protection. Deny missing/unknown policy and unauthorized lifecycle states; do not reduce a selection to its action payload before authorization.

Supply verified approval through `InvokeOptions(human_consent=True)`, following the [execution contract](/concepts/#evaluation-and-execution).

For SDK-managed dispatch, register reviewed handlers and use `await client.decide_and_execute(context=context, decision_point="support.before_response", options=InvokeOptions(dry_run=True), auto_ack=True)`. The synchronous equivalent is `client.decide_and_execute_sync(...)`. The [runnable quickstart](/getting-started/) includes registration, execution, and ACK.

For a dangerous decision point, an empty selection or client error should resolve to a safe documented outcome. Catching every exception and continuing with the model proposal defeats the control boundary.

## Ingest material events

Events are written separately and queried by temporal triggers:

```python
from datetime import datetime, timezone
from memrail.atoms import event

await client.emit_event(
    event(
        "agent.sent.email",
        ts=datetime.now(timezone.utc),
        attributes={
            "customer_id": customer.id,
            "message_id": message.id,
            "activation_id": item.activation_id,
        },
    ),
    workspace="production",
    project="support-agent",
)
```

Follow the [outcome contract](/concepts/#outcomes-and-acknowledgment). Event retention defaults to 90 days and is configurable by organization; check it before choosing lookback windows.

For an ACK-gated cooldown, emitting an event is not enough. Acknowledge the successful activation separately:

```python
await client.ack(
    activation_id=item.activation_id,
    status="acknowledged",
    code="SUCCESS",
)
```

Follow the [outcome and acknowledgment contract](/concepts/#outcomes-and-acknowledgment) when reconciling delivery failures.

## Idempotent calls

Use a stable key when repeated requests represent the same logical decision:

```python
response = await client.decide(
    decision_point="billing.capture_payment",
    project="billing",
    context=[
        state("order.id", order.id),
    ],
    idempotency_key=f"billing:billing.capture_payment:live:{operation_id}",
)
```

`operation_id` identifies one immutable request. See [two layers of idempotency](/concepts/#two-layers-of-idempotency) for cache lifetime, conflicts, and executor protection.

## Validate and inspect from the CLI

```bash
memrail list-emus -w production -p support-agent
memrail get-emu support.critical_enterprise -w production -p support-agent
memrail emu-validate -w production -p support-agent
memrail action-connectivity -w production -p support-agent
memrail tool-get-schema -w production -p support-agent
```

These are the Python-installed CLI commands. For production changes, use the [strict JSONL workflow](/production/) rather than SDK write methods. For programmatic read-only preflight, use `await client.validate_emu_candidates(candidates, workspace=..., project=...)` with complete EMU definitions.

## Error handling

Catch specific errors when recovery differs. Error classes below are from the SDK; `your_app.ConfigurationError`, `your_app.ControlEvaluationError`, and `your_app.safe_control_fallback` belong to the application:

```python
from memrail.errors import AMIUnauthorized, AMINotFound, AMIServerError, AMIError

try:
    response = await client.decide(context=context, decision_point="support.before_response")
except AMIUnauthorized:
    raise your_app.ConfigurationError("Memrail API key was rejected")
except AMINotFound:
    raise your_app.ConfigurationError("Memrail workspace or project does not exist")
except AMIServerError:
    return your_app.safe_control_fallback()
except AMIError as exc:
    raise your_app.ControlEvaluationError(str(exc)) from exc
```

Retry transient service errors with bounded exponential backoff only when the surrounding operation is still safe to retry. Never log the API key or full sensitive context.

## Workspace reset

`purge-workspace` deletes workspace data and requires an organization-level key. Use it only for development or explicit test resets:

```bash
memrail purge-workspace development --targets emus,traces,events --yes
```

This is destructive and is not part of ordinary deployment.