Python example
examples/python/axiom_agent.py is a single-file agent that speaks the action contract directly, without the TypeScript SDK. It depends only on cryptography and requests, and it is meant to be read: every rule of the contract is visible in a few lines. It is a reference client, not a supported Python SDK (a second SDK will be prioritised by pilots).
What it demonstrates
| Contract rule | Where in axiom_agent.py |
|---|---|
private_key_jwt assertion + DPoP at the token | AxiomAgent.token |
DPoP proof with ath on every call | AxiomAgent.dpop, AxiomAgent.request |
| Lost response → same body, same operation key | AxiomAgent.submit |
Bounded wait, retry_after_seconds | AxiomAgent.wait_while |
| Persist ID, actor and deadline before returning | save_pending |
| Resume only when approved, as the same agent | follow, the resume command's agent_id check |
| After dispatch: read status, never resubmit | follow (IN_FLIGHT) |
| Receipt for offline verification | the receipt command |
Only ES256 (P-256) keys are handled; axiom keygen produces those by default. EdDSA works the same way with Ed25519PrivateKey.
Run it against local development
Start the API and the sample provider from the repository root:
shnpm run dev-keys -w @axiom/api # once: local signing keys npm run db:migrate:local -w @axiom/api npm run db:migrate:local -w @axiom/sample-billing npm run dev:api # http://localhost:8787 npm run dev:sample # http://localhost:8788In the console (
npm run dev:web,http://localhost:5173), create a project with the sample enabled. Note the sample tenant ID, service ID and billing agent ID.Generate a key and enroll its public key for the billing agent with proof of possession (see registering and admitting agents):
shnode packages/sdk/bin/axiom.mjs keygen --agent agt_… export AXIOM_SESSION_COOKIE='…' # axiom_session cookie of your signed-in console session node packages/sdk/bin/axiom.mjs agent bind agt_…Run the example:
shcd examples/python python3 -m venv .venv && . .venv/bin/activate pip install -r requirements.txt export AXIOM_API_URL=http://localhost:8787 export AXIOM_AGENT_ID=agt_… export AXIOM_KEY_FILE=~/.config/axiom/keys/KID.jwk export AXIOM_TENANT=ten_… export AXIOM_SERVICE=svc_… python axiom_agent.py submit --amount 25000 # needs finance approvalThe refund is above the sample's USD 500 threshold, so after the warm wait the script prints the pending action ID, who can review it and how to check it. The record (action ID, operation key, approved parameters, deadline) is saved to
axiom-pending.json(0600; override withAXIOM_STATE_FILE).Approve it on the approval page, then:
shpython axiom_agent.py status act_… python axiom_agent.py resume act_… python axiom_agent.py receipt act_… node ../../packages/sdk/bin/axiom.mjs verify receipt-act_….jsonA refund of USD 500 or less (
--amount 120) runs straight through within the standing grant.
Commands
| Command | Options |
|---|---|
submit | --amount (default 25000), --currency (USD), --beneficiary (acct_acme_ops), --key (reuse to retry the same intent), --wait (warm-wait budget, default 20 s, capped by the server's 25 s). Always against sample invoice 417. |
status ACTION_ID | Read status. |
resume ACTION_ID | Resume from the saved record; refuses if the record belongs to another agent. |
receipt ACTION_ID | Save receipt-ACTION_ID.json for offline verification. |
The core of it
Token request (abridged from AxiomAgent.token):
now = int(time.time())
assertion = self.key.sign(
{"typ": "JWT", "kid": self.key.kid},
{"iss": self.agent_id, "sub": self.agent_id, "aud": self.token_endpoint,
"jti": secrets.token_urlsafe(16), "iat": now, "exp": now + 60},
)
res = self.http.post(
self.token_endpoint,
headers={"DPoP": self.dpop("POST", self.token_endpoint)},
data={
"grant_type": "client_credentials",
"client_id": self.agent_id,
"client_assertion_type": "urn:ietf:params:oauth:client-assertion-type:jwt-bearer",
"client_assertion": assertion,
"scope": "actions:submit actions:read",
},
)DPoP proof (from AxiomAgent.dpop). ECDSA signatures from cryptography are DER; JWS needs the raw r || s form, which Es256Key.sign converts:
claims = {"jti": secrets.token_urlsafe(16), "htm": method.upper(), "htu": strip_query(url), "iat": int(time.time())}
if access_token:
claims["ath"] = b64url(hashlib.sha256(access_token.encode()).digest())
return self.key.sign({"typ": "dpop+jwt", "jwk": self.key.public_jwk}, claims)Driving an action (from follow): wait while pending, resume once if approved, and after dispatch only read status:
if s["state"] == "pending_approval":
s = agent.wait_while(s["action_id"], {"pending_approval"}, budget)
if s["state"] == "approved":
approved = {k: record[k] for k in ("operation_key", "resource", "params")}
try:
s = agent.resume(s["action_id"], approved)
except (requests.ConnectionError, requests.Timeout):
s = agent.status(s["action_id"]) # the resume may have reached the server: read, do not resume again
if s["state"] in {"executing", "outcome_unknown"}:
s = agent.wait_while(s["action_id"], {"executing", "outcome_unknown"}, budget)A 401 invalid_token is retried once with a fresh token, because authentication fails before any handler runs. A 409 prints changed_fields.