nodius — Python SDK

Python client for the Nodius service.

Install

Install from PyPI: https://pypi.org/project/nodiusxyz/

pip install nodiusxyz

Or install from source:

cd sdk-python
pip install -e .

Quickest Start

from nodiusxyz import NodiusClient

client = NodiusClient.from_base58("https://rpc.nodius.xyz", secret_key="your-base58-secret-key")
# Credits are tied to your pubkey — deposit USDC to start making calls.
# generate_api_key() also creates the account record if it doesn't exist yet.
client.generate_api_key()   # auto-creates account if needed; auto-stores API key
slot = client.get_slot()
print(slot)

x402 Auto-Pay (Zero-Setup Onboarding)

Pass auto_pay=True and the SDK handles the entire payment flow automatically. When a request returns 402 Payment Required, the SDK:

1. Parses the payment-required header from the server
2. Builds and signs a USDC SPL transfer to the deposit address
3. Submits it via a public Solana RPC (no credits needed)
4. Confirms the deposit with Nodius to credit the account
5. Retries the original request — transparently

from nodiusxyz import NodiusClient

client = NodiusClient.from_base58(
    "https://rpc.nodius.xyz",
    secret_key="your-base58-secret-key",
    auto_pay=True,          # enable x402 auto-pay
    # solana_rpc="https://api.mainnet-beta.solana.com",  # optional: override the RPC used for deposit tx
)
# The first RPC call triggers an automatic $0.01 USDC deposit if the
# account has no credits. The wallet must have enough USDC and SOL for gas.
slot = client.get_slot()  # works immediately, even with zero prior balance

Requires the solders extra: pip install "nodiusxyz[solders]"

from_base58 is the recommended constructor. It accepts both 32-byte seeds and 64-byte expanded secret keys in base58 encoding and handles decoding internally — no extra imports needed.

generate_api_key() auto-creates the account record if it doesn't yet exist and auto-stores the returned API key on the client. Subsequent calls silently switch to API-key auth for lower latency. Wallet-signature auth works without ever calling it — it is only needed when an explicit API key is desired. Billable RPC calls require a positive credit balance; deposit USDC first or enable x402 auto-pay.

Quick Start

Keypair Auth (no API key required)

from nodiusxyz import NodiusClient

keypair = bytes([...])  # 64 bytes: secret(32) + public(32)
client = NodiusClient("https://rpc.nodius.xyz", keypair=keypair)

client.generate_api_key()   # auto-creates account if needed; auto-stores API key for lower latency

# Solana RPC
slot = client.get_slot()
balance = client.get_balance("So11111111111111111111111111111111111111112")

# Any JSON-RPC method
result = client.call("getHealth")

# Batch calls
results = client.call_batch([
    {"method": "getSlot"},
    {"method": "getBalance", "params": ["So11..."]},
])

print(f"Credits remaining: {client.credits_remaining}")
client.close()

Keypair auth signs requests with Ed25519 and does not require bots to store generated API keys. Consider using NodiusClient.from_base58() for a simpler setup when you have a base58-encoded secret key.

API Key Auth

from nodiusxyz import NodiusClient

client = NodiusClient("https://rpc.nodius.xyz", api_key="your-key")

# Solana RPC
slot = client.get_slot()
balance = client.get_balance("So11111111111111111111111111111111111111112")

print(f"Credits remaining: {client.credits_remaining}")
client.close()

Session Token Auth

client = NodiusClient("https://rpc.nodius.xyz", session_token="your-session-token")
info = client.get_billing_account()

Context Manager

with NodiusClient("https://rpc.nodius.xyz", api_key="key") as client:
    slot = client.get_slot()

Account Management

client.generate_api_key()                   # Auto-create account + new API key
client.get_billing_account()                 # Balance, deposit address, usage stats
client.get_history(offset=0, limit=50)      # Usage history
client.confirm_deposit("tx-signature...")    # Confirm USDC deposit

Enhanced Methods

Enhanced history helpers are available when the connected endpoint profile enables archive/history service. Hot-node endpoints return a no-charge service-gate error for disabled profiles.

# Priority fee estimation
client.get_priority_fee_estimate({"accountKeys": ["..."]})
client.suggest_priority_fee()

# DAS (Digital Asset Standard)
client.get_assets_by_owner("wallet-address")

# Transaction enrichment
client.get_enriched_transaction("tx-sig")
client.explain_transaction("tx-sig")

# Bundle simulation
client.simulate_bundle({"transactions": [...]})

Jito Integration

