Orchestration (Python)

The Python exposes the Machine Markets API on the existing PeaqosClient via client.orchestration. Flat methods, sync, snake_case, frozen-slotted dataclasses.

client.orchestration works only on a client constructed without tokenomics20. On a Tokenomics 2.0 client (tokenomics20 passed to the constructor, or TOKENOMICS_DEPLOYMENT_ID set in the environment that from_env() reads, which peaqos init writes), list_machines, get_machine, create_machine, update_machine, create_machine_identity_challenge, create_agent_pairing_challenge, and search_market raise TokenomicsIntegrationUnavailableError before any request is sent, so pairing and ordering cannot start. Pairing also needs an active, bonded Tokenomics 1.0 machine; see Scale.

Setup

from dotenv import load_dotenv
from peaq_os_sdk import PeaqosClient

load_dotenv()

client = PeaqosClient.from_env()
machines = client.orchestration.list_machines()
for m in machines.items:
    print(m.id, m.display_name)

Functional form, handy for testing:

from peaq_os_sdk.orchestration import list_machines

resp = list_machines(client, limit=20)

Accessing client.orchestration without orchestration_url raises OrchestrationConfigError.

Configuration

Constructor kwargs

PeaqosClient(
    *,
    rpc_url: str,
    private_key: str,
    # ... on-chain contract addresses ...
    api_url: str = "http://127.0.0.1:8000",  # MCR API
    orchestration_url: str | None = None,
    api_key: str | None = None,              # platform x-api-key
    verbose: bool = False,
)

Env vars (via `PeaqosClient.from_env()`)

Var	Required	Purpose
`PEAQOS_RPC_URL`	yes	peaq RPC for on-chain
`PEAQOS_PRIVATE_KEY`	yes	signer for on-chain
`IDENTITY_REGISTRY_ADDRESS`, `IDENTITY_STAKING_ADDRESS`, `EVENT_REGISTRY_ADDRESS`, `MACHINE_NFT_ADDRESS`, `DID_REGISTRY_ADDRESS`, `BATCH_PRECOMPILE_ADDRESS`	yes	on-chain contract addresses
`PEAQOS_ORCHESTRATION_URL`	optional	enables `client.orchestration`
`PEAQOS_API_KEY`	optional	platform API key (`x-api-key` header)
`PEAQOS_MCR_API_URL`	optional	MCR base URL
`TOKENOMICS_DEPLOYMENT_ID`	optional	selects Tokenomics 2.0 mode; leave it unset for a client that uses `client.orchestration`

Transport constants:

DEFAULT_READ_TIMEOUT_MS  = 30_000   # GET
DEFAULT_WRITE_TIMEOUT_MS = 60_000   # everything else
API_BASE_PATH            = "/api/v1"
HEADER_API_KEY              = "x-api-key"
HEADER_AGENT_PAIRING_TOKEN  = "x-agent-pairing-token"

User-Agent: peaqos-sdk-py/<version>. Per-method timeout overrides are not exposed on public methods.

Envelopes

@dataclass(frozen=True, slots=True)
class ListResponse(Generic[T]):
    items: tuple[T, ...]
    next_cursor: str | None         # None when there is no next page

@dataclass(frozen=True, slots=True)
class ItemResponse(Generic[T]):
    item: T

LIST_QUERY_DEFAULT_LIMIT = 100
LIST_QUERY_MAX_LIMIT = 500

@dataclass(frozen=True, slots=True)
class ListQuery:
    limit:  int | None = None       # 1..500
    cursor: str | None = None       # omit on first page

204 No Content returns None. validate_list_limit rejects limit < 1 or limit > 500 with OrchestrationValidationError.

Machines

client.orchestration.list_machines(
    *, limit: int | None = None, cursor: str | None = None,
) -> ListResponse[Machine]

client.orchestration.create_machine(params: CreateMachineRequest) -> ItemResponse[Machine]
client.orchestration.get_machine(machine_id: str) -> ItemResponse[Machine]
client.orchestration.update_machine(machine_id: str, params: UpdateMachineRequest) -> ItemResponse[Machine]

Machine shape:

