Python SDK reference

__all__ = ['AkashaClient', 'AkashaConfig', 'Tenant', 'TenantPlan', 'TenantStatus', 'Backup', 'BackupStatus', 'HealthStatus', 'UsageMetrics', 'ServiceStatus', 'AkashaError', 'AuthenticationError', 'NotFoundError', 'ValidationError', 'RateLimitError', 'ServerError']

class AkashaClient()

Main client for interacting with Akasha APIs.

Provides access to both Control Plane APIs (tenant management, backups,
monitoring) and Runtime APIs (KV, graph, vector operations).

Example:
    >>> client = AkashaClient("https://api.akasha.l1fe.ai", api_key="your-key")
    >>> # Control Plane operations
    >>> tenant = client.control_plane.tenants.get("tenant-123")
    >>> # Runtime operations (requires instance endpoint)
    >>> runtime = client.runtime("https://my-instance.akasha.ooo")
    >>> value = runtime.kv.get("my-key")

Args:
    base_url: Base URL of the Akasha Control Plane API.
    api_key: API key for authentication.
    timeout: Request timeout in seconds.
    max_retries: Maximum number of retries for failed requests.
    verify_ssl: Whether to verify SSL certificates.

def __init__(self, base_url: str, api_key: str, timeout: float=30.0, max_retries: int=3, verify_ssl: bool=True) -> None

def control_plane(self) -> ControlPlaneClient

Access Control Plane APIs.

Provides access to tenant management, backups, and monitoring.

Returns:
    ControlPlaneClient instance.

def runtime(self, instance_url: str) -> RuntimeClient

Get a Runtime client for a specific Akasha instance.

Args:
    instance_url: URL of the Akasha instance (e.g., "https://my-instance.akasha.ooo").

Returns:
    RuntimeClient for the specified instance.

Example:
    >>> runtime = client.runtime("https://my-instance.akasha.ooo")
    >>> value = runtime.kv.get("my-key")

def request(self, method: str, path: str, json: Optional[dict[str, Any]]=None, params: Optional[dict[str, Any]]=None) -> dict[str, Any]

Make a raw HTTP request to the API.

Args:
    method: HTTP method (GET, POST, PUT, DELETE, etc.).
    path: API path (e.g., "/api/v1/tenants").
    json: JSON body for POST/PUT requests.
    params: Query parameters.

Returns:
    Parsed JSON response.

Raises:
    AkashaError: On API errors.

def close(self) -> None

Close the HTTP client and release resources.

class ControlPlaneClient()

Control Plane client providing access to all control plane APIs.

Attributes:
    tenants: Client for tenant management operations.
    backups: Client for backup operations.
    monitoring: Client for monitoring and metrics.

def __init__(self, http_client: httpx.Client) -> None

def tenants(self) -> TenantClient

Access tenant management APIs.

def backups(self) -> BackupClient

Access backup management APIs.

def monitoring(self) -> MonitoringClient

Access monitoring and metrics APIs.

__all__ = ['ControlPlaneClient', 'TenantClient', 'BackupClient', 'MonitoringClient']

class BackupClient()

Client for backup management operations.

Provides methods for creating, listing, and restoring backups.

Example:
    >>> backups = client.control_plane.backups
    >>> backup = backups.create("tenant-123", name="pre-migration")
    >>> all_backups = backups.list("tenant-123")

def __init__(self, http_client: httpx.Client) -> None

def create(self, tenant_id: str, name: Optional[str]=None, include_services: Optional[list[ServiceType]]=None, retention_days: int=30) -> Backup

Create a new backup for a tenant.

Args:
    tenant_id: The tenant's unique identifier.
    name: Backup name (auto-generated if not provided).
    include_services: Services to include (all if not specified).
    retention_days: How long to retain the backup (default: 30 days).

Returns:
    Created Backup object.

Raises:
    NotFoundError: If tenant doesn't exist.

Example:
    >>> backup = backups.create(
    ...     "tenant-123",
    ...     name="pre-migration",
    ...     include_services=[ServiceType.KV, ServiceType.GRAPH],
    ...     retention_days=90
    ... )

def get(self, tenant_id: str, backup_id: str) -> Backup

Get a backup by ID.

Args:
    tenant_id: The tenant's unique identifier.
    backup_id: The backup's unique identifier.

Returns:
    Backup object.

Raises:
    NotFoundError: If backup doesn't exist.

Example:
    >>> backup = backups.get("tenant-123", "backup-456")

def list(self, tenant_id: str, page: int=1, page_size: int=20, status: Optional[BackupStatus]=None) -> list[Backup]

List all backups for a tenant.

Args:
    tenant_id: The tenant's unique identifier.
    page: Page number (1-indexed).
    page_size: Number of results per page.
    status: Filter by backup status.

Returns:
    List of Backup objects.

Example:
    >>> completed = backups.list("tenant-123", status=BackupStatus.COMPLETED)

def delete(self, tenant_id: str, backup_id: str) -> None

Delete a backup.

Warning: This permanently deletes the backup data.

Args:
    tenant_id: The tenant's unique identifier.
    backup_id: The backup's unique identifier.

Raises:
    NotFoundError: If backup doesn't exist.

