Skip to main content
The distribution surface ships in @peaqos/peaq-os-sdk and peaq-os-sdk 0.4.0+ and is flagged @experimental. It builds on the stream crypto module (0.3.0+): everything here moves chunks that buildChunkChain already signed and encrypted.
The stream module makes machine data sellable; this surface actually sells and ships it. It covers the whole buyer-to-seller loop: a buyer creates a purchase against a listing, pays over a transfer or rail, and the seller — notified over a control socket — re-wraps the purchased chunk keys and delivers through a pluggable delivery channel: S3 (the buyer-access files plus a pre-signed download URL, with the ciphertext staying at each chunk’s storageRef) or the peaqos-p2p transport (the encrypted chunks themselves, streamed machine-to-machine). The seller never decrypts or re-encrypts chunk data at any point; only the buyer-access documents (wrapped per-chunk keys) are generated during distribution.

​
The distribution lifecycle

Every step is a stateless API call — the SDK holds no purchase state. The buyer authenticates with an agent-pairing token (x-agent-pairing-token), the seller’s machine agent with x-machine-agent-id + x-machine-agent-token, and platform reads with x-api-key.

​
Delivery setup (before any purchase)

Both sides discover what the orchestration service supports instead of hardcoding it, and machines register what they can handle:
FunctionPurpose
listPaymentRailsReturns the supported rails — X402PaymentRail (chains, tokens, requiresProof) or TransferPaymentRail (chains, tokens, verification), discriminated by type.
listDeliveryTransportsReturns the available transports, discriminated by mode: p2p (transportId, version, features, protocols, encoding), storage (provider: "walrus" | "s3"), or tunnel (provider: "cloudflare").
putDeliveryCapabilitiesRegisters what a machine can support (buyer or seller side) — after a machine is provisioned, its capability advertisement is what makes it reachable for delivery. The server populates mode and signaling; capabilities are short-lived and renewable.
updateStreamListingPurchaseFieldsAdds paymentRails and deliveryOptions to an existing stream listing. Additive — existing listing fields are untouched.
Capability advertisements carry transport identity only. Never put secrets — provider API keys, tokens, private keys — into a capability advertisement or listing; sensitive values stay in local secrets or encrypted payloads. Shared value unions across this surface: PurchasePaymentRail ("x402" | "transfer"), PurchaseChain ("peaq" | "base" | "ethereum" | "solana"), PurchaseToken ("USDC" | "USDT" | "PEAQ"), PurchaseDeliveryMode ("p2p" | "storage" | "tunnel"), PurchaseResourceType ("stream.bundle" | "stream.live") — each with SCREAMING_SNAKE_CASE constants (PURCHASE_CHAIN_BASE, PURCHASE_TOKEN_USDC, …).

​
Buyer: purchases

import {
  createPurchase,
  getPurchase,
  createPurchasePaymentIntent,
  submitPurchasePaymentProof,
  getPurchaseDelivery,
  createPurchaseEvent,
} from "@peaqos/peaq-os-sdk";

// 1. Declare intent: listing, optional chunk selection, buyer identity + X25519 key
const purchase = await createPurchase(client, {
  resource: {
    type: "stream.listing",
    listingId: "listing-abc",
    selection: { chunkIds: ["chunk-1", "chunk-2"] },
  },
  buyer: {
    id: "did:peaq:0xBUYER",
    publicKey: { type: "x25519", publicKeyHex: BUYER_X25519_PUB },
  },
}, pairingToken);

// 2. Poll until delivery_reserved
const state = await getPurchase(client, purchase.item.id);

// 3. Declare the payment rail
const intent = await createPurchasePaymentIntent(client, purchase.item.id, {
  rail: { type: "transfer", destination: { payTo: "0xSELLER" } },
}, pairingToken);
// intent.instructions is a union:
//   { send: { chain, token, amount, to } }   — transfer rail
//   { protocol: "x402", message }            — x402 rail

// 4-5. Pay on-chain, then submit proof
const proof = await submitPurchasePaymentProof(client, purchase.item.id, {
  rail: "transfer",
  transactionHash: "0xtxhash...",
  chain: "base",
  token: "USDC",
  verificationMode: "rpc",       // optional: "recorded" | "rpc"
  rpcUrl: "https://rpc.base.org", // optional, for RPC verification
}, pairingToken);
// proof.deliveryStatus: "blocked" | "releasable"
// proof.receipt: { verificationMode, status, blockNumber?, verifiedAt }

