Skip to content

Python SDK

The official Python SDK for CertusOrdo.

Installation

pip install certusrodo

Requirements

  • Python 3.8+
  • httpx (HTTP client)
  • pydantic (data validation)

Quick Start

from certusrodo import CertusOrdoClient

# Initialize with API key
client = CertusOrdoClient(api_key="aa_your_api_key")

# Create an agent
agent = client.agents.create(
    name="my-agent",
    scopes=["read", "write"]
)

# Authenticate
tokens = client.auth.get_token(
    agent_id=agent.id,
    secret=agent.secret
)

# Begin transaction with rollback capability
txn = client.transactions.begin(
    pre_state={"balance": 1000},
    action_type="transfer"
)

# Commit or rollback
client.transactions.commit(txn.id, post_state={"balance": 750})
# OR
# client.transactions.rollback(txn.id, reason="Error")

Client Configuration

client = CertusOrdoClient(
    api_key="aa_your_key",
    base_url="https://web-production-b910f.up.railway.app",  # Optional
    timeout=30.0  # Optional, seconds
)

Resource Handlers

Organizations

# List organizations
orgs = client.organizations.list()

# Get specific organization
org = client.organizations.get("org-id")

# Create organization (returns API key)
# Note: Usually done once via API directly

Agents

# Create agent
agent = client.agents.create(
    name="processor",
    scopes=["read", "write"]
)

# List agents
agents = client.agents.list()
agents = client.agents.list(status="active")

# Get agent
agent = client.agents.get("agent-id")

# Revoke agent
client.agents.revoke("agent-id")

Authentication

# Get tokens
tokens = client.auth.get_token(
    agent_id="agent-id",
    secret="agent-secret"
)

# Access the tokens
tokens.access_token   # JWT for API calls
tokens.refresh_token  # For refreshing
tokens.expires_in     # Seconds until expiry

# Refresh tokens
new_tokens = client.auth.refresh_token(tokens.refresh_token)

# Revoke tokens
client.auth.revoke_token(tokens.refresh_token)

Sessions

# Create session with limits
session = client.sessions.create(
    agent_id="agent-id",
    max_actions=100,
    max_value_usd=5000.00,
    expires_in_seconds=3600
)

# List sessions
sessions = client.sessions.list()
sessions = client.sessions.list(status="active")

# Get session
session = client.sessions.get("session-id")

# End session
client.sessions.end("session-id")

Transactions

# Begin transaction
txn = client.transactions.begin(
    pre_state={"key": "value"},
    action_type="update",
    session_id="optional-session-id",
    rollback_confidence="guaranteed"
)

# Commit transaction
completed = client.transactions.commit(
    txn.id,
    post_state={"key": "new_value"}
)

# Rollback transaction
rollback = client.transactions.rollback(
    txn.id,
    reason="Something went wrong"
)
original_state = rollback.pre_state_snapshot

# List transactions
txns = client.transactions.list()
txns = client.transactions.list(status="completed")

# Get rollbackable transactions
rollbackable = client.transactions.list_rollbackable()

Audit

# List audit entries
entries = client.audit.list()
entries = client.audit.list(event_type="txn.rolled_back")
entries = client.audit.list(agent_id="agent-id")

# Verify chain integrity
verification = client.audit.verify()
print(f"Valid: {verification.is_valid}")

Exception Handling

from certusrodo import (
    CertusOrdoClient,
    CertusOrdoError,
    AuthenticationError,
    ValidationError,
    NotFoundError,
    RateLimitError,
    TransactionError,
    SessionError
)

client = CertusOrdoClient(api_key="aa_your_key")

try:
    agent = client.agents.get("invalid-id")
except NotFoundError:
    print("Agent not found")
except AuthenticationError:
    print("Invalid API key")
except RateLimitError as e:
    print(f"Rate limited, retry after {e.retry_after}s")
except CertusOrdoError as e:
    print(f"API error: {e}")

Models

All responses are Pydantic models with full type hints:

from certusrodo.models import Agent, Transaction, Session

agent: Agent = client.agents.create(name="test", scopes=["read"])
print(agent.id)          # UUID
print(agent.name)        # str
print(agent.status)      # str
print(agent.created_at)  # datetime

Source Code