OmegaEngine Python SDK

Decision infrastructure for autonomous AI.

OmegaEngine evaluates whether AI-generated actions should be allowed to execute. This SDK provides a Python client for OmegaEngine authorization and decision APIs.

Installation

pip install omegaengine

Quick Start

from omegaengine import OmegaClient

client = OmegaClient.create(
    api_key="your-api-key",
    base_url="https://omegaengine.ai",
)

# Authorize execution before side effects (recommended first path)
authz = client.authorize(
    agent_id="sales-agent",
    action="send_email",
    tool_name="email.send",
    tool_args={"to": "customer@example.com", "template": "follow-up"},
    vendor="SendGrid",
    amount=0,
)
print("Authorize decision:", authz["decision"])

# Authorization retries are idempotent by default. After human approval:
if authz["decision"] == "escalated":
    wait_for_human_approval(authz["approval_id"])
    claimed = client.claim_authorization(
        authorization_id=authz["authorization_id"],
        agent_id="sales-agent",
    )

# Stateful authorization session (start -> step -> end)
session = client.authorize_session_start(
    agent_id="sales-agent",
    metadata={"runId": "run-42"},
)
session_id = session["session"]["session_id"]

step = client.authorize_session_step(
    session_id=session_id,
    action="send_email",
    tool_name="email.send",
    tool_args={"to": "customer@example.com", "template": "follow-up"},
    vendor="SendGrid",
)

client.authorize_session_end(
    session_id=session_id,
    reason="workflow complete",
)

# Advanced: full scenario judgment (v2/judge)
decision = client.judge(
    "AI agent wants to send a $5,000 wire transfer to a new vendor",
    context="First-time vendor, no previous transactions",
    domain="FINANCIAL",
    risk_tolerance="low",
)

judge = decision.get("judge", {})
if judge.get("riskLevel") == "HIGH" or judge.get("needsHumanReview"):
    print(f"⚠️ Escalating: {judge.get('summary')}")

API Reference

client.authorize(**kwargs)

Authorize an agent action before it executes (POST /api/authorize).

authz = client.authorize(
    agent_id="payments-agent",
    action="wire_transfer",
    tool_name="bank.wire_transfer",
    tool_args={"destination_account": "vendor-acme", "amount": 25000, "currency": "USD"},
    amount=25000,
    vendor="acme-bank",
    risk_level="high",
    metadata={"ticket": "finops-4421"},
)
print(authz["decision"], authz["audit_id"])

Approved exact-action calls include authorization_id, action_hash, and a short-lived execution_capability. Escalated calls can be finalized with client.claim_authorization(...) after the linked human approval; the server re-checks budget before issuing the capability.

client.authorize_session_start(**kwargs)

Start a stateful authorization session (POST /api/authorize/session/start).

client.authorize_session_step(**kwargs)

Authorize a step within a session (POST /api/authorize/session/step).

client.authorize_session_end(**kwargs)

Finalize a session and return summary (POST /api/authorize/session/end).

client.judge(scenario, **kwargs)

Advanced method — evaluates a scenario through the full v2/judge decision pipeline.

result = client.judge(
    "Transfer $50,000 to vendor",
    context="Quarterly payroll processing",
    risk_tolerance="low",
    domain="finance",
    metadata={"dept": "HR", "amount": 50000},
    agent_name="payroll-bot",
)

Parameters:

Parameter	Type	Description
scenario	str	The scenario to evaluate (required)
context	str	Additional context
risk_tolerance	str	"low", "medium", or "high"
domain	str	Domain classification
metadata	dict	Arbitrary metadata
agent_name	str	Name of the AI agent
client_tag	str	Client identifier
privacy_mode	bool	Enable privacy-preserving mode
trace_id	str	External trace ID for correlation

client.feedback(request_id, **kwargs)

Submit feedback on a previous decision for RLHF.

client.feedback(
    "req-abc123",
    verdict="accurate",
    rating=5,
    comment="Decision was correct",
)

client.verify(request_id)

Verify the cryptographic proof chain for a decision.

result = client.verify("req-abc123")
print("Valid:", result["valid"])

client.health()

Check API health status.

status = client.health()
print("Status:", status["status"])

Error Handling

The SDK provides typed exceptions for common error scenarios:

from omegaengine import (
    OmegaError,
    OmegaAuthError,
    OmegaRateLimitError,
    OmegaSafetyBlockError,
)

try:
    result = client.judge("scenario")
except OmegaSafetyBlockError as e:
    print(f"Blocked by safety: {e.violations}")
except OmegaRateLimitError as e:
    print(f"Rate limited. Retry after: {e.retry_after}s")
except OmegaAuthError:
    print("Invalid API key")
except OmegaError as e:
    print(f"API error ({e.status_code}): {e}")

Configuration

from omegaengine import OmegaClient, OmegaClientOptions

client = OmegaClient(OmegaClientOptions(
    api_key="your-key",
    base_url="http://localhost:3000",  # Custom endpoint
    admin_key="admin-key",            # For admin operations
    max_retries=3,                     # Retry count (default: 2)
    timeout_sec=15.0,                  # Request timeout (default: 10s)
))

TypeScript SDK Parity

This Python SDK mirrors the TypeScript SDK API:

TypeScript	Python
client.authorize({agent_id, action})	client.authorize(agent_id=..., action=...)
client.judge({scenario})	client.judge(scenario)
client.feedback({requestId, verdict})	client.feedback(request_id, verdict=verdict)
client.verify(requestId)	client.verify(request_id)

License

Apache-2.0 — see LICENSE.