Example:
    >>> backups.delete("tenant-123", "backup-456")

def restore(self, tenant_id: str, backup_id: str, target_tenant_id: Optional[str]=None, services: Optional[list[ServiceType]]=None) -> dict[str, Any]

Restore from a backup.

Args:
    tenant_id: The source tenant's identifier.
    backup_id: The backup's unique identifier.
    target_tenant_id: Target tenant for restore (same tenant if not provided).
    services: Services to restore (all if not specified).

Returns:
    Restore operation status.

Raises:
    NotFoundError: If backup doesn't exist.
    ConflictError: If restore conflicts with existing data.

Example:
    >>> # Restore to same tenant
    >>> result = backups.restore("tenant-123", "backup-456")
    >>> # Restore to different tenant
    >>> result = backups.restore(
    ...     "tenant-123",
    ...     "backup-456",
    ...     target_tenant_id="tenant-789"
    ... )

class MonitoringClient()

Client for monitoring and metrics operations.

Provides methods for retrieving usage metrics, health status,
and other observability data.

Example:
    >>> monitoring = client.control_plane.monitoring
    >>> health = monitoring.get_health("tenant-123")
    >>> usage = monitoring.get_usage("tenant-123")

def __init__(self, http_client: httpx.Client) -> None

def get_health(self, tenant_id: str) -> InstanceHealth

Get health status for a tenant's instance.

Args:
    tenant_id: The tenant's unique identifier.

Returns:
    InstanceHealth object with overall and per-service status.

Raises:
    NotFoundError: If tenant doesn't exist.

Example:
    >>> health = monitoring.get_health("tenant-123")
    >>> if health.status == HealthStatus.HEALTHY:
    ...     print("All systems operational")

def get_usage(self, tenant_id: str, start_time: Optional[datetime]=None, end_time: Optional[datetime]=None) -> UsageMetrics

Get usage metrics for a tenant.

Args:
    tenant_id: The tenant's unique identifier.
    start_time: Start of the reporting period.
    end_time: End of the reporting period.

Returns:
    UsageMetrics object with resource consumption data.

Raises:
    NotFoundError: If tenant doesn't exist.

Example:
    >>> usage = monitoring.get_usage("tenant-123")
    >>> print(f"Storage used: {usage.storage_used_bytes / 1e9:.2f} GB")

def get_metrics(self, tenant_id: str, metric_names: Optional[list[str]]=None, start_time: Optional[datetime]=None, end_time: Optional[datetime]=None, granularity: str='hour') -> dict[str, Any]

Get detailed metrics for a tenant.

Args:
    tenant_id: The tenant's unique identifier.
    metric_names: Specific metrics to retrieve (all if not specified).
    start_time: Start of the reporting period.
    end_time: End of the reporting period.
    granularity: Data point granularity ("minute", "hour", "day").

Returns:
    Dictionary with metric time series data.

Raises:
    NotFoundError: If tenant doesn't exist.

Example:
    >>> metrics = monitoring.get_metrics(
    ...     "tenant-123",
    ...     metric_names=["read_ops", "write_ops"],
    ...     granularity="hour"
    ... )
    >>> for point in metrics["read_ops"]:
    ...     print(f"{point['timestamp']}: {point['value']}")

def get_services_status(self, instance_id: str) -> dict[str, Any]

Get status of all services for an instance.

Args:
    instance_id: The instance's unique identifier.

Returns:
    Dictionary with per-service status information.

Example:
    >>> status = monitoring.get_services_status("instance-123")
    >>> for service, info in status["services"].items():
    ...     print(f"{service}: {info['status']}")

def get_alerts(self, tenant_id: str, severity: Optional[str]=None, active_only: bool=True) -> list[dict[str, Any]]

Get alerts for a tenant.

Args:
    tenant_id: The tenant's unique identifier.
    severity: Filter by severity ("critical", "warning", "info").
    active_only: Only return active (unresolved) alerts.

Returns:
    List of alert objects.

Example:
    >>> alerts = monitoring.get_alerts("tenant-123", severity="critical")
    >>> for alert in alerts:
    ...     print(f"{alert['name']}: {alert['message']}")

class TenantClient()

Client for tenant management operations.

Provides methods for creating, reading, updating, and deleting tenants,
as well as managing tenant services.

Example:
    >>> tenants = client.control_plane.tenants
    >>> tenant = tenants.create(name="my-tenant", plan=TenantPlan.PRO, region="us-east-1")
    >>> all_tenants = tenants.list()

def __init__(self, http_client: httpx.Client) -> None

def create(self, name: str, plan: TenantPlan=TenantPlan.FREE, region: str='us-east-1', services: Optional[TenantServices]=None, metadata: Optional[dict[str, Any]]=None) -> Tenant

Create a new tenant.

Args:
    name: Unique tenant name (1-63 characters).
    plan: Subscription plan (free, starter, pro, enterprise).
    region: Deployment region (e.g., "us-east-1").
    services: Services to enable (defaults based on plan).
    metadata: Custom metadata key-value pairs.

Returns:
    Created Tenant object.

Raises:
    ValidationError: If request validation fails.
    ConflictError: If tenant name already exists.