client.jito_send_bundle(["signed-tx-1", "signed-tx-2"])
client.jito_get_bundle_statuses(["bundle-id"])
client.jito_get_tip_accounts()
client.jito_get_tip_floor()

Streaming: WebSocket & Yellowstone gRPC

Install the transport extras:

pip install nodiusxyz[ws]      # WebSocket (websocket-client)
pip install nodiusxyz[grpc]    # Yellowstone gRPC (grpcio)

WebSocket

Authenticates with your existing session token or API key and exposes the full subscription surface (accountSubscribe, logsSubscribe, programSubscribe, signatureSubscribe, slotSubscribe, rootSubscribe, slotsUpdatesSubscribe) plus the decoded subscribeSignal firehose. blockSubscribe and voteSubscribe are also implemented, but the live profile may not serve them — check GET /capabilities before relying on either:

from nodiusxyz import NodiusClient, NodiusWebSocket

rpc = NodiusClient("https://nodius.xyz", api_key="...")

with NodiusWebSocket.for_client(rpc) as ws:
    sub = ws.slot_subscribe()
    for notice in ws.notifications(subscription=sub):
        print("slot:", notice["params"]["result"]["slot"])
    ws.slot_unsubscribe(sub)

For the new-pair signal firehose: ws.subscribe_signal() then filter notifications(method="signalNotification").

Yellowstone gRPC

Opens a TLS channel to grpc.nodius.xyz:443 with the x-token metadata. Generate your Yellowstone stubs from the rpcpool/yellowstone-grpc protos and attach them:

from nodiusxyz import NodiusClient, YellowstoneGrpc

rpc = NodiusClient("https://nodius.xyz", api_key="...")

with YellowstoneGrpc.for_client(rpc) as ys:
    ys.with_stubs(my_geyser_stubs)  # your generated stub module
    stream = ys.subscribe(subscribe_request)
    for update in stream:
        ...

Bulk Operations

balances = client.bulk_get_balances(["addr1", "addr2", "addr3"])
token_bals = client.bulk_get_token_balances(["token-acc-1", "token-acc-2"])
txs = client.bulk_get_transactions(["sig1", "sig2"])

Trading helpers (non-custodial)

execute_sell builds an unsigned sell for a specified raw base-unit token amount. sell_all(mint, owner, slippage_bps) reads that owner's token inventory first, then builds a full-balance unsigned sell. Use the authenticated wallet's pubkey as owner. It rejects multiple token accounts (the builder uses the canonical ATA); a concurrent transfer between the read and build can make it fail. The inventory read is a separate billable RPC call. Neither helper signs or sends:

unsigned = client.sell_all(mint, owner_pubkey, slippage_bps=500)
# Sign unsigned["unsigned_tx_base64"] locally, then submit the signed tx.

send_and_confirm(signed_base64) is one HTTP write to the server, which handles the send and confirmation wait. Do not add wait_for_confirmation to that normal flow. If you already broadcast a signed transaction separately, wait_for_confirmation(signature, timeout=30, poll_interval=0.4) polls getSignatureStatuses (billable reads) without resending. It returns the confirmed/finalized status or raises TransactionFailedError on an on-chain error; ConfirmationTimeoutError.signature preserves the signature when landing is still unknown. JSON-RPC -32602 raises InvalidParamsError, a subclass of RpcError with .code and .data.

Webhooks

Register webhooks to receive HTTP callbacks when new signatures are detected on watched addresses.

Supported filter_type values:

• "account" — watch a specific account address
• "program" — watch a program address
• "my_account" — watch your own pubkey (auto-set)

Note: "transaction" is not a supported filter type.

# Watch an account for new signatures
client.create_webhook(
    url="https://my.app/hook",
    filter_type="account",
    filter_value="SomePubkey...",
)

# Watch your own account
client.create_webhook(
    url="https://my.app/hook",
    filter_type="my_account",
)

hooks = client.list_webhooks()
client.delete_webhook("webhook-id")

Error Handling

from nodiusxyz import (
    NodiusError,
    AuthenticationError,
    InsufficientCreditsError,
    ServiceProfileDisabledError,
    RateLimitError,
    RpcError,
)

try:
    result = client.get_balance("...")
except RateLimitError as e:
    print(f"Rate limited, retry after {e.retry_after}s")
except ServiceProfileDisabledError as e:
    print(f"Route to another endpoint profile or skip: {e.service}")
except InsufficientCreditsError:
    print("Need more credits")
except AuthenticationError:
    print("Auth failed")
except RpcError as e:
    print(f"RPC error {e.code}: {e}")
except NodiusError as e:
    print(f"General error: {e}")

Health Check

# No auth required
health = client.health()