Python SDK

rekall-search — the same shape as the TypeScript SDK, Pythonic surface. Sync-first with AsyncRekall, search_many for eval loops, and LangChain / LlamaIndex adapters in-package.

rekall-search gives you a Pythonic client over the REST API. Sync-first — calls are ~4 ms, so async matters less than for LLM RAG — with AsyncRekall for async apps.

Install

pip install rekall-search

uv add rekall-search

Client

from rekall import Rekall

rekall = Rekall()                       # reads REKALL_API_KEY / REKALL_BASE_URL

# Or configure explicitly (e.g. self-hosted):
rekall = Rekall(api_key="rk_…", base_url="http://localhost:9009/v1")

Prop

Type

Stores

rekall.store("name") is get-or-create. Explicit rekall.stores.* methods exist for full control.

store = rekall.store("contracts")            # get-or-create

s    = rekall.stores.create("contracts")
g    = rekall.stores.get("contracts")        # by name or id
page = rekall.stores.list(limit=20)          # .stores, .next_cursor
rekall.stores.delete("contracts")

Ingest

Whole documents — no chunking. Searchable the moment ingest returns.

res = store.ingest([
    {"id": "supplier_a", "text": supplier_text, "metadata": {"team": "procurement"}},
    {"id": "policy_2026", "text": policy_text},
])
print(res.ingested, res.doc_ids)

Files and streaming (big corpora)

# Extraction parity with the CLI — walks the glob, extracts pdf/md/html/docx/txt:
store.ingest_files("docs/", glob="**/*.pdf")

# O(N), backpressure-aware; shows a tqdm bar if tqdm is installed:
def documents():
    for path in corpus_paths:
        yield {"id": path, "text": read(path), "metadata": {"path": path}}

store.ingest_stream(documents(), batch_size=256)

Search

result = store.search(
    "which supplier contract conflicts with the new policy?",
    max_hops=4, max_results=5, trace=True, filter={"team": "procurement"},
)

for p in result.passages:
    print(p.doc_id, p.score, p.hop, p.text[:60])

print(result.confidence, result.hops_used, result.latency_ms)
print(result.trace)  # [TraceStep(hop, doc_id, why), ...]

Prop

Type

Dataclass results

SearchResult, Passage, and TraceStep are dataclasses — attribute access, and a repr that prints the latency / hops / confidence summary line first:

>>> result
SearchResult(confidence=0.87, hops_used=2, latency_ms=4.2, engine='skim-v1',
             passages=[Passage(doc_id='supplier_a', score=0.92, hop=1, …), …],
             trace=[TraceStep(hop=1, doc_id='supplier_a', why='matches query'), …])

search_many — batched eval loops

At ~4 ms per query, evaluating 1,000 queries takes seconds — which makes offline eval loops and regression suites a killer use case.

results = store.search_many([
    "which supplier contract conflicts with the 2026 policy?",
    "which vendor has the earliest renewal date?",
    # … 998 more
])
mean_latency = sum(r.latency_ms for r in results) / len(results)

See the full RAG evaluation harness recipe.

Documents

page = store.documents.list(limit=100)       # .documents, .next_cursor
doc  = store.documents.get("supplier_a", include_text=True)
store.documents.delete("supplier_a")

Health

health = rekall.health()   # .status, .version, .engine

Async

import asyncio
from rekall import AsyncRekall

async def main():
    async with AsyncRekall() as rekall:
        store = await rekall.store("contracts")
        result = await store.search("which contract conflicts with the 2026 policy?")
        print(result.confidence, result.hops_used, result.latency_ms)

asyncio.run(main())

Exceptions

RekallError is the base; each error type maps to a subclass carrying .type and the prescriptive .fix.

Prop

Type

from rekall import RekallError, StoreNotFoundError

try:
    rekall.stores.get("contracts")
except StoreNotFoundError as e:
    print(e.fix)   # "Create it: … rekall stores create contracts"
except RekallError as e:
    print(e.type, e.fix)

Retries built in

Idempotent operations (and 429s) retry automatically with jittered backoff, up to max_retries. Everything else raises immediately.

Integrations

Framework adapters ship in-package — one import each. Meet builders where they are.

# LangChain — drop-in retriever
from rekall.integrations.langchain import RekallRetriever
retriever = RekallRetriever(store="contracts", max_hops=4, max_results=5)

# LlamaIndex — drop-in retriever
from rekall.integrations.llamaindex import RekallRetriever as LlamaRekallRetriever
retriever = LlamaRekallRetriever(store="contracts")

# Generic agent tool — returns an Anthropic/OpenAI tool-call schema
tool = store.as_tool()
# {
#   "name": "rekall_search",
#   "description": "Reasoning-grade retrieval over the 'contracts' store.",
#   "input_schema": {
#     "type": "object",
#     "properties": {
#       "query":    {"type": "string"},
#       "max_hops": {"type": "integer", "default": 4},
#       "max_results":    {"type": "integer", "default": 5}
#     },
#     "required": ["query"]
#   }
# }

See the full LangChain & LlamaIndex retriever swap recipe.

Method summary

Method	Returns
`rekall.store(name)`	`Store` (get-or-create)
`rekall.stores.create / get / list / delete`	`Store` / page / `None`
`store.ingest(docs)`	`IngestResult(ingested, doc_ids)`
`store.ingest_files(dir, glob=…)`	`IngestResult`
`store.ingest_stream(iter, batch_size=…)`	`None` (streamed)
`store.search(query, …)`	`SearchResult`
`store.search_many(queries)`	`list[SearchResult]`
`store.documents.list / get / delete`	page / info / `None`
`rekall.health()`	`Health(status, version, engine)`

On this page