Example:
    >>> tenant = tenants.create(
    ...     name="my-app",
    ...     plan=TenantPlan.PRO,
    ...     region="us-east-1",
    ...     metadata={"environment": "production"}
    ... )

def get(self, tenant_id: str) -> Tenant

Get a tenant by ID.

Args:
    tenant_id: The tenant's unique identifier.

Returns:
    Tenant object.

Raises:
    NotFoundError: If tenant doesn't exist.

Example:
    >>> tenant = tenants.get("tenant-abc123")

def list(self, page: int=1, page_size: int=20, status: Optional[str]=None, plan: Optional[TenantPlan]=None) -> list[Tenant]

List all tenants with optional filtering.

Args:
    page: Page number (1-indexed).
    page_size: Number of results per page.
    status: Filter by status (active, suspended, etc.).
    plan: Filter by subscription plan.

Returns:
    List of Tenant objects.

Example:
    >>> active_tenants = tenants.list(status="active")
    >>> pro_tenants = tenants.list(plan=TenantPlan.PRO)

def update(self, tenant_id: str, name: Optional[str]=None, plan: Optional[TenantPlan]=None, services: Optional[TenantServices]=None, metadata: Optional[dict[str, Any]]=None) -> Tenant

Update a tenant.

Args:
    tenant_id: The tenant's unique identifier.
    name: New tenant name.
    plan: New subscription plan.
    services: Services to update.
    metadata: Metadata to update (merged with existing).

Returns:
    Updated Tenant object.

Raises:
    NotFoundError: If tenant doesn't exist.
    ValidationError: If validation fails.

Example:
    >>> updated = tenants.update("tenant-123", plan=TenantPlan.ENTERPRISE)

def delete(self, tenant_id: str) -> None

Delete a tenant.

Warning: This permanently deletes the tenant and all associated data.

Args:
    tenant_id: The tenant's unique identifier.

Raises:
    NotFoundError: If tenant doesn't exist.

Example:
    >>> tenants.delete("tenant-123")

def enable_services(self, tenant_id: str, services: list[str]) -> Tenant

Enable services for a tenant.

Args:
    tenant_id: The tenant's unique identifier.
    services: List of service names to enable (e.g., ["graph", "vector"]).

Returns:
    Updated Tenant object.

Example:
    >>> tenant = tenants.enable_services("tenant-123", ["graph", "analytics"])

def disable_services(self, tenant_id: str, services: list[str]) -> Tenant

Disable services for a tenant.

Args:
    tenant_id: The tenant's unique identifier.
    services: List of service names to disable.

Returns:
    Updated Tenant object.

Example:
    >>> tenant = tenants.disable_services("tenant-123", ["ml"])

def get_status(self, tenant_id: str) -> dict[str, Any]

Get detailed status for a tenant.

Args:
    tenant_id: The tenant's unique identifier.

Returns:
    Status information including instance state, services, and resources.

Example:
    >>> status = tenants.get_status("tenant-123")
    >>> print(status["instance_state"])

class AkashaError(Exception)

Base exception for all Akasha SDK errors.

def __init__(self, message: str, status_code: Optional[int]=None, response_body: Optional[dict[str, Any]]=None) -> None

class AuthenticationError(AkashaError)

Raised when authentication fails (401 Unauthorized).

def __init__(self, message: str='Authentication failed. Check your API key.', response_body: Optional[dict[str, Any]]=None) -> None

class AuthorizationError(AkashaError)

Raised when authorization fails (403 Forbidden).

def __init__(self, message: str='Access denied. Insufficient permissions.', response_body: Optional[dict[str, Any]]=None) -> None

class NotFoundError(AkashaError)

Raised when a resource is not found (404 Not Found).

def __init__(self, resource: str='Resource', resource_id: Optional[str]=None, response_body: Optional[dict[str, Any]]=None) -> None

class ValidationError(AkashaError)

Raised when request validation fails (400 Bad Request).

def __init__(self, message: str='Invalid request parameters.', errors: Optional[list[dict[str, Any]]]=None, response_body: Optional[dict[str, Any]]=None) -> None

class ConflictError(AkashaError)

Raised when there's a conflict (409 Conflict).

def __init__(self, message: str='Resource conflict.', response_body: Optional[dict[str, Any]]=None) -> None

class RateLimitError(AkashaError)

Raised when rate limit is exceeded (429 Too Many Requests).

def __init__(self, message: str='Rate limit exceeded. Please retry later.', retry_after: Optional[int]=None, response_body: Optional[dict[str, Any]]=None) -> None

class ServerError(AkashaError)

Raised when a server error occurs (5xx).

def __init__(self, message: str='Internal server error.', status_code: int=500, response_body: Optional[dict[str, Any]]=None) -> None

class ConnectionError(AkashaError)

Raised when unable to connect to the Akasha API.

def __init__(self, message: str='Unable to connect to Akasha API.', original_error: Optional[Exception]=None) -> None

class TimeoutError(AkashaError)

Raised when a request times out.

def __init__(self, message: str='Request timed out.', timeout_seconds: Optional[float]=None) -> None

class RuntimeClient()

Runtime client providing access to all Akasha instance APIs.

