Python SDK

Computer

Computer is the API the Quickstart uses. A computer is a versioned environment where authority, resources, and work live. You open one, fork it, grant an agent part of the fork, run work, read the signed receipts, and apply the changes you keep. Your kernel enforces every grant and signs every decision.

Calls are synchronous and return when your kernel has decided. Event streams are iterators. Every keyword argument after the first positional one is keyword-only. Composition uses & for parallel and | for sequence, with & binding tighter.

This program is Quickstart part 1 in one file. It runs inside the computer that chio new quickstart creates.

from chio import Computer, USD

project = Computer.open("quickstart")
fixer = project.agent("fixer")
fork = project.fork()
authority = fork.grant(
    to=fixer,
    resources={"/workspace": ["read", "write"], "/models/default": ["invoke"]},
    budget=USD(2),
    expires_in="20m",
)

run = fixer.run("fix issue #412", on=fork, authority=authority)
for event in run.events():
    print(event.summary)

refusal = project.history.receipts(outcome="deny")[0]
print(refusal.describe())

00:00  admitted   fixer runs "fix issue #412" on fork 51c07ae
00:03  read       /workspace/src/pricing.py
00:05  read       /workspace/tests/test_pricing.py
00:09  refused    read /secrets/.env: outside the grant
00:12  invoke     /models/default, USD 0.21
00:41  write      /workspace/src/pricing.py
00:56  test       pytest -k issue_412: passed
spent  USD 0.48 of 2.00

receipt    rcpt-01a8…9c04
request    fixer read /secrets/.env
verdict    deny: outside the grant
grant      cap-01a8…31f2
policy     sha256 3c7b…8010
signed     your kernel 4f1c…a2e9, signature verifies

The examples below continue from it: project is the quickstart computer, fixer is its agent, and fork and authority are the fork and grant above. The service examples use the pricing service from Quickstart part 2, and the partner examples use the partner card, treaty, and reviews mailbox from part 3.

Computer.open

Computer.open(name: str) -> Computer

Opens a computer your operator identity owns and returns its handle. The handle reads the computer's current revision; nothing is copied.

from chio import Computer

project = Computer.open("quickstart")
print(project.describe())

computer   quickstart
owner      did:chio:z6Mkf3…Qp9 (you)
revision   main @ 7c2e1d0
resources  /workspace         git, 14 files
           /models/default    model route, metered in USD
           /secrets           3 files

Computer.connect

Computer.connect(name: str) -> Partner

Resolves a paired partner by the name you gave its card, and authenticates its kernel before returning.

partner = Computer.connect("partner")
print(partner.describe())

partner    partner, did:chio:z6MkpA7…c2T
operator   Halden, halden.example
kernel     9b07…d31c, pinned from its card
transport  iroh, authenticated at handshake

Computer.partners

Computer.partners() -> list[Partner]

Lists every partner you have paired with, each with its pinned kernel key and the treaty in force.

for partner in Computer.partners():
    print(partner.summary)

partner    did:chio:z6MkpA7…c2T   Halden   kernel 9b07…d31c   treaty trt-01aa…4c19

computer.agent

computer.agent(name: str) -> Agent

Returns the agent with this name, issuing it an identity from your kernel the first time. An agent holds no authority until you grant it some.

fixer = project.agent("fixer")
print(fixer.describe())

agent      fixer, did:chio:z6MkrT2…8wd
issued by  your kernel 4f1c…a2e9
grants     none

computer.agents

computer.agents(role: str, *, count: int, reads: Mailbox | None = None) -> list[Agent]

Starts several agents of one role, numbered from 1, optionally reading a shared mailbox.

reviewers = project.agents("review", count=2, reads=reviews)
for reviewer in reviewers:
    print(reviewer.summary)

reviewer-1   did:chio:z6MkvR1…3mc   reads reviews
reviewer-2   did:chio:z6MkvR2…9tf   reads reviews

computer.fork

computer.fork(*, boot: Program | None = None) -> Fork

Branches the computer's managed state at its current revision. Work on the fork stays there until you apply it. A fork does not reset authority, history, or spending.

fork = project.fork()
print(fork.describe())

fork       51c07ae of quickstart, base main 7c2e1d0
writes     stay in the fork until you apply them

fork.grant

fork.grant(*, to: Agent | Partner | str, resources: dict[str, list[str]], budget: Money | None = None, expires_in: str, under: Treaty | None = None, delegation: Delegation | None = None) -> Authority

Issues a capability token over part of the fork: resources and their rights, a budget charged at your door, and an expiry. to takes an agent, a partner, or a DID. delegation lets the holder delegate narrower grants onward, and under issues a grant within a treaty. Issuance completes durably before the call returns.

authority = fork.grant(
    to=fixer,
    resources={"/workspace": ["read", "write"], "/models/default": ["invoke"]},
    budget=USD(2),
    expires_in="20m",
)
print(authority.describe())

grant      cap-01a8…31f2, signed by your kernel
to         fixer, did:chio:z6MkrT2…8wd
on         quickstart, fork 51c07ae
allows     /workspace         read, write
           /models/default    invoke
budget     USD 2.00, charged at your door
expires    in 20 minutes

authority.reissue

authority.reissue(*, on: Fork, resources: dict[str, list[str]] | None = None, budget: Money | None = None, expires_in: str | None = None) -> Authority

Issues a grant on another fork with the same terms, or with the resources, budget, or lifetime you override. Your kernel signs the new grant; the original is unchanged.

next_fork = project.fork()
again = authority.reissue(on=next_fork)
print(again.describe())

grant      cap-01a8…5b3e, signed by your kernel
to         fixer, did:chio:z6MkrT2…8wd
on         quickstart, fork 7d2e914
allows     /workspace         read, write
           /models/default    invoke
budget     USD 2.00, charged at your door
expires    in 20 minutes

authority.delegate

authority.delegate(*, to: Agent, resources: dict[str, list[str]], budget: Money | None = None, expires_in: str | None = None) -> Authority

Issues a child grant from this one to another agent. The child can drop resources, rights, money, and time, never add them, and its budget is held from the parent's. The parent must carry a delegation allowance. A child that asks for more raises chio.AttenuationError, and nothing is issued.

# authority was granted with delegation=Delegation(depth=1, fanout=2)
tester = project.agent("tester")
child = authority.delegate(
    to=tester,
    resources={"/workspace": ["read"]},
    budget=USD("0.50"),
    expires_in="10m",
)
print(child.describe())

grant      cap-01d2…6c13, delegated by fixer from cap-01d2…2a90
to         tester, did:chio:z6MkvH5…3qe
allows     /workspace         read
budget     USD 0.50, held from the fixer's USD 2.00
expires    in 10 minutes
chain      you, fixer, tester

authority.revoke

authority.revoke() -> None

Revokes the grant and every grant delegated from it. The kernel checks each ancestor at admission, so a run under any of them is refused from its next call.

authority.revoke()

revoked    cap-01d2…2a90, by you
cascade    cap-01d2…6c13 (tester), refused from its next call

authority.token

authority.token -> CapabilityToken

The signed capability token behind the grant, in its wire form. The Computer and the Kernel reads one field by field.

print(authority.token.to_json(indent=2))

