Akasha Python SDK

The official Python client for Akasha — The Shared Cognitive Fabric for Intelligent Agent Systems.

Installation

pip install akasha-client

Quick Start

from akasha import AkashaHttpClient

# Connect (auto-TLS with self-signed certs → verify_ssl=False for dev)
client = AkashaHttpClient(
    "https://localhost:7777",
    api_key="ak_your_api_key",
    verify_ssl=False,
)

# Write a record
record = client.put("agents/planner/state", {
    "status": "thinking",
    "task": "summarize document",
    "confidence": 0.85,
})

print(f"Created: {record.path} (v{record.version})")

# Read it back
state = client.get("agents/planner/state")
print(f"Status: {state.value['status']}")

# Query with glob patterns
all_states = client.query("agents/*/state")
for r in all_states:
    print(f"  {r.path}: {r.value}")

# Delete
client.delete("agents/planner/state")
client.close()

CAS (Compare-And-Swap) — Optimistic Concurrency

Prevent lost updates when multiple agents write to the same path:

from akasha import AkashaHttpClient, CasConflictError

client = AkashaHttpClient("https://localhost:7777", verify_ssl=False)

# Read current version
record = client.get("shared/counter")

try:
    # Only update if nobody else changed it since our read
    updated = client.put_cas(
        "shared/counter",
        {"count": record.value["count"] + 1},
        expected_version=record.version,
    )
    print(f"Updated to v{updated.version}")
except CasConflictError as e:
    print(f"Conflict! Expected v{e.expected_version}, got v{e.actual_version}")
    print(f"Current value: {e.current}")

Memory Layers

Akasha organizes knowledge into four cognitive layers:

from akasha import AkashaHttpClient, MemoryLayer

client = AkashaHttpClient("https://localhost:7777", verify_ssl=False)

# Working memory (volatile, 30 min TTL)
client.put(
    f"{MemoryLayer.WORKING.prefix}agent-1/scratch",
    {"current_task": "analyze"},
    ttl_seconds=MemoryLayer.WORKING.default_ttl_seconds,
)

# Episodic memory (event log, 24h TTL)
client.put(
    f"{MemoryLayer.EPISODIC.prefix}agent-1/events/2024-01-15",
    {"action": "summarized", "input_tokens": 4500},
    ttl_seconds=MemoryLayer.EPISODIC.default_ttl_seconds,
)

# Semantic memory (permanent knowledge)
client.put(
    f"{MemoryLayer.SEMANTIC.prefix}domain/users/preferences",
    {"theme": "dark", "language": "es"},
)

# Query all episodic memories
episodes = client.query(MemoryLayer.EPISODIC.glob_all)

Authentication

# API Key (recommended for agents)
client = AkashaHttpClient(
    "https://akasha.example.com",
    api_key="ak_live_abc123",
)

# JWT Token (for user sessions)
client = AkashaHttpClient(
    "https://akasha.example.com",
    token="eyJhbGciOiJIUzI1NiIs...",
)

Context Manager

with AkashaHttpClient("https://localhost:7777", verify_ssl=False) as client:
    client.put("test/hello", {"message": "world"})
    record = client.get("test/hello")
    print(record.value)
# Connection auto-closed

Async Support

import asyncio
from akasha import AsyncAkashaClient  # gRPC-based async client

async def main():
    async with AsyncAkashaClient("localhost:50051") as client:
        await client.put("agents/worker/status", {"busy": True})
        record = await client.get("agents/worker/status")
        print(record.value)

asyncio.run(main())

API Reference

AkashaHttpClient

Record

CasConflictError

Requirements

• Python ≥ 3.10
• httpx ≥ 0.27 (HTTP client)
• grpcio ≥ 1.60 (gRPC client)
• msgpack ≥ 1.0 (serialization)

Links

• Akasha GitHub
• Docker Hub
• Changelog