ainative-agent 0.1.0

Python SDK for AINative Agent APIs - agents, swarm, memory, graph, vectors, chat, files, tables, threads, events

ainative-agent

Python SDK for all AINative Agent APIs -- agents, swarm tasks, memory, graph, vectors, chat, files, tables, threads, and real-time events.

Install

pip install ainative-agent

Quick Start

import asyncio
from ainative_agent import AINativeClient

async def main():
    async with AINativeClient(api_key="sk-your-key") as client:
        # Register an agent
        agent = await client.agents.register(
            "my-agent", "worker", ["read", "write"]
        )

        # Store a memory
        memory = await client.memory.remember(
            "User prefers dark mode",
            entity_id="user-123",
            importance=0.8,
        )

        # Recall memories
        results = await client.memory.recall("user preferences", entity_id="user-123")

        # Submit a swarm task
        task = await client.tasks.submit(
            "Analyze error logs from last 24h",
            agent_types=["analyst"],
        )

        # Poll until complete
        async for update in client.tasks.poll(task.task_id):
            print(f"Status: {update.status}")

asyncio.run(main())

Modules

Module	Description
client.agents	Agent registration, keys, and audit (AAP)
client.tasks	Swarm task submission and polling
client.memory	ZeroMemory: remember, recall, forget, reflect
client.graph	Context graph: entities, edges, GraphRAG, ontology
client.vectors	Vector embeddings: upsert, search, delete
client.files	File storage: upload, download, presigned URLs
client.tables	NoSQL tables: create, query, insert, update
client.chat	Chat completions with streaming
client.threads	Conversation threads
client.events	Real-time WebSocket events

Authentication

The SDK supports two authentication methods:

# API Key (recommended for server-side)
client = AINativeClient(api_key="sk-your-key")

# JWT Token (for user-scoped access)
client = AINativeClient(token="eyJ...")

Error Handling

from ainative_agent import AINativeClient, AuthenticationError, RateLimitError, APIError

async with AINativeClient(api_key="sk-key") as client:
    try:
        agent = await client.agents.get("nonexistent")
    except AuthenticationError:
        print("Invalid credentials")
    except RateLimitError as e:
        print(f"Rate limited, retry after {e.retry_after}s")
    except APIError as e:
        print(f"API error {e.status_code}: {e.message}")

Retry Behavior

The SDK automatically retries requests on transient errors (429, 5xx) with exponential backoff, up to 3 attempts.

License

Apache-2.0