HydraDB Python SDK

The official Python SDK for HydraDB — a managed retrieval engine that combines vector search, full‑text search, and a knowledge graph behind a single API.

• Package: hydradb-sdk
• Import module: hydra_db
• Version: 2.1.1 (API version 2)
• Docs: https://docs.hydradb.com
• Python: 3.10+

---

Table of contents

• Installation
• Quick start
• Client configuration
• Core concepts
• Endpoints
  • query — unified retrieval
  • Context (client.context)
    • ingest
    • list
    • inspect
    • status
    • relations
    • update_source_metadata
    • delete
  • Databases (client.databases)
    • create
    • list
    • collections
    • stats
    • status
    • delete
  • Webhooks (client.webhooks)
    • register
    • get
    • test
    • delete
    • list_deliveries
    • get_delivery
    • retry_delivery
• Async usage
• Error handling
• Advanced
  • Request options (timeouts & retries)
  • Raw responses
  • Custom HTTP client

---

Installation

pip install hydradb-sdk

Quick start

from hydra_db import HydraDB

client = HydraDB(
    token="YOUR_API_KEY",      # your HydraDB API key (bearer token)
    api_version="2",           # optional, defaults to "2"
)

# Run a hybrid search over a database ("tenant")
result = client.query(
    query="What is our refund policy?",
    database="acme-corp",
    type="knowledge",
    max_results=5,
)

for chunk in result.data.chunks:
    print(chunk)

Every method returns a typed response object (Pydantic models). Responses are wrapped in a HandlerEnvelope… type — the payload lives on .data, with request metadata on .meta.

---

Client configuration

from hydra_db import HydraDB
from hydra_db.environment import HydraDBEnvironment

client = HydraDB(
    token="YOUR_API_KEY",
    api_version="2",
    environment=HydraDBEnvironment.DEFAULT,   # https://api.hydradb.com
    # base_url="https://api.hydradb.com",     # override for self-hosted / staging
    timeout=60.0,                             # seconds; default 60
    headers={"X-Custom-Header": "value"},     # sent on every request
)

Parameter	Type	Default	Notes
token	str | Callable[[], str]	–	Bearer token. Pass a callable for dynamic/refreshing tokens.
api_version	str	"2"	Sets the API version header.
environment	HydraDBEnvironment	DEFAULT	DEFAULT → https://api.hydradb.com.
base_url	str	–	Explicit URL; overrides environment.
timeout	float	60	Per‑request timeout in seconds.
headers	dict[str, str]	–	Extra headers on every request.
follow_redirects	bool	True	–
httpx_client	httpx.Client	–	Bring your own configured client.

---

Core concepts

Database vs. Collection (tenant vs. sub‑tenant). HydraDB v2 renamed the isolation scopes:

v2 name (canonical)	v1 alias (deprecated, still accepted)	Meaning
database	tenant_id	Top‑level isolation boundary.
collection	sub_tenant_id	A namespace within a database.

The server’s TenantAliases middleware reconciles the two, so you can pass either — but new code should use database / collection. The legacy aliases will be removed in a future release.

Corpora (type). Data is split into two corpora you can target independently: "knowledge" (documents), "memory" (agent memories), or "all".

---

Endpoints

query — unified retrieval

POST /query → HandlerEnvelopeSearchV2RetrievalResult

The single retrieval endpoint. Dispatches across corpus (type) and retrieval method (query_by), optionally enriching results with knowledge‑graph context.

from hydra_db import HydraDB

client = HydraDB(token="YOUR_API_KEY")

result = client.query(
    query="How do I rotate API keys?",
    database="acme-corp",           # v2 name for the tenant scope
    type="knowledge",               # "knowledge" | "memory" | "all"
    query_by="hybrid",              # "hybrid" | "text"
    mode="auto",                    # "fast" | "thinking" | "auto"
    operator="or",                  # "or" | "and" | "phrase"
    max_results=10,
    num_related_chunks=3,
    graph_context=True,             # include KG context (default True)
    recency_bias=0.2,
    metadata_filters={              # exact-match on tenant/document metadata
        "department": "security",
        "additional_metadata": {"author": "ada"},
    },
)

print(result.data)

Scoping to specific collections (preferred over the deprecated sub_tenant_ids):

# Equal weighting across collections
client.query(query="pricing", database="acme-corp", collections=["eu", "us"])

# Weighted ranking (one decimal place max)
client.query(query="pricing", database="acme-corp", collections={"eu": 1.0, "us": 0.5})

