SDK: Python

peaq-os-sdk on PyPI is the Python equivalent of @peaqos/peaq-os-sdk. Every method and type mirrors the JS SDK; the key differences are listed below.

Differences from the JavaScript SDK

Aspect	JavaScript	Python
Constructor	`new PeaqosClient({ config })`	`PeaqosClient(**kwargs)` (keyword args)
Keypairs	Frozen `{ address, privateKey }` object	Tuple `(address, private_key)`
Faucet methods	Global `fetch`	`requests.Session` as first arg
Response keys	`camelCase`	`snake_case`
Error hierarchy	`RuntimeError` with `.code`	Separate `RpcError` + `ApiError`
Extra accessors	n/a	`w3`, `web3`, `session`, `account`, `identity_registry`, `identity_staking`, `event_registry`, `machine_nft`, `did_precompile`, `batch_precompile`

Install

pip install "peaq-os-sdk>=0.4.0" python-dotenv
# Optional, only for OWS wallet lifecycle helpers:
pip install "peaq-os-sdk[ows]>=0.4.0"

Python: ≥ 3.10
Dependencies: web3>=6.0, eth-account, requests
Optional wallet dependency: open-wallet-standard through the [ows] extra
Virtualenv recommended.

python-dotenv is optional but recommended: PeaqosClient.from_env() reads from dotenv, so load_dotenv() at the top of your entry file is the simplest way to load .env.

Environment variables