@dataclass(frozen=True, slots=True)
class Machine:
    id:              str
    display_name:    str
    status:          Literal["draft", "active", "degraded", "blocked", "archived"]
    owner_id:        str
    machine_type:    str
    runtime_profile: str
    capabilities:    tuple[str, ...]
    labels:          Mapping[str, str]
    policy_ids:      tuple[str, ...]
    skill_keys:      tuple[str, ...]
    created_at:      str
    updated_at:      str
    identity_ref:    str | None = None

Machine agents

client.orchestration.enroll_machine_agent(machine_id: str, params: EnrollMachineAgentRequest | None = None) -> ItemResponse[MachineAgent]
client.orchestration.machine_agent_heartbeat(params: MachineAgentHeartbeatRequest) -> MachineAgentHeartbeatResponse

enroll_machine_agent returns ItemResponse[MachineAgent]. The server sends the one-time token as agentToken, and the Python SDK maps item.provisioning_token from a provisioning_token key that the server does not send, so it comes back None. To capture the token, call POST /machines/:machineId/agents/enrollment directly (see the orchestration API) and read item.agentToken. machine_agent_heartbeat uses no auth header: agent_token rides in the JSON body.

Machine identity challenges

client.orchestration.create_machine_identity_challenge(
    params: CreateMachineIdentityChallengeRequest,
) -> ItemResponse[MachineIdentityChallengeResponse]

Requests a server-issued challenge tied to a did:peaq:0x... or peaqos:machine:<id> identity reference. The DID controller signs item.message with EIP-191 personal_sign and submits the resulting {challengeId, signature} as identity_proof when creating or updating the machine record.

Agent pairings

client.orchestration.list_agent_pairings(machine_id: str, *, limit: int | None = None, cursor: str | None = None) -> ListResponse[AgentPairing]

client.orchestration.create_agent_pairing_challenge(
    machine_id: str,
    params: CreateAgentPairingChallengeRequest,
) -> ItemResponse[AgentPairingChallengeResponse]

client.orchestration.create_agent_pairing(
    machine_id: str,
    params: CreateAgentPairingRequest,
) -> ItemResponse[AgentPairing]

client.orchestration.create_agent_pairing_session(
    machine_id: str,
    pairing_id: str,
    params: CreateAgentPairingSessionRequest,
) -> ItemResponse[AgentPairing]

client.orchestration.update_agent_pairing(
    machine_id: str,
    pairing_id: str,
    params: UpdateAgentPairingRequest,
) -> ItemResponse[AgentPairing]

client.orchestration.revoke_agent_pairing(machine_id: str, pairing_id: str) -> None

Pairing is challenge-based end to end:

create_agent_pairing_challenge returns a server-issued challenge keyed to agent_address, agent_provider, agent_role, and optional agent_did.
The item.message (EIP-191) with the key behind agent_address.
create_agent_pairing accepts the proof in params.agent_proof and returns the pairing with a signed HS256 session JWT in pairing_token. The token is returned once at create.
create_agent_pairing_session rotates the session token before expiry with a fresh proof. Required after any update_agent_pairing to the , since policy changes invalidate the current token’s delegation_policy_hash.

Both create methods set pairing_token exactly once each. AgentPairing.__repr__ redacts the token to keep it out of logs. CreateAgentPairingRequest requires agent_proof: AgentProofInput and accepts an optional agent_did. delegation_policy carries service-level allow/deny lists (allowed_service_ids, denied_service_ids) next to the skill-level ones.

Runtime endpoints

client.orchestration.list_runtime_endpoints(machine_id: str, *, limit: int | None = None, cursor: str | None = None) -> ListResponse[RuntimeEndpoint]
client.orchestration.get_runtime_endpoint(machine_id: str, provider_key: str) -> ItemResponse[RuntimeEndpoint]
client.orchestration.upsert_runtime_endpoint(machine_id: str, provider_key: str, params: UpsertRuntimeEndpointRequest) -> ItemResponse[RuntimeEndpoint]
client.orchestration.delete_runtime_endpoint(machine_id: str, provider_key: str) -> None

