Developers › SDK › Python

Python SDK

The official async Python SDK for MoltChain. Built on httpx and websockets with full type hints.

Installation

bash

pip install moltchain

Requires Python 3.10+. Dependencies: httpx, websockets, PyNaCl, base58.

Quick Start

Python

import asyncio
from moltchain import Connection, Keypair, PublicKey, TransactionBuilder

async def main():
    # Connect to a local validator
    conn = Connection("http://localhost:8899", ws_url="ws://localhost:8900")

    # Generate a keypair
    kp = Keypair.generate()
    print(f"Address: {kp.public_key()}")

    # Check balance
    balance = await conn.get_balance(kp.public_key())
    print(f"Balance: {balance['shells']} shells")

    # Build and send a transfer
    recipient = PublicKey("RecipientBase58Address...")
    blockhash = await conn.get_recent_blockhash()

    tx = (TransactionBuilder()
        .add(TransactionBuilder.transfer(kp.public_key(), recipient, 1_000_000_000))
        .set_recent_blockhash(blockhash)
        .build_and_sign(kp))

    signature = await conn.send_transaction(tx)
    print(f"TX Signature: {signature}")

    await conn.close()

asyncio.run(main())

Connection Class

Connection(rpc_url, ws_url=None)

Create an async connection to a MoltChain validator node.

Parameters

Name	Type	Required	Description
rpc_url	str	Yes	HTTP JSON-RPC endpoint URL
ws_url	Optional[str]	No	WebSocket endpoint URL (required for subscriptions)

Python

conn = Connection("http://localhost:8899", ws_url="ws://localhost:8900")

Chain Queries

await get_slot() -> int

Get the current slot number.

Python

slot = await conn.get_slot()  # 42

await get_latest_block() -> Dict[str, Any]

Get the most recently produced block.

await get_block(slot: int) -> Dict[str, Any]

Get a block by slot number.

await get_recent_blockhash() -> str

Get a recent blockhash for signing transactions.

await get_metrics() -> Dict[str, Any]

Get chain performance metrics — TPS, total transactions, total blocks, average block time.

await get_chain_status() -> Dict[str, Any]

Get comprehensive chain status.

await health() -> Dict[str, str]

Liveness probe. Returns {"status": "ok"}.

await get_network_info() -> Dict[str, Any]

Get network information — chain ID, network ID, version, slot, validator/peer count.

await get_peers() -> List[Dict[str, Any]]

Get connected P2P peers.

await get_total_burned() -> Dict[str, int]

Get total burned MOLT — {"shells": ..., "molt": ...}.

Account Methods

await get_balance(pubkey: PublicKey) -> Dict[str, int]

Get account balance in shells and MOLT.

Python

balance = await conn.get_balance(PublicKey("7xKXtg2CW87d..."))
print(balance["shells"])  # 5000000000
print(balance["molt"])    # 5.0

await get_account(pubkey: PublicKey) -> Dict[str, Any]

Get raw account data — shells, owner, executable, data.

await get_account_info(pubkey: PublicKey) -> Dict[str, Any]

Get enhanced account information with metadata.

await get_transaction_history(pubkey: PublicKey, limit: int = 10) -> Dict[str, Any]

Get paginated transaction history for an account.

Transaction Methods

await send_transaction(transaction: Transaction) -> str

Serialize and submit a signed transaction. Returns the signature string.

Python

blockhash = await conn.get_recent_blockhash()
tx = (TransactionBuilder()
    .add(TransactionBuilder.transfer(kp.public_key(), recipient, 1_000_000_000))
    .set_recent_blockhash(blockhash)
    .build_and_sign(kp))

sig = await conn.send_transaction(tx)
print(f"Signature: {sig}")

await get_transaction(signature: str) -> Dict[str, Any]

Look up a transaction by its signature hash.

Validator & Staking

await get_validators() -> List[Dict[str, Any]]

Get all registered validators.

