Python Client

pip install nebula-sdk

# Set environment variable
export NEBULA_API_KEY="your_api_key_here"

from nebula import Nebula

# Automatically uses NEBULA_API_KEY environment variable
nebula = Nebula()

# Create a collection
collection = nebula.collections.create(name="my-collection").results

# Store a memory
created = nebula.memories.create(
    collection_id=collection.id,
    raw_text="Nebula makes memory management easy",
    metadata={"topic": "nebula"},
).results
memory_id = created.id

# Search memories
results = nebula.memories.search(
    query="memory management",
    collection_ids=[collection.id],
).results

for fact in results.semantic or []:
    print(fact)

import asyncio
from nebula import AsyncNebula

async def main():
    # Create async client (uses NEBULA_API_KEY env var)
    nebula = AsyncNebula()

    # All operations are async
    collection = (
        await nebula.collections.create(name="async_notes", description="Async example")
    ).results

    created = (
        await nebula.memories.create(
            collection_id=collection.id,
            raw_text="Async operations are fast",
            metadata={"type": "example"},
        )
    ).results

    results = (
        await nebula.memories.search(
            query="async",
            collection_ids=[collection.id],
        )
    ).results

    for fact in results.semantic or []:
        print(fact)

    # Clean up
    await nebula.close()

asyncio.run(main())

import asyncio
from nebula import AsyncNebula

async def main():
    # Automatically handles cleanup
    async with AsyncNebula() as nebula:
        collection = (
            await nebula.collections.create(name="notes", description="My notes")
        ).results

        await nebula.memories.create(
            collection_id=collection.id,
            raw_text="Context managers are clean",
        )

        results = (
            await nebula.memories.search(
                query="clean",
                collection_ids=[collection.id],
            )
        ).results

        print(f"Found {len(results.semantic or [])} facts")

asyncio.run(main())