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.
The Python SDK 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)
from peaq_os_sdk.orchestration import list_machines
resp = list_machines(client, limit=20)
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 |
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]
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
create_agent_pairing_challenge returns a server-issued challenge keyed to agent_address, agent_provider, agent_role, and optional agent_did.- The Machine Agent signs
item.message (EIP-191) with the wallet 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 delegation policy, 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.
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():
...
*_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
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:
list_policies, create_policy, get_policy, update_policy
- Observability:
list_audit_events, list_runs, get_run, get_readiness, get_health
- Order family extras:
cancel_market_order, retry_market_order
Types for policies, observability, and 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 now exported (same dev-orch release as the methods above):
@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.
