Python SDK

pip install aira-sdk

from aira import Aira, AiraError

aira = Aira(api_key="aira_live_...")

try:
    auth = aira.authorize(
        action_type="wire_transfer",
        details="Send 75,000 EUR to vendor X",
        agent_id="payments-agent",
    )
except AiraError as e:
    if e.code == "POLICY_DENIED":
        # A policy blocked this action. Do not execute.
        log.error(f"Blocked: {e.message}")
        return
    raise

if auth.status == "authorized":
    # Execute the actual action
    result = send_wire(75000, to="vendor")

    # Step 2: notarize the outcome and mint the receipt
    receipt = aira.notarize(
        action_uuid=auth.action_uuid,
        outcome="completed",
        outcome_details=f"Sent successfully. ref={result.id}",
    )
    print(receipt.signature)  # Ed25519 signature

elif auth.status == "pending_approval":
    # Held for human review. Agent must wait.
    # When the human approves, the action moves to status="approved"
    # and the agent can call notarize() (e.g., from a webhook handler).
    queue.enqueue(auth.action_uuid)

# Step 1: request authorization
auth = aira.authorize(
    action_type: str,
    details: str,
    agent_id: str | None = None,
    agent_version: str | None = None,
    instruction_hash: str | None = None,
    model_id: str | None = None,
    model_version: str | None = None,
    parent_action_uuid: str | None = None,
    endpoint_url: str | None = None,
    store_details: bool = False,
    idempotency_key: str | None = None,
    require_approval: bool = False,
    approvers: list[str] | None = None,
)
# Returns Authorization with:
#   .action_uuid        (UUID string)
#   .status           ("authorized" | "pending_approval")
#   .created_at
#   .request_id
#   .warnings         (list[str] | None)
#
# Raises AiraError(code="POLICY_DENIED") on policy deny.

# Step 2: notarize the outcome after execution
receipt = aira.notarize(
    action_uuid: str,
    outcome: str = "completed",   # "completed" or "failed"
    outcome_details: str | None = None,
)
# Returns ActionReceipt with:
#   .action_uuid
#   .status           ("notarized" | "failed")
#   .receipt_uuid       (None when failed)
#   .payload_hash     (None when failed)
#   .signature        (Ed25519, None when failed)
#   .timestamp_token  (RFC 3161, may be None on TSA failure)
#   .created_at
#   .request_id
#   .warnings

from aira import Aira, AiraError

try:
    auth = aira.authorize(action_type="wire_transfer", details="...")
except AiraError as e:
    if e.code == "POLICY_DENIED":
        # Action blocked by policy. Do not execute.
        log.error(f"Blocked by policy: {e.message}")
        log.error(f"Policy ID: {e.details.get('policy_uuid')}")
        log.error(f"Action audit row: {e.details.get('action_uuid')}")
        return
    if e.code == "DUPLICATE_REQUEST":
        # Idempotency key collision
        return
    if e.code == "ENDPOINT_NOT_WHITELISTED":
        # Endpoint not on the org whitelist
        return
    raise

from aira import AsyncAira, AiraError

async with AsyncAira(api_key="aira_live_...") as aira:
    try:
        auth = await aira.authorize(
            action_type="contract_signed",
            details="Signed vendor agreement #1234",
            agent_id="procurement-agent",
        )
    except AiraError as e:
        if e.code == "POLICY_DENIED":
            return
        raise

    if auth.status == "authorized":
        result = await execute_contract_signing()
        await aira.notarize(
            action_uuid=auth.action_uuid,
            outcome="completed",
            outcome_details=f"Signed. doc_id={result.id}",
        )

# Register
agent = aira.register_agent(
    agent_slug="support-agent-v2",
    display_name="Customer Support Agent",
    capabilities=["email", "chat", "tickets"],
    public=True,
)

# Publish version
aira.publish_version(
    slug="support-agent-v2",
    version="1.0.0",
    model_id="claude-sonnet-4-6",
    changelog="Initial release",
)

# Update
aira.update_agent("support-agent-v2", description="Updated description")

# List versions
versions = aira.list_versions("support-agent-v2")

# Decommission
aira.decommission_agent("old-agent")

