Trust OS Python SDK

Official Python SDK for the Trust OS Execution Governance and Decision Verification APIs.

Requires Python 3.9 or later.

---

What is Trust OS?

Trust OS is an Execution Governance Platform that intercepts high-impact operations before they run — AI agent actions, financial transfers, enterprise operations — and returns a governance verdict (APPROVE, REVIEW, or DENY) in real time.

• Real-time governance verdicts
• Immutable audit trails via append-only event streams
• Risk and policy evaluation
• Explainable decisions
• Observe mode for non-blocking recording

---

Installation

pip install trustos

---

Authentication

Sign up at trust-os.io, then provision an API key from your Dashboard. The key is prefixed with trst_live_. Store it as TRUSTOS_API_KEY in your server environment.

Security: This is a server-side SDK. Never import or use it in browser or frontend code. Always call from your backend:

Browser  →  Customer Backend  →  Trust OS SDK  →  Trust OS API

---

Quick Start

import os
import uuid
from trustos import TrustOS

# TrustOS is the preferred alias for TrustOSClient
client = TrustOS(api_key=os.environ["TRUSTOS_API_KEY"])

result = client.executions.create(
    execution_type="execute_tool",
    mode="govern",
    actor={"id": "agent_alpha_001", "type": "ai_agent"},
    context={"tool": "write_record", "target_ref": "table_ref_customers"},
    external_id=f"op-{uuid.uuid4()}",   # caller-generated deduplication ID
)

if result.is_approved():
    agent.execute_tool("write_record")
elif result.requires_review():
    # governance == "REVIEW" — pause until a human approver acts
    queue_for_human_review(result.id)
elif result.is_denied():
    agent.abort("Trust OS governance denial")

Reads TRUSTOS_API_KEY from the environment when no api_key argument is passed:

client = TrustOS()  # equivalent to TrustOS(api_key=os.environ["TRUSTOS_API_KEY"])

---

Decision Governance (V3 Alpha)

Trust OS V3 models governance around a Decision as the root object. A Decision can contain evidence snapshots, policy evaluations, recommendations, human decisions, and zero, one, or many linked executions.

Human Review is intentionally not exposed through API-key-based SDK methods. Human decisions are performed through authenticated workspace/dashboard review flows.

import os
from trustos import TrustOS

client = TrustOS(api_key=os.environ["TRUSTOS_API_KEY"])

decision = client.decision_governance.create(
    intent={
        "resource": "invoice",
        "action": "approve_payment",
    },
    mode="govern",
)

print(decision)

client.decision_governance also provides get(decision_id), list(**kwargs), add_event(decision_id, event), add_events(decision_id, events), list_executions(decision_id, **kwargs), and get_timeline(decision_id). See trustos/decision_governance.py for full parameter and response details.

A Decision's decision_trace_hash (returned by client.decision_governance.get()) is a deterministic current-state fingerprint. It is not a blockchain proof, immutable seal, or tamper-proof proof, and may change as the Decision graph/state changes.

---

Execution Governance API (v2)

client.executions.create()

Submit an operation for governance. Returns an ExecutionResult.

from trustos import TrustOS, Workflows

client = TrustOS()

result = client.executions.create(
    execution_type=Workflows.EXECUTE_TOOL,  # required — or use workflow= alias
    channel="api",                           # "voice"|"web"|"api"|"mobile"|"batch"|"email"|"chat"
    mode="govern",                           # "govern" (default) | "observe"
    actor={"type": "ai_agent", "id": "agent_alpha_001"},
    subject={
        "subject_ref": "cust_hash_a1b2c3",  # opaque customer ref — never raw PII
        "classification": "restricted",
    },
    context={"tool": "write_record", "target_ref": "table_ref_orders"},
    external_id="op-7f3a9b2c-...",          # caller-generated deduplication reference
)

print(result.id)           # "exec_01jz..."
print(result.governance)   # "APPROVE" | "REVIEW" | "DENY" | "PENDING"
print(result.can_proceed("govern"))  # True only when governance == "APPROVE"

Govern mode — Trust OS synchronously gates execution. Only proceed when can_proceed("govern") returns True.

Observe mode — Trust OS records and evaluates asynchronously. can_proceed("observe") always returns True; the caller owns the execution decision.

