RushDB — Python SDK

The memory layer for AI agents and apps.

Push any JSON. Get graph relationships and vector search — automatically. No schema. No pipeline. No glue code.

📖 Documentation • 🌐 Website • ☁️ Cloud

---

Why RushDB

Agents need memory. Apps need connected data. The standard answer involves multiple databases, schema design, and an embedding pipeline before you write a single useful line of business logic.

RushDB skips all of that. Push any JSON — nested structure becomes a traversable graph, string properties become semantically searchable, type inference happens automatically.

Works with LangChain, CrewAI, AutoGen, or any Python AI framework.

---

Installation

pip install rushdb

---

RushDB Cloud or a self-hosted engine

The SDK works with RushDB in one of two ways:

	RushDB Cloud	Self-hosted RushDB engine (direct mode)
What you run	nothing: the managed service at app.rushdb.com	one rushdb binary, on your own infrastructure
Connect with	an API key	the engine URL and the engine token

The engine is what you can self-host. RushDB Cloud runs that engine for you and adds the dashboard, managed embeddings and the AI features (natural-language search, relationship-pattern suggestions), which exist only in RushDB Cloud.

from rushdb import RushDB

# RushDB Cloud
db = RushDB("RUSHDB_API_KEY")

# A self-hosted engine
db = RushDB(
    "ENGINE_TOKEN",
    direct=True,
    engine_url="http://127.0.0.1:3000",
    scope="tenant-1",  # only for an engine started with --multitenant
)

What each offers:

Feature	Cloud	Self-hosted engine
Records: create, read, update, set, upsert, delete, find with where, orderBy, select, groupBy	✓	✓
Relationships: attach/detach, relationships.create_many/delete_many/find	✓	✓
Labels, properties, property values	✓	✓
Import JSON and CSV, export CSV	✓	✓
External vector indexes; vector search with a queryVector you supply	✓	✓
Raw openCypher (query.raw; the engine needs --languages cypher)	✓	✓
Schema (ai.get_schema, ai.get_schema_markdown)	✓	✓
Keyword (BM25) search (records.text_search), text indexes (text_indexes.*), hybrid search	✓	✓
Transactions	✓	record creates, updates, sets, deletes by id, attaches and imports are staged; other writes refuse a transaction
Managed embeddings: managed indexes, records.vector_search by text query	✓	—
Natural-language search (ai.search)	✓	—
Relationship-pattern suggestions (relationships.patterns.*)	✓	—
Project settings (settings.get), dashboard, team access	✓	—

In direct mode a Cloud-only method raises CloudOnlyError (with method and reason) before sending anything, and no request ever goes to RushDB Cloud. A write the engine cannot stage into a transaction raises DirectModeError, also before sending anything. The list is available in code:

from rushdb import CLOUD_ONLY_METHODS, CloudOnlyError, is_cloud_only_method

db.engine.execute() runs any Protocol v1 operation directly.

---

Agent memory in 3 lines

Get an API key at app.rushdb.com.

from rushdb import RushDB

db = RushDB('RUSHDB_API_KEY')

# Store an agent action — graph links sessions and context automatically
db.records.create(
    label='MEMORY',
    data={
        'agent_id': 'agent-42',
        'session_id': 'sess-001',
        'action': 'summarized',
        'topic': 'Q4 results',
        'output': summary_text,
    },
)

# Recall — traverse relationships, filter by properties
results = db.records.find({
    'labels': ['MEMORY'],
    'where': {
        'agent_id': 'agent-42',
        'topic': {'$contains': 'Q4'},
    },
    'limit': 10,
})

for memory in results:
    print(memory.get('output'))

---

Graph traversal

# Push nested JSON — relationships created automatically
db.records.create_many('COMPANY', {
    'name': 'Acme Corp',
    'DEPARTMENT': [{
        'name': 'Engineering',
        'EMPLOYEE': [{
            'name': 'Alice',
            'role': 'Staff Engineer',
        }]
    }]
})

# Traverse the auto-created graph
engineers = db.records.find({
    'labels': ['EMPLOYEE'],
    'where': {
        'role': {'$contains': 'Engineer'},
        'DEPARTMENT': {'COMPANY': {'name': 'Acme Corp'}},
    },
})

# Constrain by relationship type and direction
authored_posts = db.records.find({
    'labels': ['USER'],
    'where': {
        'POST': {
            '$relation': {'type': 'AUTHORED', 'direction': 'out'},
            'title': {'$contains': 'graph'},
        }
    },
    'limit': 10,
})

# Multi-hop: add hops to $relation — everyone in Alice's reporting chain, up to 4 levels
chain = db.records.find({
    'labels': ['EMPLOYEE'],
    'where': {
        'EMPLOYEE': {
            '$relation': {'type': 'REPORTS_TO', 'direction': 'out', 'hops': {'min': 1, 'max': 4}},
            'name': {'$contains': 'Alice'},
        }
    },
})

