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.

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

Verify the import without printing credentials:

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

Configure#

Environment configuration keeps scope consistent across calls:

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.

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:

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#

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:

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#

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.

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

Dry run and trace#

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; 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:

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.

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

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. 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:

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

Follow the outcome and acknowledgment contract when reconciling delivery failures.

Idempotent calls#

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

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 for cache lifetime, conflicts, and executor protection.

Validate and inspect from the CLI#

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

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:

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

This is destructive and is not part of ordinary deployment.