Orchestration (Python)

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

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 for tree-shaking or 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

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

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]
client.orchestration.archive_machine(machine_id: str) -> None

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.list_machine_agents(machine_id: str) -> ListResponse[MachineAgent]
client.orchestration.enroll_machine_agent(machine_id: str, params: EnrollMachineAgentRequest | None = None) -> ItemResponse[MachineAgent]
client.orchestration.revoke_machine_agent(machine_id: str, agent_id: str) -> None
client.orchestration.machine_agent_heartbeat(params: MachineAgentHeartbeatRequest) -> MachineAgentHeartbeatResponse

enroll_machine_agent returns ItemResponse[MachineAgent] whose item.provisioning_token is populated once at enrollment. 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 now requires agent_proof: AgentPairingProof and accepts optional agent_did. delegation_policy.allowed_service_ids and delegation_policy.denied_service_ids join the existing skill-level allow/deny lists.

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]
client.orchestration.update_skill_config(skill_key: str, params: UpdateSkillConfigRequest) -> ItemResponse[SkillConfigResponse]

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?
    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 a union narrowed by execution.status_code: 200 for a native run (with run + outcome), 202 for an external handoff (with handoff: { label, url, notes }).

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]

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 ship in this release:

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():
    ...
for e in client.orchestration.list_runtime_endpoints_all():
    ...

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 now 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(
    *, limit: int | None = None, cursor: str | None = None,
) -> ListResponse[Policy]

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

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

get_policy is not wired yet — fetch via list_policies until it ships.

Observability

client.orchestration.check_health() -> HealthResponse        # GET /health (host root)

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

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

Coming next

These endpoints ship on the HTTP API (see Machine Markets API: Orchestration). SDK method bindings follow shortly:

Tasks: create_task, discover_task, resolve_task, execute_task, get_task, list_tasks
Graph: get_machine_graph, create_graph_node, update_graph_node, delete_graph_node, create_graph_edge, update_graph_edge, delete_graph_edge
Policies: get_policy
Runs: list_runs, get_run
Order family extras: cancel_market_order, retry_market_order

Types for the remaining order-family operations already ship in peaq_os_sdk so application code can prepare for the methods landing.

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, PAYMENT_REQUIRED, PAYMENT_RPC_REQUIRED, PAYMENT_RPC_ERROR, PAYMENT_TX_FAILED, EXECUTION_UNSUPPORTED, ENDPOINT_UNREACHABLE. New types exported in 0.3.0:

@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-adapter credentials. Shape is adapter-specific and
    # documented per-adapter on robotic.sh. Always redacted
    # before request bodies are persisted.
    providers: Mapping[str, Mapping[str, str]] | None = None

MarketPaymentRailType adds "wdk-usdt-transfer" alongside the existing

"x402" | "mpp" | "wallet" | "vault-stripe" | "escrow" | "onchain-escrow" | "offchain-record" | "external" | "not-required"

set. 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 on this branch. Every public function is plain def over requests.Session. The transport reuses client.session, so users can wrap retries externally. First-class async support is on the roadmap.

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 to search_market as the second argument. 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

​Coming next

​Errors

​Sync vs async

​Credential handling

​Related