# Cycle detection: accounts on a circular transfer ring (fraud rings, circular ownership)
ring_members = db.records.find({
    'labels': ['ACCOUNT'],
    'where': {
        'RING': {  # key is a display name — the $cycle block holds only $relation
            '$cycle': True,
            '$relation': {'type': 'TRANSFERRED_TO', 'direction': 'out', 'hops': {'min': 2, 'max': 6}},
        }
    },
})

# Manage relationships explicitly
user = db.records.find_uniq({'labels': ['USER'], 'where': {'name': 'Alice'}})
company = db.records.find_uniq({'labels': ['COMPANY'], 'where': {'name': 'Acme Corp'}})
user.attach(
    target=company,
    options={'type': 'WORKS_AT', 'direction': 'out', 'properties': {'source': 'profile'}},
)

# Relationship search: where filters edge type/properties, source/target filter endpoint records
relationships = db.relationships.find({
    'source': {'labels': ['USER'], 'where': {'name': 'Alice'}},
    'target': {'labels': ['COMPANY']},
    'where': {'type': 'WORKS_AT', 'source': 'profile'},
})

---

Importing CSV

csv_data = "name,email,age\nJohn,john@example.com,30\nJane,jane@example.com,25"

db.records.import_csv(
    label='USER',
    data=csv_data,
    # skipEmptyValues: treat empty cells ("" / []) as unset instead of storing them (0/False are kept)
    options={'returnResult': True, 'suggestTypes': True, 'skipEmptyValues': True},
    parse_config={'header': True, 'skipEmptyLines': True, 'dynamicTyping': True},
)

---

SearchResult

db.records.find() returns a SearchResult — a list-like container with pagination metadata.

result = db.records.find({
    'where': {'status': 'active'},
    'limit': 10,
    'skip': 0,
})

# List-like usage
print(f"Loaded {len(result)} of {result.total} total")
print(f"Has more: {result.has_more}")

for record in result:
    print(record.get('name'))

# Indexing and slicing
first = result[0]
top_five = result[:5]

# Boolean check
if result:
    process(result[0])

Property	Type	Description
data	List[Record]	The result items
total	int	Total matching records in the database
has_more	bool	Whether more records exist beyond this page
search_query	dict	The query that produced this result

---

Vector Search And Smart Search

Use db.records.vector_search() for direct semantic/vector retrieval over an embedding index:

results = db.records.vector_search({
    'labels': ['MEMORY'],
    'propertyName': 'content',
    'query': 'how agents remember things',
    'where': {'agent_id': 'agent-42'},
    'limit': 5,
})

for record in results:
    print(record.score, record.get('content'))

Use db.ai.search() when you want RushDB to turn a natural-language request into a SearchQuery and execute it:

results = db.ai.search('Find active memories about Q4 results for agent-42')
print(results.search_query)

db.ai.search({...}) still works as a deprecated vector-search alias, but new code should use db.records.vector_search({...}).

Keyword And Hybrid Search

Declare a BM25 text index, then search it by keywords:

db.text_indexes.create({'label': 'MEMORY', 'property': 'content'})

results = db.records.text_search({
    'query': 'quarterly results',
    'labels': ['MEMORY'],
    'where': {'agent_id': 'agent-42'},
    'limit': 5,
})
print(results.total, results.meta['textSearch'])

Add a text leg to a vector search to fuse keyword and vector ranking. fusion is rrf (default) or dbsf; weight and prefetchLimit tune each leg:

results = db.records.vector_search({
    'labels': ['MEMORY'],
    'propertyName': 'content',
    'queryVector': embedding,
    'text': {'query': 'quarterly results', 'weight': 0.5},
    'fusion': 'dbsf',
    'limit': 5,
})

---

Record API

user = db.records.create('USER', {
    'name': 'Alice',
    'email': 'alice@example.com',
})

# Safe field access
name = user.get('name')                  # 'Alice'
phone = user.get('phone', 'N/A')         # 'N/A'

# Clean data (excludes internal __id, __label fields)
data = user.get_data()                   # {'name': 'Alice', 'email': '...'}
full = user.get_data(exclude_internal=False)  # includes __id, __label, etc.

# Existence check (no exception if record was deleted)
if user.exists:
    user.update({'status': 'active'})

# String representations
repr(user)   # Record(id='abc-123', label='USER')
str(user)    # USER: Alice

---

Transactions

with db.transactions.begin() as tx:
    record_a = db.records.create('NODE', {'value': 1}, transaction=tx)
    record_b = db.records.create('NODE', {'value': 2}, transaction=tx)
    record_a.attach(target=record_b, options={'type': 'LINKED'}, transaction=tx)
# auto-committed on exit, rolled back on exception

---

Configuration

from rushdb import RushDB

db = RushDB(
    'RUSHDB_API_KEY',
    url='https://api.rushdb.com/api/v1',  # a different RushDB Cloud API endpoint; this is the default
    timeout=30,
)

---

Documentation

docs.rushdb.com/python-sdk — full API reference, vector search, aggregations, and more.

---

Support

• GitHub Issues — bug reports and feature requests
• Email — direct support

---

Contributing

See CONTRIBUTING.md. Issues and PRs welcome.