// 6. Retrieve the delivery session + access entries
// (409 STREAM_DELIVERY_NOT_READY until the seller prepares access)
const delivery = await getPurchaseDelivery(client, purchase.item.id, pairingToken);
// delivery.delivery: sessionId, transportId, protocols, encoding, connect handoff
// delivery.access: peaq.stream.buyer-access.v1 entries (wrapped chunk keys)
For the x402 rail, the proof body is { rail: "x402", paymentHeader } — the signed payment header, no on-chain transfer of your own. Purchase status progresses delivery_reserving → delivery_reserved → payment_pending → payment_verified → delivery_released → active → completed (or cancelled / failed). Payment status is pending | verified; delivery status is requested | ready | active | completed | failed. Both sides record lifecycle progress with createPurchaseEvent — event types delivery.started, delivery.completed, delivery.failed, buyer.connected, seller.ready, seller.offline. Invalid types are rejected client-side.

​
Buyer: paying on-chain

Helpers for the transfer rail on peaq, Base, and Solana. Token decimals resolve from a well-known registry (USDC/USDT on peaq, USDC on Base/Solana) or an explicit tokenDecimals override — no on-chain decimals() lookups.
FunctionPurpose
transferTokenNative or token transfer routed by chain; returns TokenPaymentResult with the tx hash / signature.
submitPaymentProofPOSTs the proof JSON to a confirmationUrl; expects { accepted: boolean }. (Distinct from client.orchestration.submitPaymentProof().)
payAndSubmitProofChains the two. On partial failure returns { payment, proof: null, proofError } so the tx hash can be resubmitted instead of paying twice.
ChainExtra paramsOptional peer deps
peaq—viem (required peer)
baserpcUrlviem
solanarpcUrl, solanaSigner@solana/web3.js, @solana/spl-token (SPL only)
A Solana signer comes from the OWS wallet via PeaqosClient.solanaSignerFromWallet() — OWS-native Solana signing also ships in 0.4.0.

​
Seller: getting paid and preparing access

The seller’s primary signal is the control socket — a WebSocket to the orchestration service that carries short control notifications (never bulk data): 
import { connectControlSocket, prepareBuyerAccess } from "@peaqos/peaq-os-sdk";

const conn = await connectControlSocket(client, machineId, agentId, agentToken, {
  onConnected(msg) { console.log("connected:", msg.machineId); },
  async onNotification(msg) {
    if (msg.type === "purchase.payment_proof.recorded") {
      const { purchaseId } = msg.payload as { purchaseId: string };
      // payment verified — prepare buyer access now
    }
  },
  onAck() {}, onError(err) { console.error(err); }, onClose() {},
});
When the control socket is unavailable, PollingConfirmationProvider is the fallback: it polls a confirmationUrl (appending ?orderId=...) every pollInterval ms (default 30 000) until status === "confirmed" or timeout seconds (default 3 600) elapse, then resolves a PaymentConfirmation (orderId, buyerId, buyerPublicKeyHex, txHash, status, confirmedAt). HTTP 5xx retries silently; 4xx throws StreamValidationError; timeout throws StreamTimeoutError; stop() aborts immediately. The polling surface (and DistributePackage) tracks the sale as orderId; in the orchestration purchases flow the same identifier is the purchaseId — peaqos stream distribute --order-id documents it as exactly that. On confirmation the seller submits the re-wrapped chunk keys plus a short-lived connection handoff: 
const result = await prepareBuyerAccess(client, purchaseId, {
  machineId, agentId, agentToken,
  items: buyerAccessEntries,               // from buildBuyerAccessFiles
  delivery: {
    mode: "p2p",
    transportId: "peaqos-p2p",
    connect: { type: "peaqos-p2p-url", url: connectUrl, expiresAt },
  },
});
// result.access — the submitted entries; result.deliverySession — session + connect handoff
Every request on this path is authenticated against the machine’s agent identity, and delivery only proceeds for a purchase whose payment has been verified — requests against an unpaid or unprepared purchase are rejected. The delivery.connect.url is a secret handoff: never log it.

​
distributeData and delivery channels

distributeData packages the seller side into one call — generate the buyer access files and hand them to a DeliveryChannel: 
import { distributeData, S3DeliveryChannel } from "@peaqos/peaq-os-sdk";

const channel = new S3DeliveryChannel({
  bucket: "my-data-bucket",
  prefix: "stream/",
  region: "us-east-1",     // optional — AWS credential chain otherwise
  endpoint: "https://…",   // optional — MinIO, R2, Spaces
  presignExpiry: 7200,     // seconds, default 3 600
});

const result = await distributeData({
  chunks: chain.chunks,          // from buildChunkChain
  ownerPrivateKeyHex,            // never logged; wiped after key wrapping
  payment: confirmation,         // must be status "confirmed"
  deliveryChannel: channel,
  // maxFileSize: 512_000        // bytes per access file before splitting
});
// result.chunksDistributed, result.buyerAccessFiles,
// result.delivery.downloadUrl   — pre-signed S3 URL ("p2p" channels return null)
S3DeliveryChannel uploads each access file to {prefix}{buyerId}/{fileName} and needs @aws-sdk/client-s3 + @aws-sdk/s3-request-presigner (optional peer deps — missing packages throw StreamConfigError with the install hint). Custom channels implement the single-method DeliveryChannel interface: deliver(pkg: DistributePackage) => Promise<DeliveryResult>, where pkg carries buyerAccessFiles, encryptedDataRefs, buyerId, and orderId.

