Push any JSON. Get graph relationships and vector search — automatically. No schema. No pipeline. No glue code.
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.
pip install rushdbGet 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'))# 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'},
})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},
)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 |
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({...}).
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: Alicewith 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 exceptionfrom rushdb import RushDB
db = RushDB(
'RUSHDB_API_KEY',
url='http://your-rushdb-server.com/api/v1', # default: https://api.rushdb.com/api/v1
timeout=30,
)docs.rushdb.com/python-sdk — full API reference, vector search, aggregations, and more.
- GitHub Issues — bug reports and feature requests
- Email — direct support
See CONTRIBUTING.md. Issues and PRs welcome.