await get_validator_info(pubkey: PublicKey) -> Dict[str, Any]

Get detailed validator information.

await get_validator_performance(pubkey: PublicKey) -> Dict[str, Any]

Get validator performance metrics.

await stake(from_keypair: Keypair, validator: PublicKey, amount: int) -> str

Build, sign, and send a stake transaction. Amount in shells.

Python

validator = PublicKey("ValidatorPubkey...")
sig = await conn.stake(kp, validator, 10_000_000_000)  # 10 MOLT

await unstake(from_keypair: Keypair, validator: PublicKey, amount: int) -> str

Build, sign, and send an unstake request.

await get_staking_status(pubkey: PublicKey) -> Dict[str, Any]

Get staking status for an account.

await get_staking_rewards(pubkey: PublicKey) -> Dict[str, Any]

Get accumulated staking rewards.

Contract Methods

await get_contract_info(contract_id: PublicKey) -> Dict[str, Any]

Get information about a deployed smart contract.

await get_contract_logs(contract_id: PublicKey) -> Dict[str, Any]

Get execution logs emitted by a contract.

await get_all_contracts() -> Dict[str, Any]

List all deployed contracts.

Program Methods

await get_program(program_id: PublicKey) -> Dict[str, Any]

Get program information.

await get_programs() -> Dict[str, Any]

List all deployed programs.

await get_program_stats(program_id: PublicKey) -> Dict[str, Any]

Get execution statistics for a program.

await get_program_calls(program_id: PublicKey) -> Dict[str, Any]

Get recent call history for a program.

await get_program_storage(program_id: PublicKey) -> Dict[str, Any]

Get on-chain storage summary for a program.

NFT Methods

await get_collection(collection_id: PublicKey) -> Dict[str, Any]

Get NFT collection metadata.

await get_nft(collection_id: PublicKey, token_id: int) -> Dict[str, Any]

Get a specific NFT by collection and token ID.

await get_nfts_by_owner(owner: PublicKey) -> Dict[str, Any]

Get all NFTs owned by an address.

await get_nfts_by_collection(collection_id: PublicKey) -> Dict[str, Any]

Get all NFTs in a collection.

await get_nft_activity(collection_id: PublicKey, token_id: int) -> Dict[str, Any]

Get activity history for a specific NFT.

WebSocket Subscriptions

All subscription methods are async and require ws_url in the Connection constructor. Each on_* method returns a subscription ID for unsubscribing with the corresponding off_* method. Callbacks can be sync or async.

await on_slot(callback: Callable[[int], None]) -> int

Subscribe to slot updates. Unsubscribe with off_slot(sub_id).

Python

sub_id = await conn.on_slot(lambda slot: print(f"Slot: {slot}"))
# ... later
await conn.off_slot(sub_id)

await on_block(callback) -> int

Subscribe to new blocks. Unsubscribe with off_block(sub_id).

await on_transaction(callback) -> int

Subscribe to all confirmed transactions. Unsubscribe with off_transaction(sub_id).

await on_account_change(pubkey: PublicKey, callback) -> int

Watch a specific account for balance changes. Unsubscribe with off_account_change(sub_id).

await on_logs(callback, contract_id: Optional[PublicKey] = None) -> int

Stream contract logs. Optional contract filter. Unsubscribe with off_logs(sub_id).

await on_program_updates(callback) / on_program_calls(callback, program_id?)

Program deploy/upgrade events and invocations. Unsubscribe with off_program_updates / off_program_calls.

await on_nft_mints(callback, collection_id?) / on_nft_transfers(callback, collection_id?)

NFT mint and transfer events. Optional collection filter. Unsubscribe with off_nft_mints / off_nft_transfers.

await on_market_listings(callback) / on_market_sales(callback)

Marketplace listing and sale events. Unsubscribe with off_market_listings / off_market_sales.

await close()

Close the WebSocket connection, cancel background tasks, and clear subscriptions.

Keypair Class