upsert_runtime_endpoint validates that endpoint_base_url parses as http/https with a non-empty host. auth_token is treated as a secret in error redaction.

Skills

class ListSkillsOptions:
    scope:     SkillScope     | None = None  # core-product | bootstrap-reference | ecosystem
    direction: SkillDirection | None = None  # machine-offers | machine-consumes
    source:    SkillSource    | None = None  # peaq | partner | clawhub | external-marketplace

client.orchestration.list_skills(options: ListSkillsOptions | None = None, *, limit: int | None = None, cursor: str | None = None) -> ListResponse[SkillSummary]
client.orchestration.get_skill(skill_key: str) -> ItemResponse[SkillSummary]
client.orchestration.get_skill_manifest(skill_key: str) -> ItemResponse[SkillManifest]

Market services

class ListMarketServicesOptions:
    machine_id:     str           | None = None
    service_type:   ServiceType   | None = None
    execution_mode: ExecutionMode | None = None
    provider_key:   str           | None = None

class GetMarketServiceOptions:
    machine_id: str | None = None

client.orchestration.list_market_services(options: ListMarketServicesOptions | None = None, *, limit: int | None = None, cursor: str | None = None) -> ListResponse[MarketService]
client.orchestration.get_market_service(service_id: str, options: GetMarketServiceOptions | None = None) -> ItemResponse[MarketService]

Market search

client.orchestration.search_market(params: MarketSearchRequest, pairing_token: str) -> ItemResponse[MarketSearch]
client.orchestration.get_market_search(search_id: str) -> ItemResponse[MarketSearch]

search_market uses x-agent-pairing-token auth: the token rides as the second argument and is sent per call. get_market_search uses platform auth.

Market orders

client.orchestration.list_market_orders(
    *, machine_id: str, limit: int | None = None, cursor: str | None = None,
) -> ListResponse[MarketOrder]

client.orchestration.create_market_order(
    params: CreateMarketOrderRequest,        # machine_id, agent_pairing_id, service_id required
    pairing_token: str,
) -> ItemResponse[MarketOrder]

client.orchestration.get_market_order(order_id: str) -> ItemResponse[MarketOrder]

client.orchestration.execute_market_order(
    order_id: str,
    params: ExecuteMarketOrderRequest,       # input?, payment? (x402)
    pairing_token: str,
) -> ExecuteMarketOrderResponse              # union narrowed by execution.status_code

client.orchestration.confirm_market_order(
    order_id: str,
    pairing_token: str,
) -> ConfirmMarketOrderResponse              # { order, payment: MarketPayment | None }

client.orchestration.dispute_market_order(
    order_id: str,
    params: DisputeMarketOrderRequest,       # reason required; evidence optional
    pairing_token: str,
) -> DisputeMarketOrderResponse              # { order, payment, dispute }; payment → "frozen"

list_market_orders and get_market_order use platform auth; the four mutating calls require the agent’s pairing_token. execute_market_order returns execution.run and execution.outcome for a native run (execution.status_code == 200). When the order resolves to an external handoff the server answers 202, the order moves to handoff, and the call raises OrchestrationApiError with code INVALID_RESPONSE_SHAPE. Catch it and read the handoff from the order: once client.orchestration.get_market_order(order_id).item.status == "handoff", item.handoff carries label, url and notes. For x402-rail services, sign_x402_payment signs the provider challenge from create_payment_intent locally, submit_x402_payment_proof records the signature, and the same signature goes on ExecuteMarketOrderRequest.payment (X402ExecutePayment).

Payment settlement

client.orchestration.create_payment_intent(
    order_id: str,
    params: CreatePaymentIntentRequest,      # rail?, amount?, currency?
    pairing_token: str,
) -> ItemResponse[MarketPayment]

client.orchestration.get_market_payment(order_id: str) -> ItemResponse[MarketPayment | None]  # the server answers 404 NOT_FOUND when the order has no payment, so expect OrchestrationApiError rather than None

client.orchestration.submit_payment_proof(
    order_id: str,
    params: SubmitPaymentProofRequest,       # EVM: transaction_hash | Solana: transaction_signature; verification_mode required; chain, token, payer_address, payee_address, amount required
    pairing_token: str,
) -> ItemResponse[MarketPayment]