Attributes:
    core: Core storage operations (keyspaces, transactions, maintenance)
    kv: Key-value store operations
    graph: Graph database operations
    vector: Vector search operations
    analytics: SQL analytics operations
    memory: Multi-modal memory operations
    ml: Machine learning operations
    snn: Spiking neural network operations
    neurosymbolic: Neurosymbolic reasoning operations
    continual: Continual learning operations
    mhn: Modern Hopfield Network operations

Example:
    >>> runtime = client.runtime("https://my-instance.akasha.ooo")
    >>> # Core operations
    >>> runtime.core.create_keyspace("users", model="document")
    >>> # KV operations
    >>> runtime.kv.set("key", "value")
    >>> # Analytics
    >>> result = runtime.analytics.sql("SELECT * FROM users")
    >>> # Memory
    >>> runtime.memory.episodic.insert("event", {"data": "..."})
    >>> # SNN
    >>> runtime.snn.build(architecture)

def __init__(self, http_client: httpx.Client) -> None

def core(self) -> CoreClient

Access core storage operations (keyspaces, transactions, maintenance).

def kv(self) -> KVClient

Access key-value store operations.

def graph(self) -> GraphClient

Access graph database operations.

def vector(self) -> VectorClient

Access vector search operations.

def analytics(self) -> AnalyticsClient

Access SQL analytics operations.

def memory(self) -> MemoryClient

Access multi-modal memory operations.

def ml(self) -> MLClient

Access machine learning operations.

def snn(self) -> SNNClient

Access spiking neural network operations.

def neurosymbolic(self) -> NeurosymbolicClient

Access neurosymbolic reasoning operations.

def continual(self) -> ContinualClient

Access continual learning operations.

def mhn(self) -> MHNClient

Access Modern Hopfield Network operations.

def brain(self) -> CognitiveBrainClient

Access MCP cognitive brain operations (remember, recall, forget, reflect, introspect).

def close(self) -> None

Close the HTTP client.

__all__ = ['RuntimeClient', 'CoreClient', 'KVClient', 'GraphClient', 'VectorClient', 'AnalyticsClient', 'MemoryClient', 'MLClient', 'SNNClient', 'NeurosymbolicClient', 'ContinualClient', 'MHNClient', 'CognitiveBrainClient']

class AnalyticsClient()

Client for SQL analytics operations.

Provides methods for executing SQL queries, registering data sources,
and managing the analytics cache.

Example:
    >>> runtime = client.runtime("https://my-instance.akasha.ooo")
    >>> result = runtime.analytics.sql("SELECT * FROM users WHERE age > 21")

def __init__(self, http_client: httpx.Client) -> None

def sql(self, query: str, parameters: Optional[dict[str, Any]]=None, timeout_ms: Optional[int]=None) -> dict[str, Any]

Execute a SQL query.

Args:
    query: SQL query string.
    parameters: Query parameters for prepared statements.
    timeout_ms: Query timeout in milliseconds.

Returns:
    Query results with columns and rows.

Example:
    >>> result = runtime.analytics.sql(
    ...     "SELECT name, age FROM users WHERE department = $1",
    ...     parameters={"$1": "Engineering"}
    ... )
    >>> for row in result["rows"]:
    ...     print(row)

def register_csv_source(self, name: str, path: str, has_header: bool=True, delimiter: str=',', schema: Optional[dict[str, str]]=None) -> dict[str, Any]

Register a CSV file as a queryable data source.

Args:
    name: Table name to use in SQL queries.
    path: Path to the CSV file.
    has_header: Whether the CSV has a header row.
    delimiter: Field delimiter character.
    schema: Optional column type overrides.

Returns:
    Registration result.

Example:
    >>> runtime.analytics.register_csv_source(
    ...     "sales",
    ...     "/data/sales.csv",
    ...     schema={"amount": "DECIMAL", "date": "DATE"}
    ... )

def register_parquet_source(self, name: str, path: str, schema: Optional[dict[str, str]]=None) -> dict[str, Any]

Register a Parquet file as a queryable data source.

Args:
    name: Table name to use in SQL queries.
    path: Path to the Parquet file.
    schema: Optional column type overrides.

Returns:
    Registration result.

Example:
    >>> runtime.analytics.register_parquet_source("events", "/data/events.parquet")

def register_core_source(self, name: str, keyspace: str, schema: Optional[dict[str, str]]=None) -> dict[str, Any]

Register an Akasha keyspace as a queryable data source.

Args:
    name: Table name to use in SQL queries.
    keyspace: Name of the Akasha keyspace.
    schema: Optional column type overrides.

Returns:
    Registration result.

Example:
    >>> runtime.analytics.register_core_source("users", "user_keyspace")
    >>> result = runtime.analytics.sql("SELECT * FROM users")

def set_options(self, batch_size: Optional[int]=None, parallel_degree: Optional[int]=None, memory_limit_mb: Optional[int]=None, enable_caching: Optional[bool]=None) -> dict[str, Any]

Configure analytics engine options.

Args:
    batch_size: Processing batch size.
    parallel_degree: Degree of parallelism.
    memory_limit_mb: Memory limit in MB.
    enable_caching: Enable query result caching.

Returns:
    Current options after update.

def clear_cache(self) -> dict[str, Any]

Clear the analytics query cache.