# Transfer ownership
aira.transfer_agent("my-agent", to_org_uuid="org-uuid", reason="M&A")

# Create sealed evidence package
package = aira.create_evidence_package(
    title="Q1 2026 Lending Audit Trail",
    action_uuids=["act-uuid-1", "act-uuid-2", "act-uuid-3"],
    description="All lending decisions for BaFin review",
)

# Time-travel query
result = aira.time_travel(
    point_in_time="2026-03-20T00:00:00Z",
    agent_slug="lending-agent",
)

# Liability chain
chain = aira.liability_chain("act-uuid", max_depth=10)

# Set succession plan
aira.set_agent_will(
    slug="support-agent",
    successor_slug="support-agent-v3",
    succession_policy="transfer_to_successor",
    data_retention_days=2555,
    notify_emails=["compliance@acme.com"],
    instructions="Transfer conversation history. Delete PII after 7 years.",
)

# Get will
will = aira.get_agent_will("support-agent")

# Issue death certificate (agent must be decommissioned first)
aira.decommission_agent("old-agent")
cert = aira.issue_death_certificate("old-agent", reason="Replaced by v3")

# Compliance snapshot
snapshot = aira.create_compliance_snapshot(
    framework="eu-ai-act",
    agent_slug="lending-agent",
    findings={"art_12_logging": "pass", "art_14_oversight": "pass"},
)

# Create account
account = aira.create_escrow_account(
    purpose="Vendor contract #4521",
    currency="EUR",
)

# Record commitment
aira.escrow_deposit(account.id, amount=5000.00, description="Liability commitment")

# Release
aira.escrow_release(account.id, amount=5000.00)

# Dispute
aira.escrow_dispute(account.id, amount=5000.00, description="Agent error in contract terms")

response = aira.ask("How many email actions were notarized this week?")
print(response["content"])
print(response["tools_used"])  # ["count_actions"]

result = aira.verify_action("action-uuid")
print(result.valid)     # True
print(result.message)   # "Action receipt exists and signing key is valid."

# DID is assigned on registration
agent = aira.register_agent(
    agent_slug="my-agent",
    display_name="My Agent",
    capabilities=["email", "chat"],
)
# agent.did → "did:web:airaproof.com:agents:my-agent"

# Rotate keys
aira.rotate_agent_keys("my-agent")

# Get the agent's W3C Verifiable Credential
vc = aira.get_agent_credential("my-agent")

# Verify it
result = aira.verify_credential(vc)
print(result["valid"])  # True

# Revoke
aira.revoke_credential("my-agent", reason="Agent deprecated")

# Agent A initiates
request = aira.request_mutual_sign(
    action_uuid="act-uuid",
    counterparty_did="did:web:partner.com:agents:their-agent"
)

# Agent B completes
receipt = aira.complete_mutual_sign(
    action_uuid="act-uuid",
    did="did:web:partner.com:agents:their-agent",
    signature="z...",
    signed_payload_hash="sha256:..."
)

rep = aira.get_reputation("my-agent")
print(rep["score"])  # 84
print(rep["tier"])   # "Verified"

aira.set_endpoint_policy(
    mode="strict",
    whitelist=["https://api.stripe.com", "https://api.openai.com"],
)

from aira import Aira
from aira.extras.langchain import AiraCallbackHandler

aira = Aira(api_key="aira_live_xxx")
handler = AiraCallbackHandler(client=aira, agent_id="research-agent", model_id="gpt-5.2")

# Every tool call and chain completion gets a signed receipt
result = chain.invoke({"input": "Analyze Q1 revenue"}, config={"callbacks": [handler]})

from aira import Aira
from aira.extras.crewai import AiraCrewHook

aira = Aira(api_key="aira_live_xxx")
callbacks = AiraCrewHook.for_crew(client=aira, agent_id="research-crew")

crew = Crew(
    agents=[researcher, writer],
    tasks=[research_task, write_task],
    **callbacks,  # task_callback + step_callback — each notarized
)
crew.kickoff()

from aira import Aira
from aira.extras.openai_agents import AiraGuardrail

aira = Aira(api_key="aira_live_xxx")
guardrail = AiraGuardrail(client=aira, agent_id="assistant-agent")