client.orchestration.lock_escrow_payment(
    order_id: str,
    params: LockEscrowPaymentRequest,        # transaction_hash, chain, escrow_address all required
    pairing_token: str,
) -> ItemResponse[MarketPayment]

client.orchestration.release_payment(
    order_id: str,
    params: ReleasePaymentRequest | None,    # transaction_hash?, notes?
    pairing_token: str,
) -> ItemResponse[MarketPayment]

client.orchestration.refund_payment(
    order_id: str,
    params: RefundPaymentRequest | None,     # transaction_hash?, reason?
    pairing_token: str,
) -> ItemResponse[MarketPayment]

submit_payment_proof accepts both EVM and Solana proof shapes; verification_mode is "recorded" | "rpc". RPC verification cross-checks the ERC-20 Transfer log against expected token, payer_address, payee_address, and amount.

Pagination

Six sync auto-paginated iterators:

for m in client.orchestration.list_machines_all():
    ...
for s in client.orchestration.list_skills_all():
    ...
for svc in client.orchestration.list_market_services_all():
    ...
for o in client.orchestration.list_market_orders_all(machine_id="m-1"):
    ...
for p in client.orchestration.list_agent_pairings_all("m-1"):
    ...
for e in client.orchestration.list_runtime_endpoints_all("m-1"):
    ...

Each *_all accepts the same filter kwargs as its non-iterator counterpart, minus cursor. limit controls page size; cursor is managed internally. For manual pagination, the underlying list_* methods accept cursor and limit as keyword-only args:

cursor = None
while True:
    page = client.orchestration.list_market_orders(
        machine_id="m-1", limit=50, cursor=cursor,
    )
    for order in page.items:
        handle(order)
    if page.next_cursor is None:
        break
    cursor = page.next_cursor

MarketSearch:

@dataclass(frozen=True, slots=True)
class MarketSearch:
    id: str
    machine_id: str
    request_payload: Mapping[str, Any]
    normalized_task: NormalizedTask
    status: Literal["completed", "no_match"]
    quotes: tuple[MarketQuote, ...]
    created_at: str
    agent_pairing_id: str | None = None

Policies

Cross-cutting policy records above and beyond a pairing’s delegation policy. Platform auth.

client.orchestration.list_policies() -> ListResponse[Policy]

client.orchestration.create_policy(params: CreatePolicyRequest) -> ItemResponse[Policy]

client.orchestration.update_policy(policy_id: str, params: UpdatePolicyRequest) -> ItemResponse[Policy]

There is no get_policy method. Fetch policies with list_policies.

Observability

client.orchestration.check_health() -> HealthResponse        # GET /api/v1/health

client.orchestration.get_readiness() -> ReadinessResponse    # GET /api/v1/readiness

client.orchestration.list_audit_events(
    *,
    machine_id: str | None = None,
    resource_type: str | None = None,
    limit: int | None = None,
) -> ListResponse[AuditEvent]

check_health requests /api/v1/health, which returns 404 on the hosted orchestrator: the health route is GET /health at the host root, so call it over HTTP. list_audit_events sends machine_id, but the server ignores it and filters only on type, resourceType and resourceId; filter the returned events by machine yourself.

HTTP-only endpoints

These endpoints exist on the HTTP API (see Machine Markets API: Orchestration) and have no SDK method. Call them over HTTP:

Tasks: create, list, get, discover, resolve, execute
Graph: read a machine graph, create/update/delete nodes and edges
Policies: get a single policy
Runs: list runs, get a run

Errors

All extend peaq_os_sdk.exceptions.base.PeaqosError:

class OrchestrationError(PeaqosError):
    """Base. Catch this for any orchestration failure."""

class OrchestrationApiError(OrchestrationError):
    code:        str        # CommonErrorCode literal or any service string
    status_code: int        # HTTP status; 0 when raised by client-side validation
    details:     Any | None # snake-cased payload; sensitive fields redacted

class OrchestrationNetworkError(OrchestrationError):
    """Transport-level failure. __cause__ carries the original requests.RequestException."""