Returns:
    Cache clear result.

def get_cache_stats(self) -> dict[str, Any]

Get analytics cache statistics.

Returns:
    Cache statistics including hit rate, size, etc.

class CognitiveBrainClient()

Client for MCP cognitive tool operations.

Provides a high-level interface to Akasha's cognitive memory tools
via the daemon's MCP HTTP API.

Example:
    >>> brain = runtime.brain
    >>> # Store a memory
    >>> result = brain.remember("User prefers dark mode")
    >>> # Recall memories
    >>> memories = brain.recall("dark mode preference")
    >>> # Forget a memory
    >>> brain.forget([result["memory_id"]], reason="User requested removal")

def __init__(self, http_client: httpx.Client) -> None

def remember(self, content: str, *, source: Optional[str]=None, importance: Optional[float]=None, tags: Optional[list[str]]=None, memory_type_hint: Optional[str]=None) -> dict[str, Any]

Store a memory with automatic cognitive processing.

The system automatically detects duplicates, checks for contradictions,
infers relationships, and classifies the memory type.

Args:
    content: The memory content to store.
    source: Where this memory came from (e.g., "user_input", "observation").
    importance: Importance hint (0.0 to 1.0).
    tags: Tags for categorization.
    memory_type_hint: Classification hint ("auto", "episodic", "semantic", "procedural").

Returns:
    Dict with stored, memory_id, memory_type, was_duplicate, importance, etc.

def recall(self, query: str, *, limit: int=10, memory_types: Optional[list[str]]=None, tags: Optional[list[str]]=None, min_importance: Optional[float]=None, time_after: Optional[str]=None, time_before: Optional[str]=None) -> dict[str, Any]

Retrieve memories using hybrid search.

Combines semantic similarity, keyword matching, and temporal relevance
for intelligent memory retrieval.

Args:
    query: What to remember.
    limit: Maximum number of results (1-100).
    memory_types: Filter by types ("episodic", "semantic", "procedural").
    tags: Filter by tags.
    min_importance: Minimum importance threshold (0.0 to 1.0).
    time_after: Only memories after this ISO 8601 timestamp.
    time_before: Only memories before this ISO 8601 timestamp.

Returns:
    Dict with memories list, total_matches, and search_info.

def forget(self, memory_ids: list[str], reason: str, *, confirm_bulk: Optional[bool]=None, mode: str='soft') -> dict[str, Any]

Remove memories with audit trail.

Args:
    memory_ids: IDs of memories to forget.
    reason: Why these memories are being forgotten (stored in audit log).
    confirm_bulk: Must be True to forget multiple memories at once.
    mode: Delete mode - "soft" (recoverable) or "hard" (permanent).

Returns:
    Dict with deleted_count, audit_id, skipped list, and warnings.

def reflect(self, topic: Optional[str]=None) -> dict[str, Any]

Generate reflections on stored memories.

Args:
    topic: Optional topic to focus reflection on.

Returns:
    Dict with insights, patterns, connections, and recommendations.

def introspect(self, aspect: Optional[str]=None) -> dict[str, Any]

Get memory system introspection and health.

Args:
    aspect: Optional aspect to focus on (e.g., "health", "statistics").

Returns:
    Dict with total_memories, by_type counts, health status, etc.

def call_tool(self, tool_name: str, arguments: dict[str, Any]) -> dict[str, Any]

Call any MCP tool by name with raw arguments.

Args:
    tool_name: Name of the MCP tool to call.
    arguments: Tool arguments as a dictionary.

Returns:
    Tool execution result.

def list_tools(self) -> list[dict[str, Any]]

List available MCP tools.

Returns:
    List of tool descriptions with name, description, and tier.

class ContinualClient()

Client for continual learning operations.

Provides methods for managing checkpoints, recovery, and component
registration for lifelong learning without catastrophic forgetting.

Example:
    >>> runtime = client.runtime("https://my-instance.akasha.ooo")
    >>> checkpoint = runtime.continual.create_checkpoint("daily_backup")
    >>> runtime.continual.register_component("user_model", model_state)

def __init__(self, http_client: httpx.Client) -> None

def create_checkpoint(self, name: Optional[str]=None, components: Optional[list[str]]=None, metadata: Optional[dict[str, Any]]=None) -> dict[str, Any]

Create a learning checkpoint.

Saves the current state of registered components for later recovery.

Args:
    name: Checkpoint name (auto-generated if not provided).
    components: Specific components to checkpoint (all if not specified).
    metadata: Additional metadata.

Returns:
    Checkpoint info with ID.

Example:
    >>> checkpoint = runtime.continual.create_checkpoint(
    ...     name="before_new_task",
    ...     components=["user_model", "embedding_model"]
    ... )

def get_checkpoint_status(self, checkpoint_id: Optional[str]=None) -> dict[str, Any]

Get checkpoint status.

Args:
    checkpoint_id: Specific checkpoint (latest if not specified).

Returns:
    Checkpoint status and metadata.

def recover(self, checkpoint_id: str, components: Optional[list[str]]=None, strategy: str='full') -> dict[str, Any]

Recover from a checkpoint.

Args:
    checkpoint_id: ID of checkpoint to recover from.
    components: Specific components to recover (all if not specified).
    strategy: Recovery strategy ("full", "selective", "merge").

