Skip to content

Chain & State

Get the unique identifier of the connected chain.

async def get_chain_identifier() -> str

Returns: str. Chain identifier.

Get the server’s software version.

async def get_server_version() -> str

Returns: str

Get the current protocol version number.

async def get_protocol_version() -> int

Returns: int

Get the current model architecture version required by the protocol.

async def get_architecture_version() -> int

Returns: int. Must match ARCHITECTURE_VERSION in soma-models.

Get the required embedding dimension for the current architecture.

async def get_embedding_dim() -> int

Returns: int. For example, 2048 for V1.

Get the minimum stake required to register a model, in shannons.

async def get_model_min_stake() -> int

Returns: int. Minimum stake in shannons.

Verify that the SDK and server API versions are compatible. Raises an exception on mismatch.

async def check_api_version() -> None
await client.check_api_version() # raises on mismatch

Fetch a network object by ID.

async def get_object(object_id: str) -> ObjectRef
ParameterTypeRequiredDefaultDescription
object_idstrYes-Hex-encoded object ID

Returns: ObjectRef

Fetch a specific version of a network object.

async def get_object_with_version(object_id: str, version: int) -> ObjectRef
ParameterTypeRequiredDefaultDescription
object_idstrYes-Hex-encoded object ID
versionintYes-Object version number

Returns: ObjectRef

Get the SOMA balance for an address.

async def get_balance(address: str) -> float
ParameterTypeRequiredDefaultDescription
addressstrYes-SOMA address

Returns: float. Balance in SOMA.

balance = await client.get_balance("0x1234...")
print(f"{balance} SOMA")

List objects owned by an address.

async def list_owned_objects(
owner: str,
object_type: Optional[str] = None,
limit: Optional[int] = None,
) -> list[ObjectRef]
ParameterTypeRequiredDefaultDescription
ownerstrYes-Owner address
object_typestrNoNoneFilter by object type
limitintNoNoneMaximum number of results

Returns: list[ObjectRef]

Fetch the full system state for the current epoch, including validators, parameters, and model registry.

async def get_latest_system_state() -> SystemState

Returns: SystemState

Get information about a specific epoch, or the current epoch if not specified.

async def get_epoch(epoch: Optional[int] = None) -> EpochInfo
ParameterTypeRequiredDefaultDescription
epochintNoNoneEpoch number (current if omitted)

Returns: EpochInfo

Get the most recent finalized checkpoint.

async def get_latest_checkpoint() -> CheckpointSummary

Returns: CheckpointSummary

Get a checkpoint by sequence number.

async def get_checkpoint_summary(sequence_number: int) -> CheckpointSummary
ParameterTypeRequiredDefaultDescription
sequence_numberintYes-Checkpoint sequence number

Returns: CheckpointSummary

Block until the next epoch begins.

async def wait_for_next_epoch(timeout: float = 120.0) -> int
ParameterTypeRequiredDefaultDescription
timeoutfloatNo120.0Maximum seconds to wait

Returns: int. The new epoch number.

new_epoch = await client.wait_for_next_epoch(timeout=300)
print(f"Epoch {new_epoch} started")

Get the estimated UNIX timestamp (ms) when the next epoch starts. Computed as epoch_start_timestamp_ms + epoch_duration_ms from the current system state.

async def get_next_epoch_timestamp() -> int

Returns: int. Estimated UNIX timestamp in milliseconds.

Get the estimated UNIX timestamp (ms) when the epoch after next starts. Computed as epoch_start_timestamp_ms + 2 * epoch_duration_ms. Useful for scheduling actions that require two epoch boundaries (e.g., claim_rewards after submit_data).

async def get_following_epoch_timestamp() -> int

Returns: int. Estimated UNIX timestamp in milliseconds.