Scoping to specific source IDs — ids applies a hard source_id in [...] pre‑filter; if nothing matches it returns empty rather than widening to the whole corpus:

client.query(query="onboarding", database="acme-corp", ids=["doc_123", "doc_456"])

Key parameters:

Parameter	Type	Notes
query	str	The search text.
database	str	Tenant scope (v2). Alias: tenant_id.
collection / collections	str / list | dict	Sub‑tenant scope. Prefer over sub_tenant_id(s).
type	"knowledge" | "memory" | "all"	Corpus to query.
query_by	"hybrid" | "text"	Retrieval method.
mode	"fast" | "thinking" | "auto"	Recall mode.
operator	"or" | "and" | "phrase"	Text‑match operator.
max_results	int	Result cap.
num_related_chunks	int	Neighboring chunks to attach.
graph_context	bool	Include KG context. Default True.
query_apps	bool	App‑aware knowledge retrieval.
query_forceful_relations	bool	Force relation expansion. Default True.
metadata_filters	dict[str, Any]	Exact‑match on metadata (nest under additional_metadata for doc metadata).
recency_bias	float	Boost newer sources.
ids	list[str]	Restrict to specific source IDs.

---

Context (client.context)

Everything about the data inside a database: ingesting, listing, inspecting, updating metadata, checking processing status, reading graph relations, and deleting.

context.ingest

POST (multipart) → HandlerEnvelopeIngestionV2SourceUploadResponse

Ingest knowledge documents or memories. documents is a list of file uploads (one request may carry several); the other fields are form fields (JSON strings where structured).

from hydra_db import HydraDB

client = HydraDB(token="YOUR_API_KEY")

# Ingest a document file
with open("handbook.pdf", "rb") as f:
    resp = client.context.ingest(
        database="acme-corp",           # required
        documents=[f],                  # list: file-like / (filename, bytes) / bytes, one entry per file
        collection="hr",
        type="knowledge",
        # document_metadata is a JSON *array* — one object per uploaded file. Your own
        # per-document fields go under "additional_metadata"; the item's other top-level keys
        # are the API's (metadata, evidence_kind, evidence_subject, id, relations, ...) —
        # an unknown top-level key such as "title" is rejected with 400.
        document_metadata='[{"additional_metadata": {"title": "Employee Handbook", "author": "HR"}}]',
        upsert="true",                  # form field is a string
    )

print(resp.data)

# Ingest memories (no file). Each item needs "text" (or "user_assistant_pairs").
client.context.ingest(
    database="acme-corp",
    memories='[{"text": "User prefers dark mode"}]',
    type="memory",
)

Parameter	Type	Notes
database	str (required)	Database (tenant scope). Alias: tenant_id.
documents	List[core.File]	File uploads (path handles, bytes, or (name, bytes) tuples), one entry per file.
memories	str	JSON array string; each item needs text (or user_assistant_pairs).
document_metadata	str	JSON array string — one object per uploaded file (count must match). Put your own fields under additional_metadata; the API's item keys (metadata, evidence_kind, evidence_subject, id, relations, ...) may sit alongside it. Unknown top-level keys return 400.
app_knowledge	str	App‑knowledge items as a JSON array string (not raw text).
graph_payload	str	Pre‑computed graph payload.
collection	str	Collection (sub‑tenant scope). Alias: sub_tenant_id.
type	str	"knowledge" or "memory".
upsert	str	"true" to upsert on existing IDs.

context.list

GET /context/list → HandlerEnvelopeListV2SourceListResponse

List sources or memories (IDs + metadata) for a database, with filtering and pagination.

from hydra_db import HydraDB
from hydra_db import ListContentFilter

client = HydraDB(token="YOUR_API_KEY")

resp = client.context.list(
    database="acme-corp",
    collection="hr",
    type="knowledge",
    page=1,
    page_size=50,
    include_fields=["title", "type", "timestamp"],
    filters=ListContentFilter(
        metadata={"department": "finance"},          # tenant/source metadata
        additional_metadata={"author": "ada"},       # document metadata
        source_fields={"type": "pdf"},               # well-known source fields
    ),
)

for source in resp.data.sources:
    print(source)

context.inspect

GET → HandlerEnvelopeFetchV2SourceFetchResponse

Fetch a single ingested source: its content, inferred content, and a presigned download URL.

resp = client.context.inspect(
    id="doc_1234",                 # required — source ID
    database="acme-corp",          # required
    collection="hr",
    expiry_seconds=3600,           # presigned URL lifetime
    mode="both",                   # fetch mode: "content", "url", or "both"
)
print(resp.data.presigned_url)

