API Reference¶
Complete reference for all public classes and methods.
Core¶
ErgoNode¶
ergo_agent.core.node.ErgoNode(node_url=PUBLIC_NODE_URL, explorer_url=PUBLIC_EXPLORER_URL, api_key=None, timeout=15.0)
¶
Synchronous client for the Ergo blockchain. Uses the public Explorer API by default — no node required.
Usage
node = ErgoNode() # uses public API node = ErgoNode(node_url="http://localhost:9053", api_key="secret")
Functions¶
get_height()
¶
Return current blockchain height.
get_network_info()
¶
Return full network state info.
get_balance(address)
¶
Return the ERG and token balance for an address.
Returns:
| Name | Type | Description |
|---|---|---|
Balance |
Balance
|
structured balance with ERG float and token list. |
get_unspent_boxes(address, limit=50)
¶
Return unspent boxes (UTXOs) for an address.
Tries the local node's blockchain UTXO endpoint first (fastest, uses validated state), then falls back to the explorer API.
get_transaction_history(address, offset=0, limit=20)
¶
Get recent transaction history for an address.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
address
|
str
|
Ergo address |
required |
offset
|
int
|
pagination offset |
0
|
limit
|
int
|
max transactions to return (default 20) |
20
|
Returns:
| Type | Description |
|---|---|
list[Transaction]
|
list[Transaction]: recent transactions, newest first |
get_mempool_transactions(address)
¶
Return pending (unconfirmed) transactions for an address.
get_oracle_pool_box(oracle_pool_nft_id)
¶
Fetch the live oracle pool box by its NFT ID. Used to read ERG/USD price and other data feeds.
The price is stored in register R4 as nanoERG per 1 USD cent.
submit_transaction(signed_tx)
¶
Submit a signed transaction to the network. If the local node rejects it due to an unsynchronized UTXO set, it automatically falls back to broadcasting via the Explorer API.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
signed_tx
|
dict[str, Any]
|
signed transaction as a dict (ErgoTransaction format) |
required |
Returns:
| Name | Type | Description |
|---|---|---|
str |
str
|
transaction ID |
close()
¶
Close the underlying HTTP client.
Wallet¶
ergo_agent.core.wallet.Wallet(address, _private_key_hex=None, _node_wallet=False, read_only=False)
¶
Ergo wallet — holds the private key material and signs transactions.
Usage
wallet = Wallet.from_mnemonic("word1 word2 ...") wallet = Wallet.from_node(node, wallet_password="secret") # uses node's built-in wallet wallet = Wallet.read_only(address="9f...") # no signing, query only
Attributes¶
address = address
instance-attribute
¶
Functions¶
read_only(address)
classmethod
¶
Read-only wallet -- can query balances and build transactions, but cannot sign. Useful for monitoring agents.
from_node_wallet(node_address)
classmethod
¶
Use the Ergo node's built-in wallet for signing. The node handles key storage and signing -- this SDK just triggers it.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
node_address
|
str
|
the address from the node's loaded wallet |
required |
from_mnemonic(mnemonic, passphrase='')
classmethod
¶
Create a wallet from a BIP39 mnemonic phrase.
Not yet implemented. Requires ergo-lib (sigma-rust) Python bindings for proper BIP32/44 key derivation (path m/44'/429'/0'/0/0).
For now, use Wallet.from_node_wallet() or Wallet.read_only() instead.
sign_transaction(unsigned_tx, node=None, secrets=None)
¶
Sign an unsigned transaction.
If using a node wallet, the node signs it. If using local keys, signs with the private key.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
unsigned_tx
|
dict[str, Any]
|
the unsigned transaction dict |
required |
node
|
Any | None
|
ErgoNode instance (required for node_wallet mode) |
None
|
secrets
|
dict[str, Any] | None
|
Optional signing secrets for protocols requiring extra proofs (e.g., ring signatures). Dict with: - 'dlog': list of secret hex strings for proveDlog - 'dht': list of DH tuple dicts with keys: 'secret', 'g', 'h', 'u', 'v' (all compressed hex) |
None
|
Returns:
| Name | Type | Description |
|---|---|---|
dict |
dict[str, Any]
|
signed transaction ready for submission |
TransactionBuilder¶
ergo_agent.core.builder.TransactionBuilder(node, wallet)
¶
Fluent builder for Ergo unsigned transactions.
Usage (simple transfer): tx = ( TransactionBuilder(node, wallet) .send(to="9f...", amount_erg=1.5) .with_fee(0.001) .build() )
Usage (contract interaction with context extensions): tx = ( TransactionBuilder(node, wallet) .with_input(pool_box, extension={"0": key_image_hex, "1": proof_hex}) .add_output_raw(ergo_tree=..., value_nanoerg=..., tokens=[...], registers={...}) .build() )
The builder: - Validates all destination addresses (checksum + network byte) - Supports explicit input boxes (for spending contract UTXOs) - Supports context extensions on inputs (for getVar() in ErgoScript) - Automatically selects additional wallet UTXOs if explicit inputs aren't enough - Converts addresses to ErgoTree hex (P2PK direct, P2S/P2SH via API) - Calculates change output - Returns an unsigned tx dict ready for signing
Functions¶
send(to, amount_erg)
¶
Add a simple ERG transfer output. (Deprecated: use send_funds)
send_token(to, token_id, amount)
¶
Add a token transfer output (also sends minimum ERG dust to the box).
send_funds(to, amount_erg, tokens=None)
¶
Add a transfer output sending ERG and optionally multiple tokens.
mint_token(name, description, amount, decimals)
¶
Mint a new native token (EIP-004 compliant).
The token ID will be the ID of the first input box in the built transaction. The caller's wallet receives the minted tokens.
add_output_raw(ergo_tree, value_nanoerg, tokens=None, registers=None)
¶
Add a raw output box (for custom contract interactions like DEX swaps).
with_input(box, extension=None)
¶
Add an explicit input box (for spending contract boxes like PoolBoxes).
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
box
|
Box | str
|
a Box object or a box ID string. If a string, the box will be fetched from the node when build() is called. |
required |
extension
|
dict[str, str | Constant] | None
|
context extension variables for this input.
Keys are variable IDs (as strings, e.g. "0", "1"),
values are either serialized hex strings or native
|
None
|
Example
from ergo_lib_python.chain import Constant builder.with_input(pool_box, extension={ "0": Constant.from_i64(100), "1": bytes(Constant(b"metadata")).hex(), # manual hex works too })
build()
¶
Build the unsigned transaction dict.
Supports three input modes: 1. Auto-selection only (default): selects wallet UTXOs to cover outputs + fee 2. Explicit only: uses only with_input() boxes 3. Mixed: explicit inputs first, then auto-selects wallet UTXOs for the remainder
Returns:
| Name | Type | Description |
|---|---|---|
dict |
dict[str, Any]
|
unsigned transaction in Ergo API format |
Address Utilities¶
ergo_agent.core.address
¶
Ergo address utilities: validation, encoding, ErgoTree resolution.
Ergo uses a custom Base58 encoding with Blake2b-256 checksums. Address format: network_byte + content_bytes + checksum (4 bytes)
Network types
- 0x01 = mainnet P2PK
- 0x02 = mainnet P2SH
- 0x03 = mainnet P2S
- 0x10 = testnet P2PK etc.
Reference: https://docs.ergoplatform.com/dev/wallet/address/
Classes¶
AddressError
¶
Bases: Exception
Raised for invalid Ergo addresses.
Functions¶
is_valid_address(address)
¶
Check if an Ergo address is valid without raising exceptions.
validate_address(address)
¶
Validate an Ergo address using the native Sigma-Rust bindings.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
address
|
str
|
Ergo address string (Base58 encoded) |
required |
Returns:
| Type | Description |
|---|---|
bool
|
True if valid |
Raises:
| Type | Description |
|---|---|
AddressError
|
if the address is malformed or has a bad checksum |
is_mainnet_address(address)
¶
Check if address is a mainnet address.
is_p2pk_address(address)
¶
Check if address is a P2PK (pay-to-public-key) address.
get_address_type(address)
¶
Return a human-readable address type.
address_to_ergo_tree(address, node_url='ignored')
¶
Convert an Ergo address to its ErgoTree hex representation.
Uses natively compiled ergo_lib rust bindings to generate the exact ErgoTree
offline, eliminating the need to query the node API.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
address
|
str
|
valid Ergo address |
required |
node_url
|
str
|
Kept for backwards compatibility but ignored. |
'ignored'
|
Returns:
| Name | Type | Description |
|---|---|---|
str |
str
|
ErgoTree hex string |
Privacy Protocols¶
ergo_agent.core.privacy
¶
Privacy module for privacy pool — application-level ring-signature privacy pools.
This module provides: - The NUMS second generator H (for key images / nullifiers) - The ErgoScript source for the PoolContract (deposit + withdrawal paths) - High-level helpers for constructing deposit and withdrawal transactions
Architecture
PoolBox { tokens: [(poolTokenId, totalReserve)] R4: Coll[GroupElement] — deposit keys R5: AvlTree — nullifier set (used key images) R6: Long — denomination R7: Int — maxDeposits (8 or 16) }
Deposit: add a key to R4, add tokens to the pool Withdrawal: ring-signature proves ownership of one key, key image as nullifier
Attributes¶
POOL_CONTRACT_SCRIPT = '\n{\n val poolKeys = SELF.R4[Coll[GroupElement]].get\n val denom = SELF.R6[Long].get\n val maxRing = SELF.R7[Int].get\n val poolOut = OUTPUTS(0)\n val H = decodePoint(fromBase16("' + NUMS_H_HEX + '"))\n val tokenDiff = poolOut.tokens(0)._2 - SELF.tokens(0)._2\n\n if (tokenDiff > 0L) {\n val scriptOk = poolOut.propositionBytes == SELF.propositionBytes\n val denomOk = poolOut.R6[Long].get == denom\n val maxOk = poolOut.R7[Int].get == maxRing\n val treeOk = poolOut.R5[AvlTree].get.digest == SELF.R5[AvlTree].get.digest\n val tokenIdOk = poolOut.tokens(0)._1 == SELF.tokens(0)._1\n val tokenOk = (tokenDiff % denom) == 0L\n val tickets: Int = (tokenDiff / denom).toInt\n val newKeys = poolOut.R4[Coll[GroupElement]].get\n val sizeOk = newKeys.size == poolKeys.size + tickets\n val spaceOk = newKeys.size <= maxRing\n val oldKeysOk = poolKeys.indices.forall { (i: Int) => newKeys(i) == poolKeys(i) }\n val newKey = newKeys(poolKeys.size)\n val newKeyValid = newKey != groupGenerator\n val uniqueKeyOk = poolKeys.forall { (pk: GroupElement) => pk != newKey }\n sigmaProp(scriptOk && denomOk && maxOk && treeOk && tokenIdOk && tokenOk &&\n sizeOk && spaceOk && oldKeysOk && newKeyValid && uniqueKeyOk)\n } else if (tokenDiff < 0L) {\n val ringOk = poolKeys.size >= 2\n val scriptOk = poolOut.propositionBytes == SELF.propositionBytes\n val denomOk = poolOut.R6[Long].get == denom\n val maxOk = poolOut.R7[Int].get == maxRing\n val keysOk = poolOut.R4[Coll[GroupElement]].get == poolKeys\n val tokenIdOk = poolOut.tokens(0)._1 == SELF.tokens(0)._1\n val tokenOk = tokenDiff == -denom\n val keyImage = getVar[GroupElement](0).get\n val insertProof = getVar[Coll[Byte]](1).get\n val keyImageSafe = keyImage != groupGenerator\n val keyImageNotH = keyImage != H\n val curTree = SELF.R5[AvlTree].get\n val newTree = curTree.insert(\n Coll((keyImage.getEncoded, Coll[Byte]())),\n insertProof\n ).get\n val treeOk = poolOut.R5[AvlTree].get.digest == newTree.digest\n val noteOutOk = if (OUTPUTS.size > 1 && OUTPUTS(1).tokens.size > 0) {\n OUTPUTS(1).tokens(0)._1 == SELF.tokens(0)._1 && OUTPUTS(1).tokens(0)._2 == denom\n } else { false }\n val ringProof = atLeast(1, poolKeys.map { (pk: GroupElement) =>\n proveDlog(pk) && proveDHTuple(groupGenerator, H, pk, keyImage)\n })\n sigmaProp(ringOk && scriptOk && denomOk && maxOk && keysOk &&\n tokenIdOk && tokenOk && noteOutOk && treeOk &&\n keyImageSafe && keyImageNotH) && ringProof\n } else {\n val hasTokens = SELF.tokens(0)._2 > 0L\n val ageOk = HEIGHT - SELF.creationInfo._1 >= 788400\n val scriptOk = poolOut.propositionBytes == SELF.propositionBytes\n val keysOk = poolOut.R4[Coll[GroupElement]].get == poolKeys\n val nullOk = poolOut.R5[AvlTree].get.digest == SELF.R5[AvlTree].get.digest\n val denomOk = poolOut.R6[Long].get == denom\n val maxOk = poolOut.R7[Int].get == maxRing\n val tokenIdOk = poolOut.tokens(0)._1 == SELF.tokens(0)._1\n val tokenAmtOk = tokenDiff == 0L\n sigmaProp(hasTokens && ageOk && scriptOk && keysOk && nullOk &&\n denomOk && maxOk && tokenIdOk && tokenAmtOk)\n }\n}\n'
module-attribute
¶
NUMS_H_HEX = '022975f1d28b92b6e84499b83b0797ef5235553eeb7edaa0cea243c1128c2fe739'
module-attribute
¶
Functions¶
generate_fresh_secret()
¶
Securely generates a fresh mathematical secret (private key) and its corresponding public key (compressed secp256k1 point) for a pool deposit.
Returns:
| Name | Type | Description |
|---|---|---|
tuple |
tuple[str, str]
|
(private_key_hex, public_key_compressed_hex) |
compute_key_image(secret_hex)
¶
Compute the key image (nullifier) M = secret * H for a withdrawal.
The key image is used to prevent double-spending: once a key image is inserted into the AvlTree nullifier set, the same secret cannot be used again.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
secret_hex
|
str
|
The 32-byte secret (private key) as hex string. |
required |
Returns:
| Type | Description |
|---|---|
str
|
Compressed GroupElement hex of the key image (33 bytes / 66 chars). |
Raises:
| Type | Description |
|---|---|
ValueError
|
If the secret is not a valid 32-byte hex string or is zero. |
generate_avl_insert_proof(key_image_hex, current_r5_hex=None)
¶
Generate an AvlTree insert proof for the key image nullifier.
Uses the ergo_avltree Python extension (PyO3 wrapper around
ergo_avltree_rust) to create and insert into the authenticated
AVL+ tree.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
key_image_hex
|
str
|
Compressed GroupElement hex of the key image. |
required |
current_r5_hex
|
str | None
|
Current R5 register hex (AvlTree). If None, assumes an empty tree. |
None
|
Returns:
| Name | Type | Description |
|---|---|---|
tuple |
tuple[bytes, str]
|
(proof_bytes, new_r5_hex) where new_r5_hex is the full Sigma-serialized AvlTree constant for the output box R5. |
serialize_context_extension(key_image_hex, proof_bytes)
¶
Build the context extension dict for a withdrawal transaction input.
Var 0: GroupElement (type 0x07 + 33 compressed bytes) Var 1: Coll[Byte] (type 0x0e + VLQ length + raw bytes)
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
key_image_hex
|
str
|
Compressed key image hex (66 chars). |
required |
proof_bytes
|
bytes
|
Raw AvlTree insert proof bytes. |
required |
Returns:
| Type | Description |
|---|---|
dict[str, str]
|
dict mapping string var indices to hex-encoded Sigma values. |
find_optimal_pool(node, pool_ergo_tree, pool_token_id, denomination)
¶
Finds an optimal, un-congested PoolBox for a deposit to mitigate UTXO contention.
Queries the blockchain for all unspent PoolBoxes matching the ErgoTree contract, filters for those serving the requested denomination that have available capacity, and returns a randomly selected pool to spread out concurrent deposits.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
node
|
ErgoNode
|
Connected ErgoNode client |
required |
pool_ergo_tree
|
str
|
ErgoTree hex of the pool contract |
required |
pool_token_id
|
str
|
The privacy pool token ID |
required |
denomination
|
int
|
The denomination amount requested |
required |
Returns:
| Name | Type | Description |
|---|---|---|
Box |
Box
|
A matched, unsaturated PoolBox. |
Raises:
| Type | Description |
|---|---|
ValueError
|
If no valid pools with available capacity are found. |
build_pool_deposit_tx(builder, pool_box, note_box, depositor_public_keys_hex, pool_ergo_tree, pool_token_id, denomination)
¶
Build a deposit transaction: send a NoteBox's tokens into a PoolBox.
The depositor's public keys are appended to the pool's key list (R4). The pool's token reserve increases by (denomination * number of keys).
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
builder
|
TransactionBuilder
|
a TransactionBuilder instance |
required |
pool_box
|
Box
|
the PoolBox to deposit into |
required |
note_box
|
Box
|
the NoteBox being deposited (must contain enough tokens) |
required |
depositor_public_keys_hex
|
list[str]
|
list of the depositor's one-time public keys |
required |
pool_ergo_tree
|
str
|
the ErgoTree hex of the pool contract |
required |
pool_token_id
|
str
|
the privacy pool token ID |
required |
denomination
|
int
|
the denomination amount (must match pool's R6) |
required |
Returns:
| Type | Description |
|---|---|
dict[str, Any]
|
unsigned transaction dict |
build_pool_withdraw_tx(builder, pool_box, secret_hex, recipient_ergo_tree, pool_ergo_tree, pool_token_id, denomination)
¶
Build a withdrawal transaction: ring-signature proof to withdraw from a PoolBox.
Computes the key image from the secret, generates the AvlTree insert proof, and serializes the context extension. The ring proof itself is generated by the node's prover at signing time.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
builder
|
TransactionBuilder
|
a TransactionBuilder instance |
required |
pool_box
|
Box
|
the PoolBox to withdraw from |
required |
secret_hex
|
str
|
the depositor's 32-byte secret key (hex) |
required |
recipient_ergo_tree
|
str
|
ErgoTree of the withdrawal recipient |
required |
pool_ergo_tree
|
str
|
ErgoTree hex of the pool contract |
required |
pool_token_id
|
str
|
the privacy pool token ID |
required |
denomination
|
int
|
amount to withdraw (must match pool's R6) |
required |
Returns:
| Type | Description |
|---|---|
dict[str, Any]
|
unsigned transaction dict |
decompose_into_tiers(amount)
¶
Greedy decomposition algorithm for privacy pool "Auto-Route" mechanism. Breaks any amount into the optimal number of deposits across the 4 pool tiers.
Tiers (in tokens): 1,000,000 | 100,000 | 10,000 | 1,000
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
amount
|
int
|
Total token amount to anonymize |
required |
Returns:
| Type | Description |
|---|---|
dict[int, int]
|
dict mapping denomination to number of required tickets (keys) |
dict[int, int]
|
e.g., 15,300,000 -> {1000000: 15, 100000: 3, 10000: 0, 1000: 0} |
Note: Any "loose change" remainder (< 1,000) cannot be deposited and is ignored here.
Data Models¶
ergo_agent.core.models.Box
¶
Bases: BaseModel
An unspent Ergo UTXO box.
Attributes¶
value_erg
property
¶
ERG value (human-readable).
Functions¶
decode_register(register_id)
¶
Dynamically decode a raw Box register hex string natively using ergo-lib-python's
Constant.from_bytes schema back into its typed Python equivalent.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
register_id
|
str
|
The register to fetch (e.g. 'R4', 'R5') |
required |
Returns:
| Type | Description |
|---|---|
Any | None
|
The parsed python value, or None if the register is missing or unparseable. |
ergo_agent.core.models.Balance
¶
ergo_agent.core.models.Token
¶
ergo_agent.core.models.SwapQuote
¶
ergo_agent.core.models.Transaction
¶
Bases: BaseModel
A submitted or confirmed transaction.
DeFi¶
OracleReader¶
ergo_agent.defi.oracle.OracleReader(node)
¶
Read live prices from the Ergo Oracle Pool v2.
Usage
oracle = OracleReader(node) price = oracle.get_erg_usd_price() # e.g. 0.87 print(f"ERG/USD: ${price:.2f}")
Functions¶
get_erg_usd_price()
¶
Return the current ERG/USD price from the oracle pool.
The oracle R4 value is nanoERG per 1 USD. We convert: price_usd = NANOERG_PER_ERG / nanoERG_per_USD
Returns:
| Name | Type | Description |
|---|---|---|
float |
float
|
ERG price in USD (e.g. 0.31 means 1 ERG = $0.31) |
get_erg_usd_nanoerg_per_usd()
¶
Return the raw oracle value: nanoERG per 1 USD. This is the value used directly in ErgoScript contracts.
get_oracle_box_id(pair='erg_usd')
¶
Return the current oracle pool box ID. Used when adding an oracle box as a data input to a transaction.
SpectrumDEX¶
ergo_agent.defi.spectrum.SpectrumDEX(node, api_url=SPECTRUM_API, timeout=15.0)
¶
Adapter for Spectrum Finance DEX on Ergo.
Usage
dex = SpectrumDEX(node) markets = dex.get_pools() quote = dex.get_quote(token_in="ERG", token_out="SigUSD", amount_erg=10.0) print(f"You get: {quote.token_out_amount} SigUSD")
Functions¶
get_pools(offset=0, limit=100)
¶
Fetch active Spectrum markets (trading pairs).
Returns:
| Type | Description |
|---|---|
list[Market]
|
list[Market]: sorted by volume descending |
get_erg_price_in_sigusd()
¶
Convenience: get ERG/USD price from DEX pool directly. Note: use OracleReader for the canonical on-chain oracle price.
get_quote(token_in, token_out, amount_erg=None, amount_token=None)
¶
Get a swap quote for a given input.
This uses the Spectrum API's last price data to estimate the output. Note: for exact on-chain output, pool box reserves should be read directly. The API price gives a good approximation.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
token_in
|
str
|
input token (name or ID, use "ERG" for native ERG) |
required |
token_out
|
str
|
output token (name or ID) |
required |
amount_erg
|
float | None
|
if token_in is ERG, specify amount in ERG |
None
|
amount_token
|
int | None
|
if token_in is a token, specify raw amount |
None
|
Returns:
| Type | Description |
|---|---|
SwapQuote
|
SwapQuote with expected output amount and price impact |
build_swap_order(token_in, token_out, amount_erg, return_address, min_output=None, max_slippage_pct=1.0)
¶
Build a Spectrum DEX swap order dict for TransactionBuilder.add_output_raw().
On Ergo's eUTXO model, DEX swaps work via order boxes: 1. User creates an order box containing ERG + order parameters 2. Spectrum off-chain bots detect this box and execute the swap 3. User receives output tokens in a new box at their address
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
token_in
|
str
|
input token name (e.g. "ERG") |
required |
token_out
|
str
|
output token name (e.g. "SigUSD") |
required |
amount_erg
|
float
|
amount of ERG to swap |
required |
return_address
|
str
|
address to receive output tokens |
required |
min_output
|
int | None
|
minimum acceptable output amount (auto-calculated if None) |
None
|
max_slippage_pct
|
float
|
max slippage for auto min_output calculation |
1.0
|
Returns:
| Type | Description |
|---|---|
dict[str, Any]
|
dict with 'ergo_tree', 'value_nanoerg', 'tokens', 'registers' |
dict[str, Any]
|
ready for TransactionBuilder.add_output_raw() |
close()
¶
SigmaUSD¶
ergo_agent.defi.sigmausd.SigmaUSD(node=None)
¶
Client for interacting with the SigmaUSD / AgeUSD protocol on Ergo. Provides read-only access to bank state, reserve ratio, prices, and transaction builders for minting/redeeming stablecoins.
Initialize the SigmaUSD client.
Functions¶
get_bank_state()
¶
Fetch the current state of the SigmaUSD Bank. Returns the reserve ratio, SigUSD price, and SigRSV price in nanoERG.
build_mint_sigusd_tx(amount_sigusd, wallet)
¶
Build an unsigned transaction to mint SigUSD in exchange for ERG.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
amount_sigusd
|
int
|
The integer amount of SigUSD cents to mint (e.g. 100 for 1.00 SigUSD). |
required |
wallet
|
Any
|
The Wallet instance of the user. |
required |
Returns:
| Name | Type | Description |
|---|---|---|
Dict |
dict[str, Any]
|
Unsigned transaction dict ready for signing. |
build_redeem_sigusd_tx(amount_sigusd, wallet)
¶
Build an unsigned transaction to redeem SigUSD for ERG.
RosenBridge¶
ergo_agent.defi.rosen.RosenBridge(node=None)
¶
Client for interacting with the Rosen Bridge. Provides read-only access to global TVL and transaction builders for sending assets cross-chain to Cardano, Bitcoin, etc.
Functions¶
get_bridge_status()
¶
Fetch the current status and TVL of the Rosen Bridge.
build_bridge_tx(to_chain, to_address, amount_erg, tokens, wallet)
¶
Build an unsigned transaction to bridge assets via Rosen.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
to_chain
|
str
|
Destination network (e.g. "Cardano", "Bitcoin") |
required |
to_address
|
str
|
Destination address on the target network |
required |
amount_erg
|
float
|
ERG amount to send |
required |
tokens
|
dict[str, int]
|
Dictionary of Ergo native tokens to send |
required |
wallet
|
Any
|
User's Wallet instance |
required |
ErgoTreasury¶
ergo_agent.defi.treasury.ErgoTreasury(node=None)
¶
Client for interacting with Ergo DAO Treasury or MultiSig wallets. Provides methods to build governance proposals, vote on proposals, and execute approved multi-sig transactions.
Functions¶
build_proposal_tx(treasury_address, target_address, amount_erg, description, wallet)
¶
Build an unsigned transaction that submits a new funding proposal to the DAO.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
treasury_address
|
str
|
The P2S address of the Treasury/MultiSig contract |
required |
target_address
|
str
|
The recipient address for the proposed funding |
required |
amount_erg
|
float
|
The amount of ERG requested in the proposal |
required |
description
|
str
|
A short string describing the proposal (stored in R4) |
required |
wallet
|
Any
|
The proposer's wallet instance |
required |
build_vote_tx(proposal_box_id, vote, wallet)
¶
Build an unsigned transaction to cast a vote on an active proposal.
build_execute_tx(proposal_box_id, treasury_address, wallet)
¶
Build an unsigned transaction to execute a proposal that has reached consensus.
Tools¶
ErgoToolkit¶
ergo_agent.tools.toolkit.ErgoToolkit(node, wallet, safety=None)
¶
Unified AI agent toolkit for the Ergo blockchain.
All methods are safe to call directly from an LLM's tool-calling loop. Every state-changing action is validated by the SafetyConfig before execution.
Functions¶
get_wallet_balance()
¶
Get the current ERG and token balance of the agent's wallet.
Returns:
| Type | Description |
|---|---|
dict[str, Any]
|
dict with 'erg' (float), 'tokens' (list), and 'summary' (str) |
get_erg_price()
¶
Get the current ERG/USD price from the Ergo Oracle Pool.
Returns:
| Type | Description |
|---|---|
dict[str, Any]
|
dict with 'erg_usd' (float) and 'source' (str) |
get_swap_quote(token_in, token_out, amount_erg=None, amount_token=None)
¶
Get a swap quote from Spectrum DEX without executing it.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
token_in
|
str
|
Input token (e.g. "ERG", "SigUSD") |
required |
token_out
|
str
|
Output token (e.g. "SigUSD", "ERG") |
required |
amount_erg
|
float | None
|
Amount in ERG if token_in is ERG |
None
|
amount_token
|
int | None
|
Raw token amount if token_in is a token |
None
|
Returns:
| Type | Description |
|---|---|
dict[str, Any]
|
dict with expected output, price impact, and fee info |
get_mempool_status()
¶
Check for pending transactions from this wallet.
Returns:
| Type | Description |
|---|---|
dict[str, Any]
|
dict with pending transaction count and tx IDs |
get_safety_status()
¶
Get current safety limits and usage status.
Returns:
| Type | Description |
|---|---|
dict[str, Any]
|
dict with remaining daily budget, rate limit status, etc. |
swap_erg_for_token(token_out, amount_erg, max_slippage_pct=1.0)
¶
Swap ERG for a token on Spectrum DEX.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
token_out
|
str
|
Token to receive (e.g. "SigUSD") |
required |
amount_erg
|
float
|
Amount of ERG to spend |
required |
max_slippage_pct
|
float
|
Maximum acceptable price impact (default 1%) |
1.0
|
Returns:
| Type | Description |
|---|---|
dict[str, Any]
|
dict with quote info and tx_id (or dry_run confirmation) |
mint_sigusd(amount_sigusd)
¶
Mint SigUSD using ERG from the wallet.
redeem_sigusd(amount_sigusd)
¶
Redeem SigUSD for ERG.
mint_sigmrsv(amount_sigrsv)
¶
Mint SigRSV (Reserve coin) using ERG from the wallet.
redeem_sigmrsv(amount_sigrsv)
¶
Redeem SigRSV (Reserve coin) for ERG.
bridge_assets(to_chain, to_address, amount_erg=0.0, tokens=None)
¶
Bridge ERG or tokens to another blockchain via Rosen Bridge.
create_treasury_proposal(treasury_address, target_address, amount_erg, description)
¶
Create a new funding proposal for an Ergo Treasury or MultiSig wallet.
execute_tool(tool_name, tool_input)
¶
Execute a tool by name with given inputs. Used by LLM frameworks to dispatch tool calls.
Returns:
| Name | Type | Description |
|---|---|---|
str |
str
|
JSON-encoded result |
to_openai_tools()
¶
Generate OpenAI function-calling tool definitions.
to_anthropic_tools()
¶
Generate Anthropic tool-use definitions.
to_langchain_tools()
¶
Generate LangChain BaseTool instances.
SafetyConfig¶
ergo_agent.tools.safety.SafetyConfig(max_erg_per_tx=10.0, max_erg_per_day=50.0, allowed_contracts=(lambda: ['spectrum', 'sigmausd', 'rosen', 'privacy_pool'])(), rate_limit_per_hour=20, dry_run=False, min_withdrawal_delay_blocks=100, min_pool_ring_size=4, _action_timestamps=deque(), _daily_spend_log=list(), _deposit_heights=dict())
dataclass
¶
Configuration for agent spending limits and operational boundaries.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
max_erg_per_tx
|
float
|
Hard cap on ERG per single transaction |
10.0
|
max_erg_per_day
|
float
|
Rolling 24h cap on total ERG spent |
50.0
|
allowed_contracts
|
list[str]
|
Whitelist of allowed interaction targets. Use protocol names ("spectrum", "sigmausd", "rosen") or raw Ergo addresses. |
(lambda: ['spectrum', 'sigmausd', 'rosen', 'privacy_pool'])()
|
rate_limit_per_hour
|
int
|
Max number of state-changing actions per hour |
20
|
dry_run
|
bool
|
If True, build transactions but never submit them |
False
|
Functions¶
validate_send(amount_erg, destination)
¶
Validate a send action. Raises SafetyViolation if any limit is exceeded.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
amount_erg
|
float
|
amount to send in ERG |
required |
destination
|
str
|
target address or protocol name |
required |
validate_rate_limit()
¶
Check that the agent hasn't exceeded the hourly action rate.
record_action(erg_spent=0.0)
¶
Record a completed action for rate limiting and daily spend tracking.
get_status()
¶
Return current safety status for agent awareness.
ergo_agent.tools.safety.SafetyViolation
¶
Bases: Exception
Raised when an agent action violates the configured safety rules.