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.0.2" python-dotenv
# Optional, only for OWS wallet lifecycle helpers:
pip install "peaq-os-sdk[ows]>=0.0.2"

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.

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: ...

rpc_url

str

required

RPC endpoint.

private_key

str

required

0x + 64 hex.

identity_registry

str

required

Identity Registry contract address.

identity_staking

str

required

Identity Staking contract address.

event_registry

str

required

Event Registry contract address.

machine_nft

str

required

Machine NFT contract address (ONFT).

did_registry

str

required

DID Registry precompile address.

batch_precompile

str

required

Batch precompile address.

machine_account_factory

str | None

MachineAccountFactory contract address. Required only for deploy_smart_account and get_smart_account_address.

machine_nft_adapter

str | None

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

api_url

str

MCR API base URL. Defaults to DEFAULT_API_URL.

operational_limits

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.

WalletInfo

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: ...

machine_address

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: ...

owner_address

str

required

Owner to enroll.

faucet_base_url

str

required

Gas Station base URL.

qr_format

'svg' | 'png'

QR format. Defaults to "svg".

FaucetSetupResponse

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: ...

owner_address

str

required

2FA-enrolled owner address.

target_wallet_address

str

required

Machine EOA to fund.

chain_id

str

required

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

two_factor_code

str

required

Current TOTP.

faucet_base_url

str

required

Gas Station base URL.

request_id

str | None

UUID idempotency key. Auto-generated if omitted.

FaucetFundResponse

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.

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: ...

machine_id

int

required

Registered machine ID. Must be a positive integer.

recipient

str

required

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

tx_hash

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: ...

machine_id

int

required

Registered machine ID. Must be a positive integer.

token_id

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).

`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: ...

machine_id

int

required

Registered machine ID.

nft_token_id

int

required

NFT token ID assigned to the machine.

operator_did

str

required

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

documentation_url

str

required

Non-empty ASCII URL, ≤ 2560 bytes.

data_api

str

required

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

data_visibility

'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: ...

proxy_machine_id

int

required

The proxy operator’s registered machine ID.

machine_ids

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: ...

did_address

str

required

The machine address whose DID is being read.

name

str

required

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

DIDAttributeResult

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: ...

owner

str

required

EOA that will own the smart account.

machine

str

required

Machine EOA the account is scoped to.

salt

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: ...

token_id

int

required

Positive NFT id to bridge.

source

'peaq' | 'base'

required

Origin chain.

destination

'peaq' | 'base'

required

Target chain. Must differ from source.

recipient

str

required

Destination-chain recipient address.

base_rpc_url

str | None

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

base_nft_address

str | None

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

options

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: ...

dst_rpc_url

str

required

Destination-chain RPC endpoint.

dst_nft_address

str

required

MachineNFT contract address on the destination.

token_id

int

required

The NFT id expected to arrive.

timeout

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.

(tx_hash, data_hash)

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]: ...

events

list[dict | SubmitEventParams]

required

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

tx_hashes

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: ...

params

SubmitEventParams

required

Event with machine_id and value.

limits

OperationalLimits

required

Configured max_value_per_tx, rate_limit_max_events, rate_limit_window_seconds.

tracker

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: ...

did

str

required

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

MCRResponse

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), average_revenue_per_event: float (USD), 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: ...

did

str

required

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

MachineProfileResponse

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: ...

did

str

required

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

OperatorMachinesResponse

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.

SDK Reference

Documentation Index

​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