Returns:
    Recovery result.

Example:
    >>> result = runtime.continual.recover(
    ...     checkpoint["id"],
    ...     strategy="selective",
    ...     components=["user_model"]
    ... )

def register_component(self, name: str, state: dict[str, Any], component_type: str='model', ewc_importance: Optional[float]=None) -> dict[str, Any]

Register a component for continual learning.

Args:
    name: Unique component name.
    state: Component state (weights, parameters, etc.).
    component_type: Type of component ("model", "embeddings", "custom").
    ewc_importance: Elastic Weight Consolidation importance weight.

Returns:
    Registration confirmation.

Example:
    >>> runtime.continual.register_component(
    ...     "sentiment_model",
    ...     {"weights": model.state_dict()},
    ...     ewc_importance=0.5
    ... )

def unregister_component(self, name: str) -> bool

Unregister a component.

Args:
    name: Component name.

Returns:
    True if unregistered, False if not found.

def list_components(self) -> list[dict[str, Any]]

List all registered components.

Returns:
    List of component info.

class CoreClient()

Client for core storage operations.

Provides keyspace management, item CRUD, transactions,
namespaces, and maintenance operations.

Example:
    >>> runtime = client.runtime("https://my-instance.akasha.ooo")
    >>> runtime.core.create_keyspace("users", model="document")
    >>> runtime.core.store("users", {"id": "1", "name": "Alice"})

def __init__(self, http_client: httpx.Client) -> None

def get_stats(self) -> dict[str, Any]

Get engine statistics.

def list_keyspaces(self) -> list[dict[str, Any]]

List all keyspaces.

def create_keyspace(self, name: str, model: str='document', config: Optional[dict[str, Any]]=None) -> dict[str, Any]

Create a new keyspace.

Args:
    name: Keyspace name.
    model: Data model ("document", "kv", "graph", "vector", "timeseries").
    config: Keyspace configuration.

Returns:
    Created keyspace info.

def get_keyspace(self, name: str) -> Optional[dict[str, Any]]

Get keyspace info.

def delete_keyspace(self, name: str) -> bool

Delete a keyspace.

def store(self, keyspace: str, item: dict[str, Any]) -> dict[str, Any]

Store an item in a keyspace.

Args:
    keyspace: Target keyspace.
    item: Item to store (must have 'id' field).

Returns:
    Stored item info.

def get(self, keyspace: str, item_id: str) -> Optional[dict[str, Any]]

Get an item by ID.

def delete(self, keyspace: str, item_id: str) -> bool

Delete an item.

def query(self, keyspace: str, query: dict[str, Any], limit: int=100, offset: int=0) -> list[dict[str, Any]]

Query items in a keyspace.

Args:
    keyspace: Target keyspace.
    query: Query filter.
    limit: Maximum results.
    offset: Result offset.

Returns:
    Matching items.

def text_search(self, query: str, keyspaces: Optional[list[str]]=None, limit: int=100) -> list[dict[str, Any]]

Full-text search across keyspaces.

Args:
    query: Search query.
    keyspaces: Keyspaces to search (all if not specified).
    limit: Maximum results.

Returns:
    Search results.

def begin_transaction(self) -> dict[str, Any]

Begin a new transaction.

def commit_transaction(self, tx_id: str) -> dict[str, Any]

Commit a transaction.

def abort_transaction(self, tx_id: str) -> dict[str, Any]

Abort a transaction.

def list_dataspaces(self) -> list[dict[str, Any]]

List all dataspaces.

def switch_namespace(self, namespace: str) -> dict[str, Any]

Switch to a different namespace.

def analyze(self, keyspace: Optional[str]=None) -> dict[str, Any]

Analyze storage for optimization hints.

Args:
    keyspace: Specific keyspace (all if not specified).

Returns:
    Analysis results.

def compact(self, keyspace: Optional[str]=None) -> dict[str, Any]

Compact storage to reclaim space.

Args:
    keyspace: Specific keyspace (all if not specified).

Returns:
    Compaction results.

def reindex(self, keyspace: Optional[str]=None) -> dict[str, Any]

Rebuild indexes.

Args:
    keyspace: Specific keyspace (all if not specified).

Returns:
    Reindex results.

def clear_cache(self) -> dict[str, Any]

Clear all caches.

def create_backup(self, name: Optional[str]=None, keyspaces: Optional[list[str]]=None) -> dict[str, Any]

Create a backup.

Args:
    name: Backup name.
    keyspaces: Keyspaces to backup (all if not specified).

Returns:
    Backup info.

def list_backups(self) -> list[dict[str, Any]]

List all backups.

def restore_backup(self, backup_id: str) -> dict[str, Any]

Restore from a backup.

def delete_backup(self, backup_id: str) -> bool

Delete a backup.

def create_replication_link(self, target_url: str, keyspaces: Optional[list[str]]=None, mode: str='async') -> dict[str, Any]

Create a replication link to another instance.

Args:
    target_url: Target instance URL.
    keyspaces: Keyspaces to replicate (all if not specified).
    mode: Replication mode ("sync", "async").

Returns:
    Link info.

