​
Python Client

The official Python client library for the Nebula Memory API provides a simple and intuitive interface for storing, retrieving, and searching memories with advanced clustering capabilities.

​
Installation & Setup

Install: 
pip install nebula-client
Environment setup: 
export NEBULA_API_KEY="your_api_key_here"

​
Client Initialization

Basic usage: 
from nebula_client import NebulaClient

# Using environment variable
client = NebulaClient()

# Or with explicit API key
client = NebulaClient(api_key="your_api_key_here")
Advanced configuration: 
client = NebulaClient(
    api_key="your_api_key_here",
    base_url="https://api.nebulacloud.app",  # Default
    timeout=30.0  # Default timeout in seconds
)

​
Cluster Management

Create cluster: 
cluster = client.create_cluster(
    name="research_notes",
    description="Cluster for research-related memories",
    metadata={"category": "academic", "priority": "high"}
)
Get cluster: 
cluster = client.get_cluster("cluster_id_here")
List clusters: 
clusters = client.list_clusters(limit=100, offset=0)
Update cluster: 
updated_cluster = client.update_cluster(
    cluster_id="cluster_id_here",
    name="new_name",  # Optional
    description="new_description",  # Optional
    metadata={"updated": True}  # Optional
)
Delete cluster: 
success = client.delete_cluster("cluster_id_here")

​
Memory Storage

Store memory: 
# Using dict (recommended)
memory_id = client.store_memory({
    "cluster_id": "user_preferences",
    "content": "User prefers email communication over phone calls",
    "metadata": {
        "user_id": "user_123",
        "preference_type": "communication",
        "importance": "high"
    }
})

# Using Memory class
from nebula_client import Memory
memory = Memory(
    cluster_id="user_preferences",
    content="User prefers email communication over phone calls",
    metadata={"user_id": "user_123"}
)
memory_id = client.store_memory(memory)

​
Memory Retrieval

Get specific memory: 
memory = client.get_memory("memory_id_here")
List memories: 
memories = client.list_memories(
    cluster_ids=["cluster_123"],  # Required - list of cluster IDs
    limit=100,      # Default: 100
    offset=0        # Default: 0
)

​
Search Operations

Search memories: 
results = client.search(
    query="artificial intelligence",
    cluster_ids=["research_cluster"],            # Required - list of cluster IDs
    limit=10,                                    # Default: 10
    retrieval_type="simple",                     # Default: "simple"
    filters={"metadata.topic": "ai"}             # Optional
)
Search types:
  • simple: Basic semantic search
  • reasoning: Enhanced reasoning-based search
  • planning: Planning-oriented search

​
Memory Management

Delete memory: 
success = client.delete("memory_id_here")

​
Data Models

Memory class (for writing): 
from nebula_client import Memory

# Memory class structure
memory = Memory(
    cluster_id="user_preferences",  # Required: cluster to store in
    content="User prefers email communication",  # Required: text content
    role="user",  # Optional: 'user', 'assistant', or custom (for conversations)
    parent_id="conv_123",  # Optional: conversation ID (for conversations)
    metadata={"user_id": "user_123"}  # Optional: additional metadata
)
MemoryResponse class (for reading): 
# Returned by get_memory(), list_memories(), etc.
class MemoryResponse:
    id: str                    # Memory/document ID
    content: str              # Memory content (retrieved from chunks)
    metadata: dict            # Additional metadata
    cluster_ids: List[str]    # Associated clusters
    created_at: str           # Creation timestamp
    updated_at: str           # Last update timestamp

​
Error Handling

Exception types: 
from nebula_client.exceptions import (
    NebulaException,              # Base exception
    NebulaClientException,        # Client-side errors
    NebulaAuthenticationException, # Authentication errors
    NebulaRateLimitException,     # Rate limiting errors
    NebulaValidationException     # Validation errors
)
Error handling: 
try:
    memory = client.store_memory({
        "content": "Test memory",
        "cluster_id": "test_cluster"
    })
except NebulaAuthenticationException:
    print("Invalid API key")
except NebulaRateLimitException:
    print("Rate limit exceeded")
except NebulaValidationException as e:
    print(f"Validation error: {e}")
except NebulaException as e:
    print(f"API error: {e}")

​
Quick Example

from nebula_client import NebulaClient

# Initialize
client = NebulaClient(api_key="your_api_key")

# Create cluster and store memory
cluster = client.create_cluster("my_cluster", "Example cluster")
memory_id = client.store_memory({
    "cluster_id": cluster.id,
    "content": "Hello world",
    "metadata": {"test": True}
})

# Search
results = client.search("hello", cluster_ids=[cluster.id])

​
Next Steps