context.status

GET → HandlerEnvelopeIngestionV2BatchProcessingStatus

Check processing status for one or more source IDs.

# Single source
client.context.status(database="acme-corp", id="doc_1234", collection="hr")

# Batch
resp = client.context.status(
    database="acme-corp",
    ids=["doc_1", "doc_2", "doc_3"],
)
print(resp.data)

context.relations

GET → HandlerEnvelopeGraphGraphRelationsResponse

Return knowledge‑graph relations for a whole database or a single source.

resp = client.context.relations(
    database="acme-corp",          # required
    collection="hr",
    id="doc_1234",                 # omit for database-wide relations
    type="knowledge",              # "knowledge" | "memory"
    limit=100,
    cursor=0,                      # pagination cursor (float)
)
for triplet in resp.data.relations:
    print(triplet)

context.update_source_metadata

PATCH → HandlerEnvelope…MetadataEditResult

Merge/upsert tenant_metadata and additional_metadata for one source. collection (alias sub_tenant_id) is required by the server.

resp = client.context.update_source_metadata(
    id="doc_1234",                 # required — source ID
    database="acme-corp",
    collection="hr",               # required by the server
    # tenant_metadata keys must be declared in the database's tenant_metadata_schema
    # (and match the declared type). Use additional_metadata for free-form fields.
    tenant_metadata={"department": "finance"},
    additional_metadata={"author": "ada", "tags": ["policy", "2026"], "reviewed": True},
)
print(resp.data)

Note: although the SDK exposes a document_metadata parameter here, this endpoint rejects it (HTTP 400 "document_metadata is not accepted; use additional_metadata"). Put per-document fields in additional_metadata instead.

context.delete

DELETE → HandlerEnvelopeSourcesMemoryDeleteResponse

Delete one or more sources or memories by ID.

resp = client.context.delete(
    database="acme-corp",
    collection="hr",
    ids=["doc_1234", "doc_5678"],
    type="knowledge",
)
print(resp.data)

---

Databases (client.databases)

Manage databases (tenants) and inspect their collections, stats, and provisioning status.

databases.create

POST → HandlerEnvelopeTenantsTenantCreateAcceptedResponse

Create a new database, optionally with a custom metadata schema for its collections.

from hydra_db import HydraDB
from hydra_db import TenantsCustomPropertyDefinition

client = HydraDB(token="YOUR_API_KEY")

resp = client.databases.create(
    database="acme-corp",
    embeddings_dimension=1536,
    database_metadata_schema=[
        TenantsCustomPropertyDefinition(
            name="department",
            data_type="VARCHAR",   # BOOL | INT8..INT64 | FLOAT | DOUBLE | VARCHAR | JSON | ARRAY
            max_length=128,
            enable_match=True,
        ),
        TenantsCustomPropertyDefinition(name="priority", data_type="INT32"),
    ],
)
print(resp.data)

Creation is asynchronous — poll databases.status until infrastructure is provisioned before ingesting.

databases.list

GET → HandlerEnvelopeTenantsTenantIdsResponse

List all databases for the authenticated user.

resp = client.databases.list()
print(resp.data)

databases.collections

GET → HandlerEnvelopeTenantsSubTenantIdsResponse

List all collections within a database.

resp = client.databases.collections(database="acme-corp")
print(resp.data)

databases.stats

GET → HandlerEnvelopeTenantsTenantStatsResponse

Get collection statistics for a database.

resp = client.databases.stats(database="acme-corp")
print(resp.data)

databases.status

GET → HandlerEnvelopeTenantsInfraStatusResponseV2

Check infrastructure provisioning status for a database.

resp = client.databases.status(database="acme-corp")
print(resp.data)

databases.delete

DELETE → HandlerEnvelopeTenantsTenantDeleteResponse

Delete a database and all associated data.

resp = client.databases.delete(database="acme-corp")
print(resp.data)

---

Webhooks (client.webhooks)

Register a single indexing webhook per org and inspect/replay its deliveries.

webhooks.register

POST → HandlerEnvelopeWebhooksWebhookRegisterResponse

Register (or update) the indexing webhook for this API key’s org.

resp = client.webhooks.register(
    url="https://example.com/hooks/hydradb",
    event_types=["indexing.status_changed"],   # the only supported event type
    signing_secret="whsec_at_least_16_chars",  # must be >= 16 characters
)
print(resp.data)

webhooks.get

GET → HandlerEnvelopeWebhooksWebhookGetResponse

Fetch the currently registered webhook.

resp = client.webhooks.get()
print(resp.data)

