Python SDK Production Guide

Your service (FastAPI, a worker, an agent loop)
        │
        ▼
ParmanaClient (Python SDK)
        │  bearer-key auth, typed errors
        ▼
Parmana Runtime (your deployment, see Production deployment)
        │
        ▼
Policy Engine + registered connectors

import os
from dataclasses import dataclass


@dataclass(frozen=True)
class ParmanaSettings:
    endpoint: str
    api_key: str
    fail_closed: bool = True

    @classmethod
    def from_env(cls) -> "ParmanaSettings":
        endpoint = os.environ["PARMANA_ENDPOINT"]
        api_key = os.environ["PARMANA_API_KEY"]
        if not endpoint.startswith("https://") and os.environ.get("ENVIRONMENT") == "production":
            raise ValueError("PARMANA_ENDPOINT must use HTTPS in production")
        return cls(endpoint=endpoint, api_key=api_key)


settings = ParmanaSettings.from_env()

from parmana import ParmanaClient

client = ParmanaClient(endpoint=settings.endpoint, api_key=settings.api_key)

import logging
from typing import Optional

from parmana import (
    ExecutionRejectedError,
    NetworkError,
    ParmanaClient,
    TimeoutError as ParmanaTimeoutError,
)

logger = logging.getLogger(__name__)


class SafeParmanaClient:
    """Wraps ParmanaClient so every non-success path returns None instead of propagating."""

    def __init__(self, client: ParmanaClient, fail_closed: bool = True):
        self.client = client
        self.fail_closed = fail_closed

    def execute(self, transaction):
        try:
            return self.client.execution.execute(transaction)
        except ExecutionRejectedError as exc:
            logger.warning("execution denied", extra={"reason": str(exc)})
            return None
        except (NetworkError, ParmanaTimeoutError) as exc:
            # Unknown, not "did not execute": the request may still have run.
            # Read it with the same id before doing anything else.
            logger.error("parmana outcome unknown", extra={"error": str(exc)})
            raise
        except Exception:
            logger.exception("unexpected error calling parmana")
            if self.fail_closed:
                return None
            raise

from parmana import ServerError

NOTHING_RAN = {"SIGNING_UNAVAILABLE", "EXECUTION_INTENT_UNAVAILABLE", "CONNECTOR_NOT_REGISTERED"}

try:
    record = client.execution.execute(transaction)
except ServerError as exc:
    if exc.server_code in NOTHING_RAN:
        # Nothing was executed. Retry later with a NEW business_transaction_id....
    else:
        # EXECUTION_RECORD_INCOMPLETE (500) or EXECUTION_OUTCOME_UNKNOWN (502): the action
        # was released. Never resend; the operator finalizes or resolves the Execution Intent.
        raise

import json
from datetime import datetime, timezone


class AuditLogger:
    def __init__(self, db_connection):
        self.db = db_connection

    def log_outcome(self, *, business_transaction_id: str, principal_id: str, action: str,
                     outcome: str, trust_record_id: str | None = None, reason: str | None = None):
        self.db.execute(
            """
            INSERT INTO parmana_audit_log
                (business_transaction_id, principal_id, action, outcome, trust_record_id, reason, occurred_at)
            VALUES (%s, %s, %s, %s, %s, %s, %s)
            """,
            (
                business_transaction_id,
                principal_id,
                action,
                outcome,
                trust_record_id,
                reason,
                datetime.now(timezone.utc),
            ),
        )

CREATE TABLE parmana_audit_log (
    id BIGSERIAL PRIMARY KEY,
    business_transaction_id TEXT NOT NULL,
    principal_id TEXT NOT NULL,
    action TEXT NOT NULL,
    outcome TEXT NOT NULL,          -- 'APPROVED' | 'DENIED' | 'ERROR'
    trust_record_id TEXT,
    reason TEXT,
    occurred_at TIMESTAMPTZ NOT NULL DEFAULT now()
);
CREATE INDEX ON parmana_audit_log (business_transaction_id);
CREATE INDEX ON parmana_audit_log (occurred_at);

import time
from prometheus_client import Counter, Histogram

parmana_execute_total = Counter(
    "parmana_execute_total", "Total execute() calls", ["outcome"]
)
parmana_execute_latency_seconds = Histogram(
    "parmana_execute_latency_seconds", "execute() latency"
)

start = time.monotonic()
try:
    trust_record = safe_client.execute(transaction)
    parmana_execute_total.labels(outcome="approved" if trust_record else "denied").inc()
finally:
    parmana_execute_latency_seconds.observe(time.monotonic() - start)

from collections import deque
from datetime import datetime, timedelta


class RateLimiter:
    def __init__(self, max_requests: int = 100, window_seconds: int = 60):
        self.max_requests = max_requests
        self.window = timedelta(seconds=window_seconds)
        self.requests: deque[datetime] = deque()

    def allow(self) -> bool:
        now = datetime.utcnow()
        cutoff = now - self.window
        while self.requests and self.requests[0] < cutoff:
            self.requests.popleft()
        if len(self.requests) >= self.max_requests:
            return False
        self.requests.append(now)
        return True

from parmana import ParmanaClient

def smoke_test(endpoint: str, api_key: str) -> None:
    client = ParmanaClient(endpoint=endpoint, api_key=api_key)
    health = client.execution.health()
    assert health["status"] == "UP", f"unexpected health status: {health['status']}"
    version = client.execution.version()
    print(f"Connected to {endpoint}, server version {version}")