The mintlayer.node module is a JSON-RPC 2.0 client for the Mintlayer node
daemon. Every method is synchronous and thread-safe.
from mintlayer.node import Client
c = Client(
"http://127.0.0.1:3030",
username="user", # optional; Basic Auth applied only when set
password="pass", # optional
timeout=10.0, # optional, seconds (default 30.0)
session=None, # optional requests.Session (client owns it otherwise)
)Default ports: 3030 (mainnet), 13030 (testnet).
Transport security: Basic Auth credentials travel in cleartext over plain
http://. Bind the daemon to localhost or front it with an HTTPS reverse proxy / SSH tunnel when connecting over a network.
There is no per-call cancellation: configure timeout on the client (Go's
per-call context has no direct requests equivalent). Errors from the daemon
raise RPCError; transport and decode failures raise JSONRPCError:
from mintlayer.node import RPCError, JSONRPCError
try:
height = c.best_block_height()
except RPCError as e:
print(e.code, e.message)
except JSONRPCError as e:
... # HTTP failure or malformed response bodyThe HTTP status code is never inspected — a JSON-RPC error object in the body
is the error contract. JSON null results map to None for every
"not-found"-style method (block_id_at_height, stake_pool_balance, …).
def chainstate_info(self) -> ChainstateInfo: ...Returns a summary of the current chain state.
@dataclass(frozen=True)
class ChainstateInfo:
best_block_height: int
best_block_id: str
best_block_timestamp: Timestamp # Timestamp(timestamp: int), Unix seconds
median_time: Timestamp
is_initial_block_download: booldef best_block_id(self) -> str: ...Returns the hex block ID of the current tip.
def best_block_height(self) -> int: ...Returns the height of the current tip.
def block_id_at_height(self, height: int) -> str | None: ...Returns the block ID at a given height, or None if no block exists at that
height.
def block_height_in_main_chain(self, block_id: str) -> int | None: ...Returns the mainchain height for a block ID, or None if the block is not on
the main chain.
def get_block(self, block_id: str) -> str | None: ...Returns the hex-encoded raw block (None if unknown; the genesis block is not
retrievable).
def get_block_json(self, block_id: str) -> Any: ...Returns the block as decoded JSON. Useful for inspection without custom deserialization.
def get_mainchain_blocks(self, from_height: int, max_count: int) -> list[str]: ...Returns up to max_count mainchain block IDs starting at from_height.
def get_utxo(self, outpoint: Outpoint) -> Any: ...Returns the output at a given outpoint as decoded JSON (None if
spent/unknown).
@dataclass(frozen=True)
class Outpoint:
source_id: OutpointSourceID
index: int
@dataclass
class OutpointSourceID:
type: str # "Transaction" or "BlockReward"
content: Any # {"tx_id": "<hex>"} or {"block_id": "<hex>"}Build the content payloads with the helpers:
from mintlayer.node import Outpoint, OutpointSourceID, tx_source_content, block_source_content
op = Outpoint(
source_id=OutpointSourceID(type="Transaction", content=tx_source_content(tx_id)),
index=0,
)
utxo = c.get_utxo(op)def submit_block(self, block_hex: str) -> None: ...Submits a hex-encoded block. Used by block producers.
def stake_pool_balance(self, pool_address: str) -> Amount | None: ...Returns the total balance of a pool (staker pledge plus all delegations).
Returns None if the pool is not found.
def staker_balance(self, pool_address: str) -> Amount | None: ...Returns the staker's own balance, excluding delegations. Returns None if the
pool is not found.
def pool_decommission_destination(self, pool_address: str) -> str | None: ...Returns the address that receives funds when the pool is decommissioned.
def delegation_share(self, pool_address: str, delegation_address: str) -> Amount | None: ...Returns the amount owned by a specific delegation in a pool.
Amounts are decimal atom strings — 1 ML = 100,000,000,000 atoms (11
decimal places). Amount is a frozen dataclass with a single atoms: str
field.
def token_info(self, token_id: str) -> TokenInfo | None: ...Returns on-chain token metadata (None if unknown).
@dataclass
class TokenInfo:
type: str # "FungibleToken" or "NonFungibleToken"
content: Any # raw decoded JSONdef tokens_info(self, token_ids: list[str]) -> list[TokenInfo]: ...Batch version of token_info. More efficient than calling token_info in a
loop.
def order_info(self, order_id: str) -> OrderInfo | None: ...Returns the current state of an order.
@dataclass
class OrderInfo:
conclude_key: str
initially_asked: Any
initially_given: Any
ask_balance: Amount
give_balance: Amount
nonce: int | None # None for active orders (daemon sends null)
is_frozen: boolQuirk: nonce is None for active orders — the daemon sends JSON null for
the field. This fixes a known Go SDK incompatibility, where the null broke
uint64 decoding.
def orders_info_by_currencies(
self, ask: Currency | None, give: Currency | None
) -> dict[str, OrderInfo]: ...Returns all orders matching the given currency pair, as a dict from order ID to
OrderInfo. Pass None for either currency to match any (both keys are always
sent; None serialises as JSON null).
@dataclass(frozen=True)
class Currency:
type: str # "Coin" or "Token"
content: str | None # bech32 token ID when type is "Token"Construct with the helpers:
from mintlayer.node import Currency
orders = c.orders_info_by_currencies(Currency.coin(), Currency.token("ttml1..."))
any_coin = c.orders_info_by_currencies(None, Currency.coin())def contains_tx(self, tx_id: str) -> bool: ...Returns True if the mempool contains the transaction.
def contains_orphan_tx(self, tx_id: str) -> bool: ...Returns True if the orphan pool contains the transaction.
def get_transaction(self, tx_id: str) -> MempoolTx | None: ...Returns a mempool transaction (None if not present).
@dataclass(frozen=True)
class MempoolTx:
id: str
status: str
transaction: strdef mempool_submit_transaction(self, tx_hex: str, trust_policy: TrustPolicy | str) -> None: ...Submits a transaction to the local mempool only, without broadcasting to peers.
Use TrustPolicy.UNTRUSTED for transactions you constructed yourself; use
TrustPolicy.TRUSTED to skip some fee checks. A plain string is accepted too:
from mintlayer.node import TrustPolicy
c.mempool_submit_transaction(signed_hex, TrustPolicy.UNTRUSTED)class TrustPolicy(str, enum.Enum):
TRUSTED = "Trusted"
UNTRUSTED = "Untrusted"def get_fee_rate(self, in_top_x_mb: int) -> FeeRate | None: ...Returns the fee rate needed to land in the top in_top_x_mb megabytes of the
mempool.
@dataclass(frozen=True)
class FeeRate:
amount_per_kb: Amount # atoms per kilobytedef get_fee_rate_points(self) -> list[FeeRatePoint]: ...Returns the mempool fee-rate curve as a list of (size, rate) pairs.
Wire quirk: the daemon sends each point as a two-element array
[size, {"amount_per_kb": {...}}], decoded into:
@dataclass(frozen=True)
class FeeRatePoint:
size: int
rate: FeeRatedef memory_usage(self) -> int: ...Returns the current mempool memory usage in bytes.
def get_peer_count(self) -> int: ...Returns the number of currently connected peers.
def get_connected_peers(self) -> list[PeerInfo]: ...Returns details about all connected peers.
@dataclass(frozen=True)
class PeerInfo:
peer_id: int
address: str
peer_role: str
ban_score: int
user_agent: str
software_version: str
ping_wait: int | None = None
ping_last: int | None = None
ping_min: int | None = None
last_tip_block_time: int | None = Nonedef get_bind_addresses(self) -> list[str]: ...Returns the addresses the node is listening on for P2P connections.
def add_reserved_node(self, addr: str) -> None: ...Adds a persistent peer that the node always attempts to reconnect to.
def remove_reserved_node(self, addr: str) -> None: ...Removes a persistent peer.
def connect(self, addr: str) -> None: ...Makes a one-time connection attempt to a peer address.
def disconnect(self, peer_id: int) -> None: ...Closes the connection to a peer by ID.
def list_banned(self) -> list[BannedPeer]: ...Returns the list of banned peers.
Wire quirk: each entry is a two-element array
["<address>", {"time": [secs, nanos]}], decoded into:
@dataclass(frozen=True)
class BannedPeer:
address: str
ban_time: tuple[int, int] # (seconds, nanoseconds)def ban(self, address: str, duration: timedelta) -> None: ...Bans a peer for the specified duration. Durations are datetime.timedelta
values, split into the daemon's [seconds, nanoseconds] wire form:
from datetime import timedelta
c.ban("192.0.2.1", timedelta(hours=24))def unban(self, address: str) -> None: ...Removes a peer from the ban list.
def p2p_submit_transaction(self, tx_hex: str, trust_policy: TrustPolicy | str) -> None: ...Submits a transaction to the mempool and broadcasts it to peers. This is the
normal path for publishing a transaction to the network (the alternative — the
indexer's submit_transaction — requires --enable-post-routes, see
indexer.md).
def node_version(self) -> str: ...Returns the node software version string.
def node_shutdown(self) -> None: ...Initiates a graceful node shutdown.
- indexer.md — read-only chain queries and
submit_transaction - transactions.md — building and signing transactions
- wallet.md — the wallet daemon client