Metadata-Version: 2.4
Name: trustos
Version: 0.2.0
Summary: Python SDK for the Trust OS Execution Governance and Decision Verification APIs
Author-email: "Trustfolio Inc." <admin@trust-os.io>
License-Expression: MIT
Project-URL: Homepage, https://trust-os.io
Project-URL: Documentation, https://trust-os.io/docs
Project-URL: Source, https://github.com/trustos-trustfolio/trustos-python-sdk
Project-URL: Issues, https://github.com/trustos-trustfolio/trustos-python-sdk/issues
Project-URL: OpenAPI, https://trust-os.io/openapi.json
Keywords: trust-os,decision-verification,ai-agents,fintech,api-client
Classifier: Development Status :: 5 - Production/Stable
Classifier: Intended Audience :: Developers
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.9
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Topic :: Software Development :: Libraries :: Python Modules
Classifier: Topic :: Internet :: WWW/HTTP
Requires-Python: >=3.9
Description-Content-Type: text/markdown
License-File: LICENSE
Requires-Dist: requests>=2.31.0
Provides-Extra: dev
Requires-Dist: pytest>=8.0.0; extra == "dev"
Requires-Dist: responses>=0.25.0; extra == "dev"
Dynamic: license-file

# 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

```bash
pip install trustos
```

---

## Authentication

Sign up at [trust-os.io](https://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

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

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

---

## Execution Governance API (v2)

### `client.executions.create()`

Submit an operation for governance. Returns an `ExecutionResult`.

```python
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.

```python
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.

```python
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.

```python
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.

```python
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.

```python
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)

```python
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)

```python
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

```python
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

```python
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`

```python
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
