Python SDK

halvorai-ingest is the Python client. It signs events the same way as the TypeScript SDK, so an event signed in one language verifies in the other. It needs Python 3.10 or later and is open source (Apache-2.0).

pip install halvorai-ingest

Scope in this release

Included: the client with batching and signing, agent context, redaction, local instruction hashing, a sink for self-hosting, the x402 wrapper for httpx, and emit(). Not yet included: the Stripe, MPP and MCP adapters. For those, build the event yourself and pass it to emit(), or use the TypeScript SDK.

Create a client and emit

import os
from halvorai_ingest import HalvorClient, agent_context

halvor = HalvorClient(
    api_key=os.environ["HALVOR_API_KEY"],
    source_key=os.environ["HALVOR_SOURCE_KEY"],
)

with agent_context(
    agent={"id": "agt_research_07", "name": "Market research agent"},
    owner={"id": "own_acme", "type": "business"},
    instruction={"id": "task_1", "text": "Pull the latest EU power prices"},
):
    halvor.emit({
        "type": "payment.settled",
        "payment": {
            "payment_id": "pi_3QxAbCdEfGhIjKlM",
            "rail": "stripe",
            "amount": {"value": "41200", "unit": "minor", "currency": "USD"},
            "status": "settled",
            "rail_ref": {"payment_intent": "pi_3QxAbCdEfGhIjKlM"},
        },
        "payee": {"type": "merchant", "name": "Example Airways"},
    })

halvor.flush()

emit returns the event id. The agent, owner and instruction come from agent_context, which uses Python’s contextvars, so it follows async code and threads started inside the block. The instruction text is hashed locally.

Client options

Argument	Default	Purpose
api_key	none	Bearer key, hk_test_... or hk_live_....
source_key	none	Private Ed25519 JWK as a JSON string or a dict. Its kid is your source id.
environment	"test"	test or live.
endpoint	the API URL	Override the base URL.
sink	none	A function that receives signed envelopes instead of sending them.
batch_size	100	Events per request.
on_error	none	A function called with an error dict.

Call flush() before a short-lived process exits and close() when you are done. stats() returns counters for emitted, accepted, duplicate, rejected and dropped events.

x402 with httpx

Wrap the httpx client that already pays, and HalvorAI observes the x402 headers. It stays outside the payment.

import httpx
from halvorai_ingest.x402 import wrap_httpx

paying_client = httpx.Client()          # your x402-capable client
client = wrap_httpx(paying_client, halvor)

with agent_context(agent={"id": "agt_research_07"}, owner={"id": "own_acme", "type": "business"}):
    response = client.get("https://api.example.com/premium-data")

It emits payment.requested when a 402 with PAYMENT-REQUIRED arrives, payment.authorized for a request carrying PAYMENT-SIGNATURE, and payment.settled or payment.failed from PAYMENT-RESPONSE. See the x402 guide for the event details.

Redaction and hashing helpers

from halvorai_ingest import redact_text, hash_instruction, evidence_sha256

redact_text("card 4242 4242 4242 4242")   # "card [redacted:pan]"
hash_instruction("Book a flight")          # "sha256:..."

Self-hosting

store = []
halvor = HalvorClient(source_key=os.environ["HALVOR_SOURCE_KEY"], environment="test",
                      sink=lambda envelopes: store.extend(envelopes))