class OrchestrationConfigError(OrchestrationError):
    """Misconfiguration: orchestration_url missing or invalid."""

class OrchestrationValidationError(OrchestrationError):
    field:      str
    constraint: str

CommonErrorCode literal: AUTH_REQUIRED, AUTH_INVALID, AGENT_AUTH_REQUIRED, AGENT_AUTH_INVALID, AGENT_AUTH_EXPIRED, VALIDATION_ERROR, NOT_FOUND, MACHINE_NOT_ACTIVE, MACHINE_NOT_ACTIVATED, MACHINE_IDENTITY_EXISTS, MACHINE_IDENTITY_IMMUTABLE, MACHINE_IDENTITY_PROOF_REQUIRED, MACHINE_IDENTITY_PROOF_INVALID, MACHINE_IDENTITY_PROOF_EXPIRED, PEAQOS_IDENTITY_UNAVAILABLE, AGENT_PAIRING_PROOF_REQUIRED, AGENT_PAIRING_PROOF_INVALID, AGENT_PAIRING_PROOF_EXPIRED, AGENT_PAIRING_UNAVAILABLE, AGENT_PAIRING_REQUIRED, AGENT_PAIRING_INACTIVE, AGENT_POLICY_DENIED, AGENT_SPEND_LIMIT_EXCEEDED, AGENT_DAILY_LIMIT_EXCEEDED, QUOTE_EXPIRED, ORDER_CLOSED, ORDER_NOT_DELIVERED, EXECUTION_UNSUPPORTED, PAYMENT_REQUIRED, PAYMENT_RPC_REQUIRED, PAYMENT_RPC_ERROR, PAYMENT_NOT_MINED, PAYMENT_TX_FAILED, PAYMENT_TRANSFER_NOT_FOUND. Proof and credential types:

@dataclass(frozen=True, slots=True)
class MachineIdentityProofInput:
    challenge_id: str
    signature: str

@dataclass(frozen=True, slots=True)
class AgentPairingChallengeResponse:
    challenge_id: str
    agent_address: str
    message: str
    expires_at: str
    verification_method: Literal["eip191"]
    machine_id: str | None = None
    machine_identity_ref: str | None = None
    agent_did: str | None = None

@dataclass(frozen=True, slots=True)
class AgentProofInput:
    challenge_id: str
    signature: str

@dataclass(frozen=True, slots=True)
class ProviderCredentials:
    # Per-provider credentials sent with market search requests.
    # Always redacted before request bodies are persisted.
    walrus:   WalrusCredentials  | None = None
    akash:    AkashCredentials   | None = None
    world_id: WorldIdCredentials | None = None   # wire: worldId
    geodnet:  GeodnetCredentials | None = None
    aethir:   AethirCredentials  | None = None

MarketPaymentRailType literal: "not-required", "wallet", "xvv42", "x402", "vault-stripe", "escrow", "offchain-record", "external", "onchain-escrow", "wdk-usdt-transfer". The transport synthesises BAD_RESPONSE (non-JSON body) and INVALID_RESPONSE_SHAPE (envelope mismatch) when the server response is malformed.

Sync vs async

Sync only. Every public function is plain def over requests.Session. The transport reuses client.session, so you can wrap retries externally.

Credential handling

Platform (x-api-key): set once on the client via api_key, sent on every call by default.
Agent pairing (x-agent-pairing-token): passed per call as the last argument to search_market, the order mutations, and the payment mutations. The SDK does not store or rotate it.
None: used only by machine_agent_heartbeat (token rides in the body).

_sanitize_body redacts any field whose name contains token, key, secret, password, credential, or auth (case-insensitive) before attaching the body to an error.

​Setup

​Configuration

​Constructor kwargs

​Env vars (via PeaqosClient.from_env())

​Envelopes

​Machines

​Machine agents

​Machine identity challenges

​Agent pairings

​Runtime endpoints

​Skills

​Market services

​Market search

​Market orders

​Payment settlement

​Pagination

​Policies

​Observability

​HTTP-only endpoints

​Errors

​Sync vs async

​Credential handling

​Related