Python SDK
The official Python SDK for the Alsorn Protocol. It wraps the REST API with typed models, automatic retries, and async support so you can integrate Alsorn into any Python application.
Installation
Requires Python 3.9 or later.
pip install alsornQuick Start
Create a client, register an agent, and query its trust score in a few lines.
from alsorn import AlsornClient
client = AlsornClient(api_key="als_your_api_key")
# Register an agentagent = client.agents.register( name="order-fulfillment-agent", authorized_actions=["execute_trade", "transfer_funds"], spending_limit=10000, currency="USD", environment="sandbox",)
# Query trust scoretrust = client.trust.query(agent.id)print(f"Score: {trust.score}, Assessment: {trust.assessment}")Client Configuration
The AlsornClient constructor accepts the following options.
Constructor Options
| Parameter | Type | Required | Description |
|---|---|---|---|
| api_key | str | required | Your Alsorn API key. Falls back to the ALSORN_API_KEY environment variable if not provided. |
| base_url | str | optional | Base URL for the API. Defaults to "https://api.alsorn.com". Override for sandbox or self-hosted deployments. |
| timeout | float | optional | Request timeout in seconds. Defaults to 30. |
The client can also be used as a context manager to ensure connections are cleaned up.
with AlsornClient(api_key="als_your_api_key") as client: agents = client.agents.list() for agent in agents: print(agent.name)Agents
All agent management methods are available on client.agents.
register()
Register a new agent on the protocol.
agent = client.agents.register( name="data-pipeline-agent", authorized_actions=["execute_trade"], spending_limit=5000, currency="USD", environment="production",)print(agent.id, agent.fingerprint)list()
List all agents in your organization.
agents = client.agents.list(status="active", limit=50)for agent in agents: print(agent.id, agent.name, agent.status)get()
Retrieve a single agent by ID.
agent = client.agents.get("agent_7f3k9x2m")print(agent.name, agent.status, agent.created_at)verify()
Verify an agent's cryptographic fingerprint.
result = client.agents.verify("agent_7f3k9x2m")print(result.verified) # Trueprint(result.fingerprint) # sha256:9f86d08...update()
Update an agent's configuration.
agent = client.agents.update( "agent_7f3k9x2m", name="renamed-agent", spending_limit=20000,)print(agent.name) # renamed-agentsuspend()
Suspend an active agent.
agent = client.agents.suspend("agent_7f3k9x2m")print(agent.status) # suspendedget_audit_chain()
Retrieve the full audit chain for an agent.
chain = client.agents.get_audit_chain("agent_7f3k9x2m")for entry in chain: print(entry.action, entry.timestamp, entry.actor)Trust
Trust scoring methods are available on client.trust.
query()
Query the current trust profile for an agent.
trust = client.trust.query("agent_7f3k9x2m")print(trust.score) # 0–1000print(trust.assessment) # "Verified", "Moderate", or "High Risk"print(trust.factors) # { completion_rate: 0.98, dispute_ratio: 0.01, ... }evaluate()
Run a trust evaluation before executing a transaction.
evaluation = client.trust.evaluate( agent_id="agent_7f3k9x2m", action="transfer_funds", amount=5000, currency="USD",)print(evaluation.allowed) # Trueprint(evaluation.reason) # "Trust score meets threshold"batch_query()
Query trust scores for multiple agents at once.
results = client.trust.batch_query([ "agent_7f3k9x2m", "agent_2b8m4p1q", "agent_9c1n5r3s",])for r in results: print(r.agent_id, r.score, r.assessment)get_history()
Get the trust score history for an agent.
history = client.trust.get_history( "agent_7f3k9x2m", since="2025-01-01T00:00:00Z",)for snapshot in history: print(snapshot.timestamp, snapshot.score, snapshot.assessment)report()
File a trust dispute against an agent.
dispute = client.trust.report( agent_id="agent_7f3k9x2m", reason="Agent did not deliver agreed-upon results.", evidence_ids=["tx_r4s5t6u7v8"],)print(dispute.dispute_id) # disp_g9h0i1j2k3Transactions
Transaction methods are available on client.transactions.
execute()
Execute a new transaction between agents.
tx = client.transactions.execute( from_agent="agent_7f3k9x2m", to_agent="agent_2b8m4p1q", action="transfer_funds", amount=500, currency="USD", memo="Payment for data processing job",)print(tx.id, tx.status) # tx_abc123... completedlist()
List transactions with optional filters.
txns = client.transactions.list( agent_id="agent_7f3k9x2m", status="completed", limit=25,)for tx in txns: print(tx.id, tx.amount, tx.currency, tx.status)get()
Retrieve a single transaction by ID.
tx = client.transactions.get("tx_r4s5t6u7v8")print(tx.from_agent, tx.to_agent, tx.amount, tx.status)reverse()
Reverse a completed transaction.
reversal = client.transactions.reverse( "tx_r4s5t6u7v8", reason="Duplicate payment",)print(reversal.id, reversal.status) # tx_b4c5d6... reversedexport()
Export transactions as CSV or JSON.
export = client.transactions.export( format="csv", since="2025-01-01T00:00:00Z", until="2025-06-01T00:00:00Z",)with open("transactions.csv", "wb") as f: f.write(export.content)Webhooks
Webhook management methods are available on client.webhooks.
register()
Register a new webhook endpoint.
webhook = client.webhooks.register( url="https://example.com/webhooks/alsorn", events=["agent.registered", "transaction.completed"],)print(webhook.id) # wh_x1y2z3print(webhook.secret) # whsec_a1b2c3... (only returned once)list()
List all registered webhook endpoints.
webhooks = client.webhooks.list()for wh in webhooks: print(wh.id, wh.url, wh.active)get()
Retrieve a single webhook by ID.
webhook = client.webhooks.get("wh_x1y2z3")print(webhook.url, webhook.events, webhook.consecutive_failures)update()
Update a webhook's URL or subscribed events.
webhook = client.webhooks.update( "wh_x1y2z3", events=["agent.registered", "transaction.completed", "trust.score_changed"],)print(webhook.events)delete()
Delete a webhook endpoint.
client.webhooks.delete("wh_x1y2z3")test()
Send a test event to a webhook endpoint.
result = client.webhooks.test("wh_x1y2z3")print(result.status_code) # 200print(result.success) # TrueError Handling
The SDK raises typed exceptions for different error conditions. Catch specific exceptions to handle each case appropriately.
from alsorn import AlsornClientfrom alsorn.errors import ( AlsornError, AlsornRateLimitError, AlsornNotFoundError, AlsornAuthError, AlsornValidationError,)
client = AlsornClient(api_key="als_your_api_key")
try: agent = client.agents.get("agent_nonexistent")except AlsornNotFoundError as e: print(f"Agent not found: {e.message}")except AlsornRateLimitError as e: print(f"Rate limited. Retry after {e.retry_after} seconds.")except AlsornAuthError as e: print(f"Authentication failed: {e.message}")except AlsornValidationError as e: print(f"Invalid request: {e.message}") for field_error in e.field_errors: print(f" {field_error.field}: {field_error.message}")except AlsornError as e: # Catch-all for any other API error print(f"API error {e.status_code}: {e.message}")Automatic retries
max_retries to the client constructor.