Same set as the JS SDK: 8 required core vars (RPC URL, private key, and the 6 contract addresses), plus PEAQOS_MCR_API_URL (defaults to http://127.0.0.1:8000), plus 2 optional vars for smart-account deploy (MACHINE_ACCOUNT_FACTORY_ADDRESS) and cross-chain Machine NFT bridging (MACHINE_NFT_ADAPTER_ADDRESS). See JS environment variables and peaq mainnet contracts. For agung testnet addresses see Install → Agung testnet contracts. Bridging is mainnet-only since LayerZero has no DVN routes to agung. For OWS wallet lifecycle helpers, pass a passphrase argument or set OWS_PASSPHRASE. For Scale, set PEAQOS_ORCHESTRATION_URL (and optionally PEAQOS_API_KEY) before PeaqosClient.from_env() to enable client.orchestration. Full method reference: Orchestration (Python).

Client

`PeaqosClient`

class PeaqosClient:
    def __init__(
        self,
        *,
        rpc_url: str,
        private_key: str,
        identity_registry: str,
        identity_staking: str,
        event_registry: str,
        machine_nft: str,
        did_registry: str,
        batch_precompile: str,
        machine_account_factory: str | None = None,
        machine_nft_adapter: str | None = None,
        api_url: str = DEFAULT_API_URL,
        operational_limits: OperationalLimits | None = None,
    ) -> None: ...

str

required

RPC endpoint.

str

required

0x + 64 hex.

str

required

Identity Registry contract address.

str

required

Identity Staking contract address.

str

required

Event Registry contract address.

str

required

Machine NFT contract address (ONFT).

str

required

DID Registry precompile address.

str

required

Batch precompile address.

str | None

MachineAccountFactory contract address. Required only for deploy_smart_account and get_smart_account_address.

str | None

MachineNFTAdapter (LayerZero ONFT adapter) contract address on peaq. Required only for bridge_nft when source="peaq".

str

MCR API base URL. Defaults to DEFAULT_API_URL.

OperationalLimits | None

Per-tx and rate-limit caps.

Returns a PeaqosClient instance. __repr__ redacts the private key. Other RPC endpoints are available. See Public RPC endpoints.

Show Example

from peaq_os_sdk import PeaqosClient

client = PeaqosClient(
    rpc_url="https://peaq.api.onfinality.io/public",
    private_key="0xabc...def",
    identity_registry="0x...",
    identity_staking="0x...",
    event_registry="0x...",
    machine_nft="0x...",
    did_registry="0x...",
    batch_precompile="0x...",
)

print(client.address)

Errors: ValidationError: missing/invalid constructor args.

`from_env`

@classmethod
def from_env(cls) -> "PeaqosClient": ...

Show Example

from dotenv import load_dotenv
from peaq_os_sdk import PeaqosClient

load_dotenv() # load envs from .env file

client = PeaqosClient.from_env()

Errors: ValidationError: any required env var missing or empty.

`from_wallet`

@classmethod
def from_wallet(
    cls,
    name_or_id: str,
    passphrase: str | None = None,
    ows_signing: bool = True,
    vault_path: str | None = None,
    **config_kwargs: Any,
) -> "PeaqosClient": ...

Builds a client whose signing identity is an OWS vault wallet. Mirrors the JS SDK’s PeaqosClient.fromWallet. The remaining keyword arguments (rpc_url, contract addresses, etc.) match the regular PeaqosClient constructor.

Param	Default	Meaning
`name_or_id`	required	Wallet name or UUID in the OWS vault.
`passphrase`	`OWS_PASSPHRASE` env	Vault passphrase. Falls back to the env var when `None`; raises `PeaqosError` when both are missing.
`ows_signing`	`True`	When `True`, transactions sign through OWS (key never held by the SDK process). When `False`, the key is exported and decrypted at construction so the client is identical to one built with `private_key=...`.
`vault_path`	`~/.ows/`	Optional custom vault directory.

In OWS-native mode the SDK never holds the private key — OWS decrypts it inside the Rust FFI for each sign_hash call and wipes it immediately. The passphrase is verified eagerly: a wrong or missing one raises PeaqosError at construction. Each transaction’s chainId (from the tx dict) drives both the CAIP-2 OWS arg and the EIP-155 v value, so the same client transparently signs both peaq (eip155:3338) and Base (eip155:8453) bridge transactions.

Show Example

from peaq_os_sdk import PeaqosClient

# OWS-native (default): private key never decrypted in this process.
client = PeaqosClient.from_wallet(
    "my-machine",
    passphrase="s3cret",
    rpc_url="https://peaq-rpc.example.com",
    identity_registry="0x...",
    identity_staking="0x...",
    event_registry="0x...",
    machine_nft="0x...",
    did_registry="0x0000000000000000000000000000000000000800",
    batch_precompile="0x0000000000000000000000000000000000000805",
)

# Eager-decrypt: behaves identically to PeaqosClient(private_key=...).
eager = PeaqosClient.from_wallet(
    "my-machine",
    passphrase="s3cret",
    ows_signing=False,
    rpc_url="https://peaq-rpc.example.com",
    identity_registry="0x...",
    # ...
)

Errors: PeaqosError when the wallet is missing, the passphrase is wrong (eager-mode) or missing, or OWS itself rejects the request. See OWS signing error codes for the canonical 5 codes mapped from OWSAccount.sign_transaction.

Wallets (OWS)

Wallet lifecycle helpers (create_wallet, import_wallet, import_wallet_mnemonic, list_wallets, get_wallet, export_wallet, delete_wallet) back the Open Wallet Standard integration: mnemonic-backed encrypted vault, multi-chain accounts (peaq, Base, Ethereum, Solana, Bitcoin, etc.). Available under peaq_os_sdk.wallet and as @staticmethods on PeaqosClient. Install with the optional [ows] extra (pip install 'peaq-os-sdk[ows]'). The raw-key constructor and from_env flow keep working unchanged. Full reference on the Wallets page. Wallet returns are typed as WalletInfo (frozen dataclass) with an accounts: list[AccountInfo] field. AccountInfo carries account_id (CAIP-10), address, chain_id (CAIP-2), network (human-readable name like "peaq", "base", "ethereum", "solana"), and derivation_path. The SDK synthesizes accounts for every supported EVM chain from the first EVM source returned by OWS: peaq, ethereum, base, polygon, arbitrum, and optimism all surface in accounts even when OWS only returns one EVM key. Both classes are re-exported from the package root.

`generate_keypair`

@staticmethod
def generate_keypair() -> tuple[Address, str]: ...

Returns tuple (address, private_key). Both are 0x-prefixed hex strings; Address is a NewType over str (checksummed).

Show Example

address, private_key = PeaqosClient.generate_keypair()
print(address)

OWS wallet lifecycle

OWS wallet helpers are available as static PeaqosClient methods. They derive multi-chain accounts, keep wallet material in an encrypted OWS vault, and return public WalletInfo metadata.

@staticmethod
def create_wallet(name: str, passphrase: str | None = None, words: int = 12, vault_path: str | None = None) -> WalletInfo: ...

@staticmethod
def import_wallet(name: str, private_key: str, passphrase: str | None = None, chain: ImportChain = IMPORT_CHAIN_EVM, vault_path: str | None = None) -> WalletInfo: ...

@staticmethod
def import_wallet_mnemonic(name: str, mnemonic: str, passphrase: str | None = None, index: int = 0, vault_path: str | None = None) -> WalletInfo: ...

@staticmethod
def list_wallets(vault_path: str | None = None) -> list[WalletInfo]: ...

@staticmethod
def get_wallet(name_or_id: str, vault_path: str | None = None) -> WalletInfo: ...

@staticmethod
def export_wallet(name_or_id: str, passphrase: str | None = None, vault_path: str | None = None) -> str: ...

@staticmethod
def delete_wallet(name_or_id: str, vault_path: str | None = None) -> None: ...

Install peaq-os-sdk[ows] before using these helpers. create_wallet, import_wallet, import_wallet_mnemonic, and export_wallet require a passphrase argument or OWS_PASSPHRASE. vault_path can point at a custom OWS vault directory.

dataclass

Frozen dataclass with id, name, created_at, key_type, peaq_address, and accounts. Each account has account_id, address, chain_id, network, and derivation_path. Supported EVM accounts include peaq, Ethereum, Base, Polygon, Arbitrum, and Optimism; missing EVM chain entries are synthesized from the first available EVM address.

Show Example

from peaq_os_sdk import PeaqosClient, IMPORT_CHAIN_EVM
from peaq_os_sdk.wallet.utils import extract_peaq_address

wallet = PeaqosClient.create_wallet("robot-001")
print(wallet.peaq_address)

imported = PeaqosClient.import_wallet(
    "legacy-machine",
    "0xabc...def",
    chain=IMPORT_CHAIN_EVM,
)

all_wallets = PeaqosClient.list_wallets()
same = PeaqosClient.get_wallet(imported.id)
print(len(all_wallets), extract_peaq_address(same.accounts))

export_wallet returns mnemonic or private-key material. Keep it in local administrative tooling; do not expose it through robot control channels.

Accessors (Python-only)

Accessor	Type	Description
`address`	`str`	Checksummed owner address
`w3` / `web3`	`Web3`	Web3 instance
`session`	`requests.Session`	HTTP session for API calls
`account`	`LocalAccount` \| `OWSAccount`	Bound signing account; `OWSAccount` when constructed via `from_wallet(ows_signing=True)`
`identity_registry`	`Contract`	Bound IdentityRegistry contract
`identity_staking`	`Contract`	Bound IdentityStaking contract
`event_registry`	`Contract`	Bound EventRegistry contract
`machine_nft`	`Contract`	Bound MachineNFT contract
`did_precompile`	`Contract`	Bound peaq DID precompile
`batch_precompile`	`Contract`	Bound peaq Batch precompile

Registration

`register_machine`

def register_machine(self) -> int: ...

Returns the newly allocated machine_id. The SDK decodes it from the Registered event in the transaction receipt. 1 PEAQ is sent as msg.value: the payable register() function auto-bonds the machine.

Show Example

machine_id = client.register_machine()
print("registered as", machine_id)

Errors: RpcError: chain revert (AlreadyRegistered), insufficient balance for gas + 1 PEAQ bond, or any other on-chain failure. See errors.

`register_for`

def register_for(self, machine_address: str) -> int: ...

str

required

Machine EOA to register. The caller acts as proxy operator.

Returns the newly allocated machine_id for the proxied machine. 1 PEAQ is sent as msg.value.

Show Example

machine_address, machine_key = PeaqosClient.generate_keypair()
machine_id = client.register_for(machine_address)
print(f"registered machine {machine_address} with machine ID {machine_id}")

Errors: ValidationError on invalid machine_address. RpcError on chain revert (AlreadyRegistered, InvalidMachineAddress).

Gas Station

`setup_faucet_2fa`

def setup_faucet_2fa(
    self,
    owner_address: str,
    faucet_base_url: str,
    qr_format: Literal["svg", "png"] = "svg",
) -> FaucetSetupResponse: ...

str

required

Owner to enroll.

str

required

Gas Station base URL.

'svg' | 'png'

QR format. Defaults to "svg".

TypedDict

Keys: owner_address: str, otpauth_uri: str (OTP auth URI for authenticator apps), qr_image_url: str (expires after ~2 minutes).

Show Example

setup = client.setup_faucet_2fa(
    owner_address=client.address,
    faucet_base_url="https://depinstation.peaq.network",
)
print(setup["otpauth_uri"])

Errors: ValidationError if owner_address or faucet_base_url is empty. ApiError for INVALID_OWNER_ADDRESS, QR_GENERATION_FAILED, NETWORK_ERROR, or an unexpected response envelope. See errors.

`confirm_faucet_2fa`

def confirm_faucet_2fa(
    self,
    owner_address: str,
    faucet_base_url: str,
    two_factor_code: str,
) -> None: ...

Show Example

client.confirm_faucet_2fa(
    owner_address=client.address,
    faucet_base_url="https://depinstation.peaq.network",
    two_factor_code="123456",
)

Errors: ValidationError, ApiError (INVALID_2FA, 2FA_NOT_CONFIGURED, 2FA_LOCKED).

`fund_from_gas_station`

def fund_from_gas_station(
    self,
    owner_address: str,
    target_wallet_address: str,
    chain_id: str,
    two_factor_code: str,
    faucet_base_url: str,
    request_id: str | None = None,
) -> FaucetFundResponse: ...

str

required

2FA-enrolled owner address.

str

required

Machine EOA to fund.

str

required

Chain identifier configured on the faucet, e.g. "peaq".

str

required

Current TOTP.

str

required

Gas Station base URL.

str | None

UUID idempotency key. Auto-generated if omitted.

union

Discriminated union on status. Either a FundedResponse ({status: "success", tx_hash: str, funded_amount: str}, funded_amount is decimal wei) or a SkippedResponse ({status: "skipped", current_balance: str, min_gas_balance: str}). request_id is a request-side idempotency key passed by the caller; the response does not echo it back.

Cross-SDK behavior: The Python SDK does not include request_id in the response. The JS SDK echoes requestId back in both success and skipped responses. Do not write code that reads requestId / request_id from the response and expects it to work in both SDKs.

Show Example

result = client.fund_from_gas_station(
    owner_address=client.address,
    target_wallet_address=machine_address,
    chain_id="peaq",
    two_factor_code="123456",
    faucet_base_url="https://depinstation.peaq.network",
)

if result["status"] == "success":
    print("tx:", result["tx_hash"], "amount:", result["funded_amount"])
else:
    print("skipped; balance:", result["current_balance"])

Errors: ValidationError, ApiError (any of the 20 faucet codes). See errors.

NFT & DID

Machine NFT minting, token-ID lookup, and the two canonical DID attribute writers. The DID writes are submitted as a single atomic batchAll transaction via the peaq Batch precompile.

`mint_nft`

Mints a Machine NFT on the MachineNFT contract for a registered, bonded machine.

def mint_nft(self, machine_id: int, recipient: str) -> str: ...

int

required

Registered machine ID. Must be a positive integer.

str

required

0x-prefixed 20-byte hex address that will receive the NFT.

str

Transaction hash as a 0x-prefixed hex string.

Show Example

tx_hash = client.mint_nft(machine_id=42, recipient=client.address)

Errors: ValidationError on non-positive machine_id or invalid recipient. RpcError on chain revert (MachineNotBonded, AlreadyMinted, NotMachineOwner) or transaction failure.

`token_id_of`

Reads the NFT token ID assigned to a registered machine via a view call.

def token_id_of(self, machine_id: int) -> int: ...

int

required

Registered machine ID. Must be a positive integer.

int

The NFT token ID, as a positive integer.

Show Example

token_id = client.token_id_of(machine_id=42)

Errors: ValidationError if machine_id is not positive. RpcError when the machine has no NFT minted (contract reverts).

Cross-SDK behavior: The Python SDK raises RpcError when no NFT has been minted for the machine (contract reverts). The JS SDK returns 0 instead. In polyglot codebases, catch RpcError in Python and check for 0 in JS — do not assume the same pattern works in both.

`write_machine_did_attributes`

Atomically writes the six canonical Machine DID attributes to the caller’s DID via a single batchAll transaction.

def write_machine_did_attributes(
    self,
    machine_id: int,
    nft_token_id: int,
    operator_did: str,
    documentation_url: str,
    data_api: str,
    data_visibility: str,
) -> str: ...

int

required

Registered machine ID.

int

required

NFT token ID assigned to the machine.

str

required

Operator DID reference. May be an empty string. ASCII, ≤ 2560 bytes.

str

required

Non-empty ASCII URL, ≤ 2560 bytes.

str

required

Non-empty ASCII URL for the machine’s data API, ≤ 2560 bytes.

'public' | 'private' | 'onchain'

required

Visibility setting.

Show Example

tx_hash = client.write_machine_did_attributes(
    machine_id=42,
    nft_token_id=1,
    operator_did="",
    documentation_url="https://docs.example.com",
    data_api="https://api.example.com",
    data_visibility="public",
)

`write_proxy_did_attributes`

Atomically writes the two canonical Proxy DID attributes (machineId, machines) to the caller’s DID.

def write_proxy_did_attributes(
    self,
    proxy_machine_id: int,
    machine_ids: list[int],
) -> str: ...

int

required

The proxy operator’s registered machine ID.

list[int]

required

Non-empty list of positive machine IDs. The JSON-encoded array must be ≤ 2560 bytes.

Show Example

tx_hash = client.write_proxy_did_attributes(
    proxy_machine_id=10,
    machine_ids=[42, 43, 44],
)

`read_attribute`

Reads a single DID attribute directly from the peaq DID precompile. Most consumers should prefer the /machine/{did} API, which composes the full attribute set; this helper is the on-chain escape hatch. Not exported from the package root: import from peaq_os_sdk.did.did_precompile.

from peaq_os_sdk.did.did_precompile import read_attribute

def read_attribute(
    client: PeaqosClient,
    did_address: str,
    name: str,
) -> DIDAttributeResult: ...

str

required

The machine address whose DID is being read.

str

required

Attribute key, e.g. "machineId", "data_visibility", "machines".

TypedDict

Keys: name: str, value: str, validity: int (seconds; 0 = no expiry), created: int (block timestamp). Raises RpcError if the attribute does not exist on the precompile.

Smart accounts

ERC-4337 smart accounts deployed via the MachineAccountFactory. Requires the client to be constructed with machine_account_factory (or MACHINE_ACCOUNT_FACTORY_ADDRESS when using from_env).

`deploy_smart_account`

Deploys a smart account via MachineAccountFactory.createAccount and returns the deployed address.

def deploy_smart_account(
    self,
    owner: str,
    machine: str,
    salt: int,
) -> Address: ...

str

required

EOA that will own the smart account.

str

required

Machine EOA the account is scoped to.

int

required

Non-negative CREATE2 salt.

Show Example

address = client.deploy_smart_account(
    owner=client.address,
    machine=machine_address,
    salt=0,
)

Errors: ValidationError on invalid param or client constructed without machine_account_factory. RpcError on revert or missing AccountCreated event.

`get_smart_account_address`

Read-only equivalent: computes the CREATE2 address without deploying. Identical result to deploy_smart_account for the same inputs.

def get_smart_account_address(
    self,
    owner: str,
    machine: str,
    salt: int,
) -> Address: ...

Same parameters as deploy_smart_account. No transaction, no gas.

Show Example

predicted = client.get_smart_account_address(
    owner=client.address,
    machine=machine_address,
    salt=0,
)

Bridge

Supported routes: peaq ↔ Base. Additional peaqOS chains are added as peer contracts deploy. See Machine NFT cross-chain portability.

LayerZero v2 Machine NFT bridging between peaq and Base. Requires machine_nft_adapter (or MACHINE_NFT_ADAPTER_ADDRESS) when sending from peaq. The SDK’s source / destination string union expands as peer contracts deploy on new chains.

`bridge_nft`

Bridges a Machine NFT from source to destination. On the peaq→Base path the SDK runs an ERC-721 approval pre-flight: it reads MachineNFT.getApproved(token_id) and submits a one-shot approve(adapter, token_id) if the token isn’t already cleared for the adapter. The Base→peaq path uses burn-and-unlock and needs no approval. Callers don’t handle approvals themselves.

def bridge_nft(
    self,
    token_id: int,
    source: str,
    destination: str,
    recipient: str,
    *,
    base_rpc_url: str | None = None,
    base_nft_address: str | None = None,
    options: bytes = b"",
) -> Hex32: ...

int

required

Positive NFT id to bridge.

'peaq' | 'base'

required

Origin chain.

'peaq' | 'base'

required

Target chain. Must differ from source.

str

required

Destination-chain recipient address.

str | None

Base RPC URL. Required only when source == "base".

str | None

MachineNFTBase address on Base. Required only when source == "base".

bytes

Raw LayerZero v2 extraOptions bytes.

Show Example

tx = client.bridge_nft(
    token_id=42,
    source="peaq",
    destination="base",
    recipient="0xabc...",
)

`wait_for_bridge_arrival`

Static method that polls the destination chain’s MachineNFT.ownerOf(token_id) every 10 seconds until a non-zero owner returns or the timeout elapses. Does not require a client instance.

@staticmethod
def wait_for_bridge_arrival(
    dst_rpc_url: str,
    dst_nft_address: str,
    token_id: int,
    timeout: int = 300,
) -> bool: ...

str

required

Destination-chain RPC endpoint.

str

required

MachineNFT contract address on the destination.

int

required

The NFT id expected to arrive.

int

Wait budget in seconds. Defaults to 300 (5 min).

Show Example

arrived = PeaqosClient.wait_for_bridge_arrival(
    dst_rpc_url="https://mainnet.base.org",
    dst_nft_address="0x...",
    token_id=42,
)

Events (Qualify)

`submit_event`

Submits a single event to EventRegistry. Returns (tx_hash, data_hash).

def submit_event(
    self,
    *,
    machine_id: int,
    event_type: int,
    value: int,
    timestamp: int,
    raw_data: bytes | None,
    trust_level: int,
    source_chain_id: int,
    source_tx_hash: Hex32 | None,
    metadata: bytes,
    currency: str | None = ...,  # omit for smart default: revenue → "USD", activity → ""
) -> tuple[str, bytes]: ...

Param shape mirrors validate_submit_event_params below. value is an ISO 4217 minor-unit integer: cents for USD/HKD, whole units for JPY/KRW/VND, thousandths for BHD. The MCR pipeline FX-normalizes to USD cents using the rate at timestamp. currency is a 3-10 char uppercase alphanumeric code on revenue events and "" on activity events; omitting the kwarg applies the smart default. batch_submit_events is strict: currency must be supplied per event.

tuple[str, bytes]

tx_hash is a 0x-prefixed hash string; data_hash is exactly 32 bytes.

Show Example

import time

tx_hash, data_hash = client.submit_event(
    machine_id=1024,
    event_type=EVENT_TYPE_REVENUE,
    value=12500,                # $125.00 in cents
    currency="USD",
    timestamp=int(time.time()),
    raw_data=b"\x01\x02\x03",
    trust_level=TRUST_ON_CHAIN_VERIFIABLE,
    source_chain_id=SUPPORTED_CHAINS["peaq"],
    source_tx_hash=None,
    metadata=b"",
)

Errors: ValidationError, ValueCapExceeded, RateLimitExceeded, RpcError.

`batch_submit_events`

Submits multiple events atomically through the peaq Batch precompile. All succeed or all revert.

def batch_submit_events(
    self,
    events: list[dict[str, object] | SubmitEventParams],
) -> list[str]: ...

list[dict | SubmitEventParams]

required

Non-empty list of event payloads. Items may be SubmitEventParams instances or dicts with matching keys.

list[str]

One transaction hash per input event. All hashes are identical (same batch tx).

Errors: ValidationError on empty list or invalid event. ValueCapExceeded / RateLimitExceeded when operational limits hit. RpcError on revert or transport failure.

`validate_submit_event_params`

def validate_submit_event_params(params: SubmitEventParams) -> None: ...

SubmitEventParams is a frozen dataclass (imported from peaq_os_sdk.types.events):

@dataclass(frozen=True, slots=True)
class SubmitEventParams:
    machine_id: int
    event_type: int           # 0 revenue, 1 activity
    value: int                # ISO 4217 subunit integer
    timestamp: int
    raw_data: bytes | None
    trust_level: int          # 0, 1, 2
    source_chain_id: int
    source_tx_hash: Hex32 | None
    metadata: bytes
    currency: str = ...       # revenue: ^[A-Z0-9]{3,10}$; activity: ""

The validator uses attribute access, so callers must construct a SubmitEventParams instance. Dict literals will raise AttributeError.

Show Example

import time

from peaq_os_sdk import (
    EVENT_TYPE_REVENUE,
    TRUST_ON_CHAIN_VERIFIABLE,
    SUPPORTED_CHAINS,
)
from peaq_os_sdk.types.events import SubmitEventParams
from peaq_os_sdk.validation import validate_submit_event_params

params = SubmitEventParams(
    machine_id=1024,
    event_type=EVENT_TYPE_REVENUE,
    value=1250,                  # $12.50 in cents
    currency="USD",
    timestamp=int(time.time()),
    raw_data=b"revenue: $12.50",
    trust_level=TRUST_ON_CHAIN_VERIFIABLE,
    source_chain_id=SUPPORTED_CHAINS["base"],
    source_tx_hash="0xabc...",
    metadata=b"",
)
validate_submit_event_params(params)

Errors: ValidationError: any field out of range.

`compute_data_hash`

def compute_data_hash(raw_data: bytes) -> bytes: ...

Returns the 32-byte keccak256 digest of raw_data. Pair with submit_event’s data_hash output (also bytes) to compare on-chain payloads.

Show Example

from peaq_os_sdk.utils import compute_data_hash
data_hash = compute_data_hash(b"revenue: $12.50")

`check_operational_limits`

def check_operational_limits(
    params: SubmitEventParams,
    limits: OperationalLimits,
    tracker: EventTracker | None,
) -> None: ...

SubmitEventParams

required

Event with machine_id and value.

OperationalLimits

required

Configured max_value_per_tx, rate_limit_max_events, rate_limit_window_seconds.

EventTracker | None

required

Current rate-tracking state for the machine, or None when tracking is disabled.

Show Example

from peaq_os_sdk.validation import check_operational_limits, EventTracker

tracker = EventTracker(machine_id=1, count=5, window_start=1700000000.0)
check_operational_limits(params, client.operational_limits, tracker)

Errors: ValueCapExceeded if value > max_value_per_tx. RateLimitExceeded when the event window is exceeded.

Queries

Read-only helpers backed by the off-chain MCR API server (client.api_url). Each function validates the DID, issues a single GET through client.session, and returns a shape-checked TypedDict. The default per-request timeout is 30 seconds, enforced inside peaq_os_sdk.query.http_client.get_json; the public query_* functions do not currently expose an override.

`query_mcr`

Fetches the Machine Credit Rating for a machine DID. See GET /mcr/{did}.

def query_mcr(client: PeaqosClient, did: str) -> MCRResponse: ...

str

required

Machine DID. Must start with did:peaq:0x.

TypedDict

Snake_case keys: did: str, machine_id: int, mcr_score: int (0–100), mcr: str ("AAA" | "AA" | "A" | "BBB" | "BB" | "B" | "NR" | "Provisioned"), bond_status: str ("bonded" | "unbonded"), negative_flag: bool, event_count: int, revenue_event_count: int, activity_event_count: int, revenue_trend: str ("up" | "stable" | "down" | "insufficient"), total_revenue: float (USD cents; divide by 100 for display), average_revenue_per_event: float (USD cents; divide by 100 for display), last_updated: int | None, mcr_degraded: bool (true when ≥1 scored event used a stale or unavailable FX source).

mcr_score is always an int. The SDK coerces a null score from the API to 0. Provisioned machines (bonded but not yet scored) surface as 0, matching the JavaScript SDK.

Show Example

from peaq_os_sdk.query import query_mcr

mcr = query_mcr(client, "did:peaq:0xMachineAddress")
print(mcr["mcr_score"], mcr["mcr"])

Errors: ValidationError if did does not start with did:peaq:0x. ApiError with code in NOT_FOUND (HTTP 404), SERVICE_UNAVAILABLE (503), SERVER_ERROR (other 5xx), HTTP_ERROR (other non-2xx), BAD_RESPONSE (malformed body), TIMEOUT, NETWORK_ERROR.

`query_machine`

Fetches the full machine profile (NFT Metadata JSON v1.0) and validates the response shape against a strict TypedDict. The SDK raises BAD_RESPONSE if the server payload does not match the schema. See GET /machine/{did}.

def query_machine(client: PeaqosClient, did: str) -> MachineProfileResponse: ...

str

required

Machine DID. Must start with did:peaq:0x.

TypedDict

Top-level keys: schema_version: str, name: str, peaqos: PeaqosData. PeaqosData always carries machine_id: int, did: str, operator: str | None, mcr: str, mcr_score: int, bond_status: str, negative_flag: bool, event_count: int, data_visibility: str, documentation_url: str | None. Visibility-dependent extras (data_api: str, event_data: list[EventEntry], partner_data: dict[str, Any], partner_data_error: str) are present only when the API includes them. Use "data_api" in profile["peaqos"] (etc.) to test for presence rather than truthiness.

Show Example

from peaq_os_sdk.query import query_machine

profile = query_machine(client, "did:peaq:0xMachineAddress")
print(profile["schema_version"], profile["name"])
print(profile["peaqos"]["mcr"], profile["peaqos"]["mcr_score"])

Errors: ValidationError on invalid DID. ApiError with the same codes as query_mcr; BAD_RESPONSE if any required field is missing, has the wrong type, or is out of range.

`query_operator_machines`

Fetches the fleet of machines managed by a proxy operator. Each entry is individually validated. See GET /operator/{did}/machines.

def query_operator_machines(
    client: PeaqosClient,
    did: str,
) -> OperatorMachinesResponse: ...

str

required

Operator DID. Must start with did:peaq:0x.

TypedDict

Keys: operator_did: str, machines: list[OperatorMachine], and pagination: Pagination. Each OperatorMachine has did: str, machine_id: int, mcr_score: int (0–100), mcr: str, and negative_flag: bool. Pagination carries offset: int, limit: int, and total: int.

Show Example

from peaq_os_sdk.query import query_operator_machines

fleet = query_operator_machines(client, "did:peaq:0xProxyAddress")
for m in fleet["machines"]:
    print(m["machine_id"], m["mcr_score"], m["mcr"])

Errors: ValidationError on invalid DID. ApiError with the same codes as query_mcr; BAD_RESPONSE if the body or any machines entry is malformed.

Error classes

from peaq_os_sdk import (
    PeaqosError,
    ValidationError,
    RpcError,
    ApiError,
    ValueCapExceeded,
    RateLimitExceeded,
)

See errors for the full hierarchy and the 20-code faucet table.

Constants

Machine status

from peaq_os_sdk import (
    MACHINE_STATUS_NONE,         # 0
    MACHINE_STATUS_PENDING,      # 1
    MACHINE_STATUS_VERIFIED,     # 2
    MACHINE_STATUS_REJECTED,     # 3
    MACHINE_STATUS_DEACTIVATED,  # 4
)

MCR ratings (Python-only)

from peaq_os_sdk import (
    MCR_RATING_AAA,
    MCR_RATING_AA,
    MCR_RATING_A,
    MCR_RATING_BBB,
    MCR_RATING_BB,
    MCR_RATING_B,
    MCR_RATING_NR,
)

# Bonded-but-unscored machines are returned by the MCR API with the literal
# rating string "Provisioned" and mcr_score coerced to 0 by the SDK.
# No SDK constant yet for the "Provisioned" rating string.

Event types, trust levels & chain IDs

Identical to the JS SDK: EVENT_TYPE_REVENUE, EVENT_TYPE_ACTIVITY, TRUST_SELF_REPORTED, TRUST_ON_CHAIN_VERIFIABLE, TRUST_HARDWARE_SIGNED, DID_ATTR_*, DID_MAX_NAME_BYTES, DID_MAX_VALUE_BYTES, SUPPORTED_CHAINS, LAYERZERO_EIDS, DEFAULT_API_URL. Same values, Python-idiomatic names.

Data visibility (Python-only)

from peaq_os_sdk import (
    DATA_VISIBILITY_PUBLIC,   # "public"
    DATA_VISIBILITY_PRIVATE,  # "private"
    DATA_VISIBILITY_ONCHAIN,  # "onchain"
)

Pass to data_visibility= on write_machine_did_attributes. Validated server-side; using one of these constants avoids typos.

Wallet import chains & passphrase env

from peaq_os_sdk import (
    IMPORT_CHAIN_EVM,       # "evm" (default)
    IMPORT_CHAIN_SOLANA,    # "solana"
    IMPORT_CHAIN_BITCOIN,   # "bitcoin"
    IMPORT_CHAIN_COSMOS,    # "cosmos"
    IMPORT_CHAIN_TRON,      # "tron"
    IMPORT_CHAIN_TON,       # "ton"
    IMPORT_CHAIN_SUI,       # "sui"
    IMPORT_CHAIN_XRPL,      # "xrpl"
    IMPORT_CHAIN_SPARK,     # "spark"
    IMPORT_CHAIN_FILECOIN,  # "filecoin"

    OWS_PASSPHRASE_ENV,     # "OWS_PASSPHRASE"
)

Used by import_wallet(..., chain=IMPORT_CHAIN_SOLANA). The JS SDK exports identical constants. See JS Constants.

​Differences from the JavaScript SDK

​Install

​Environment variables

​Client

​PeaqosClient

​from_env

​from_wallet

​Wallets (OWS)

​generate_keypair

​OWS wallet lifecycle

​Accessors (Python-only)

​Registration

​register_machine

​register_for

​Gas Station

​setup_faucet_2fa

​confirm_faucet_2fa

​fund_from_gas_station

​NFT & DID

​mint_nft

​token_id_of

​write_machine_did_attributes

​write_proxy_did_attributes

​read_attribute

​Smart accounts

​deploy_smart_account

​get_smart_account_address

​Bridge

​bridge_nft

​wait_for_bridge_arrival

​Events (Qualify)

​submit_event

​batch_submit_events

​validate_submit_event_params

​compute_data_hash

​check_operational_limits

​Queries

​query_mcr

​query_machine

​query_operator_machines

​Error classes

​Constants

Differences from the JavaScript SDK

Install

Environment variables

Client

`PeaqosClient`

`from_env`

`from_wallet`

Wallets (OWS)

`generate_keypair`

OWS wallet lifecycle

Accessors (Python-only)

Registration

`register_machine`

`register_for`

Gas Station

`setup_faucet_2fa`

`confirm_faucet_2fa`

`fund_from_gas_station`

NFT & DID

`mint_nft`

`token_id_of`

`write_machine_did_attributes`

`write_proxy_did_attributes`

`read_attribute`

Smart accounts

`deploy_smart_account`

`get_smart_account_address`

Bridge

`bridge_nft`

`wait_for_bridge_arrival`

Events (Qualify)

`submit_event`

`batch_submit_events`

`validate_submit_event_params`

`compute_data_hash`

`check_operational_limits`

Queries

`query_mcr`

`query_machine`

`query_operator_machines`

Error classes

Constants