keystoneos 0.1.0

Python SDK for the KeyStone OS settlement orchestration API

KeyStone OS Python SDK

Python SDK for the KeyStone OS settlement orchestration API. Provides both async and sync clients with full type safety, automatic token management, and retry logic.

Installation

pip install keystoneos

Quick Start

Async (recommended)

import asyncio
from keystoneos import KeystoneClient

async def main():
    async with KeystoneClient(
        client_id="your-client-id",
        client_secret="your-client-secret",
    ) as client:
        # List settlements
        settlements = await client.settlements.list(state="SETTLED", limit=10)
        for s in settlements.items:
            print(f"{s.id} [{s.state}]")

        # Get a specific settlement
        settlement = await client.settlements.get("settlement-uuid")

        # List available templates
        templates = await client.templates.list()

asyncio.run(main())

Sync

from keystoneos import KeystoneSyncClient

with KeystoneSyncClient(
    client_id="your-client-id",
    client_secret="your-client-secret",
) as client:
    settlements = client.settlements.list()
    for s in settlements.items:
        print(f"{s.id} [{s.state}]")

Features

Bilateral Settlement Instructions

Submit settlement instructions from both counterparties. When both sides match, a settlement is created automatically.

from keystoneos.types.instructions import (
    InstructionCreate,
    InstructionLegInput,
    InstructionPartyInput,
)
from decimal import Decimal
from datetime import datetime, timedelta, timezone

# Seller submits first (no trade_reference - one is generated)
seller = await client.instructions.submit(
    InstructionCreate(
        template_slug="dvp-immediate",
        role="seller",
        party=InstructionPartyInput(
            external_reference="SELLER-001",
            name="Acme Corp",
            wallet_address="0x1234...",
        ),
        legs=[
            InstructionLegInput(
                instrument_id="BOND-001",
                quantity=Decimal("1000"),
                direction="deliver",
            ),
        ],
        timeout_at=datetime.now(timezone.utc) + timedelta(hours=1),
        idempotency_key=client.generate_idempotency_key(),
    )
)

# Share seller.trade_reference with the buyer out-of-band

# Buyer submits with the same trade_reference
buyer = await client.instructions.submit(
    InstructionCreate(
        trade_reference=seller.trade_reference,
        template_slug="dvp-immediate",
        role="buyer",
        party=InstructionPartyInput(
            external_reference="BUYER-001",
            name="Beta LLC",
            wallet_address="0xabcd...",
        ),
        legs=[
            InstructionLegInput(
                instrument_id="BOND-001",
                quantity=Decimal("1000"),
                direction="receive",
            ),
        ],
        timeout_at=datetime.now(timezone.utc) + timedelta(hours=1),
        idempotency_key=client.generate_idempotency_key(),
    )
)

if buyer.settlement_id:
    print(f"Matched! Settlement: {buyer.settlement_id}")

Webhook Verification

Verify incoming webhook signatures to ensure they were sent by KeyStone.

from keystoneos.utils.webhook_verify import verify_signature

# In your webhook handler
payload = request.body  # raw bytes
signature = request.headers["X-Keystone-Signature"]
secret = "whsec_your-webhook-secret"

if verify_signature(payload, secret, signature):
    # Process the event
    ...
else:
    # Reject the request
    ...

Webhook Management

from keystoneos.types.webhooks import WebhookEndpointCreate

# Create an endpoint
endpoint = await client.webhooks.create(
    WebhookEndpointCreate(
        url="https://example.com/webhooks/keystone",
        events=["settlement.*"],
    )
)
# Save endpoint.secret - it is only returned once!

# Test the endpoint
result = await client.webhooks.test(endpoint.id)
print(f"Reachable: {result.success}")

# Rotate the secret (previous secret valid for 24h)
rotation = await client.webhooks.rotate_secret(endpoint.id)
print(f"New secret: {rotation.new_secret}")

Compliance Decisions

# Approve a settlement awaiting compliance review
settlement = await client.compliance.approve("settlement-uuid")

# Or reject it
settlement = await client.compliance.reject("settlement-uuid")

Pagination

All list endpoints return PaginatedResponse objects with convenience properties.

page = await client.settlements.list(limit=10)
print(f"Showing {len(page.items)} of {page.total}")

if page.has_more:
    next_page = await client.settlements.list(limit=10, offset=page.next_offset)

Error Handling

from keystoneos import (
    NotFoundError,
    ValidationError,
    RateLimitError,
    ConflictError,
)

try:
    settlement = await client.settlements.get("nonexistent-id")
except NotFoundError:
    print("Settlement not found")
except ValidationError as e:
    print(f"Invalid request: {e.detail}")
except RateLimitError as e:
    print(f"Rate limited. Retry after {e.retry_after}s")
except ConflictError as e:
    print(f"Conflict: {e.detail}")

Retry Configuration

Automatic retry with exponential backoff on 429 and 5xx errors.

from keystoneos import KeystoneClient, RetryConfig

client = KeystoneClient(
    client_id="...",
    client_secret="...",
    retry=RetryConfig(
        max_retries=5,
        base_delay=1.0,
        max_delay=60.0,
    ),
)

API Resources

Resource	Methods
client.settlements	create, get, list, get_events, get_related, submit_compliance_decision
client.instructions	submit, get, list, cancel
client.templates	list, get
client.webhooks	create, get, list, update, delete, test, rotate_secret, list_deliveries, verify_signature
client.compliance	approve, reject
client.environments	create, get, list, update, deactivate

Authentication

The SDK uses Auth0 M2M (Client Credentials) tokens. Tokens are automatically fetched, cached, and refreshed before expiry. No manual token management is needed.

Requirements

• Python 3.10+
• httpx >= 0.25.0
• pydantic >= 2.0.0

License

MIT