webhooks.test

POST → HandlerEnvelopeWebhooksWebhookTestResponse

Send a test delivery to the registered endpoint.

resp = client.webhooks.test()
print(resp.data)

webhooks.delete

DELETE → HandlerEnvelopeWebhooksWebhookDeleteResponse

Remove the registered webhook.

resp = client.webhooks.delete()
print(resp.data)

webhooks.list_deliveries

GET → HandlerEnvelopeWebhooksDeliveryListResponse

List recent webhook deliveries, with filtering and cursor pagination.

resp = client.webhooks.list_deliveries(
    limit=50,
    cursor=None,                    # pass the previous page's cursor to continue
    status="failed",               # filter by delivery status
)
for delivery in resp.data.deliveries:
    print(delivery)

webhooks.get_delivery

GET → HandlerEnvelopeWebhooksDeliveryItem

Fetch a single delivery by ID.

resp = client.webhooks.get_delivery(delivery_id="dlv_1234")
print(resp.data)

webhooks.retry_delivery

POST → HandlerEnvelopeWebhooksRetryResponse

Re‑attempt a failed delivery.

resp = client.webhooks.retry_delivery(delivery_id="dlv_1234")
print(resp.data)

---

Async usage

Every method has an async twin on AsyncHydraDB. The API surface is identical — just await each call.

import asyncio
from hydra_db import AsyncHydraDB

client = AsyncHydraDB(token="YOUR_API_KEY")


async def main() -> None:
    result = await client.query(query="refund policy", database="acme-corp")
    print(result.data)

    dbs = await client.databases.list()
    print(dbs.data)


asyncio.run(main())

AsyncHydraDB also accepts an async_token callable for token acquisition that involves async I/O (e.g. refreshing a token over HTTP).

---

Error handling

Non‑2xx responses raise typed exceptions, all subclasses of ApiError. Each carries status_code and the parsed body.

from hydra_db import HydraDB
from hydra_db.core.api_error import ApiError
from hydra_db.errors import (
    BadRequestError,          # 400
    ForbiddenError,           # 403
    NotFoundError,            # 404
    ConflictError,            # 409
    UnprocessableEntityError, # 422
    InternalServerError,      # 500
)

client = HydraDB(token="YOUR_API_KEY")

try:
    client.databases.status(database="does-not-exist")
except NotFoundError as e:
    print("not found:", e.body)
except ApiError as e:
    print(f"API error {e.status_code}: {e.body}")

---

Advanced

Request options (timeouts & retries)

Pass request_options to any method to override per call.

client.query(
    query="hello",
    database="acme-corp",
    request_options={
        "timeout_in_seconds": 30,
        "max_retries": 3,
        "additional_headers": {"X-Trace-Id": "abc123"},
    },
)

Raw responses

Use with_raw_response to access status code and headers alongside the parsed body.

raw = client.with_raw_response.query(query="hello", database="acme-corp")
print(raw.headers)
print(raw.data)

Custom HTTP client

import httpx
from hydra_db import HydraDB

client = HydraDB(
    token="YOUR_API_KEY",
    httpx_client=httpx.Client(
        proxy="http://localhost:8080",   # httpx >= 0.26 renamed `proxies` -> `proxy`
        transport=httpx.HTTPTransport(retries=3),
    ),
)

---

Endpoint reference

Group	Method	Description
—	query	Unified hybrid/text retrieval across corpora with optional graph context.
context	ingest	Ingest documents or memories (multipart upload).
context	list	List sources/memories with metadata filters + pagination.
context	inspect	Fetch a source’s content + presigned download URL.
context	status	Processing status for one or many source IDs.
context	relations	Knowledge‑graph relations for a database or source.
context	update_source_metadata	Merge/upsert metadata for a source.
context	delete	Delete sources/memories by ID.
databases	create	Create a database with optional metadata schema.
databases	list	List all databases for the user.
databases	collections	List collections in a database.
databases	stats	Collection statistics for a database.
databases	status	Infrastructure provisioning status.
databases	delete	Delete a database and all its data.
webhooks	register	Register/update the org indexing webhook.
webhooks	get	Get the registered webhook.
webhooks	test	Send a test delivery.
webhooks	delete	Remove the webhook.
webhooks	list_deliveries	List recent deliveries.
webhooks	get_delivery	Fetch one delivery by ID.
webhooks	retry_delivery	Retry a failed delivery.

---

This SDK is generated from the HydraDB API definition. For the full parameter reference see src/hydra_db/reference.md or https://docs.hydradb.com.