Python SDK

The Python SDK is an HTTP client that wraps the Akshi HTTP API. Use it for external tooling, dashboards, and orchestration scripts.

Installation

pip install akshi-sdk

Quick start

from akshi_sdk import AkshiClient

client = AkshiClient(base_url="http://127.0.0.1:3210", token="my-token")

if client.health().get("ok"):
    print("Runtime is healthy")

Client methods

Method	HTTP	Returns	Description
`health()`	`GET /api/v1/health`	`dict`	Runtime health check
`status()`	`GET /api/v1/status`	`list[dict]`	All agent statuses
`logs(since=0)`	`GET /api/v1/logs`	`dict`	Log entries, optionally filtered by sequence number
`findings(since=0)`	`GET /api/v1/findings`	`dict`	Findings, optionally filtered by sequence number
`post_findings(records)`	`POST /api/v1/findings`	`dict`	Submit a batch of findings
`broker_grants()`	`GET /api/v1/broker/grants`	`list[dict]`	List active broker secret grants
`revoke_broker_grant(grant_id)`	`DELETE /api/v1/broker/grants/{id}`	`dict`	Revoke a broker grant by ID
`approvals()`	`GET /api/v1/approvals`	`dict`	List pending and resolved approval requests
`post_approval(payload)`	`POST /api/v1/approvals`	`dict`	Approve or deny a pending approval request
`a2a_tasks()`	`GET /api/v1/a2a/tasks`	`dict`	List A2A task statuses
`reload()`	`POST /api/v1/reload`	`dict`	Hot-reload runtime configuration
`events_sse(callback)`	`GET /api/v1/agui/events`	–	Subscribe to SSE event stream (blocking)

FindingRecord

from dataclasses import dataclass

@dataclass(frozen=True)
class FindingRecord:
    source: str       # Agent name that produced the finding
    severity: str     # "critical" | "warning" | "info"
    file: str         # Origin file path
    summary: str      # Short description

Broker grants

# List all active grants
grants = client.broker_grants()
for grant in grants:
    print(f"{grant['agent']}: {grant['secret_name']}")

# Revoke a specific grant
client.revoke_broker_grant("grant-uuid-here")

Approvals

# List pending approvals
pending = client.approvals()

# Approve a request
client.post_approval({
    "request_id": "req-uuid",
    "decision": "approve"
})

A2A tasks

# List all A2A task statuses
tasks = client.a2a_tasks()
for task in tasks.get("tasks", []):
    print(f"{task['id']}: {task['status']}")

Hot reload

# Reload runtime.toml without restarting
result = client.reload()
print(result)

SSE event stream

def on_event(data: str):
    print(f"Event: {data}")

# Blocks until the connection closes
client.events_sse(on_event)

Error handling

All methods raise on non-2xx responses. The exception is a standard urllib.error.HTTPError.

from urllib.error import HTTPError

try:
    client.post_findings(records)
except HTTPError as e:
    print(f"{e.code}: {e.read().decode()}")

Keyboard shortcuts

Akshi Documentation