Python SDK¶

pip install akasha-client

from akasha import AkashaHttpClient

client = AkashaHttpClient(
    base_url="https://localhost:7777",
    api_key="your-api-key",       # or use token="jwt-token"
    verify_ssl=False,              # for self-signed certs
)

# CRUD
record = client.put("agents/planner/state", {"status": "active"})
state = client.get("agents/planner/state")
client.delete("agents/planner/state")

# Query
results = client.query("agents/*/state")

# CAS
from akasha import CasConflictError
try:
    client.put_cas("shared/counter", {"n": 1}, expected_version=record.version)
except CasConflictError:
    pass  # retry

# Subscribe to changes
for event in client.subscribe("agents/**"):
    print(f"[{event.kind}] {event.path}")

from akasha import AsyncAkashaHttpClient

async def main():
    client = AsyncAkashaHttpClient(
        base_url="https://localhost:7777",
        token="your-jwt-token",
        verify_ssl=False,
    )

    await client.put("memory/working/task", {"step": 1})
    record = await client.get("memory/working/task")

# API Key
client = AkashaHttpClient(api_key="your-key")

# JWT Token
client = AkashaHttpClient(token="eyJ0eXAi...")

# Login to get a token
client = AkashaHttpClient(base_url="https://localhost:7777")
# POST /api/v1/auth/login