Ed25519 keypair backed by PyNaCl. Implements signing and seed-based derivation.

Keypair.generate() -> Keypair

Generate a new random keypair.

Python

kp = Keypair.generate()
print(f"Address: {kp.public_key()}")

Keypair.from_seed(seed: bytes) -> Keypair

Derive a keypair from a 32-byte seed. Raises ValueError if seed is not 32 bytes.

Python

seed = b'\x00' * 32
kp = Keypair.from_seed(seed)

Keypair.load(path: Path) -> Keypair / keypair.save(path: Path)

Load a keypair from a JSON file or save it to one. The JSON contains seed, pubkey, and pubkey_base58 fields.

Python

from pathlib import Path

kp = Keypair.generate()
kp.save(Path("my-wallet.json"))

# Later
kp2 = Keypair.load(Path("my-wallet.json"))
assert kp.public_key() == kp2.public_key()

keypair.sign(message: bytes) -> bytes

Sign an arbitrary message. Returns the 64-byte Ed25519 detached signature.

keypair.public_key() -> PublicKey

Get the public key as a PublicKey object.

keypair.seed() -> bytes

Get the 32-byte seed for backup/restore.

PublicKey Class

A 32-byte public key with base58 encoding. Uses the base58 library.

PublicKey(value: Union[str, bytes, list])

Create from a base58 string, bytes, or list of ints. Raises ValueError if not 32 bytes.

Python

pk1 = PublicKey("7xKXtg2CW87d97TXJSDpbD5jBkheTqA83TZRuJosgAsU")
pk2 = PublicKey(b'\x00' * 32)
pk3 = PublicKey([0] * 32)

Methods

Method	Returns	Description
to_base58()	str	Base58-encoded string representation
to_bytes()	bytes	Raw 32-byte bytes
__str__()	str	Same as to_base58()
__eq__(other)	bool	Byte-level equality
__hash__()	int	Hashable for use in sets/dicts

Class Methods

Method	Returns	Description
PublicKey.from_base58(s)	PublicKey	Create from base58 string
PublicKey.from_bytes(b)	PublicKey	Create from bytes
PublicKey.new_unique()	PublicKey	Random key (testing only)

Transaction Types & Builder

Dataclasses

Python

@dataclass
class Instruction:
    program_id: PublicKey
    accounts: List[PublicKey]
    data: bytes

@dataclass
class Message:
    instructions: List[Instruction]
    recent_blockhash: str

@dataclass
class Transaction:
    signatures: List[str]  # hex-encoded
    message: Message

TransactionBuilder

Fluent builder for constructing and signing transactions.

Method	Returns	Description
add(instruction)	TransactionBuilder	Append an instruction
set_recent_blockhash(hash)	TransactionBuilder	Set the recent blockhash
build()	Message	Build unsigned message
build_and_sign(keypair)	Transaction	Build, sign, and return a Transaction

Static Instruction Builders

Method	Description
TransactionBuilder.transfer(from_pubkey, to_pubkey, amount)	Create a MOLT transfer instruction (type 0). Amount in shells.
TransactionBuilder.stake(from_pubkey, validator, amount)	Create a stake instruction (type 9)
TransactionBuilder.unstake(from_pubkey, validator, amount)	Create an unstake instruction (type 10)

Serialization helpers: TransactionBuilder.message_to_bincode(msg) and TransactionBuilder.transaction_to_bincode(tx) are available for custom serialization workflows.

Error Handling

All RPC methods raise standard Python Exception with descriptive messages from the server.

Python

try:
    balance = await conn.get_balance(pubkey)
except Exception as e:
    if "Account not found" in str(e):
        print("Account does not exist yet")
    elif "Rate limit" in str(e):
        await asyncio.sleep(1)
    else:
        raise

Common error codes:

-32601 — Method not found
-32602 — Invalid params
-32003 — Admin auth required
-32005 — Rate limit / subscription limit exceeded