def stop_replication(self, link_id: str) -> dict[str, Any]

Stop a replication link.

def get_replication_status(self, link_id: Optional[str]=None) -> dict[str, Any]

Get replication status.

class GraphClient()

Client for graph database operations.

Provides methods for creating, querying, and traversing graph data.

Example:
    >>> runtime = client.runtime("https://my-instance.akasha.ooo")
    >>> node = runtime.graph.create_node(labels=["Person"], properties={"name": "Alice"})
    >>> runtime.graph.create_edge(node.id, other_id, "KNOWS")

def __init__(self, http_client: httpx.Client) -> None

def create_node(self, labels: list[str], properties: Optional[dict[str, Any]]=None, node_id: Optional[str]=None) -> GraphNode

Create a new graph node.

Args:
    labels: List of labels for the node (e.g., ["Person", "User"]).
    properties: Node properties as key-value pairs.
    node_id: Optional custom ID (auto-generated if not provided).

Returns:
    Created GraphNode object.

Example:
    >>> node = runtime.graph.create_node(
    ...     labels=["Person"],
    ...     properties={"name": "Alice", "age": 30}
    ... )

def get_node(self, node_id: str) -> Optional[GraphNode]

Get a node by ID.

Args:
    node_id: The node's unique identifier.

Returns:
    GraphNode if found, None otherwise.

Example:
    >>> node = runtime.graph.get_node("node-123")

def update_node(self, node_id: str, properties: Optional[dict[str, Any]]=None, add_labels: Optional[list[str]]=None, remove_labels: Optional[list[str]]=None) -> GraphNode

Update a node's properties or labels.

Args:
    node_id: The node's unique identifier.
    properties: Properties to set (merged with existing).
    add_labels: Labels to add.
    remove_labels: Labels to remove.

Returns:
    Updated GraphNode object.

Example:
    >>> node = runtime.graph.update_node(
    ...     "node-123",
    ...     properties={"age": 31},
    ...     add_labels=["Premium"]
    ... )

def delete_node(self, node_id: str, detach: bool=False) -> bool

Delete a node.

Args:
    node_id: The node's unique identifier.
    detach: If True, also delete connected edges (default: False).

Returns:
    True if deleted, False if not found.

Example:
    >>> deleted = runtime.graph.delete_node("node-123", detach=True)

def create_edge(self, source_id: str, target_id: str, edge_type: str, properties: Optional[dict[str, Any]]=None, edge_id: Optional[str]=None) -> GraphEdge

Create an edge between two nodes.

Args:
    source_id: Source node ID.
    target_id: Target node ID.
    edge_type: Type/label of the edge (e.g., "KNOWS", "OWNS").
    properties: Edge properties as key-value pairs.
    edge_id: Optional custom ID (auto-generated if not provided).

Returns:
    Created GraphEdge object.

Example:
    >>> edge = runtime.graph.create_edge(
    ...     "alice-id",
    ...     "bob-id",
    ...     "KNOWS",
    ...     properties={"since": 2020}
    ... )

def get_edge(self, edge_id: str) -> Optional[GraphEdge]

Get an edge by ID.

Args:
    edge_id: The edge's unique identifier.

Returns:
    GraphEdge if found, None otherwise.

def delete_edge(self, edge_id: str) -> bool

Delete an edge.

Args:
    edge_id: The edge's unique identifier.

Returns:
    True if deleted, False if not found.

def get_neighbors(self, node_id: str, direction: str='both', edge_types: Optional[list[str]]=None, labels: Optional[list[str]]=None, limit: int=100) -> list[GraphNode]

Get neighboring nodes.

Args:
    node_id: The source node ID.
    direction: Edge direction ("outgoing", "incoming", "both").
    edge_types: Filter by edge types.
    labels: Filter neighbors by labels.
    limit: Maximum number of neighbors to return.

Returns:
    List of neighboring GraphNode objects.

Example:
    >>> friends = runtime.graph.get_neighbors(
    ...     "alice-id",
    ...     direction="outgoing",
    ...     edge_types=["KNOWS"]
    ... )

def get_edges(self, node_id: str, direction: str='both', edge_types: Optional[list[str]]=None, limit: int=100) -> list[GraphEdge]

Get edges connected to a node.

Args:
    node_id: The node ID.
    direction: Edge direction ("outgoing", "incoming", "both").
    edge_types: Filter by edge types.
    limit: Maximum number of edges to return.

Returns:
    List of GraphEdge objects.

def query(self, cypher: str, parameters: Optional[dict[str, Any]]=None) -> dict[str, Any]

Execute a Cypher query.

Args:
    cypher: The Cypher query string.
    parameters: Query parameters.

Returns:
    Query results with columns and rows.

Example:
    >>> result = runtime.graph.query(
    ...     "MATCH (p:Person)-[:KNOWS]->(f) WHERE p.name = $name RETURN f",
    ...     parameters={"name": "Alice"}
    ... )
    >>> for row in result["rows"]:
    ...     print(row)

def shortest_path(self, source_id: str, target_id: str, max_depth: int=10, edge_types: Optional[list[str]]=None) -> Optional[list[dict[str, Any]]]

Find the shortest path between two nodes.