Idempotency: Pass idempotency_key to set the Idempotency-Key request header. Requests with the same key and identical body return the cached response, making retries safe on network failures. The same key with a different body returns a 409 error.

import uuid

idempotency_key = str(uuid.uuid4())  # generate once per operation; persist on your side

result = client.executions.create(
    execution_type="execute_tool",
    context={"tool": "write_record"},
    idempotency_key=idempotency_key,  # safe to retry with the same key
)

create() parameters

Parameter	Type	Description
execution_type	str	Workflow identifier (e.g. "execute_tool"). Required unless workflow= is passed.
workflow	str	Alias for execution_type. Pass one or the other, not both.
mode	str	"govern" (default) or "observe"
channel	str	Request channel; defaults to "api"
actor	dict	Actor invoking the execution
subject	dict	Subject of the operation — use opaque refs, never raw PII
context	dict	Workflow-specific context
intent	dict	Detected intent payload
evidence	dict	Creation-time evidence hints
external_id	str	Customer-supplied deduplication reference
metadata	dict	Arbitrary key-value metadata
idempotency_key	str	Sets the Idempotency-Key HTTP header (1–255 chars)

---

client.executions.get(execution_id)

Retrieve a single Execution Record with its rebuilt event DAG.

result = client.executions.get("exec_01jz...")
print(result.governance)       # "APPROVE" | "REVIEW" | "DENY" | "PENDING"
print(result.execution_status) # "running" | "completed" | "failed" | "cancelled"

---

client.executions.list(**kwargs)

List Execution Records for your organization.

page = client.executions.list(
    limit=25,
    execution_type="execute_tool",
    governance="REVIEW",
)
for execution in page["executions"]:
    print(execution.id, execution.governance)

# Fetch the next page
page2 = client.executions.list(cursor=page["next_cursor"])

---

client.executions.append_event(execution_id, event)

Append a single event to an Execution Record's audit trail.

client.executions.append_event(
    result.id,
    {
        "type": "api.succeeded",
        "node_id": "write_record",
        "data": {"rows_written": 3},
        "idempotency_key": f"{result.id}:write-succeeded",  # safe to retry
    },
)

Common event types: execution.started, execution.completed, execution.failed, execution.cancelled, intent.detected, identity.verified, identity.failed, fraud.completed, aml.completed, approval.requested, approval.approved, approval.rejected, policy.evaluated, policy.blocked, llm.requested, llm.completed, api.succeeded, api.failed.

Approval events (approval.approved, approval.rejected) require an API key with key_role: approver or key_role: admin. Agent keys receive a 403 insufficient_role error — this prevents AI agents from self-approving human-gated operations.

---

client.executions.append_events(execution_id, events)

Append a batch of up to 100 events in one call.

client.executions.append_events(result.id, [
    {"type": "identity.verified", "node_id": "kyc",   "idempotency_key": f"{result.id}:kyc"},
    {"type": "fraud.completed",   "node_id": "fraud", "data": {"score": 0.02}, "idempotency_key": f"{result.id}:fraud"},
])

---

ExecutionResult helpers

Method / Property	Description
result.id	The execution_id
result.governance	Verdict: APPROVE, REVIEW, DENY, PENDING
result.execution_status	Downstream status: running, completed, failed, …
result.is_approved()	True when governance == "APPROVE"
result.requires_review()	True when governance == "REVIEW"
result.is_denied()	True when governance == "DENY"
result.can_proceed(mode)	In govern mode: True only on APPROVE. In observe mode: always True.
result.required_actions	Required human actions list from governance block
result.policy_evaluations	Per-policy evaluation list from governance block
result.to_dict()	Raw API response as dict

---

Examples

AI Agent (govern mode)

from trustos import TrustOS, Workflows, GovernanceVerdicts

client = TrustOS()

result = client.executions.create(
    execution_type=Workflows.EXECUTE_TOOL,
    mode="govern",
    actor={"id": "agent_alpha_001", "type": "ai_agent"},
    context={"tool": "write_record", "target_ref": "table_ref_orders"},
)

if result.governance == GovernanceVerdicts.APPROVE:
    agent.execute_tool("write_record")
elif result.governance == GovernanceVerdicts.REVIEW:
    notify_approver(result.id)
elif result.governance == GovernanceVerdicts.DENY:
    agent.abort("Blocked by Trust OS policy")