# Wrap tools — every call and result gets a signed receipt
search = guardrail.wrap_tool(search_tool, tool_name="web_search")
execute = guardrail.wrap_tool(code_executor, tool_name="code_exec")

from aira import Aira
from aira.extras.google_adk import AiraPlugin

aira = Aira(api_key="aira_live_xxx")
plugin = AiraPlugin(client=aira, agent_id="adk-agent", model_id="gemini-2.0-flash")

# Hook into ADK tool lifecycle — receipts at invocation and completion
plugin.before_tool_call("search_documents", args={"query": "contract terms"})
result = search_documents(query="contract terms")
plugin.after_tool_call("search_documents", result=result)

import boto3
from aira import Aira
from aira.extras.bedrock import AiraBedrockHandler

aira = Aira(api_key="aira_live_xxx")
handler = AiraBedrockHandler(client=aira, agent_id="bedrock-agent")

bedrock = boto3.client("bedrock-runtime")
bedrock.invoke_model = handler.wrap_invoke_model(bedrock)

# Every invoke_model call now produces a cryptographic receipt
response = bedrock.invoke_model(modelId="anthropic.claude-v2", body=payload)

export AIRA_API_KEY="aira_live_xxx"
aira-mcp

{
  "mcpServers": {
    "aira": {
      "command": "aira-mcp",
      "env": { "AIRA_API_KEY": "aira_live_xxx" }
    }
  }
}

pip install aira-sdk[cli]

aira version                          # SDK version
aira verify <action-uuid>             # Verify a receipt
aira actions list --agent lending-agent --limit 20
aira agents list                      # List registered agents
aira agents create my-agent --name "My Agent"
aira snapshot create eu-ai-act lending-agent
aira package create --title "Q1 Audit Trail" --actions "uuid-1,uuid-2"

aira = Aira(api_key="aira_live_...", offline=True)

# Queue a two-step record locally. authorize and notarize are serialized.
auth = aira.authorize(
    action_type="scan_completed",
    details="Scanned document batch #77",
    agent_id="scanner-agent",
)
# Execute the real work offline...
aira.notarize(
    action_uuid=auth.action_uuid,
    outcome="completed",
    outcome_details="Batch 77 scanned, 412 pages",
)

print(aira._queue.pending_count)

# Flush to API when back online. Policies are evaluated server-side at flush time.
results = aira.sync()

with aira.session(agent_id="onboarding-agent", model_id="claude-sonnet-4-6") as sess:
    auth = sess.authorize(action_type="identity_verified", details="Verified customer ID #4521")
    if auth.status == "authorized":
        verify_identity_in_db(4521)
        sess.notarize(
            action_uuid=auth.action_uuid,
            outcome="completed",
            outcome_details="KYC check passed",
        )

    # Subsequent calls inherit the agent_id and model_id defaults
    contract_auth = sess.authorize(
        action_type="document_generated",
        details="Generated contract for customer #4521",
    )
    if contract_auth.status == "authorized":
        contract = build_contract(4521)
        sess.notarize(
            action_uuid=contract_auth.action_uuid,
            outcome="completed",
            outcome_details=f"Contract {contract.id} generated",
        )

from aira.extras.webhooks import verify_signature, parse_event

is_valid = verify_signature(
    payload=request.body,
    signature=request.headers["X-Aira-Signature"],
    secret="whsec_xxx",
)

if is_valid:
    event = parse_event(request.body)
    print(event.event_type)   # "action.notarized"
    print(event.data)         # Action data with cryptographic receipt

from aira.client import AiraError

try:
    auth = aira.authorize(
        action_type="email_sent",
        details="Send onboarding email",
        agent_id="support-agent",
    )
except AiraError as e:
    if e.code == "POLICY_DENIED":
        # A policy blocked this action
        print(f"Blocked: {e.message}")
    elif e.code == "PLAN_LIMIT_EXCEEDED":
        print("Monthly operation limit reached")
    else:
        print(f"[{e.code}] {e.message} (status={e.status})")

aira = Aira(
    api_key="aira_live_xxx",
    base_url="https://your-self-hosted.com",  # Self-hosted
    timeout=60.0,                               # Request timeout
)