Args:
    source_id: Source node ID.
    target_id: Target node ID.
    max_depth: Maximum path length.
    edge_types: Allowed edge types (all if not specified).

Returns:
    List of path segments, or None if no path exists.

Example:
    >>> path = runtime.graph.shortest_path("alice-id", "charlie-id")
    >>> if path:
    ...     print(f"Path length: {len(path)}")

class KVClient()

Client for key-value store operations.

Provides methods for getting, setting, and deleting key-value pairs,
as well as batch operations and transactions.

Example:
    >>> runtime = client.runtime("https://my-instance.akasha.ooo")
    >>> runtime.kv.set("user:123", {"name": "Alice", "age": 30})
    >>> user = runtime.kv.get("user:123")

def __init__(self, http_client: httpx.Client) -> None

def get(self, key: str, namespace: str='default') -> Optional[Any]

Get a value by key.

Args:
    key: The key to retrieve.
    namespace: Key namespace (default: "default").

Returns:
    The value if found, None otherwise.

Example:
    >>> value = runtime.kv.get("user:123")
    >>> if value is not None:
    ...     print(f"Found: {value}")

def set(self, key: str, value: Any, namespace: str='default', ttl_seconds: Optional[int]=None) -> None

Set a key-value pair.

Args:
    key: The key to set.
    value: The value to store (must be JSON-serializable).
    namespace: Key namespace (default: "default").
    ttl_seconds: Time-to-live in seconds (None for no expiration).

Example:
    >>> runtime.kv.set("user:123", {"name": "Alice"})
    >>> runtime.kv.set("session:abc", {"token": "xyz"}, ttl_seconds=3600)

def delete(self, key: str, namespace: str='default') -> bool

Delete a key.

Args:
    key: The key to delete.
    namespace: Key namespace (default: "default").

Returns:
    True if the key was deleted, False if it didn't exist.

Example:
    >>> deleted = runtime.kv.delete("user:123")

def exists(self, key: str, namespace: str='default') -> bool

Check if a key exists.

Args:
    key: The key to check.
    namespace: Key namespace (default: "default").

Returns:
    True if the key exists, False otherwise.

Example:
    >>> if runtime.kv.exists("user:123"):
    ...     print("User exists")

def list_keys(self, prefix: Optional[str]=None, namespace: str='default', limit: int=100, cursor: Optional[str]=None) -> dict[str, Any]

List keys matching a prefix.

Args:
    prefix: Key prefix filter.
    namespace: Key namespace (default: "default").
    limit: Maximum number of keys to return.
    cursor: Pagination cursor from previous response.

Returns:
    Dictionary with 'keys' list and optional 'cursor' for pagination.

Example:
    >>> result = runtime.kv.list_keys(prefix="user:")
    >>> for key in result["keys"]:
    ...     print(key)

def batch_get(self, keys: list[str], namespace: str='default') -> dict[str, Any]

Get multiple keys in a single request.

Args:
    keys: List of keys to retrieve.
    namespace: Key namespace (default: "default").

Returns:
    Dictionary mapping keys to their values (missing keys omitted).

Example:
    >>> results = runtime.kv.batch_get(["user:1", "user:2", "user:3"])
    >>> for key, value in results.items():
    ...     print(f"{key}: {value}")

def batch_set(self, items: dict[str, Any], namespace: str='default', ttl_seconds: Optional[int]=None) -> None

Set multiple key-value pairs in a single request.

Args:
    items: Dictionary of key-value pairs to set.
    namespace: Key namespace (default: "default").
    ttl_seconds: TTL for all keys (None for no expiration).

Example:
    >>> runtime.kv.batch_set({
    ...     "user:1": {"name": "Alice"},
    ...     "user:2": {"name": "Bob"},
    ... })

def batch_delete(self, keys: list[str], namespace: str='default') -> int

Delete multiple keys in a single request.

Args:
    keys: List of keys to delete.
    namespace: Key namespace (default: "default").

Returns:
    Number of keys that were deleted.

Example:
    >>> deleted_count = runtime.kv.batch_delete(["user:1", "user:2"])

def increment(self, key: str, delta: int=1, namespace: str='default') -> int

Atomically increment a numeric value.

Args:
    key: The key to increment.
    delta: Amount to increment by (can be negative).
    namespace: Key namespace (default: "default").

Returns:
    The new value after incrementing.

Example:
    >>> new_count = runtime.kv.increment("page_views", delta=1)

class EpisodicMemoryClient()

Client for episodic memory operations (temporal events).

def __init__(self, http_client: httpx.Client) -> None

def insert(self, event_type: str, content: dict[str, Any], timestamp: Optional[datetime]=None, tags: Optional[list[str]]=None, metadata: Optional[dict[str, Any]]=None) -> dict[str, Any]

Insert an episodic event.

Args:
    event_type: Type of event (e.g., "user_action", "system_event").
    content: Event content/payload.
    timestamp: Event timestamp (defaults to now).
    tags: Optional tags for filtering.
    metadata: Additional metadata.

Returns:
    Created event with ID.

def get(self, event_id: str) -> Optional[dict[str, Any]]

Get an episodic event by ID.

def update(self, event_id: str, content: dict[str, Any]) -> dict[str, Any]

Update an episodic event.