Skip to main content

Installation

pip install nebula-client

Quick Start

# Set environment variable
export NEBULA_API_KEY="your_api_key_here"
from nebula import Nebula, Memory

# Automatically uses NEBULA_API_KEY environment variable
nebula = Nebula()

# Create a collection
collection = nebula.create_collection(name="my-collection")

# Store a memory
memory_id = nebula.store_memory(
    Memory(
        collection_id=collection.id,
        content="Nebula makes memory management easy",
        metadata={"topic": "nebula"}
    )
)

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

for fact in results.semantics:
    print(fact)
For more detailed examples, see Memory Operations.

Async Support

The Python SDK includes a dedicated async client:
import asyncio
from nebula import AsyncNebula, Memory

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

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

    memory_id = await nebula.store_memory(
        Memory(
            collection_id=collection.id,
            content="Async operations are fast",
            metadata={"type": "example"}
        )
    )

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

    for fact in results.semantics:
        print(fact)

    # Clean up
    await nebula.aclose()

asyncio.run(main())

Async Context Manager

import asyncio
from nebula import AsyncNebula, Memory

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

        memory_id = await nebula.store_memory(
            Memory(collection_id=collection.id, content="Context managers are clean")
        )

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

        print(f"Found {len(results.semantics)} facts")

asyncio.run(main())

Next Steps