Python SDK

pip install pacspace-sdk

Python 3.10 or later. The calls on this page use only the standard library. Answers are plain dicts carrying the wire's field names (entityId, seq), while the calls take the SDK's words (record, entry). Every entry the SDK takes counts from 1, as the pages do; it sends seq = entry - 1 on the wire and raises ValidationError for an entry below 1 before anything is sent.

The client#

Create the client where your system already logs what an agent did, such as the harness or the tool router, so the key stays with your code and the agent never holds it.

import os
from pacspace_sdk import PacSpace

pac = PacSpace(
    api_key=os.environ["PACSPACE_API_KEY"],
    webhook_secret=os.environ.get("PACSPACE_WEBHOOK_SECRET"),  # only if you verify webhooks with this client
)

fingerprint#

from pacspace_sdk import fingerprint, write_blinding_file, read_blinding_file, blinding_file_text

fingerprint(source, alg=None, *, blind=None, blinding=None) -> Fingerprint   # .ref: {"alg": "hmac-sha-256", "digest": "...", "byteLength": "..."}; .blinding: Blinding or None
write_blinding_file(file_path, blinding) -> str   # writes <file_path>.pacspace.json and returns that path
read_blinding_file(path) -> Blinding
blinding_file_text(blinding) -> str

source is bytes, a file object opened in binary mode, or any iterable of bytes. A file object is read in chunks and never held whole. Blinded by default: .blinding is None only with blind=False or alg="sha-256". Pass blinding= to fingerprint the same file again and get the same .ref. A Blinding shows no value when printed. See Fingerprint a file.

records.emit#

pac.records.emit(
    *,
    record: str,                 # your id for the record (wire: entityId)
    title: str,                  # at most 120 characters
    occurred_at: str,            # RFC 3339, UTC, ending in Z
    actor_id: str,
    payloads: list[dict],        # at least one fingerprint's .ref
    kind: str | None = None,     # lower case with hyphens
    instructed_by: str | None = None,
    description: str | None = None,
    note: str | None = None,
    references: list[dict] | None = None,   # each {"recordKey": ..., "entry": ...}, entries counted from 1
    amends: dict | None = None,             # {"recordKey": ..., "entry": ...}, the entry this one corrects; send lifecycle="amended"
    record_type: str | None = None,         # default "machine-action-record"
    lifecycle: str | None = None,           # default "recorded"
    idempotency_key: str | None = None,     # one key is one write; wire: referenceId
) -> dict

The answer is the write's body: intakeId, receiptId, recordType, entityId, recordKey, lifecycle, status (QUEUED), referenceId, receivedAt, message, and idempotent: True on a repeat of a finished key. A repeat of a key whose first write is still running returns {"code": "IDEMPOTENCY_PENDING", "retryAfterSeconds": ...} rather than raising. A payload that holds a Blinding or a whole Fingerprint is refused with RECORD_BLINDING_INCLUDED before anything is sent. See Write an entry and, for amends, Correcting an entry.

records.history#

pac.records.history(record, from_entry=None, to_entry=None, record_type=None) -> dict   # from_entry and to_entry counted from 1

Returns the history file whole: entries (each with seq, status, receiptId, and for a failed entry code and sentence), receipts, bundleDigest, and nextFromSeq when more remain. The file keeps the wire's seq, counted from 0, and nextFromSeq is a wire number: the next page's from_entry is nextFromSeq + 1. See History.

records.receipt#

pac.records.receipt(record, entry, record_type=None) -> dict   # entry counted from 1; {"recordType", "entityId", "receipt", "verify"}

See Receipts.

records.check#

pac.records.check(history, expect=None, source=None) -> dict
# {"ok": bool, "entries": [{"entry", "status", "kind", "code", "sentence", "expected"}], "failedCheck": str | None,
#  "sourceChecked": False, "reason": str | None, "note": str | None, "blindingFileNeeded": [int]}
# expect[].entry and each result's entry count from 1
# expect[].payloads: fingerprints you kept; expect[].files: [{"source": bytes or a binary file object, "blinding": Blinding or None}]
# "expected" is on the rows you asked about; "blindingFileNeeded" is there only when an entry needed one

The Python check compares the fingerprints in expect, or those of the files in expect[].files, with the committed ones and reads each entry's status. ok is True when no entry failed, at least one is committed, and everything you asked about was found. It does not recompute the seals, and sourceChecked is always False. For the whole check, save the history and run the standalone checker:

import json, subprocess

with open("history.json", "w") as f:
    json.dump(history, f)
subprocess.run(["npx", "@pacspace-io/check", "history.json"], check=False)

See Check.

Each row of the result's entries carries kind: attempt for an entry the record's rules did not allow, failure for a failure record, transition for any other committed entry, and None for an entry that is queued or failed. kind is read from the entry's sealed record to help you list entries. It is not a checked fact: this check does not check seals.

Checking a file of chosen fields#

verify_record_link_copy_v1(copy) takes a kept copy, such as the file the TypeScript records.disclose makes, read with json.load. The copy shows only the fields its maker chose, and the call checks each shown field against the record's seal history inside it. verify_record_disclosure_v1(form, disclosure) does the same for a disclosure and its seal history form. Both use only the standard library and never raise; ok is True only when every entry checked. Neither recomputes the seals or reads the proof layer, and the result's chainCheckNote says so; for that, run @pacspace-io/check on the file.

webhooks#

pac = PacSpace(api_key=..., webhook_secret=...)

pac.webhooks.verify(signature, timestamp, raw_body, tolerance=300) -> dict
pac.webhooks.verify_from_headers(headers, raw_body, tolerance=300) -> dict

Both return the delivery as a dict, {"event", "timestamp", "data"}, or raise WebhookVerificationError. tolerance is in seconds. raw_body is the request body as a string, before any JSON parsing. See Signature verification.

Errors#

Every API error is a PacSpaceError with status_code, code, message, and request_path. A refusal with status 400, 403, 409, 413, 422, or 500, such as RECORD_TITLE_TOO_LONG, carries the API's code exactly. A 401, 402, 404, 412, 429, or 503 arrives as its own class instead: InvalidApiKeyError, InsufficientCreditsError, NotFoundError, ContractNotDeployedError, RateLimitError (retry_after is the seconds the API named), and ServiceUnavailableError, each with the API's message. ValidationError is a 400 with no code, or the SDK's own refusal before anything is sent, and WebhookVerificationError comes from webhooks. The full table is in Errors.

An agent harness that writes after each tool call can treat a refused entry as its own event, apart from the agent's work:

from pacspace_sdk import PacSpaceError, RateLimitError

try:
    pac.records.emit(
        record="eval-run-4417",
        title=title,
        occurred_at=occurred_at,
        actor_id="agent-7",
        payloads=[ref],
        idempotency_key=f"eval-run-4417:{event_id}",
    )
except RateLimitError as err:
    ...  # still limited after the retries: wait err.retry_after seconds, send again with the same idempotency_key
except PacSpaceError as err:
    if err.code == "RECORD_TITLE_TOO_LONG":
        ...  # nothing was queued: shorten the title, move the rest to description, and send again
    else:
        print(err.status_code, err.code, err.message)