> ## Documentation Index
> Fetch the complete documentation index at: https://docs.peaq.xyz/llms.txt
> Use this file to discover all available pages before exploring further.
# SDK: Python
> Python client for peaqOS. Class, static factories, methods, types, constants. Mirrors the JavaScript SDK in snake_case.
`peaq-os-sdk` on PyPI is the Python equivalent of [`@peaqos/peaq-os-sdk`](/peaqos/sdk-reference/sdk-js). 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
```bash theme={"theme":{"light":"github-light-default","dark":"github-dark"}}
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](/peaqos/sdk-reference/sdk-js#environment-variables) and [peaq mainnet contracts](/peaqos/sdk-reference/sdk-js#peaq-mainnet-contracts). For agung testnet addresses see [Install → Agung testnet contracts](/peaqos/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`
```python theme={"theme":{"light":"github-light-default","dark":"github-dark"}}
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 endpoint.
`0x` + 64 hex.
Identity Registry contract address.
Identity Staking contract address.
Event Registry contract address.
Machine NFT contract address (ONFT).
DID Registry precompile address.
Batch precompile address.
`MachineAccountFactory` contract address. Required only for `deploy_smart_account` and `get_smart_account_address`.
`MachineNFTAdapter` (LayerZero ONFT adapter) contract address on peaq. Required only for `bridge_nft` when `source="peaq"`.
MCR API base URL. Defaults to `DEFAULT_API_URL`.
Per-tx and rate-limit caps.
Returns a `PeaqosClient` instance. `__repr__` redacts the private key.
Other RPC endpoints are available. See [Public RPC endpoints](/peaqos/install#public-rpc-endpoints).
```python theme={"theme":{"light":"github-light-default","dark":"github-dark"}}
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`
```python theme={"theme":{"light":"github-light-default","dark":"github-dark"}}
@classmethod
def from_env(cls) -> "PeaqosClient": ...
```
```python theme={"theme":{"light":"github-light-default","dark":"github-dark"}}
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`
```python theme={"theme":{"light":"github-light-default","dark":"github-dark"}}
@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.
```python theme={"theme":{"light":"github-light-default","dark":"github-dark"}}
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](/peaqos/sdk-reference/errors#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](/peaqos/wallets) integration: mnemonic-backed encrypted vault, multi-chain accounts (peaq, Base, Ethereum, Solana, Bitcoin, etc.). Available under `peaq_os_sdk.wallet` and as `@staticmethod`s 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](/peaqos/wallets#sdk-methods).
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`
```python theme={"theme":{"light":"github-light-default","dark":"github-dark"}}
@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).
```python theme={"theme":{"light":"github-light-default","dark":"github-dark"}}
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.
```python theme={"theme":{"light":"github-light-default","dark":"github-dark"}}
@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.
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.
```python theme={"theme":{"light":"github-light-default","dark":"github-dark"}}
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`
```python theme={"theme":{"light":"github-light-default","dark":"github-dark"}}
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.
```python theme={"theme":{"light":"github-light-default","dark":"github-dark"}}
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](/peaqos/sdk-reference/errors).
### `register_for`
```python theme={"theme":{"light":"github-light-default","dark":"github-dark"}}
def register_for(self, machine_address: str) -> int: ...
```
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`.
```python theme={"theme":{"light":"github-light-default","dark":"github-dark"}}
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`
```python theme={"theme":{"light":"github-light-default","dark":"github-dark"}}
def setup_faucet_2fa(
self,
owner_address: str,
faucet_base_url: str,
qr_format: Literal["svg", "png"] = "svg",
) -> FaucetSetupResponse: ...
```
Owner to enroll.
Gas Station base URL.
QR format. Defaults to `"svg"`.
Keys: `owner_address: str`, `otpauth_uri: str` (OTP auth URI for authenticator apps), `qr_image_url: str` (expires after \~2 minutes).
```python theme={"theme":{"light":"github-light-default","dark":"github-dark"}}
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](/peaqos/sdk-reference/errors).
### `confirm_faucet_2fa`
```python theme={"theme":{"light":"github-light-default","dark":"github-dark"}}
def confirm_faucet_2fa(
self,
owner_address: str,
faucet_base_url: str,
two_factor_code: str,
) -> None: ...
```
```python theme={"theme":{"light":"github-light-default","dark":"github-dark"}}
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`
```python theme={"theme":{"light":"github-light-default","dark":"github-dark"}}
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: ...
```
2FA-enrolled owner address.
Machine EOA to fund.
Chain identifier configured on the faucet, e.g. `"peaq"`.
Current TOTP.
Gas Station base URL.
UUID idempotency key. Auto-generated if omitted.
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.
```python theme={"theme":{"light":"github-light-default","dark":"github-dark"}}
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](/peaqos/sdk-reference/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.
```python theme={"theme":{"light":"github-light-default","dark":"github-dark"}}
def mint_nft(self, machine_id: int, recipient: str) -> str: ...
```
Registered machine ID. Must be a positive integer.
`0x`-prefixed 20-byte hex address that will receive the NFT.
Transaction hash as a `0x`-prefixed hex string.
```python theme={"theme":{"light":"github-light-default","dark":"github-dark"}}
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.
```python theme={"theme":{"light":"github-light-default","dark":"github-dark"}}
def token_id_of(self, machine_id: int) -> int: ...
```
Registered machine ID. Must be a positive integer.
The NFT token ID, as a positive integer.
```python theme={"theme":{"light":"github-light-default","dark":"github-dark"}}
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.
```python theme={"theme":{"light":"github-light-default","dark":"github-dark"}}
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: ...
```
Registered machine ID.
NFT token ID assigned to the machine.
Operator DID reference. May be an empty string. ASCII, ≤ 2560 bytes.
Non-empty ASCII URL, ≤ 2560 bytes.
Non-empty ASCII URL for the machine's data API, ≤ 2560 bytes.
Visibility setting.
```python theme={"theme":{"light":"github-light-default","dark":"github-dark"}}
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.
```python theme={"theme":{"light":"github-light-default","dark":"github-dark"}}
def write_proxy_did_attributes(
self,
proxy_machine_id: int,
machine_ids: list[int],
) -> str: ...
```
The proxy operator's registered machine ID.
Non-empty list of positive machine IDs. The JSON-encoded array must be ≤ 2560 bytes.
```python theme={"theme":{"light":"github-light-default","dark":"github-dark"}}
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](/peaqos/api-reference/get-machine), 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`.
```python theme={"theme":{"light":"github-light-default","dark":"github-dark"}}
from peaq_os_sdk.did.did_precompile import read_attribute
def read_attribute(
client: PeaqosClient,
did_address: str,
name: str,
) -> DIDAttributeResult: ...
```
The machine address whose DID is being read.
Attribute key, e.g. `"machineId"`, `"data_visibility"`, `"machines"`.
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.
```python theme={"theme":{"light":"github-light-default","dark":"github-dark"}}
def deploy_smart_account(
self,
owner: str,
machine: str,
salt: int,
) -> Address: ...
```
EOA that will own the smart account.
Machine EOA the account is scoped to.
Non-negative CREATE2 salt.
```python theme={"theme":{"light":"github-light-default","dark":"github-dark"}}
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.
```python theme={"theme":{"light":"github-light-default","dark":"github-dark"}}
def get_smart_account_address(
self,
owner: str,
machine: str,
salt: int,
) -> Address: ...
```
Same parameters as `deploy_smart_account`. No transaction, no gas.
```python theme={"theme":{"light":"github-light-default","dark":"github-dark"}}
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](/peaqos/concepts/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.
```python theme={"theme":{"light":"github-light-default","dark":"github-dark"}}
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: ...
```
Positive NFT id to bridge.
Origin chain.
Target chain. Must differ from `source`.
Destination-chain recipient address.
Base RPC URL. Required only when `source == "base"`.
`MachineNFTBase` address on Base. Required only when `source == "base"`.
Raw LayerZero v2 `extraOptions` bytes.
```python theme={"theme":{"light":"github-light-default","dark":"github-dark"}}
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.
```python theme={"theme":{"light":"github-light-default","dark":"github-dark"}}
@staticmethod
def wait_for_bridge_arrival(
dst_rpc_url: str,
dst_nft_address: str,
token_id: int,
timeout: int = 300,
) -> bool: ...
```
Destination-chain RPC endpoint.
`MachineNFT` contract address on the destination.
The NFT id expected to arrive.
Wait budget in seconds. Defaults to 300 (5 min).
```python theme={"theme":{"light":"github-light-default","dark":"github-dark"}}
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)`.
```python theme={"theme":{"light":"github-light-default","dark":"github-dark"}}
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`](#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` is a `0x`-prefixed hash string; `data_hash` is exactly 32 bytes.
```python theme={"theme":{"light":"github-light-default","dark":"github-dark"}}
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.
```python theme={"theme":{"light":"github-light-default","dark":"github-dark"}}
def batch_submit_events(
self,
events: list[dict[str, object] | SubmitEventParams],
) -> list[str]: ...
```
Non-empty list of event payloads. Items may be `SubmitEventParams` instances or dicts with matching keys.
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`
```python theme={"theme":{"light":"github-light-default","dark":"github-dark"}}
def validate_submit_event_params(params: SubmitEventParams) -> None: ...
```
`SubmitEventParams` is a frozen dataclass (imported from `peaq_os_sdk.types.events`):
```python theme={"theme":{"light":"github-light-default","dark":"github-dark"}}
@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`.
```python theme={"theme":{"light":"github-light-default","dark":"github-dark"}}
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`
```python theme={"theme":{"light":"github-light-default","dark":"github-dark"}}
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.
```python theme={"theme":{"light":"github-light-default","dark":"github-dark"}}
from peaq_os_sdk.utils import compute_data_hash
data_hash = compute_data_hash(b"revenue: $12.50")
```
### `check_operational_limits`
```python theme={"theme":{"light":"github-light-default","dark":"github-dark"}}
def check_operational_limits(
params: SubmitEventParams,
limits: OperationalLimits,
tracker: EventTracker | None,
) -> None: ...
```
Event with `machine_id` and `value`.
Configured `max_value_per_tx`, `rate_limit_max_events`, `rate_limit_window_seconds`.
Current rate-tracking state for the machine, or `None` when tracking is disabled.
```python theme={"theme":{"light":"github-light-default","dark":"github-dark"}}
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}`](/peaqos/api-reference/get-mcr).
```python theme={"theme":{"light":"github-light-default","dark":"github-dark"}}
def query_mcr(client: PeaqosClient, did: str) -> MCRResponse: ...
```
Machine DID. Must start with `did:peaq:0x`.
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.
```python theme={"theme":{"light":"github-light-default","dark":"github-dark"}}
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}`](/peaqos/api-reference/get-machine).
```python theme={"theme":{"light":"github-light-default","dark":"github-dark"}}
def query_machine(client: PeaqosClient, did: str) -> MachineProfileResponse: ...
```
Machine DID. Must start with `did:peaq:0x`.
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.
```python theme={"theme":{"light":"github-light-default","dark":"github-dark"}}
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`](/peaqos/api-reference/get-operator-machines).
```python theme={"theme":{"light":"github-light-default","dark":"github-dark"}}
def query_operator_machines(
client: PeaqosClient,
did: str,
) -> OperatorMachinesResponse: ...
```
Operator DID. Must start with `did:peaq:0x`.
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`.
```python theme={"theme":{"light":"github-light-default","dark":"github-dark"}}
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
```python theme={"theme":{"light":"github-light-default","dark":"github-dark"}}
from peaq_os_sdk import (
PeaqosError,
ValidationError,
RpcError,
ApiError,
ValueCapExceeded,
RateLimitExceeded,
)
```
See [errors](/peaqos/sdk-reference/errors) for the full hierarchy and the 20-code faucet table.
***
## Constants
```python theme={"theme":{"light":"github-light-default","dark":"github-dark"}}
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
)
```
```python theme={"theme":{"light":"github-light-default","dark":"github-dark"}}
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.
```
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.
```python theme={"theme":{"light":"github-light-default","dark":"github-dark"}}
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.
```python theme={"theme":{"light":"github-light-default","dark":"github-dark"}}
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](/peaqos/sdk-reference/sdk-js#constants).