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.

Overview

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

AssetCoreClient: High-level client with async and sync methods
Operation builders: Typed functions for all 16 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.

Structure

assetcore_sdk/
  __init__.py          # Package exports
  client.py            # AssetCoreClient facade
  operations.py        # Operation builder functions
  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.

from assetcore_sdk import AssetCoreClient

client = AssetCoreClient(
    write_url="http://localhost:8080",
    read_url="http://localhost:8081",
    timeout=30.0
)

Parameter	Type	Default	Description
`write_url`	str	`http://localhost:8080`	Write daemon URL
`read_url`	str	`http://localhost:8081`	Read daemon URL
`timeout`	float	30.0	Request timeout in seconds

Commit Methods

# Async
result = await client.commit(
    operations=[...],
    idempotency_key="unique-key"
)

# Sync
result = client.commit_sync(
    operations=[...],
    idempotency_key="unique-key"
)

Parameter	Type	Required	Description
`operations`	list	Yes	List of operation dictionaries
`idempotency_key`	str	No	Deduplication key
`metadata`	dict	No	User metadata

Returns CommitResponse with global_seq, batch_seq, and minted_instances.

Read Methods

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

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

Read methods return typed response objects with freshness metadata.

Operation Builders

Import individual builders or the entire module:

from assetcore_sdk.operations import (
    create_container,
    add_fungible,
    remove_fungible,
    transfer_fungible,
    mint_instance,
    place_in_slot,
    register_class
)

Each function returns a dictionary suitable for the operations list:

op = create_container(
    container_id=1001,
    kind="Standard"
)
# Returns: {"op": "CreateContainer", "args": {...}}

Error Types

from assetcore_sdk.errors import (
    AssetCoreError,      # Base exception
    ValidationError,     # 422 errors
    NotFoundError,       # 404 errors
    ConflictError,       # 409 errors
    ConnectionError,     # Network failures
    ServiceUnavailable   # 503 errors
)

Errors include the response body with error code and details.

Examples

Basic Commit

from assetcore_sdk import AssetCoreClient
from assetcore_sdk.operations import create_container, add_fungible

client = AssetCoreClient()

operations = [
    create_container(container_id=1001, kind="Standard"),
    add_fungible(
        container_id=1001,
        class_id=100,
        key=1,
        quantity=500
    )
]

result = await client.commit(
    operations=operations,
    idempotency_key="init-container-1001"
)

print(f"Committed at sequence {result.global_seq}")

Handling Errors

from assetcore_sdk.errors import ValidationError, ConflictError

try:
    result = await client.commit(operations)
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 client.get_container_balances(1001)
for balance in balances.balances:
    print(f"Class {balance.class_id}: {balance.quantity}")

# Check freshness
print(f"Data as of sequence {balances.freshness.checkpoint_seq}")

Sync Usage

For scripts or environments without async:

client = AssetCoreClient()

# All async methods have _sync variants
result = client.commit_sync(operations)
balances = client.get_container_balances_sync(1001)

Building Complex Transactions

from assetcore_sdk.operations import (
    register_class,
    register_class_shape,
    mint_instance,
    place_in_slot
)

operations = [
    register_class(
        class_id=200,
        name="Sample Tube",
        fungible=False
    ),
    register_class_shape(
        class_id=200,
        shape_id=1,
        width=1,
        height=3
    ),
    mint_instance(class_id=200, key=1),
    place_in_slot(
        container_id=1001,
        instance_id=9001,  # Known from allocation
        slot_index=0
    )
]

result = await client.commit(operations)
print(f"Minted instances: {result.minted_instances}")