Python SDK Observability

The Python SDK records one trace for each IAP plan and sends completed traces to ArmorIQ without making telemetry part of the policy decision. Use the session API to get policy, tool, and model activity under one stable session ID.

Install the production package

You need an organization-scoped ArmorIQ API key before the SDK can ship telemetry.

python -m pip install "armoriq-sdk==0.6.4"

The import package is armoriq_sdk. Customer applications should install the production package from PyPI, not a development package or a local checkout.

Start with the defaults

Observability is enabled by default when you start an ArmorIQSession. The recorder inherits the client's API key, HTTP client, and backend base URL.

from armoriq_sdk import ArmorIQClient, SessionOptions

client = ArmorIQClient(
    api_key="YOUR_ARMORIQ_API_KEY",
    agent_id="support-agent",
)

session = client.start_session(SessionOptions(mode="sdk"))

The session creates one stable UUID for all of its traces. It uses a valid SessionOptions.session_id, then a valid context_id, or generates a UUID when neither value is valid. Set SessionOptions.session_id when you need to correlate the ArmorIQ session with an application-level UUID.

Configure observability

SessionOptions.observability accepts ObservabilityConfig from armoriq_sdk.observability.

from armoriq_sdk import SessionOptions
from armoriq_sdk.observability import ObservabilityConfig

options = SessionOptions(
    mode="sdk",
    session_id="6f7bbf5e-3f5f-46b8-86aa-f3b9107df6bb",
    observability=ObservabilityConfig(
        enabled=True,
        endpoint="https://api.armoriq.ai",
        product="armoriq-sdk",
        max_buffer_size=1000,
        flush_interval_ms=5000,
        batch_size=100,
    ),
)

endpoint is the ArmorIQ backend base URL. Do not add /observability/spans; the shipper appends that path.

SessionOptions fields

ObservabilityConfig fields

The Python configuration does not expose a durable-disk option. Session traces are buffered in memory; product integrations that need cross-process durability implement it separately.

Environment variables

An explicit config value wins over its environment fallback. In particular, ObservabilityConfig() contains enabled=True and product="armoriq-sdk", so those values override the corresponding environment variables. Omit the config block when you want the environment to control the default behavior.

To disable one session regardless of the environment:

from armoriq_sdk import SessionOptions
from armoriq_sdk.observability import ObservabilityConfig

session = client.start_session(
    SessionOptions(observability=ObservabilityConfig(enabled=False))
)

Trace and span lifecycle

The session implements the current Model A lifecycle:

1. start_plan() ends the previous plan trace, if one exists, then starts an iap.plan trace.
2. check() opens an iap.check span and records the selected local, SDK, or proxy enforcement path beneath it.
3. Policy results become policy_call spans. report() and dispatch() create tool spans.
4. record_generation() adds a generation span to the active plan trace.
5. The next start_plan(), flush_observability(), close(), dispose(), or context-manager exit ends the active trace.

If an instrumented operation runs before start_plan(), the session lazily opens an iap.plan trace. A single session can therefore contain multiple plan traces without splitting their correlation ID.

Record policy and tool activity

Session methods instrument their own operations. You do not need to construct spans for the normal plan-check-execute-report loop.

from armoriq_sdk import ReportOptions

tool_name = "crm-mcp__search_contacts"
tool_input = {"query": "Acme"}

session.start_plan(
    [{"name": tool_name, "args": tool_input}],
    goal="Find contacts at Acme",
)

decision = session.check(tool_name, tool_input)
if decision.allowed:
    result = {"matches": [{"id": "contact-1", "company": "Acme"}]}
    session.report(
        tool_name,
        tool_input,
        result,
        ReportOptions(status="success", duration_ms=18),
    )

Tool inputs and outputs are truncated to a bounded preview before being attached to session spans, but arbitrary secrets are not automatically redacted. Avoid recording credentials or sensitive customer content.

Record model generations

Call record_generation() with usage from your model provider. Prompt and completion text are optional.

session.record_generation(
    model="gpt-4o",
    input_tokens=420,
    output_tokens=96,
    cache_read_tokens=0,
    cache_write_tokens=0,
    prompt=None,
    completion=None,
    finish_reason="stop",
)

The SDK estimates cost from its built-in per-model table. Unknown model names currently produce costUsd: 0.0; treat that value as not estimated, not evidence that the call was free. Token counts still remain available. See the data model for generation fields and aggregates.

Buffering, retries, and failure behavior

The recorder protects its in-memory trace buffer with a lock. Ended traces move to a separate locked shipping queue, and a daemon thread wakes every five seconds by default to flush them in batches.

• Successful 2xx responses return backend accepted and rejected counts.
• Invalid local batches are dropped before the request.
• 4xx responses are logged and dropped without retry.
• Network failures and 5xx responses use three retries after the first attempt, with 500 ms, 1 second, and 2 second backoff.
• Exhausted batches are logged and dropped. They are not requeued or written to disk.

Both the recorder and shipper register atexit cleanup as a last safety net. Application code should still close sessions explicitly because interpreter shutdown ordering and forced process termination cannot guarantee delivery.

Flush and shut down

flush_observability() ends the active plan trace and synchronously flushes queued traces, but leaves the background thread running. close() ends the trace, stops the thread, and performs a final flush. dispose() is an alias for close().

The session is a context manager, which is the safest default:

with client.start_session(options) as session:
    # Run one or more plans.
    ...

If the block raises, context-manager exit ends the active trace with error. Otherwise it ends with ok. Close the session before closing the client so the shared HTTP client is still available for the final flush.

Complete example

import os
import time
import uuid

from armoriq_sdk import ArmorIQClient, ReportOptions, SessionOptions
from armoriq_sdk.observability import ObservabilityConfig

api_key = os.environ.get("ARMORIQ_API_KEY")
if not api_key:
    raise RuntimeError("Set ARMORIQ_API_KEY")

client = ArmorIQClient(
    api_key=api_key,
    agent_id="crm-assistant",
)

options = SessionOptions(
    mode="sdk",
    default_mcp_name="crm-mcp",
    llm="gpt-4o",
    session_id=str(uuid.uuid4()),
    observability=ObservabilityConfig(
        enabled=True,
        endpoint=os.environ.get(
            "ARMORIQ_OBSERVABILITY_ENDPOINT",
            "https://api.armoriq.ai",
        ),
        product="armoriq-sdk",
        max_buffer_size=1000,
        flush_interval_ms=5000,
        batch_size=100,
    ),
)

tool_name = "search_contacts"
tool_input = {"query": "Acme"}

try:
    with client.start_session(options) as session:
        session.start_plan(
            [{"name": tool_name, "args": tool_input}],
            goal="Find contacts at Acme",
        )

        decision = session.check(tool_name, tool_input)
        if decision.allowed:
            started = time.perf_counter()
            result = {
                "matches": [{"id": "contact-1", "company": "Acme"}],
            }
            session.report(
                tool_name,
                tool_input,
                result,
                ReportOptions(
                    status="success",
                    duration_ms=(time.perf_counter() - started) * 1000,
                ),
            )

        session.record_generation(
            model="gpt-4o",
            input_tokens=420,
            output_tokens=96,
            cache_read_tokens=0,
            cache_write_tokens=0,
            prompt=None,
            completion=None,
            finish_reason="stop",
        )

        # Optional request boundary. Context-manager exit also closes and flushes.
        session.flush_observability()
finally:
    client.close()

Next: use the observability dashboard or troubleshoot delivery. For lower-level record shapes, see the data model.