​
P2P delivery

The peaqos-p2p transport moves the encrypted chunks directly machine-to-machine — no storage bucket in the middle. The seller runs a long-lived channel; the buyer dials in with a transient node, receives the chunks, verifies, and decrypts. Delivery here is transport and hand-over only: what a machine is allowed to serve is still governed by its provisioning and registered capabilities.

​
Seller: P2PDeliveryChannel

import { P2PDeliveryChannel } from "@peaqos/peaq-os-sdk";

const channel = new P2PDeliveryChannel(
  machineId,
  agentId,
  agentToken,        // secret — never logged
  client,            // PeaqosClient
  resolveChunkData,  // (ref: string) => Promise<{ envelope: Chunk; ciphertext: Uint8Array }>
  {
    capabilityTtlSeconds: 3600,   // default 1 h
    connectTtlSeconds: 300,       // default 5 min; must be < capabilityTtlSeconds
    connectionTimeoutMs: 30_000,
    chunkTransferTimeoutMs: 60_000,
  },
);

await channel.start();                  // idempotent
const result = await channel.deliver(pkg); // { channel: "p2p", downloadUrl: null, deliveredAt }
await channel.stop();
start() discovers the peaqos-p2p transport via listDeliveryTransports, validates it against a fixed allowlist, creates the node, registers the machine’s delivery capability, schedules capability renewal, and installs a single central accept loop for inbound buyer connections. deliver() implements DeliveryChannel, so it plugs straight into distributeData. Called after payment confirmation, it resolves and validates every chunk against the access entries, builds a short-lived delivery.connect handoff (an opaque peaqos-p2p:// URL), calls prepareBuyerAccess, waits for the buyer to connect, streams the chunks as frames, and emits delivery.started / delivery.completed / delivery.failed progress events. Inbound handshakes are routed by (purchaseId, sessionId, buyerId); sessions are single-use with tombstone replay protection.

​
Buyer: P2PDeliveryReceiver

import { P2PDeliveryReceiver, getPurchaseDelivery } from "@peaqos/peaq-os-sdk";

const receiver = new P2PDeliveryReceiver(
  buyerId,
  client,
  recipientPrivateKeyHex,  // X25519 private key — never logged
  { connectionTimeoutMs: 30_000, chunkReceiveTimeoutMs: 60_000 },
);

const delivery = await getPurchaseDelivery(client, purchaseId, pairingToken);
const { data, chunksReceived, completedAt } = await receiver.receive({ purchaseId, delivery });
receive() validates the negotiated session (mode, transport, encoding, connect expiry), dials the seller over delivery.connect.url, identifies itself with a handshake frame, and then — for each chunk — verifies the ciphertext hash, signature, and chain linking before decrypting with decryptChunk, runs verifyChunkChain as a final cross-check, and reassembles the plaintext in index order. The transient node is closed in finally, success or failure.

​
Wire format and dependencies

Frames use peaqos-json-frame-v1 — newline-delimited JSON with types handshake, chunk, complete, error. The libp2p stack is wrapped entirely inside the SDK: no libp2p types, peer IDs, or multiaddrs appear in the public API, and delivery.connect URLs are redacted from all SDK-controlled logs, errors, and telemetry. libp2p ships as optional peer dependencies — install them only when you use P2P delivery: 
npm install libp2p @libp2p/tcp @chainsafe/libp2p-yamux @chainsafe/libp2p-noise @multiformats/multiaddr
Without them, the channel throws StreamConfigError carrying exactly that install hint.

​
Errors

Distribution extends the Stream error hierarchy:
ErrorWhenExtras
StreamTimeoutErrorConfirmation polling timed out..timeout (s), .orderId
StreamPaymentErrorOn-chain transfer or proof submission failed..txHash (when available), .reason
StreamConfigErrorOptional dependency missing (AWS SDK, libp2p)..installHint
StreamDeliveryErrorSeller-side P2P delivery failure..channel: "p2p", .purchaseId, .code
StreamReceiveErrorBuyer-side P2P reception failure..purchaseId, .code
.code is a coarse public StreamErrorCode: "config" (missing dependency, transport/capability negotiation failure), "protocol" (wire/frame/session violation, verification failure), or "transfer" (connection, timeout, or data-transfer failure). Error messages are sanitized — no key material, no raw transport exceptions. Purchase functions throw the standard orchestration errors (OrchestrationApiError, OrchestrationValidationError, …).

​
Python and the CLI

The Python SDK ships the same distribution surface in snake_case. From the terminal, peaqos stream distribute (payment listener + S3 delivery), peaqos stream pay / payproof (buyer transfer + proof), and peaqos stream consume --download-url (remote release packages) wrap it — P2P delivery itself is SDK-level and has no CLI flag yet.