{
  "id": "cap-01a8…31f2",
  "issuer": "4f1c…a2e9",
  "subject": "b81e…40c7",
  "scope": { "grants": [ 3 tool grants with path_prefix constraints ] },
  "issued_at": 1791763200,
  "expires_at": 1791764400,
  "signature": "a3f9…0d52"
}

agent.run

agent.run(task: str, *, on: Fork, authority: Authority) -> Run

Sends a task to an agent under a grant. Every operation the agent attempts reaches your kernel before it reaches a resource.

run = fixer.run("fix issue #412", on=fork, authority=authority)
print(run.describe())

run        run-01a8…e7c0
agent      fixer, on fork 51c07ae
authority  cap-01a8…31f2

run.events

run.events(*, include: list[Agent] | None = None) -> Iterator[RunEvent]

Streams the run's events in order: admissions, reads, writes, model calls, tests, and refusals. A refusal is an event with a receipt, not an exception, and the run continues.

for event in run.events():
    print(event.summary)

00:00  admitted   fixer runs "fix issue #412" on fork 51c07ae
00:03  read       /workspace/src/pricing.py
00:05  read       /workspace/tests/test_pricing.py
00:09  refused    read /secrets/.env: outside the grant
00:12  invoke     /models/default, USD 0.21
00:41  write      /workspace/src/pricing.py
00:56  test       pytest -k issue_412: passed
spent  USD 0.48 of 2.00

partner.exec

partner.exec(task: str, *, on: Fork, authority: Authority, report_to: Mailbox | None = None) -> Run

Sends a task to a partner's agents under a grant issued within a treaty. The partner's kernel admits the work and runs it in its own process tree. run.join_tree() waits for every agent the partner delegated to.

run = partner.exec("fix issue #414", on=fork, authority=authority, report_to=reviews)
for event in run.events():
    print(event.summary)
run.join_tree()

00:00  them  admitted   "fix issue #414" under trt-01aa…4c19
00:01  them  delegate   coordinator to fixer-1, fixer-2, tester, each narrower
00:31  them  refused    fixer-2 read /secrets/.env: outside the grant, at your door
01:47  them  sealed     fork c4e81b7 at 5b93d10
spent  USD 3.12 of 5.00 across 4 of their agents

fork.diff

fork.diff() -> str

Returns the changes the fork holds against its base, as a unified diff.

print(fork.diff())

src/pricing.py
@@ def discount(price, percent):
+    if not 0 <= percent <= 100:
+        raise ValueError("percent must be between 0 and 100")
     return price * (1 - percent / 100)

computer.apply

computer.apply(changes: ChangeSet, *, treaty: Treaty | None = None) -> Applied

Publishes the fork's exact changes if main still matches the fork's base, and returns the signed merge. Under a treaty, both kernels co-sign it.

merged = project.apply(fork.changes)
print(merged.describe())

applied    fork 51c07ae onto quickstart main
revision   main 8d4b2e7, base 7c2e1d0
receipt    rcpt-01a8…d51e, signed by your kernel

computer.history.receipts

computer.history.receipts(*, outcome: str | None = None, agent: Agent | str | None = None, rule: str | None = None, service: str | None = None, since: str | None = None) -> list[Receipt]

Reads signed receipts from the computer's history, filtered by outcome, agent, rule, service, or time. Every filter combines with the others.

refusal = project.history.receipts(outcome="deny")[0]
print(refusal.describe())

receipt    rcpt-01a8…9c04
request    fixer read /secrets/.env
verdict    deny: outside the grant
grant      cap-01a8…31f2
policy     sha256 3c7b…8010
signed     your kernel 4f1c…a2e9, signature verifies

Receipt

class Receipt: id: str; request: str; summary: str; raw: str; def describe(self) -> str

One signed decision. summary is one line for listings, describe is the full card, and raw is the canonical JSON your kernel signed.

print(refusal.raw)

{"action":{"parameter_hash":"5e0d…71c2","parameters":{"path":"/secrets/.env"}},"capability_id":"cap-01a8…31f2","decision":{"reason":"not in capability scope","verdict":"deny"},"evidence":[],"id":"rcpt-01a8…9c04","kernel_key":"4f1c…a2e9","policy_hash":"3c7b…8010","receipt_kind":"mediated_decision","signature":"9d2e…07b4","tool_name":"read","tool_server":"quickstart"}

computer.history.export

computer.history.export(path: str, *, agent: Agent | str | None = None) -> Evidence

Writes the computer's history, or one agent's part of it, to a directory as an evidence package. Verification checks every signature against pinned kernel keys, offline.

evidence = project.history.export("audit/")
print(evidence.verify())

receipts   9 signed decisions from your kernel
refusals   1: fixer read /secrets/.env
spend      USD 0.48 of 2.00
verified   offline, against your kernel key 4f1c…a2e9

evidence.verify

evidence.verify(*, ledger: bool = False) -> Verification

Checks every signature in an exported evidence package against your pinned kernel key, offline. ledger=True also rebuilds both money ledgers from the signed records: each budget hold must close as committed, released, or retained with nothing outstanding, and each escrowed deposit must satisfy paid + refunded + locked + free = deposited. A hold whose outcome is unknown stays retained until it is resolved. Money Conservation states both identities.

evidence = project.history.export("audit/")
print(evidence.verify(ledger=True))

receipts   9 signed decisions from your kernel
holds      3 on cap-01a8…31f2, each with one terminal disposition or a retained outcome
budget     reserved 2.40 = committed 0.21 + released 1.39 + retained 0.80 + outstanding 0.00
retained   rcpt-01a8…c311: outcome unknown, hold kept until resolved
funding    1 deposit: paid 200 + refunded 60 + locked 0 + free 40 = deposited 300
verified   offline, against your kernel key 4f1c…a2e9

computer.spend.stream

computer.spend.stream(*, capability: str | None = None, cursor: str | None = None) -> Iterator[SpendEvent]

Streams spend events from your kernel's signed receipts, optionally for one capability lineage. Each event copies one receipt's financial metadata, bound to it by digest, and carries its sequence number. Pass an earlier cursor to resume: the cursor is opaque and binds your tenant, the filter, the last sequence scanned, and the store generation, and a stale cursor yields a typed gap rather than skipping ahead. Comptroller Console covers webhooks and burn rate.

for event in project.spend.stream(capability="cap-01d2…6c13"):
    print(event.summary)

seq 4308  15:46  triage-418  models  USD 0.02   remaining USD 10.04 of USD 20.00
seq 4309  15:46  triage-419  models  USD 0.02   remaining USD 10.02 of USD 20.00
seq 4310  15:47  triage-417  models  USD 0.02   remaining USD 10.00 of USD 20.00
cursor    9c1d…77ae, complete through seq 4310, checkpoint 43

computer.policy.add

computer.policy.add(*, name: str, allow: dict | None = None, deny: dict | None = None, egress: dict | None = None) -> Rule

Adds a named rule to the computer's policy: a path match to allow or deny, or an egress allow list for network connections. Policy narrows every grant on the computer, and no grant widens policy.

rule = project.policy.add(name="tests-are-evidence", deny={"write": "/workspace/tests/**"})
print(rule.describe())

