Python SDK

The Asset Core Python SDK provides a typed HTTP client for interacting with the write and read daemons, plus operation builders that map directly to transaction JSON. It is a thin wrapper over the public HTTP contract, so anything you can do in the SDK can be done over raw HTTP as well.

Overview

The SDK is distributed as the assetcore-sdk package and provides:

• AssetCoreClient: High-level client with async and sync methods
• Operation builders: Typed dataclasses for all 24 operations
• Error types: Structured exceptions for error handling
• Generated client: OpenAPI-based HTTP layer

The SDK is a pure client; it invokes the public HTTP API and cannot mutate server-side state directly. This keeps the security model consistent across all integrations.

Structure

assetcore_sdk/
  __init__.py          # Package exports
  client.py            # AssetCoreClient facade
  operations.py        # Operation builder dataclasses
  errors.py            # Exception types
  models/              # Typed data classes
  utils/               # HTTP and serialization helpers
  _generated/          # OpenAPI codegen output

Fields

AssetCoreClient

The main client class for interacting with Asset Core. Use separate client instances for write and read tokens to preserve least-privilege access.

import os
from assetcore_sdk import AssetCoreClient

write_client = AssetCoreClient(
    base_url="http://localhost:8080",
    api_key=os.environ["ASSETCORE_WRITE_TOKEN"],
    timeout=30.0,
)

read_client = AssetCoreClient(
    base_url="http://localhost:8081",
    api_key=os.environ["ASSETCORE_READ_TOKEN"],
    timeout=30.0,
)

Parameter	Type	Default	Description
base_url	str	required	Base URL for either the write or read daemon
api_key	str	required	Bearer token used for Authorization
timeout	float	10.0	Request timeout in seconds
headers	dict	None	Extra headers to send with every request
refresh_token	callable	None	Optional sync refresh token provider
refresh_token_async	callable	None	Optional async refresh token provider

Commit Methods

# Async
result = await write_client.commit_operations(
    operations=[...],
    idempotency_key="unique-key-2026-01-15",
    namespace_id=5001
)

# Sync
result = write_client.commit_operations_sync(
    operations=[...],
    idempotency_key="unique-key-2026-01-15",
    namespace_id=5001
)

Parameter	Type	Required	Description
operations	list	Yes	List of typed operations or envelopes
namespace_id	int	No	Namespace (defaults to 1)
idempotency_key	str	No	Deduplication key
metadata	dict	No	User metadata
actor_id	str	No	Actor identifier for audit trails
policy_id	str	No	Policy identifier (future use)
origin	dict	No	Optional origin metadata

Returns CommitResponse with world_seq_start, world_seq_end, event_count, and correlation metadata.

Read Methods

# Get container balances
balances = await read_client.get_container_balances(1001)

# Get container (sync)
container = read_client.get_container_sync(1001)

Read methods return typed response objects with freshness metadata. You can use freshness to enforce read-your-writes semantics in client code.

Operation Builders

Import individual builders or the entire module:

from assetcore_sdk.operations import (
    CreateContainer,
    AddFungible,
    RemoveFungible,
    TransferFungible,
    AddInstance,
    PlaceInSlot,
    RegisterClass,
)

Each builder creates a dataclass that can be converted to the operation envelope:

op = CreateContainer(
    container_id=1001,
    kind={"type": "balance"},
    owner=None,
    policies=None,
)
# op.to_envelope() -> {"op": "CreateContainer", "args": {...}}

Error Types

from assetcore_sdk.errors import (
    AssetCoreError,        # Base exception
    HttpError,             # Transport failures
    ValidationError,       # 422 errors
    ConflictError,         # 409 errors
    NotFoundError,         # 404 errors
    SDKUnauthorizedError,  # 401 errors
    SDKForbiddenError,     # 403 errors
    SDKQuotaError,         # 429 errors
    TokenRefreshError      # Refresh hook failures
)

Errors include the response body with error code and details.

Examples

Basic Commit

from assetcore_sdk import AssetCoreClient
from assetcore_sdk.operations import CreateContainer, AddFungible

write_client = AssetCoreClient(
    base_url="http://localhost:8080",
    api_key=os.environ["ASSETCORE_WRITE_TOKEN"],
)

operations = [
    CreateContainer(
        container_id=1001,
        kind={"type": "balance"},
        owner=None,
        policies=None,
    ),
    AddFungible(
        class_id=100,
        key=1,
        quantity=500,
        location={"container_id": 1001, "kind": "balance"},
    ),
]

result = await write_client.commit_operations(
    operations,
    idempotency_key="init-container-2026-01-15-1001",
    namespace_id=5001,
)

print(f"Committed at world seq {result.world_seq_start}")

Handling Errors

from assetcore_sdk.errors import ValidationError, ConflictError

try:
    result = await write_client.commit_operations(operations, namespace_id=5001)
except ValidationError as e:
    print(f"Validation failed: {e.error_code}")
    print(f"Details: {e.details}")
except ConflictError as e:
    print(f"Conflict: {e.message}")

Reading State

# Get balances
balances = await read_client.get_container_balances(1001, namespace_id=5001)
for balance in balances.balances:
    print(f"Class {balance.class_id}: {balance.quantity}")

# Check freshness
print(f"Data as of world seq {balances.freshness.world_seq}")

Sync Usage

For scripts or environments without async:

import os
from assetcore_sdk import AssetCoreClient

write_client = AssetCoreClient(
    base_url="http://localhost:8080",
    api_key=os.environ["ASSETCORE_WRITE_TOKEN"],
)

read_client = AssetCoreClient(
    base_url="http://localhost:8081",
    api_key=os.environ["ASSETCORE_READ_TOKEN"],
)

# All async methods have _sync variants
result = write_client.commit_operations_sync(operations, namespace_id=5001)
balances = read_client.get_container_balances_sync(1001, namespace_id=5001)

Building Complex Transactions

from assetcore_sdk.operations import (
    RegisterClass,
    RegisterClassShape,
    AddInstance,
)

operations = [
    RegisterClass(
        request={
            "class_id": 200,
            "flags": 2,
            "name": "Sample Tube",
        }
    ),
    RegisterClassShape(
        request={
            "class_id": 200,
            "stack_key": None,
            "shape": {"width": 1, "height": 3},
        }
    ),
    AddInstance(
        class_id=200,
        key=1,
        location={"container_id": 1001, "kind": "slot", "slot_index": 1},
    ),
]

result = await write_client.commit_operations(operations, namespace_id=5001)
print(f"Committed transaction {result.commit_id}")

Related references

• Getting Started with Python SDK
• Action Reference
• Recipes