AI Agent (observe mode — non-blocking)

result = client.executions.create(
    execution_type=Workflows.EXECUTE_TOOL,
    mode="observe",
    actor={"id": "agent_beta_002", "type": "ai_agent"},
    context={"tool": "read_file", "path_ref": "file_ref_report"},
)

if result.can_proceed("observe"):  # always True in observe mode
    agent.execute_tool("read_file")

client.executions.append_event(result.id, {
    "type": "execution.completed",
    "idempotency_key": f"{result.id}:completed",
    "data": {"lines_read": 120},
})

Financial Transfer

result = client.executions.create(
    execution_type=Workflows.TRANSFER,
    mode="govern",
    actor={"id": "payments-api", "type": "service"},
    context={
        "amount": 250000,
        "currency": "USD",
        "destination_ref": "account_hash_beneficiary",
    },
)

if result.is_approved():
    execute_payment()
elif result.requires_review():
    queue_for_human_review(result.id)
else:
    raise ValueError("Transfer denied by Trust OS policy")

---

Constants

Workflows

Constant	Value	Policy Pack
Workflows.EXECUTE_TOOL	"execute_tool"	ai_agent
Workflows.TRANSFER	"transfer"	financial_v1
Workflows.BALANCE_INQUIRY	"balance_inquiry"	financial_v1
Workflows.ADDRESS_CHANGE	"address_change"	financial_v1
Workflows.CARD_SUSPENSION	"card_suspension"	financial_v1
Workflows.DOCUMENT_VERIFICATION	"document_verification"	financial_v1

GovernanceVerdicts

Constant	Value	Meaning
GovernanceVerdicts.APPROVE	"APPROVE"	Operation authorized — proceed
GovernanceVerdicts.REVIEW	"REVIEW"	Human action required before proceeding
GovernanceVerdicts.DENY	"DENY"	Operation blocked by policy — do not execute
GovernanceVerdicts.PENDING	"PENDING"	Verdict not yet computed (async governance)

---

Error Handling

from trustos import (
    TrustOS,
    TrustOSError,
    TrustOSAuthError,
    TrustOSValidationError,
    TrustOSRateLimitError,
    TrustOSNotFoundError,
    TrustOSNetworkError,
)

client = TrustOS()

try:
    result = client.executions.create(execution_type="execute_tool")
except TrustOSAuthError as e:
    # 401/403 — invalid or missing API key, or insufficient role (e.g. agent key on approval event)
    print(f"Authentication failed: {e}")
except TrustOSValidationError as e:
    # 400/422 — malformed request parameters
    print(f"Validation error: {e}")
except TrustOSRateLimitError:
    # 429 — back off and retry
    print("Rate limited — retry after a delay")
except TrustOSNotFoundError as e:
    # 404 — execution_id not found
    print(f"Not found: {e}")
except TrustOSNetworkError as e:
    # Timeout or connection failure
    print(f"Network error: {e}")
except TrustOSError as e:
    # Any other API error (including 409 idempotency conflict)
    print(f"API error {e.status_code}: {e}")

All error instances expose: status_code, response_body, error_code, request_id.

---

Known Limitations

• KI-AL-001 (High): PII guard is shallow — structured field names are rejected at execution creation (subject, context), but free-text data payloads in event appends are not scanned. Do not pass raw PII in event data fields.
• KI-AL-005 (Medium): Rate limiting is IP-based only. Authenticated per-key throughput controls are not enforced server-side. Do not rely on the API to enforce throughput controls in multi-tenant deployments.

---

Legacy API (v1)

The v1 Decision Verification API remains fully supported. New integrations should use the v2 Execution Governance API above.

client.verify_decision(payload: dict) -> dict

result = client.verify_decision({
    "action": "wire_transfer",
    "amount": 250000,
    "currency": "USD",
    "destination": "account_hash_beneficiary",
})

print(result["recommendation"])  # APPROVE | REVIEW | DENY
print(result["proof_hash"])

client.verify(payload) is an alias for verify_decision().

---

Documentation

• Website: https://trust-os.io
• Developer Docs: https://trust-os.io/docs
• OpenAPI: https://trust-os.io/openapi.json
• GitHub: https://github.com/trustos-trustfolio

---

License

MIT