Skip to main content
For the complete documentation index optimized for AI agents, see llms.txt.
Queries for HelixDB can be authored directly in Python with the helix-db package, imported as helixdb. The Python SDK pairs a query-builder DSL with a small sync HTTP client. The API is Pythonic (read_batch, write_batch, var_as, value_map) and emits the same dynamic-query JSON AST as the Rust, TypeScript, and Go SDKs. For the traversal model and query patterns themselves, see Querying and the Querying Guide.

​
Prerequisites

  • Python 3.10 or later.
  • Optional: uv for fast environment and dependency management.
  • Optional: the Helix CLI for local development and ad-hoc query testing.

​
Create a project

uv init helix-python-app
cd helix-python-app

​
Add the dependency

Install the helix-db package from PyPI. 
uv add helix-db
Import the SDK from helixdb: 
from helixdb import Client, Predicate, define_params, g, param, read_batch, write_batch
A compatibility import path, helix_db, is also available for codebases that prefer underscore package names. With uv, run your script through the managed environment: 
uv run main.py

​
Write query functions

Python query builders are normal functions returning ReadBatch or WriteBatch. Use define_params for runtime values and pass the returned refs into predicates, limits, property inputs, search inputs, and mutations. 
from helixdb import Predicate, Projection, define_params, g, param, read_batch

find_users_params = define_params({
    "tenant_id": param.string(),
    "limit": param.i64(),
})


def find_users(p=find_users_params):
    return (
        read_batch()
        .var_as(
            "users",
            g()
            .n_with_label("User")
            .where(Predicate.eq("tenantId", p.tenant_id))
            .limit(p.limit)
            .project([
                Projection.property("$id", "id"),
                Projection.property("name"),
                Projection.property("tenantId"),
            ]),
        )
        .returning(["users"])
    )
Direct values are serialized as literals in the query AST. That is useful for true constants, but values that change per request should be declared as params so the query shape stays stable and the server can reuse cached work across requests.

​
Build dynamic requests

A batch becomes a dynamic request with to_dynamic_request(...) or to_dynamic_json(...): 
request = find_users().to_dynamic_request(
    find_users_params,
    {"tenant_id": "acme", "limit": 25},
    query_name="find_users",
)

body = find_users().to_dynamic_json(
    find_users_params,
    {"tenant_id": "acme", "limit": 25},
    query_name="find_users",
)
The request includes request_type, query_name, query, parameters, and parameter_types, ready to POST to /v1/query.

​
Execute queries

Create the client once and reuse it: 
from helixdb import Client, HelixError

client = Client("http://localhost:6969")

try:
    response = client.query().dynamic(request).send()
except HelixError as error:
    if error.kind == "Remote":
        raise RuntimeError(error.details) from error
    raise

users = response["users"]
For Helix Cloud, pass the cluster URL and API key: 
client = Client("https://helix.example.com", api_key="hx_secret")
Use request-builder options when a write must hit the writer node or wait for durability: 
created = (
    client
    .query()
    .writer_only()
    .should_await_durability(True)
    .dynamic(create_user_request)
    .send()
)
Warm read-query caches with warm_only(): 
client.query().warm_only().dynamic(read_request).send()
Stored routes post to /v1/query/{name}: 
response = client.query().body({"tenant_id": "acme"}).stored("find_users").send()

​
Write queries

from helixdb import define_params, g, param, write_batch

create_user_params = define_params({
    "name": param.string(),
    "tenant_id": param.string(),
})


def create_user(p=create_user_params):
    return (
        write_batch()
        .var_as("user", g().add_n("User", {"name": p.name, "tenantId": p.tenant_id}))
        .returning(["user"])
    )
read_batch().var_as(...) rejects write traversals. Use write_batch() for any node/edge creation, property update/removal, drop, or index mutation.

​
Bundles

Python can also generate query bundles: 
from helixdb import define_queries, register_read, register_write

queries = define_queries({
    "read": {"find_users": register_read(find_users, find_users_params)},
    "write": {"create_user": register_write(create_user, create_user_params)},
})

# Dynamic request with query_name="find_users".
request = queries.call.find_users({"tenant_id": "acme", "limit": 25})

# Write queries.json.
queries.generate("queries.json")
Route names must be unique across read and write routes. Bundles serialize with version 4, matching the other SDKs.

​
Handle conflicts in application code

The Python client does not retry HTTP 409 Conflict responses automatically. Retry only when the operation is safe to replay. Remote errors are raised as HelixError with kind == "Remote", details, and status_code populated. 
for attempt in range(3):
    try:
        return client.query().dynamic(request).send()
    except HelixError as error:
        if error.kind != "Remote" or error.status_code != 409 or attempt == 2:
            raise

​
Next Steps

Querying

Dynamic query envelopes, client execution, and transactions.

Parameters & bundles

Parameter serialization across TypeScript, Rust, Go, and Python.

Local Development

Run HelixDB locally while developing queries.

Python SDK source

Python package source and README.