rule       tests-are-evidence
denies     write /workspace/tests/**
policy     sha256 9a41…e6f2, narrows every grant on quickstart

computer.policy.describe

computer.policy.describe() -> str

Describes the policy in force: its hash and the one it replaced, your rules, and the defaults every computer starts with.

print(project.policy.describe())

policy     sha256 9e42…c5d1, was 3c7b…8010
rules      tests-are-evidence   deny write /workspace/tests/**
base       forbidden paths      **/.env, **/.env.*, **/.ssh/** and 27 more
compiled   HushSpec 0.1.0, evaluated at admission

computer.policy.hash

computer.policy.hash() -> str

Returns the hash every receipt records as policy_hash. A process you build yourself passes it to the kernel so its receipts name the same policy.

print(project.policy.hash())

sha256:9e42…c5d1

computer.service

computer.service(name: str, *, tools: dict[str, str]) -> Service

Exposes project functions as named tools and publishes them over MCP and HTTP. Every caller presents a grant from your kernel.

pricing = project.service(
    "pricing",
    tools={"quote": "src/pricing.py:quote", "refund": "src/pricing.py:refund"},
)
endpoint = pricing.publish(protocols=["mcp", "http"])
print(endpoint.describe())

service    pricing, from quickstart @ main 2f60a91
tools      quote · refund
mcp        https://127.0.0.1:9440/pricing/mcp
http       https://127.0.0.1:9440/pricing/v1, OpenAPI at /openapi.json
admits     callers that hold a grant from your kernel

service.grant

service.grant(*, to: str, tools: list[str], limits: dict | None = None, budget: Money | None = None, expires_in: str) -> Authority

Grants a caller named tools on the service, with a call rate, a budget spent at each tool's price, and an expiry.

access = pricing.grant(
    to="did:chio:z6MkbY4…e1R",
    tools=["quote"],
    limits={"calls": 100, "per": "1h"},
    budget=USD(2),
    expires_in="24h",
)
print(access.describe())

grant      cap-01ab…77d1, signed by your kernel
to         did:chio:z6MkbY4…e1R
allows     pricing.quote
limits     100 calls per hour
budget     USD 2.00, USD 0.01 per call
expires    in 24 hours

service.require

service.require(*, approval: str, when: str) -> Rule

Pauses each matching call until the approver signs for that exact request.

rule = pricing.require(approval="you", when="quote.percent > 50")
print(rule.describe())

rule       quote.percent > 50 waits for did:chio:z6Mkf3…Qp9 (you)
binds      one signature to one request

service.settle

service.settle(tool: str, *, rail: str, chain: str, token: str, payee: str) -> SettlementTerms

Declares how a tool's calls are paid: the rail, the chain, the token, and where the payee comes from. The terms go inside the governed intent, so an approval required by require covers the chain, payee, amount, and token along with the request. Approval-Bound Payments walks the flow.

terms = pricing.settle(
    "refund",
    rail="eip3009",
    chain="eip155:8453",
    token="USDC",
    payee="refund.wallet",
)
print(terms.describe())

settles    refund over eip3009 on eip155:8453 in USDC, payee from refund.wallet
binding    chain, payee, amount, token contract, dispatch, capability, expiry

service.price

service.price(tool: str, *, amount: Money, economy: str | None = None, split: list[Payee] | None = None) -> ToolPricing; Payee(beneficiary: str, *, bps: int, vest: Vest | None = None, cosigners: list[str] | None = None); Vest(*, cliff: str, end: str)

Sets a tool's price per call, the economy it settles in, and the payee split, one Payee per beneficiary. Your kernel writes the terms into the tool manifest as ToolPricing and signs the manifest again. A split must sum to exactly 10,000 basis points; each share may carry a vesting window and the cosigners who must sign it. Revenue Splits covers vesting and settlement.

from chio import Payee, Vest

priced = pricing.price(
    "quote",
    amount=USD("0.01"),
    economy="urn:chio:economy:pricing",
    split=[
        Payee("did:chio:z6Mkf3…Qp9", bps=5000),
        Payee(
            "did:chio:z6MkpA7…c2T",
            bps=3333,
            vest=Vest(cliff="2026-11-01", end="2027-01-30"),
            cosigners=["did:chio:z6Mkf3…Qp9"],
        ),
        Payee("urn:chio:economy:pricing:reserve", bps=1667),
    ],
)
print(priced.describe())

price      pricing.quote USD 0.01 per call, economy urn:chio:economy:pricing
split      5000  did:chio:z6Mkf3…Qp9
           3333  did:chio:z6MkpA7…c2T, vests 2026-11-01 to 2027-01-30, cosigned by you
           1667  urn:chio:economy:pricing:reserve
manifest   re-signed, payee_split sums to 10000 bps

service.events

service.events() -> Iterator[CallEvent]

Streams calls to the service as your kernel decides them.

for call in pricing.events():
    print(call.summary)

00:00  z6MkbY4  quote   sku-118 at 10%        allow     USD 0.01
00:04  z6MkbY4  refund  order-77              refused   not in the grant
00:05  z6MkQx8  quote   sku-118 at 10%        refused   no grant from your kernel
00:09  z6MkbY4  quote   sku-301 at 80%        pending   approval required
00:41  z6MkbY4  quote   sku-301 at 80%        allow     USD 0.01

service.usage

service.usage(*, since: str) -> Usage

Summarizes each caller's calls, refusals, approvals, and spend, computed from that caller's signed receipts.

print(pricing.usage(since="1h"))

caller             calls   refused   approvals   spent
z6MkbY4…e1R        100     2         1           USD 1.00
z6MkQx8…7Ha        0       1         0           USD 0.00

computer.treaty

computer.treaty(*, partner: Partner, actions: list[str], expires_in: str) -> Treaty

Agrees terms with a partner: the actions you cooperate on and what each requires. Both kernels sign, and each computes the intersection of the two governance ladders, so the stricter requirement wins.

treaty = project.treaty(
    partner=partner,
    actions=["workspace.read", "workspace.write", "model.invoke", "merge"],
    expires_in="7d",
)
print(treaty.describe())

treaty           trt-01aa…4c19, signed by both kernels
participants     you · partner (did:chio:z6MkpA7…c2T)
action           you       them      admitted as
workspace.read   allow     allow     either kernel admits under its own policy
workspace.write  allow     allow     either kernel admits under its own policy
model.invoke     allow     allow     charged at your door
merge            co-sign   co-sign   both kernels sign
expires          in 7 days

treaty.revoke

treaty.revoke() -> None

Ends the treaty and every grant issued under it, including grants the partner's agents delegated to each other. The revocation reaches the partner's kernel over iroh, and its next call is refused.

treaty.revoke()
for event in run2.events():
    print(event.summary)

00:00  them  admitted   "add banker's rounding as an option" under trt-01aa…4c19
00:04  you   revoked    trt-01aa…4c19, revocation root sent over iroh
00:05  them  refused    fixer-1 write /workspace/src/pricing.py: treaty revoked
00:05  them  stopped    their work closed, no effects after revocation

computer.mailbox

computer.mailbox(name: str, *, partner: Partner, treaty: Treaty) -> Mailbox

Opens a mailbox shared with a partner under a treaty. Each message is sent and claimed under its sender's or reader's own grant.

reviews = project.mailbox("reviews", partner=partner, treaty=treaty)
print(reviews.describe())

mailbox    reviews, under trt-01aa…4c19
senders    partner's agents, each under its own grant
readers    your agents that hold a reads grant

first_passing

first_passing(runs: Iterable[Run], *, check: str) -> Run

Waits on several runs and returns the first whose sealed fork passes the check. The other runs are cancelled, and their receipts stay in history.

from chio import first_passing

ours = fixer.run("fix issue #414", on=fork_a, authority=grant_a)
theirs = partner.exec("fix issue #414", on=fork_b, authority=grant_b)
winner = first_passing([ours, theirs], check="pytest")
print(winner.describe())
project.apply(winner.changes)

winner     run-01ab…0c7d, partner on fork c4e81b7
check      pytest: 14 passed
others     cancelled, receipts kept in history

Program composition

Program & Program -> Program; Program | Program -> Program

Composes boot programs, the pinned descriptors your project exports. & runs programs at once and | runs them in sequence; & binds tighter, so the expression below needs no parentheses. A fork started with a boot program runs it under the fork's grant.

from programs import reproduce, survey, patch, verify

fork = project.fork(boot=reproduce & survey | patch | verify)
print(fork.describe())

boot       (reproduce & survey) | patch | verify
fork       a3d90f1 of quickstart, base main 2f60a91

USD

USD(amount: int | str | Decimal) -> Money

Constructs an exact dollar amount for budgets and prices. Pass a string or a Decimal for cents, so the amount never passes through a float.

from chio import USD

print(USD(2))
print(USD("0.01") * 100)

USD 2.00
USD 1.00

ApplyConflict

class ApplyConflict(ChioError): base: str; current: str

Apply refuses to publish when main moved after the fork was taken. Nothing changes; fork again from the new main and reapply.

from chio import ApplyConflict

try:
    project.apply(fork.changes)
except ApplyConflict as conflict:
    print(conflict)

main moved from 7c2e1d0 to 9e03a4b; fork 51c07ae was not applied

How the Computer reaches the kernel

Each grant is a chio.capability.v1 token signed by your kernel. Each operation in a run is a mediated call, and each event carries the id of a chio.receipt.v1 receipt in the store that history reads. Policy rules compile into the HushSpec policy every grant on the computer is checked against. The Computer and the Kernel maps each Computer object onto its kernel mechanism. The low-level clients below speak to the same kernel and the same records without the Computer layer.

Work

WorkClient buys checked work from another company through your owner work service. You select a partner's offer, prepare the agreement both sides sign, submit the funded commitment, and read the work view until the verifier's decision settles the claim. Your owner service holds every key; the client sends chio.work.v1 documents and never constructs authority. Verifiable Work explains the model.

The client ships in the chio-process distribution and imports from chio_process. Calls are synchronous and return when your owner service has answered. Documents are dictionaries with their wire field names, checked against the TypedDict types the package exports. The client never retries a side effect and never generates a replacement command id.

pip install chio-process

The examples follow Run Verifiable Work: work is the client on ./work, your partner sells an authentication inventory of the pricing service's OpenAPI document for USD 28.00, and json is imported.

WorkClient.open

WorkClient.open(state: str | os.PathLike[str]) -> WorkClient

Connects to the owner work service serving this state directory, over its authenticated local socket. chio work serve must be running; opening dispatches nothing.

from chio_process import WorkClient

work = WorkClient.open("./work")

work.select

work.select(offer: Signed[WorkOffer], *, program_id: str, command_id: str) -> WorkSelection

Selects the exact offer you pass: your owner service signs the selection, the partner's kernel seals a dispatch permit for one slot, and the result's commitment is a draft WorkCommitmentV1 with no funding attached.

with open("partner-offer.json") as f:
    offer = json.load(f)

draft = work.select(offer, program_id="pricing-openapi-auth", command_id="auth-select-1")
with open("auth-draft.json", "w") as f:
    json.dump(draft.commitment, f, indent=2)
print(draft.describe())

offer        openapi-explicit-auth-v1 from did:chio:z6MkpA7…c2T, USD 28.00
selection    signed by did:chio:z6Mkf3…Qp9 at revision 2
permit       sealed by kernel 9b07…d31c for slot auth-inventory
draft        WorkCommitmentV1 written to auth-draft.json, no funding yet

work.prepare

work.prepare(proposal: WorkPreparationV1) -> WorkPreparedV1

Prepares one owner artifact from a chio.work.preparation.v1 document. The proposal is one of Delegate, Offer, Select, Extend, Join, or Agreement, and the outcome is Prepared, Pending, or Refused. An agreement binds the terms both parties sign and moves no money; chio work fund locks the allocation and attaches its WorkFundingRefV1 to the commitment.

prepared = work.prepare({
    "schema": "chio.work.preparation.v1",
    "program_id": "pricing-openapi-auth",
    "command_id": "auth-agree-1",
    "proposal": {"Agreement": {
        "commitment": draft.commitment,
        "funding_profile_id": "work-claim-escrow",
        "terms_reference": "terms/openapi-auth.json",
    }},
})
print(prepared.describe())

preparation  auth-agree-1  Prepared
agreement    7e1d…c04a  SHA-256 of the RFC 8785 body
buyer        did:chio:z6Mkf3…Qp9  signed
provider     did:chio:z6MkpA7…c2T  signed through kernel.halden.example/cosign
verifier     verify.corbel.example  pinned 0x9f62…3ad5
checker      openapi-explicit-auth-v1  c7a1…02be
amount       USD 28.00 to 0x7d0c…12f4
deadlines    submitBy 2026-10-14T18:00Z  challengeUntil 2026-10-15T18:00Z
             resolveBy 2026-10-16T18:00Z  refundAfter 2026-10-17T18:00Z

work.submit

work.submit(command: WorkCommandV1) -> WorkCommandResultV1

Submits one chio.work.command.v1 document under its caller-supplied command id. The action is one of Delegate, Select, Seal, Extend, Submit, Reconcile, or Cancel. The outcome is Applied with a revision, Pending with the owner's operation reference, or Refused. Applied reports that the command was processed, not that the work succeeded. The same command id with the same body returns the original outcome; a different body is refused before dispatch.

with open("auth-commitment.json") as f:
    commitment = json.load(f)

submitted = work.submit({
    "schema": "chio.work.command.v1",
    "program_id": "pricing-openapi-auth",
    "command_id": "auth-submit-1",
    "action": {"Submit": {"commitment": commitment}},
})
print(submitted.describe())

command     auth-submit-1  Applied at revision 4
admitted    by kernel 9b07…d31c under its own policy

WorkCommandResultV1.handle

WorkCommandResultV1.handle: WorkHandleV1

The handle an applied Submit returns: program, slot, allocation hash, receiver kernel, and original request id. It is a reference, not a bearer permit. Reading it on a Pending or Refused outcome raises WorkError; query the original command id instead.

print(json.dumps(submitted.handle, indent=2))

{
  "program_id": "pricing-openapi-auth",
  "slot_id": "auth-inventory",
  "allocation_hash": "a51e…07c3",
  "receiver_kernel_id": "9b07…d31c",
  "request_id": "req-01c4…7a90"
}

work.inspect

work.inspect(handle: WorkHandleV1) -> WorkViewV1

Reads the work view for a handle. It is a query, so it mutates nothing and dispatches no checker.

print(work.inspect(submitted.handle).describe())

handle      pricing-openapi-auth/auth-inventory, request req-01c4…7a90
execution   Available      returned, rcpt-01c4…7a92 signed by kernel 9b07…d31c
acceptance  Pending        verify.corbel.example, recordable after 2026-10-15T18:00Z
results     Available      output 6f0e…b31a, 3 operations
recovery    NotApplicable
settlement  Pending        Submitted, commitment 0x2a90…f1e3
delivery    Available      co-signed by kernels 4f1c…a2e9 and 9b07…d31c

work.query

work.query(query: WorkQueryV1) -> WorkQueryResultV1

Reads by reference: Work by handle, Command or Preparation by original id, and Catalog or Profile for the entries owners provisioned for you. An original-id query answers before you hold a handle, so it resolves a lost response.

found = work.query({"Command": {"command_ref": "auth-submit-1"}})
print(found.describe())

query       Command auth-submit-1
command     auth-submit-1  Applied at revision 4
handle      pricing-openapi-auth/auth-inventory, request req-01c4…7a90

work.collect

work.collect(handle: WorkHandleV1) -> WorkViewV1

Reconciles finished work: it records the settlement and evidence the owning authorities already hold and returns the view. It never dispatches the work again and never returns protected result bytes; read those through the recovery release operation.

view = work.collect(submitted.handle)
print(view.describe())

handle      pricing-openapi-auth/auth-inventory, request req-01c4…7a90
execution   Available      returned, rcpt-01c4…7a92 signed by kernel 9b07…d31c
acceptance  Available      Accepted, predicate match, checker c7a1…02be
results     Available      output 6f0e…b31a, 3 operations
recovery    NotApplicable
settlement  Available      Paid USD 28.00, allocation 0x8c1f…40d2
delivery    Available      co-signed by kernels 4f1c…a2e9 and 9b07…d31c

WorkCommitmentV1

class WorkCommitmentV1(TypedDict): handle: WorkHandleV1; permit: Signed[DispatchPermit]; profile_ref: WorkProfileRefV1; custody_ref: str; graph_ref: str; funding: NotRequired[WorkFundingRefV1]

Links the handle, the sealed permit, the agreed profile, the custody reference for the exact invocation, and the graph. funding names the rail, the agreement digest, and the reserve; the receiver checks the allocation on the rail before it admits the work.

with open("auth-commitment.json") as f:
    commitment = json.load(f)
print(json.dumps(commitment["funding"], indent=2))

{
  "rail": "work-claim-escrow",
  "agreement_digest": "7e1d…c04a",
  "reserve_ref": "0x8c1f…40d2"
}

WorkViewV1

class WorkViewV1: handle: WorkHandleV1; execution: Observation; acceptance: Observation; results: Observation; recovery: Observation; settlement: Observation; delivery: Observation; def describe(self) -> str

Six independent observations. Each is NotApplicable, Pending with the owner's reference, Available with its source, revision, time, and value, or Unavailable with a code. Each authority revises its own observation, so acceptance and settlement can move separately from execution.

print(json.dumps(view.acceptance, indent=2))

{
  "Available": {
    "source": "verify.corbel.example",
    "revision": 1,
    "observed_at": "2026-10-15T18:04Z",
    "value": { "decision": "Accepted", "predicate": "match", "checker": "c7a1…02be" }
  }
}

WorkError

class WorkError(Exception): code: str; command_ref: str | None

A stable code and the original command reference. After a timeout, query that reference; the outcome the owner recorded is the answer.

from chio_process import WorkError

try:
    work.submit(command)
except WorkError:
    found = work.query({"Command": {"command_ref": "auth-submit-1"}})
    print(found.describe())

query       Command auth-submit-1
command     auth-submit-1  Applied at revision 4
handle      pricing-openapi-auth/auth-inventory, request req-01c4…7a90

Low-level session clients

Beneath the Computer API, chio-sdk ships the protocol-level clients: a sidecar client for services that put a colocated kernel in front of their own operations, and a hosted client for agents that open MCP sessions on a kernel edge, such as the pricing service. The import package decides which one you get.

Sidecar client (chio_sdk)

chio_sdk is an async HTTP client to the sidecar kernel plus the typed models for capabilities, scopes, verdicts, receipts, and approvals. The sidecar owns kernel evaluation and receipt signing. The client also checks selected model and response invariants locally; most methods return typed models, while mediated evaluation returns a status dictionary. Its declared dependencies are httpx, pydantic.

Confirm the sidecar connection

Start a configured HTTP sidecar before running this script. CHIO_SIDECAR_URL is optional; the local default is http://127.0.0.1:9090.

import asyncio
import os
from chio_sdk import ChioClient

async def main() -> None:
    async with ChioClient(os.getenv("CHIO_SIDECAR_URL")) as client:
        print(await client.health())

asyncio.run(main())

A health document confirms connectivity. It does not establish that a particular capability, policy, or call will be accepted.

chio_sdk.ChioClient

The constructor takes a base URL (default http://127.0.0.1:9090) and a keyword-only timeout in seconds, default 5. The class is an async context manager, and close() shuts the underlying httpx.AsyncClient down. Every method is a coroutine except the static collect_evidence.

The approvals channel has its own models: PendingApproval (the held call, its policy, subject, capability, tool, parameter hash, expiry, and the triggers that held it), ResolvedApproval (outcome, resolution time, approver key, token id), Approval (one or the other), PendingApprovalList, ApprovalResponse, and SubmitApprovalResult. The verdict enum ApprovalVerdict has the wire values approved and denied, and ApprovalVerdict.from_action accepts the shorthands approve, allow, deny, and reject.

Request pre-execution authorization

evaluate_tool_call_mediated(*, capability, tool_server, tool_name, parameters, request_id=None, governed_intent=None, approval_token=None, dpop_proof=None) -> dict checks a full token and returns the authorization state. The example assumes a configured route and a signed capability supplied by your operator.

import asyncio
import json
import os
from pathlib import Path
from chio_sdk import ChioClient

async def main() -> None:
    capability = json.loads(Path(os.environ["CHIO_CAPABILITY_FILE"]).read_text())
    parameters = json.loads(os.getenv("CHIO_TOOL_ARGUMENTS", "{}"))
    async with ChioClient(os.getenv("CHIO_SIDECAR_URL")) as client:
        result = await client.evaluate_tool_call_mediated(
            capability=capability,
            tool_server=os.environ["CHIO_TOOL_SERVER"],
            tool_name=os.environ["CHIO_TOOL_NAME"],
            parameters=parameters,
        )
        status = result.get("status")
        if status == "authorized":
            print("Authorization reserved; trusted execution and reconciliation remain")
        elif status == "pending_approval":
            print("Approval required; no execution authorized yet")
        elif status == "deny":
            print("Request denied; inspect the returned receipt and reason")
        else:
            raise RuntimeError("Unexpected authorization status")

asyncio.run(main())

The authorized response includes an execution nonce and a reservation receipt. That receipt is intentionally non-authoritative until reconciliation. A policy requiring approval or proof of possession also needs the corresponding request-bound token or proof.

Reconcile from the trusted tool server

After execution, the trusted tool server calls reconcile_mediated_authorization(*, control_token, execution_nonce, arguments, realized_cost). It submits the unchanged nonce and matching arguments with realized cost; the sidecar settles the hold and produces the reconciled receipt. The agent does not hold the reconciliation control credential. See sidecar client source for the complete request and state contract.

Testing without a sidecar

chio_sdk.testing ships a drop-in MockChioClient with allow_all(), deny_all(), and with_policy(...), so capability-checked code paths run with no sidecar. It records calls as RecordedCall and takes a MockVerdict.

async def test_allow_all_permits_every_tool_call() -> None:
    async with allow_all() as chio:
        receipt = await chio.evaluate_tool_call(
            capability_id="cap-1",
            tool_server="srv",
            tool_name="read",
            parameters={"path": "/tmp"},
        )
    assert receipt.is_allowed
    assert receipt.decision.verdict == "allow"
    assert receipt.tool_name == "read"

These are application test doubles. In particular, allow_all().evaluate_tool_call(...) returns a synthetic receipt, while the real id-only method always refuses execution authorization. A passing mock test does not verify a kernel integration.

Protecting a FastAPI application

chio-asgi puts the sidecar in front of any ASGI framework. It evaluates every request before it reaches the application, returns the receipt id on X-Chio-Receipt, and denies rather than passes when the sidecar is unreachable.

from chio_asgi import ChioASGIMiddleware, ChioASGIConfig

# Starlette / FastAPI
app.add_middleware(
    ChioASGIMiddleware,
    config=ChioASGIConfig(sidecar_url="http://127.0.0.1:9090"),
)

The worked application is examples/hello-fastapi. It excludes /healthz from evaluation, allows GET /hello, and requires a capability on POST /echo.

def build_chio_config(sidecar_url: str | None = None) -> ChioASGIConfig:
    return ChioASGIConfig(
        sidecar_url=sidecar_url
        or os.environ.get("CHIO_SIDECAR_URL", "http://127.0.0.1:9090"),
        exclude_paths=frozenset({"/healthz"}),
    )

The example's smoke run starts the trust control plane, the application, and a chio api protect sidecar, then walks the three routes. On a refusal the middleware sends a JSON body of error (the guard name, or ChioGuard when the verdict names none), message (the verdict reason, or denied), and status, with the receipt id on the configured receipt header. The status is the verdict's own HTTP status when it carries one and 403 otherwise.

$ cat hello.headers hello.json

HTTP/1.1 200 OK
date: Sat, 05 Sep 2026 11:49:07 GMT
server: uvicorn
content-length: 32
content-type: application/json
x-chio-receipt: 1dba0f38326d8eaba47d2b8e50a54fdeb0da056639cbaa20fca0100e0dc67f64

{"message":"hello from fastapi"}

$ cat deny.headers deny.json

HTTP/1.1 403 Forbidden
date: Sat, 05 Sep 2026 11:49:08 GMT
server: uvicorn
content-type: application/json
content-length: 103
x-chio-receipt: 5d31d1463fe3819b3a5ffae0fb89e01c92339c3e4da455e6701511509690aaa3

{"error": "CapabilityGuard", "message": "side-effect route requires a capability token", "status": 403}

Cognition market

chio_sdk.cognition_market carries the Finding market clients, and chio_sdk.finding the buyer-local bid ceiling. Both are re-exported from chio_sdk. Verification is not reimplemented in Python: the buyer shells out to the chio binary and parses its JSON, so the Rust verifier stays the only judge of a bundle. Each client is an async context manager over its own httpx.AsyncClient.

CognitionMarketBuyer

Constructed from a buyer profile path plus keyword-only chio_binary, status_floor_path, timeout (default 30 seconds), and transport. The profile must carry the schema chio.finding.buyer-client.v1, and the bearer token on it goes on every request. The status floor defaults to the profile path with .status-floor.json appended.

from chio_sdk import (
    CognitionMarketBuyer,
    CognitionMarketError,
    PurchasedVerifiedFix,
    VerifiedFindingProof,
)


async def buy(finding_id: str) -> PurchasedVerifiedFix:
    async with CognitionMarketBuyer("./buyer-profile.json") as buyer:
        verified: VerifiedFindingProof = await buyer.verified_proof(finding_id)

        report = await buyer.status(finding_id)
        if report["status"] != "live":
            raise CognitionMarketError("finding is retracted")

        return await buyer.purchase_verified_fix(
            verified,
            max_price_units=300,
            currency="USD",
        )

CognitionMarketSeller

Constructed from a credential path carrying the schema chio.finding.seller-client.v1. The default deadline is 720 seconds, sized for the operator's bounded packaging sandbox.

• package_verified_fix(*, repository, base, candidate, tests, topic, price=300, output=None) builds the submission identity and derives its requestId as SHA-256 over the domain-separated canonical JSON. The price must fall between 1 and the verified-fix sale exposure of 450 units, the repository must be an absolute normalized operator-side path, and a caller-supplied output is refused because the package is operator-owned.
• admit(package) posts to POST /v1/findings/operator/verified-fixes.
• retract(finding_id) posts chio.finding.voluntary-retraction-request.v1 to POST /v1/findings/operator/retractions, with a request id derived from the retraction domain and the Finding id.

HostedCognitionMarketClient

The multi-tenant transport. The constructor is positional in endpoint, tenant_id, api_key_id, and api_key_secret, with keyword-only timeout and transport. The endpoint must be an https: URL with no embedded credentials.

• publish(finding, request_id, idempotency_key): POST /v1/findings/publish.
• mutate(operation, mutation, request_id): POST /v1/findings/events/{operation}, where the operation is one of listing, delivery, challenge, verified-fix, retraction, and penalty. The mutation accepts only aggregateId, eventId, expectedRevision, expectedEventSha256, artifactSignerKey, and payload; any other key is refused, and the event id doubles as the idempotency key.
• finding(finding_id, request_id): GET /v1/findings/{id}.
• findings(request_id, *, after=None, limit=None): GET /v1/findings, with a limit from 1 through 100.

Every failure on all three clients raises CognitionMarketError, a RuntimeError subclass outside the ChioError tree: a non-2xx response, an oversized body, a timeout, a profile whose schema does not match, and a verifier answer that names a different Finding.

finding_bid_ceiling

finding_bid_ceiling computes a buyer-local ceiling with exact integer arithmetic and returns it as a decimal string. It authenticates nothing. What it does is bind one caller-carried estimate to the buyer's own expected source, context, and replay-recipe digests, to the buyer's currency, and to a validity window, then discount the estimate by three basis-point factors and cap the result at the remaining budget. It rounds down once, after the combined product.

from chio_sdk import (
    BuyerFindingEstimate,
    FindingBidCeilingError,
    FindingBidCeilingInput,
    FindingBidCeilingPolicy,
    finding_bid_ceiling,
)

estimate: BuyerFindingEstimate = {
    "units": "1200",
    "currency": "USD",
    "provenance": "buyer_metering_history_v1",
    "sourceSha256": expected_source_sha256,
    "contextSha256": expected_context_sha256,
    "replayRecipeSha256": expected_replay_recipe_sha256,
    "observedAtUnixMs": "1744500000000",
    "validUntilUnixMs": "1744503600000",
}

policy: FindingBidCeilingPolicy = {
    "budgetRemainingUnits": "5000",
    "currency": "USD",
    "wouldHaveRunBps": "9000",
    "siblingRedundancyBps": "1500",
    "guaranteeClassBps": "10000",
}

payload: FindingBidCeilingInput = {
    "estimate": estimate,
    "policy": policy,
    "expectedSourceSha256": expected_source_sha256,
    "expectedContextSha256": expected_context_sha256,
    "expectedReplayRecipeSha256": expected_replay_recipe_sha256,
    "nowUnixMs": "1744501000000",
}

try:
    ceiling = finding_bid_ceiling(payload)
except FindingBidCeilingError as error:
    print("refused:", error.code)

FindingBidCeilingError is a ValueError subclass carrying a code: invalid_decimal, u64_overflow, basis_points_out_of_range, currency_mismatch, provenance_unsupported, source_substituted, context_substituted, replay_recipe_substituted, digest_malformed, invalid_validity_window, or stale_estimate. Only two provenance values are accepted: buyer_metering_history_v1 and buyer_fresh_metered_quote_v1.

Companion distributions

Framework adapters, orchestrator operators, agent-framework bindings, the AWS Lambda runtime, and the WASM guard guest SDK ship as separate distributions on PyPI so a base install stays small. Install one with pip install chio-fastapi or its own name. Each uses a distinct top-level module (for example import chio_fastapi, not chio.fastapi). Every distribution the sdks dataset reads is below, with the version and the dependencies its own manifest declares.

The dependency column decides which of these are Chio clients at all. Every companion depends on chio-sdk and uses its sidecar package, except chio-bedrock, chio-guard-py. A companion governs a service's own operations; agents that call a service use the hosted client or the Computer API. chio-adapter-base is the shared adapter base the orchestrator and agent-framework packages build on, and chio-asgi is the one chio-fastapi builds on.

Companions integrate evaluation at different framework boundaries. Inspect the adapter’s return and exception contract; request middleware and agent wrappers do not all report refusal the same way. chio_fastapi exposes route decorators (chio_requires, chio_approval, chio_budget) plus dependency-injection helpers (get_chio_client, get_chio_receipt, get_caller_identity).

from fastapi import FastAPI, Request
from chio_fastapi import chio_requires

app = FastAPI()

@app.post("/v1/refund")
@chio_requires("pricing", "refund", ["Invoke"])
async def refund(request: Request):
    # request.state.chio_receipt is an HttpReceipt attached on success.
    receipt = request.state.chio_receipt
    return {"status": "refunded", "receipt_id": receipt.id}

Hosted MCP client (chio)

The rest of this page documents the hosted session client in chio. Its declared dependencies are httpx, pure25519.

The same package provides hosted transport, local cryptographic verification, and the Computer API.

from chio import ChioClient, ChioSession, ReceiptQueryClient
from chio.invariants import verify_receipt, verify_capability
from chio.auth import discover_oauth_metadata, perform_authorization_code_flow

The chio package exposes the submodules auth, client, errors, invariants, models, nested, receipt_query, session, transport, and version. Its __all__ re-exports the client and session types (ChioClient, ChioSession, SessionHandshake, TransportResponse, initialize_session, rpc_result), the receipt query types (ReceiptQueryClient, ReceiptQueryParams, ReceiptQueryResponse), the five errors, the OAuth helpers (StaticBearerAuth, static_bearer_auth, authorization_server_metadata_url, discover_oauth_metadata, exchange_access_token, get_json, perform_authorization_code_flow, pkce_challenge, resolve_oauth_access_token), the nested-callback router (NestedCallbackRouter, elicitation_accept_result, roots_list_result, sampling_text_result), parse_json_text, and __version__.

chio.ChioClient

chio.ChioClient opens authenticated Chio MCP HTTP sessions. Both import packages export a class called ChioClient, and they are different classes: this one takes a base URL and returns a session, chio_sdk.ChioClient takes a sidecar URL and returns receipts. The with_static_bearer classmethod is the shortest path for development; a deployment constructs the client with the token string obtained through its OAuth flow. The base URL is the edge origin: the transport appends /mcp to it on every request. initialize() is keyword-only in capabilities, client_info, on_message, and protocol_version, which defaults to 2025-11-25.

Open a session and discover tools

This is an outside buyer's agent opening a session on the pricing service from Quickstart part 2. CHIO_AUTH_TOKEN holds the token for the grant you issued to the buyer.

import os
from chio import ChioClient

client = ChioClient.with_static_bearer(
    base_url="https://127.0.0.1:9440/pricing",
    auth_token=os.environ["CHIO_AUTH_TOKEN"],
)
session = client.initialize(
    client_info={"name": "basalt-buyer", "version": "1.0.0"},
)
try:
    response = session.list_tools()
    for tool in response["result"].get("tools", []):
        print(tool["name"], tool.get("description", ""))
finally:
    session.close()

The listing names only the tools the grant covers: quote, not refund.

ChioSession

ChioSession exposes the MCP method set plus lower-level entry points for custom JSON-RPC work. It numbers its own request ids.

• list_tools(), call_tool(name, arguments=None)
• list_resources(), read_resource(uri), subscribe_resource(uri), unsubscribe_resource(uri), list_resource_templates()
• list_prompts(), get_prompt(name, arguments=None)
• complete(params) for completion/complete, and set_log_level(level) for logging/setLevel
• list_tasks(), get_task(task_id), get_task_result(task_id), cancel_task(task_id)
• request(), request_result(), notification(), send_envelope()
• close(), which returns the HTTP status of the session delete

The high-level MCP methods return the terminal JSON-RPC message, so the MCP payload is under result. The TypeScript session unwraps it; this one does not. The lower-level request, notification, and send_envelope return TransportResponse; close returns an HTTP status.

Distinguish RPC refusal from tool failure

Use this fragment inside the session’s try block. JSON-RPC errors, including a refusal, raise ChioRpcError; tool-level failures arrive in result.isError. The buyer's grant allows quote; the same call to refund raises with the refusal and its receipt id in error.data.

from chio import ChioRpcError

try:
    response = session.call_tool("quote", {"sku": "sku-118", "percent": 10})
except ChioRpcError as error:
    print("RPC rejected:", error.code, str(error), error.data)
else:
    result = response["result"]
    if result.get("isError"):
        print("Tool reported failure:", result.get("content"))
    else:
        print("Tool result:", result)

Invariants

chio.invariants holds the pure verification primitives. Every function is synchronous, dependency-light, and safe to call from any thread or async context. This is its whole __all__.

from chio.invariants import (
    canonicalize_json,
    canonicalize_json_string,
    sha256_hex_bytes,
    sha256_hex_utf8,
    is_valid_public_key_hex,
    is_valid_signature_hex,
    public_key_hex_matches,
    sign_utf8_message_ed25519,
    verify_utf8_message_ed25519,
    sign_json_string_ed25519,
    verify_json_string_signature_ed25519,
    verify_chio_signature,
    parse_receipt_json,
    receipt_body_canonical_json,
    receipt_signing_body_canonical_json,
    verify_receipt,
    verify_receipt_json,
    verify_receipt_with_trusted_signers,
    parse_capability_json,
    capability_body_canonical_json,
    capability_signing_body_canonical_json,
    verify_capability,
    verify_capability_json,
    parse_signed_manifest_json,
    signed_manifest_body_canonical_json,
    verify_signed_manifest,
    verify_signed_manifest_json,
)

Canonical JSON

from chio.invariants import canonicalize_json_string

canonical = canonicalize_json_string('{"b":2,"a":1}')
assert canonical == '{"a":1,"b":2}'

Receipts and capabilities

These receipt helpers support Ed25519 records without BBS material. Their signing projection omits BBS fields; use the Rust verifier for BBS-bound receipts. Parsing a JSON object does not validate the complete receipt schema, so validate its wire shape at ingestion.

Supply separate receipt and capability JSON documents. Obtain kernel_key_hex independently from trusted operator configuration.

import time
from chio.invariants import (
    parse_capability_json,
    parse_receipt_json,
    verify_capability,
    verify_receipt,
    verify_receipt_with_trusted_signers,
)

receipt = parse_receipt_json(receipt_json)
result = verify_receipt(receipt)
assert result["signature_valid"]
assert result["parameter_hash_valid"]
assert result["receipt_id_valid"]

# Pass the kernel keys you accept when you want an authorization answer.
trusted = verify_receipt_with_trusted_signers(receipt, [kernel_key_hex])
if not trusted["ok"]:
    raise ValueError("Receipt integrity or signer trust failed")
print("Authorizing receipt:", trusted["authorized"])

cap = parse_capability_json(capability_json)
status = verify_capability(cap, int(time.time()))
assert status["signature_valid"]
assert status["delegation_chain_shape_valid"]
print("Time valid:", status["time_valid"], status["time_status"])

Ed25519

Ed25519 signing and verification use pure25519, a pure-Python implementation chosen so the distribution installs on a restricted runner (Lambda, App Engine, a serverless container) without a native toolchain. Ed25519 is the one algorithm this build verifies. verify_chio_signature dispatches on the signature prefix and raises ChioInvariantError with code invalid_signature for a p256:, p384:, or hybrid: signature, so a receipt this build cannot check fails closed rather than passing.

sign_json_string_ed25519(input_text, seed_hex) canonicalizes the input first and returns a dict with canonical_json, public_key_hex, and signature_hex.

from chio.invariants import sign_json_string_ed25519, verify_json_string_signature_ed25519

sig = sign_json_string_ed25519('{"key":"value"}', seed_hex)
ok = verify_json_string_signature_ed25519(
    '{"key":"value"}',
    sig["public_key_hex"],
    sig["signature_hex"],
)
assert ok

ReceiptQueryClient

ReceiptQueryClient wraps GET /v1/receipts/query, injects the bearer token, and exposes both a one-shot and an iterator API. ReceiptQueryParams is a total-false TypedDict in camelCase.

from chio import ReceiptQueryClient

client = ReceiptQueryClient(
    base_url="https://receipts.example.com",
    auth_token="service-token",
)

response = client.query({
    "capabilityId": "cap-01a8…31f2",
    "agentSubject": agent_public_key_hex,
    "toolServer": "workspace",
    "toolName": "read_file",
    "outcome": "deny",
    "since": 1744500000,
    "until": 1744600000,
    "minCost": 1,
    "maxCost": 1000,
    "costCurrency": "USD",  # three uppercase letters; required with a cost bound
    "limit": 50,
})

print(response["totalCount"], response.get("nextCursor"))

Use paginate() to drive the numeric cursor automatically. The wire contract, including the values outcome accepts and the server's pagination limits, is on the Receipt Query API page.

for page in client.paginate({"toolServer": "workspace"}):
    for receipt in page:
        # decision is present only on mediated_decision receipts; trace and
        # advisory receipts omit it, and it is a tagged union when present.
        print(receipt["id"], receipt.get("decision"))

OAuth and PKCE

chio.auth ships the helpers that discover a chio edge's OAuth configuration, drive an authorization-code flow with PKCE, and run a token exchange. Both flow helpers are keyword-only and return a bare access-token string. discover_oauth_metadata returns a dict carrying protected_resource_metadata and authorization_server_metadata.

from chio.auth import (
    discover_oauth_metadata,
    perform_authorization_code_flow,
    exchange_access_token,
)

metadata = discover_oauth_metadata("https://edge.example.com")

access_token = perform_authorization_code_flow(
    base_url="https://edge.example.com",
    auth_scope="chio.read chio.invoke",
    authorization_server_metadata=metadata["authorization_server_metadata"],
    client_id="chio-cli",
    redirect_uri="http://127.0.0.1:53682/callback",
)

exchanged = exchange_access_token(
    base_url="https://edge.example.com",
    auth_scope="chio.read chio.invoke",
    authorization_server_metadata=metadata["authorization_server_metadata"],
    access_token=access_token,
)

Error types

The two distributions carry two error trees with the same base name and no shared class.

In chio:

• ChioError: base SDK exception.
• ChioTransportError: SDK-detected transport failure. Underlying HTTP-client exceptions can also propagate; handle those at the application boundary.
• ChioQueryError: non-2xx response from the receipt query endpoint.
• ChioRpcError: JSON-RPC error returned by the hosted MCP edge.
• ChioInvariantError: parsing or verification failure in the invariants layer.

In chio_sdk.errors, a different set. Transport failures and invalid responses raise. A kernel denial can instead be a returned verdict or status: inspect evaluate_http_request(...).verdict.is_allowed and evaluate_tool_call_mediated(...)["status"]. Absence of an exception is not permission.

• ChioError: base SDK exception.
• ChioConnectionError: the sidecar could not be reached.
• ChioDeniedError: the kernel refused. This is the expected outcome of an id-only evaluate_tool_call.
• ChioTimeoutError: the sidecar did not answer in time.
• ChioValidationError: the request or the response failed a model invariant.

CognitionMarketError and FindingBidCeilingError sit outside both trees: the first extends RuntimeError, the second ValueError.

from chio import ChioError, ChioQueryError, ChioTransportError

try:
    response = client.query({"capabilityId": "cap_abc"})
except ChioQueryError as err:
    print("query failed", err.status, err)
except ChioTransportError as err:
    print("network failed", err)
except ChioError as err:
    print("chio error", err)

Source

The sidecar source defines evaluation, approvals, reconciliation, market clients, and models. The hosted source defines the Computer API, MCP sessions, local invariants, OAuth helpers, and receipt queries. Distribution tables render their manifests.

The Protocol Reference and Bindings API define the shared contracts.

Conformance

The Python SDK runs against the cross-language conformance vectors: canonical JSON output, SHA-256 digests, Ed25519 signatures, receipt verification, capability verification, and manifest verification all produce byte-identical results against the Rust reference.

chio conformance run --peer python

The installed CLI runs the vectors against the published package, so a pass covers the same wheel pip install chio-sdk installs.

Related