Context Engine Python API reference

Overview

Everything is reached from one ContextEngine object, imported from context_engine. Its methods are async (only register_function_tool and with_redaction are not). Construction is cheap: no provider client is built until a call needs one. The package targets Python 3.12 or newer and is also published for TypeScript (TypeScript API).

Python

import asyncio
from context_engine import (
    ContextEngine, ContextEngineConfig, EmbeddingConfig, LLMConfig, TRUSTED,
)

config = ContextEngineConfig(
    database_url="postgresql://user:pass@localhost:5432/app",
    embedding=EmbeddingConfig(provider="openai", model="text-embedding-3-small", api_key="sk-..."),
    llm=LLMConfig(provider="openai", model="gpt-4o-mini", api_key="sk-..."),
)

async def main():
    async with ContextEngine(config) as engine:
        await engine.ingest(text="Refunds are issued within 14 days.", name="refunds.md")
        result = await engine.search("refund window", principals=TRUSTED)
        for hit in result.hits:
            print(hit.document_name, hit.chunk_text[:80])

asyncio.run(main())

The one argument to get right. Every read and tool method takes principals. TRUSTED means trusted (no ACL filtering), [] means anonymous, a list is a real caller. Leaving it out, or passing None, is deprecated: it still means trusted and returns every document, emits a DeprecationWarning, and a later release refuses it. On any path that serves a user, always pass a list.

Install and extras

The PyPI package is promptev-context-engine; the import name is context_engine. The core install carries ingestion, hybrid search, redaction and tools; each extra below adds one capability.

Shell

pip install "promptev-context-engine[postgres]"
pip install "promptev-context-engine[postgres,fastapi,graph,mcp]"

Extra	Installs	Enables
[postgres]	psycopg2-binary	The Postgres driver. Not a core dependency, but the engine needs a driver: install it unless you bring your own.
[graph]	neo4j, networkx	Graph mode (entities and communities). Neo4j itself is optional; without neo4j_uri the graph lives in Postgres.
[vision]	pillow	Page rasterization for scanned pages and images, read by whichever LLM you configure.
[ocr]	pytesseract, pillow	Tesseract OCR fallback.
[reranker]	(nothing)	Reserved. The reranker runs on the core HTTP client.
[mcp]	mcp >=2.2,<3	The official MCP Python SDK: create_mcp_app, and the client mcp tools connect with (Streamable HTTP).
[compute]	pandas	DataFrames for compute().
[pdf]	pdf-inspector	Structure-preserving PDF text: tables stay tables.
[presidio]	presidio-analyzer	Presidio entities as redaction detectors. Also needs a spaCy model (python -m spacy download en_core_web_lg).
[fastapi]	fastapi, python-multipart	create_router, create_tools_router.
[flask]	flask[async]	create_flask_blueprint.
[django]	django	create_django_urlpatterns.

A feature whose extra is missing fails with an ImportError that names the install command; it never silently does nothing.

Migrate the database first

The engine needs a Postgres with pgvector and its own tables. To try it locally, start one and migrate it, before any code runs:

Shell

docker run -d --name context-engine-pg -p 127.0.0.1:5432:5432 \
  -e POSTGRES_USER=user -e POSTGRES_PASSWORD=pass -e POSTGRES_DB=mydb \
  pgvector/pgvector:pg16
until docker exec context-engine-pg pg_isready -h localhost -U user -d mydb; do sleep 1; done
context-engine migrate --database-url postgresql://user:pass@localhost:5432/mydb --dim 1536

The second line waits until Postgres accepts connections: docker run -d returns before it does. The port is published on loopback only. If port 5432 is already taken, publish another one (-p 127.0.0.1:5433:5432) and use it in the URL (localhost:5433). migrate creates the extensions and the tables, and is safe to run again. On a database that was never migrated, the first call raises DatabaseNotMigrated, and its message is the command to run:

Text

context_engine.errors.DatabaseNotMigrated: This database has no Context Engine tables, or this role cannot see them. Run the migration once, then try again: context-engine migrate --database-url <your database URL> --dim 1536

DatabaseNotMigrated is a RuntimeError; the HTTP routers answer it with 503 and the MCP tools with one fixed sentence. A plain postgresql:// or postgres:// URL uses psycopg2, the driver the [postgres] extra installs, on any SQLAlchemy 2.x. Name another driver in the URL (postgresql+psycopg://) to use it.

Constructor

Takes the configuration and the host seams. The settings classes (ContextEngineConfig, EmbeddingConfig, LLMConfig, StorageConfig, GraphConfig, RerankerConfig, FusionConfig, SearchConfig, ExtractionConfig) are documented on the Configuration page. The fields that hold a credential (database_url, secret_key, redaction_secret_key, every api_key, neo4j_password) are pydantic SecretStr: pass plain strings, read one back with .get_secret_value(), and repr and model_dump() mask them (see Secret fields).

Signature

ContextEngine(
    config: ContextEngineConfig,
    *,
    on_usage: Callable[[UsageEvent], None] | None = None,
    on_error: Callable[[Exception, dict], None] | None = None,
    on_progress: Callable[[dict], None] | None = None,
    on_elicitation: Callable[[ElicitationRequest], ElicitationResponse | dict] | None = None,
    code_runner: CodeRunner | None = None,
    http_client_factory: HttpClientFactory | None = None,
    mcp_resolver: HostResolver | None = None,   # async (host, port) -> list[str]
    model: ModelFn | None = None,
    embedder: Any = None,
    reranker: RerankFn | None = None,
    retrieval_legs: Mapping[str, RetrievalLeg] | None = None,
    fusion: FusionFn | None = None,
    graph_backend: GraphBackend | None = None,
)

Name	Type	Default	Meaning
config	`ContextEngineConfig`	`required`	All settings. See Configuration.
on_usage	`Callable[[UsageEvent], None] \| None`	`None`	Called once per metered unit of work.
on_error	`Callable[[Exception, dict], None] \| None`	`None`	Called on errors the engine handles itself; ctx names the stage. Errors are always logged too.
on_progress	`Callable[[dict], None] \| None`	`None`	Ingest stage boundaries, for live progress.
on_elicitation	`Callable[[ElicitationRequest], ElicitationResponse \| dict] \| None`	`None`	A third-party MCP server's question during an mcp tool call, sync or async. With None the client declares no elicitation. A server that asks anyway gets a protocol error instead of decline: on a 2026-07-28 connection that fails the call, and on a handshake connection the server decides what happens next. See Connecting MCP servers.
code_runner	`CodeRunner \| None`	`None`	Where compute runs generated code. None uses the in-process sandbox.
http_client_factory	`HttpClientFactory \| None`	`None`	Your own httpx.AsyncClient per purpose. Returning None lets the engine build one. Never closed by the engine.
mcp_resolver	`HostResolver \| None`	`None`	Your own name resolution for MCP connections: an async function (host, port) that returns a list of addresses. Any callable that returns an awaitable works. Its answers are still address-checked. Anything that is not callable is refused with TypeError at construction, and a function that returns its addresses without being awaited is refused the first time it is called.
model	`ModelFn \| None`	`None`	Your own model behind every model call: (ModelRequest) -> ModelReply, sync or async. request.purpose is one of structured_extraction, entity_extraction, community_summaries, vision_pages, vision_image, map_reduce, compute. With it, llm, vision_llm and graph.extraction_llm are optional. The engine never retries or times out a host call. See Bring your own model.
embedder	`callable \| object \| None`	`None`	Your own embedding transport: (texts, *, kind) -> (vectors, tokens), or an object with that embed method. config.embedding stays required as the identity the database records (model, dim) and the batch limits; the dimension check runs on every reply.
reranker	`RerankFn \| None`	`None`	Your own reranker: (query, docs) -> list[int], indices into docs most relevant first, sync or async. Replaces config.reranker on the same widened candidate window; out-of-range and duplicate indices are dropped and omitted ones appended.
retrieval_legs	`Mapping[str, RetrievalLeg] \| None`	`None`	Extra retrieval legs by name: fn(query, *, limit, scope) -> list[str], sync or async, run beside the built-in legs. Their ids are re-filtered by the caller's permissions before fusion. Weight fusion.weights[name], 1.0 when absent, 0 means never called. A name reusing fts, trgm, ann or graph raises ValueError. See Plug in your own retrieval.
fusion	`FusionFn \| None`	`None`	Replaces reciprocal rank fusion: fn(ranked, weights, k) -> list[str]. Its answer is restricted to ids some leg produced; a fusion function that raises falls back to RRF.
graph_backend	`GraphBackend \| None`	`None`	Supplies the raw graph: expand_candidates(seeds, query_entities, , max_depth, scope) and chunk_degrees(ids, , scope), sync or async. The engine applies the caller's scope, then the caps (100 per entity, 500 in all), exactly as for the built-ins. Calls are bounded by search.plugin_timeout_s.

The instance exposes config, hooks (a Hooks), backend (the storage backend), session_factory (a SQLAlchemy sessionmaker) and embedder (a property that builds the embedding client on first use).

Python

import httpx
from context_engine import ContextEngine, ModelReply, ModelRequest

shared = httpx.AsyncClient(timeout=30)

def clients(purpose):
    # purpose is "llm", "embeddings", "reranker", "tool_http" or "mcp_oauth"
    return shared if purpose in ("llm", "embeddings") else None

async def my_model(request: ModelRequest) -> ModelReply:
    # every model call the engine makes; request.purpose says which
    return ModelReply(text="...", tokens={"input": 0, "output": 0})

engine = ContextEngine(
    config,
    on_usage=lambda event: print(event.kind, event.units),
    on_error=lambda exc, ctx: print("error", ctx, exc),
    http_client_factory=clients,
    model=my_model,
)

Ingestion

Every method is async. Chunks are stored before they are embedded, so a run that dies mid-embedding can be finished later without the source file.

ingest

Ingest one document: extract, chunk, embed and store it, plus graph and structured extraction when asked. Exactly one of file, content or text must be given. Size caps (config.ingest.max_file_bytes, config.ingest.max_rows) are checked first, before any database or provider work. Re-ingesting unchanged content is reported as skipped.

Signature

async def ingest(
    self,
    *,
    file: Any = None,
    content: bytes | None = None,
    filename: str | None = None,
    text: str | None = None,
    name: str | None = None,
    description: str | None = None,
    source_id: str | None = None,
    external_id: str | None = None,
    meta_data: dict | None = None,
    acl: list[str] | None = None,
    mode: Literal["hybrid", "graph"] | None = None,
    extract_structured: bool = False,
    field_hints: list[dict] | None = None,
    mime_type: str | None = None,
) -> IngestReport

Name	Type	Default	Meaning
file	`str \| os.PathLike \| binary file`	`None`	A path, or an open binary file object with .read(). A path's base name becomes the filename.
content	`bytes \| None`	`None`	Raw file bytes.
filename	`str \| None`	`None`	Filename used to detect the type. Always wins over the name taken from file.
text	`str \| None`	`None`	Text that is already extracted. Skips extraction.
name	`str \| None`	`None`	Display name stored on the document.
description	`str \| None`	`None`	Free-text description stored on the document.
source_id	`str \| None`	`None`	The source (your project, connector or tenant) the document belongs to.
external_id	`str \| None`	`None`	Your own id for the document. With source_id it identifies the document on re-ingest.
meta_data	`dict \| None`	`None`	Free-form metadata stored with the document.
acl	`list[str] \| None`	`None`	Principals allowed to see the document. None is unrestricted. [] is visible only to trusted callers.
mode	`"hybrid" \| "graph" \| None`	`None`	None uses config.default_mode. "graph" needs config.graph.enabled.
extract_structured	`bool`	`False`	Run structured extraction through the host model or config.llm after the chunks are stored.
field_hints	`list[dict] \| None`	`None`	Hints for structured extraction. Needs extract_structured=True.
mime_type	`str \| None`	`None`	What the document is, for example "text/plain" or "text/csv". It is believed over everything else. Without it the filename's extension decides (the name stands in for it on a text= ingest), and only a document with neither is judged by its content, where text is read as CSV only on strong evidence. The type picks the chunker, and a tabular one keeps the document out of graph mode. See how the file type is decided.

Returns: An IngestReport. A document that fails comes back with status="failed", a failure_reason code and the raw error: per-document failures never raise.

Raises: ValueError for a wrong input combination, mode="graph" without graph enabled, extract_structured without config.llm or a host model, or field_hints without extract_structured. IngestTooLarge over a size cap. OSError when a path cannot be opened.

Example

report = await engine.ingest(
    file="policies/leave-policy.pdf",
    source_id="hr",
    external_id="leave-policy",
    acl=["group:hr"],
    extract_structured=True,
)
doc = report.documents[0]
print(doc.document_id, doc.status, doc.chunks, doc.units)
print(report.totals)   # {"files": 1, "failed": 0, "units": ..., "graph_units": ...}

resume_embeddings

Fill in the vectors a failed or interrupted run did not reach. Only the chunks still missing a vector are sent, in order; nothing is re-extracted, re-chunked or re-billed, and the source file is not needed. With no id it sweeps up to 100 resumable documents, oldest first, so it is safe to call from a cron job.

Signature

async def resume_embeddings(self, document_id: Any = None) -> IngestReport

Name	Type	Default	Meaning
document_id	`str \| UUID \| None`	`None`	One document to finish. None sweeps every resumable document. A document with nothing pending is completed and otherwise left alone.

Returns: An IngestReport covering every document it touched, with summed totals.

Example

report = await engine.resume_embeddings()          # sweep
report = await engine.resume_embeddings(doc_id)    # one document

resume_documents

Ask your model or embedder again for every document parked as waiting_model after it raised ModelDeferred. It re-runs the parked stage under the same request keys; a document that defers again parks again with rounds incremented. With no ids it sweeps the parked documents whose retry_at has passed (the database clock), oldest first, in pages of 100, claiming each with one compare-and-set, so a cron on several workers never asks you twice. Every call also fills the community summaries a source is still missing. See Batch mode.

Signature

async def resume_documents(
    self,
    document_ids: Sequence[Any] | None = None,
    *,
    content_source: ContentSource | None = None,
) -> ResumeReport

Name	Type	Default	Meaning
document_ids	`Sequence[Any] \| None`	`None`	Document ids (strings or UUIDs) to resume, due or not. One that is not parked is skipped without a word, so a resume is idempotent. None sweeps the due ones.
content_source	`ContentSource \| None`	`None`	For a document parked at stage extract: called with a ParkedDocument, returns the original bytes (sync or async) or None. Without bytes the document stays parked with no_content_source; bytes whose sha256 differs leave it parked with content_mismatch. A stage enrich park needs no bytes.

Returns: A ResumeReport.

Example

report = await engine.resume_documents(content_source=lambda doc: blobs.get(doc.document_id))
for park in report.superseded:
    await batch.drop(park.request_keys)   # nobody asks for these again

Search

Hybrid retrieval runs full-text, trigram and vector legs, filtered by scope and ACL inside the query, then fuses them with reciprocal rank fusion. A graph leg is added when the deployment has a graph and the query names an entity it knows.

search

Return the top_k chunks most relevant to query. With mode=None the engine decides: the graph leg runs only when graph mode is enabled, reachable, something in scope was ingested as graph, and the query names an entity the caller can see. A graph failure on that path falls back to hybrid. An explicit mode="graph" raises instead of silently answering hybrid when the graph extra is missing or entity resolution fails.

Signature

async def search(
    self,
    query: str,
    *,
    source_ids: list[str] | None = None,
    document_ids: list[str] | None = None,
    principals: list[str] | None = None,
    top_k: int = 10,
    mode: Literal["hybrid", "graph"] | None = None,
    compress_to_tokens: int | None = None,
    redaction: RedactionPolicy | None = None,
    secret_key: str | None = None,
) -> SearchResult

Name	Type	Default	Meaning
query	`str`	`required`	The search text. Must not be empty.
source_ids	`list[str] \| None`	`None`	Limit the call to these sources. None is the whole corpus. Each source counts as one billing scope.
document_ids	`list[str] \| None`	`None`	Narrow within the sources to these documents.
principals	`list[str] \| TRUSTED`	`None (deprecated)`	Who is asking. TRUSTED skips ACL filtering, [] is an anonymous caller (unrestricted documents only), a list sees unrestricted documents plus any whose ACL overlaps it. Omitting it or passing None is deprecated: it means trusted, warns, and a later release refuses it. See sentinels.
top_k	`int`	`10`	How many hits to return (at least 1).
mode	`"hybrid" \| "graph" \| None`	`None`	None lets the engine decide. Naming a mode forces it.
compress_to_tokens	`int \| None`	`None`	Drop the weakest hits until the returned text fits this token budget.
redaction	`RedactionPolicy \| None`	`None`	A RedactionPolicy for this call only. None uses config.redaction.
secret_key	`str \| None`	`None`	The key hash redaction rules use for this call only. None uses the engine's redaction key.

Returns: A SearchResult: hits (list order is authoritative) and a usage dict with kind, units, mode (asked for), mode_used (what ran), graph_leg, degraded, scopes, legs, leg_hits, candidates, returned, reranked, compressed, provider_tokens, graph (applied, a reason from GRAPH_REASONS, entities) and redaction when a rule fired.

Raises: ValueError for an empty query, an unknown mode, or mode="graph" without config.graph.enabled. ImportError when graph mode needs the [graph] extra and it is missing.

Example

result = await engine.search(
    "parental leave eligibility",
    source_ids=["hr"],
    principals=["user:42", "group:hr"],
    top_k=5,
)
for hit in result.hits:
    print(hit.score, hit.document_name, hit.meta.get("page"))
print(result.usage["mode_used"], result.usage["units"])

Documents

Reads and writes on single documents. An absent document and one the caller may not see give the same error, so a caller cannot learn that a restricted document exists.

get_document

One document's row as a dict, with its chunk count but not the chunk texts. An output redaction policy masks text, document_type, structured_data and error.

Signature

async def get_document(
    self,
    document_id,
    *,
    principals: list[str] | None = None,
    redaction: RedactionPolicy | None = None,
    secret_key: str | None = None,
) -> dict

Name	Type	Default	Meaning
document_id	`str \| UUID`	`required`	The document id.
principals	`list[str] \| TRUSTED`	`None (deprecated)`	Who is asking. TRUSTED skips ACL filtering, [] is an anonymous caller (unrestricted documents only), a list sees unrestricted documents plus any whose ACL overlaps it. Omitting it or passing None is deprecated: it means trusted, warns, and a later release refuses it. See sentinels.
redaction	`RedactionPolicy \| None`	`None`	A RedactionPolicy for this call only. None uses config.redaction.
secret_key	`str \| None`	`None`	The key hash redaction rules use for this call only. None uses the engine's redaction key.

Returns: id, source_id, external_id, name, description, meta_data, acl, mode, mime_type, lang, text, status, error, failure_reason, chunks, document_type, structured_data, created_at, updated_at, started_at, completed_at, unreadable_reason, unreadable_pages, embedding_progress.

Raises: KeyError when the document is absent or not visible.

Example

doc = await engine.get_document(doc_id, principals=["group:hr"])
print(doc["status"], doc["chunks"], doc["acl"])

get_document_text

The whole text of one document, untruncated. config.redaction is applied; this method has no per-call redaction override.

Signature

async def get_document_text(self, document_id, *, principals: list[str] | None = None) -> str

Name	Type	Default	Meaning
document_id	`str \| UUID`	`required`	The document id.
principals	`list[str] \| TRUSTED`	`None (deprecated)`	Who is asking. TRUSTED skips ACL filtering, [] is an anonymous caller (unrestricted documents only), a list sees unrestricted documents plus any whose ACL overlaps it. Omitting it or passing None is deprecated: it means trusted, warns, and a later release refuses it. See sentinels.

Returns: The document text as a string.

Raises: EngineActionError for a malformed id, or a document that is absent or not visible.

Example

text = await engine.get_document_text(doc_id, principals=["group:hr"])

get_documents

Several whole documents at once, each ACL-checked. An id that is absent or not visible is left out of the result rather than raising.

Signature

async def get_documents(
    self,
    document_ids: list[str],
    *,
    principals: list[str] | None = None,
    redaction: RedactionPolicy | None = None,
    secret_key: str | None = None,
) -> list[dict]

Name	Type	Default	Meaning
document_ids	`list[str]`	`required`	The ids to read.
principals	`list[str] \| TRUSTED`	`None (deprecated)`	Who is asking. TRUSTED skips ACL filtering, [] is an anonymous caller (unrestricted documents only), a list sees unrestricted documents plus any whose ACL overlaps it. Omitting it or passing None is deprecated: it means trusted, warns, and a later release refuses it. See sentinels.
redaction	`RedactionPolicy \| None`	`None`	A RedactionPolicy for this call only. None uses config.redaction.
secret_key	`str \| None`	`None`	The key hash redaction rules use for this call only. None uses the engine's redaction key.

Returns: A list of dicts in the get_document shape, for the visible ids only.

Example

docs = await engine.get_documents([a_id, b_id], principals=["user:42"])

get_chunks

Read one document in order, a range of chunks at a time. At most 25 chunks come back per call; the document itself is ACL-checked first.

Signature

async def get_chunks(
    self,
    document_id,
    *,
    principals: list[str] | None = None,
    start: int = 0,
    end: int | None = None,
    redaction: RedactionPolicy | None = None,
    secret_key: str | None = None,
) -> dict

Name	Type	Default	Meaning
document_id	`str \| UUID`	`required`	The document id.
principals	`list[str] \| TRUSTED`	`None (deprecated)`	Who is asking. TRUSTED skips ACL filtering, [] is an anonymous caller (unrestricted documents only), a list sees unrestricted documents plus any whose ACL overlaps it. Omitting it or passing None is deprecated: it means trusted, warns, and a later release refuses it. See sentinels.
start	`int`	`0`	First chunk position (0-based).
end	`int \| None`	`None`	Last chunk position, inclusive. Clamped so one call returns at most 25 chunks.
redaction	`RedactionPolicy \| None`	`None`	A RedactionPolicy for this call only. None uses config.redaction.
secret_key	`str \| None`	`None`	The key hash redaction rules use for this call only. None uses the engine's redaction key.

Returns: document_id, total_chunks, start, end, chunks (each position, text, language), has_more, next_start.

Raises: EngineActionError for a malformed id, or a document that is absent or not visible.

Example

page = await engine.get_chunks(doc_id, principals=["group:hr"])
while page["has_more"]:
    page = await engine.get_chunks(doc_id, principals=["group:hr"], start=page["next_start"])

list_documents

Page through documents in scope, newest first, with keyset cursors. source_id and source_ids combine as a union; with neither, the whole corpus is listed. An output policy masks document_type.

Signature

async def list_documents(
    self,
    *,
    source_id: str | None = None,
    principals: list[str] | None = None,
    cursor: Any = None,
    limit: int = 50,
    source_ids: list[str] | None = None,
    document_ids: list[str] | None = None,
    redaction: RedactionPolicy | None = None,
    secret_key: str | None = None,
) -> dict

Name	Type	Default	Meaning
source_id	`str \| None`	`None`	List one source.
principals	`list[str] \| TRUSTED`	`None (deprecated)`	Who is asking. TRUSTED skips ACL filtering, [] is an anonymous caller (unrestricted documents only), a list sees unrestricted documents plus any whose ACL overlaps it. Omitting it or passing None is deprecated: it means trusted, warns, and a later release refuses it. See sentinels.
cursor	`dict \| str \| None`	`None`	The next_cursor of the previous page (a dict or its JSON string).
limit	`int`	`50`	Page size, clamped to 1 to 200.
source_ids	`list[str] \| None`	`None`	List several sources in one paged query. An empty list lists nothing.
document_ids	`list[str] \| None`	`None`	Narrow within the sources to these documents. Malformed ids are skipped.
redaction	`RedactionPolicy \| None`	`None`	A RedactionPolicy for this call only. None uses config.redaction.
secret_key	`str \| None`	`None`	The key hash redaction rules use for this call only. None uses the engine's redaction key.

Returns: documents (each id, source_id, external_id, name, description, acl, mode, mime_type, lang, status, failure_reason, document_type, created_at, updated_at, unreadable_reason, unreadable_pages, embedding_progress), count, has_more, and next_cursor (time, id) only when more remain.

Example

page = await engine.list_documents(source_id="hr", principals=["group:hr"])
while page.get("next_cursor"):
    page = await engine.list_documents(
        source_id="hr", principals=["group:hr"], cursor=page["next_cursor"],
    )

update_document

Change caller-declared attributes without re-sending content. PATCH semantics: only the arguments you pass change. Omitting acl leaves it alone; acl=None makes the document unrestricted. meta_data is merged onto the existing dict (None is a no-op). Passing acl always re-syncs the chunks, so retrying a failed update heals drift.

Signature

async def update_document(
    self,
    document_id,
    *,
    acl: Any = UNSET,
    name: Any = UNSET,
    description: Any = UNSET,
    meta_data: Any = UNSET,
    principals: list[str] | None = None,
) -> list[str]

Name	Type	Default	Meaning
document_id	`str \| UUID`	`required`	The document id.
acl	`list[str] \| None`	`UNSET`	New ACL. None unrestricts. Omit to keep.
name	`str \| None`	`UNSET`	Replaced, including with None.
description	`str \| None`	`UNSET`	Replaced, including with None.
meta_data	`dict \| None`	`UNSET`	Merged onto the stored dict.
principals	`list[str] \| TRUSTED`	`None (deprecated)`	Who is asking. TRUSTED skips ACL filtering, [] is an anonymous caller (unrestricted documents only), a list sees unrestricted documents plus any whose ACL overlaps it. Omitting it or passing None is deprecated: it means trusted, warns, and a later release refuses it. See sentinels.

Returns: The names of the fields that actually changed ([] for a no-op).

Raises: KeyError when the document is absent or not visible, checked before anything is written.

Example

changed = await engine.update_document(
    doc_id, acl=["group:hr", "group:legal"], principals=TRUSTED,
)

delete_document

Delete a document and its chunks. With graph mode enabled it also repairs the graph mirror; a graph cleanup failure is reported to on_error and never fails the delete.

Signature

async def delete_document(self, document_id, *, principals: list[str] | None = None) -> None

Name	Type	Default	Meaning
document_id	`str \| UUID`	`required`	The document id.
principals	`list[str] \| TRUSTED`	`None (deprecated)`	Who is asking. TRUSTED skips ACL filtering, [] is an anonymous caller (unrestricted documents only), a list sees unrestricted documents plus any whose ACL overlaps it. Omitting it or passing None is deprecated: it means trusted, warns, and a later release refuses it. See sentinels.

Returns: None.

Raises: KeyError when the document is absent or not visible, checked before anything is deleted.

Example

await engine.delete_document(doc_id, principals=["group:hr"])

stats

Corpus counters, optionally for one source, counted as the caller sees the corpus.

Signature

async def stats(self, source_id: str | None = None, *, principals: list[str] | None = None) -> dict

Name	Type	Default	Meaning
source_id	`str \| None`	`None`	Count one source only.
principals	`list[str] \| TRUSTED \| None`	`None`	The same contract as every read: TRUSTED counts everything, [] only unrestricted rows, a list the rows its acl overlaps. Omitted, it counts everything, as before.

Returns: source_id, documents, chunks, by_status (status to count), embedding (provider, model, dim).

Example

print(await engine.stats("hr"))

resync_graph

Bring Neo4j back in line with the Postgres graph. Only a deployment with graph.neo4j_uri set has anything to do here: Postgres is written first and is always complete, and a Neo4j write that still fails after its retries leaves that document out of Neo4j, reported as graph_backend_stale on the document, through on_error with stage graph_sync, and as graph_sync_pending in the document's meta_data. With no argument this replays every document recorded that way, each with its own entities and relationships, and retries every removal Neo4j still owes. full=True replays the whole corpus (every entity, every graph document, every relationship), then removes from Neo4j what Postgres no longer has; its cost follows the corpus. Run the full replay once after upgrading, and whenever the two stores must be made identical whatever happened before. Idempotent, calls no model, bills nothing. Needs graph.enabled. A maintenance call for the host: it takes no principals, and no router, tool or MCP action reaches it.

Signature

async def resync_graph(
    self,
    *,
    document_id: str | None = None,
    source_id: str | None = None,
    full: bool = False,
) -> dict

Name	Type	Default	Meaning
document_id	`str \| None`	`None`	Replay that one document. A document that was not ingested in graph mode has nothing to replay.
source_id	`str \| None`	`None`	Replay every graph document of that source.
full	`bool`	`False`	Replay the whole corpus and remove from Neo4j what Postgres no longer has. Takes no document_id or source_id.

Returns: documents, synced, failed (document ids; each keeps its record and is reported through on_error). Three more keys appear only when they have something to say: removals (done, failed); with full, removed (chunks, relationships, entities); and error, a failure of a corpus-wide step of the full replay, reported once through on_error with scope="corpus". Without a neo4j_uri the result is all zeros.

Raises: ValueError without graph mode, when both document_id and source_id are given, or when full=True is combined with either. KeyError when document_id names no document.

Example

result = await engine.resync_graph()            # documents recorded as pending
await engine.resync_graph(source_id="hr")       # every graph document of one source
await engine.resync_graph(full=True)            # once after upgrading

rebuild_communities

Detect, summarise and store one source's communities again, reading and writing only that source. Use it for a source whose communities are missing (none built yet, or removed by a delete) and that gets no graph ingest to rebuild them. Needs graph.enabled. Summaries and their embeddings are written only with graph.community_summaries on; off, it detects and stores the communities with no model call, bills 0 units, and drops the source's existing summaries. Metered to on_usage as graph units with detail["operation"] == "rebuild_communities".

Signature

async def rebuild_communities(self, source_id: str | None) -> dict

Name	Type	Default	Meaning
source_id	`str \| None`	`required`	The source to rebuild. None is the documents ingested without a source.

Returns: communities_detected, communities_summarized, primary_communities, units.

Example

await engine.rebuild_communities("hr")

enable_bm25

Opt this database in to the experimental search.lexical_rank="bm25". Installs four triggers on the chunk table (its lock is taken briefly, with a 2-second timeout and retries) and counts the chunks already stored without blocking ingest. Idempotent; running it again recounts. Until it has finished, a bm25 search ranks with ts_rank_cd, reports a LexicalRankFallback to on_error and says so in usage["lexical"]. disable_bm25() removes the triggers and statistics; compact_bm25_stats() folds the statistics rows each write adds and never blocks a writer. Also context-engine bm25-enable.

Signature

async def enable_bm25(self) -> dict

Returns: {"created_triggers": bool, "sources": [{"source_id", "chunks"}]}.

Example

await engine.enable_bm25()

Corpus questions

The building blocks behind the knowledge tool: what is in the corpus, what is inside each document, and answers computed across documents. All are ACL-checked and scoped like search.

documents_by_name

The documents whose name matches filename, by trigram similarity on the name. Metered as a search (one unit per scope) when something matched; a name that matched nothing is free and emits no usage event.

Signature

async def documents_by_name(
    self,
    filename: str,
    *,
    source_ids: list[str] | None = None,
    document_ids: list[str] | None = None,
    principals: list[str] | None = None,
    limit: int = 10,
    redaction: RedactionPolicy | None = None,
    secret_key: str | None = None,
) -> dict

Name	Type	Default	Meaning
filename	`str`	`required`	The file name to look for.
source_ids	`list[str] \| None`	`None`	Limit the call to these sources. None is the whole corpus.
document_ids	`list[str] \| None`	`None`	Narrow within the sources to these documents. Malformed ids are skipped.
principals	`list[str] \| TRUSTED`	`None (deprecated)`	Who is asking. TRUSTED skips ACL filtering, [] is an anonymous caller (unrestricted documents only), a list sees unrestricted documents plus any whose ACL overlaps it. Omitting it or passing None is deprecated: it means trusted, warns, and a later release refuses it. See sentinels.
limit	`int`	`10`	Most documents to return.
redaction	`RedactionPolicy \| None`	`None`	A RedactionPolicy for this call only. None uses config.redaction.
secret_key	`str \| None`	`None`	The key hash redaction rules use for this call only. None uses the engine's redaction key.

Returns: documents (each document_id, name, score, snippet, source_id, chunk_idx, best first) and usage (kind="search", units, matched_by="filename", scopes, empty legs and leg_hits, returned, and redaction when a rule fired).

Example

found = await engine.documents_by_name("q3-budget.xlsx", principals=["user:42"])

document_types

The corpus census. kind comes from the MIME type and is always known ("spreadsheet" or "text"); type is the LLM-written document type and exists only where structured extraction ran.

Signature

async def document_types(
    self,
    *,
    source_ids: list[str] | None = None,
    principals: list[str] | None = None,
    document_ids: list[str] | None = None,
    redaction: RedactionPolicy | None = None,
    secret_key: str | None = None,
) -> list[dict]

Name	Type	Default	Meaning
source_ids	`list[str] \| None`	`None`	Limit the call to these sources. None is the whole corpus.
principals	`list[str] \| TRUSTED`	`None (deprecated)`	Who is asking. TRUSTED skips ACL filtering, [] is an anonymous caller (unrestricted documents only), a list sees unrestricted documents plus any whose ACL overlaps it. Omitting it or passing None is deprecated: it means trusted, warns, and a later release refuses it. See sentinels.
document_ids	`list[str] \| None`	`None`	Narrow within the sources to these documents. Malformed ids are skipped.
redaction	`RedactionPolicy \| None`	`None`	A RedactionPolicy for this call only. None uses config.redaction.
secret_key	`str \| None`	`None`	The key hash redaction rules use for this call only. None uses the engine's redaction key.

Returns: A list of {kind, type, documents, with_fields}.

Example

census = await engine.document_types(principals=TRUSTED)

field_summary

Extracted structured field names, grouped by the same (kind, type) pair document_types uses.

Signature

async def field_summary(
    self,
    *,
    source_ids: list[str] | None = None,
    principals: list[str] | None = None,
    document_ids: list[str] | None = None,
    redaction: RedactionPolicy | None = None,
    secret_key: str | None = None,
) -> list[dict]

Name	Type	Default	Meaning
source_ids	`list[str] \| None`	`None`	Limit the call to these sources. None is the whole corpus.
principals	`list[str] \| TRUSTED`	`None (deprecated)`	Who is asking. TRUSTED skips ACL filtering, [] is an anonymous caller (unrestricted documents only), a list sees unrestricted documents plus any whose ACL overlaps it. Omitting it or passing None is deprecated: it means trusted, warns, and a later release refuses it. See sentinels.
document_ids	`list[str] \| None`	`None`	Narrow within the sources to these documents. Malformed ids are skipped.
redaction	`RedactionPolicy \| None`	`None`	A RedactionPolicy for this call only. None uses config.redaction.
secret_key	`str \| None`	`None`	The key hash redaction rules use for this call only. None uses the engine's redaction key.

Returns: A list of {kind, type, fields: [{field, type, documents}], more_fields}. more_fields lists the names left out by the per-group cap, and is absent when nothing was left out.

Example

fields = await engine.field_summary(source_ids=["finance"], principals=TRUSTED)

spreadsheet_schema

Sheet names, columns and row counts for the named spreadsheets, so a model can write one correct compute call. ACL and scope are applied again; a document that is not visible, not in scope or not tabular is left out. A document past the text budget is listed with sheets: None and a reason.

Signature

async def spreadsheet_schema(
    self,
    *,
    document_ids: list[str],
    source_ids: list[str] | None = None,
    principals: list[str] | None = None,
    redaction: RedactionPolicy | None = None,
    secret_key: str | None = None,
) -> list[dict]

Name	Type	Default	Meaning
document_ids	`list[str]`	`required`	The spreadsheets to describe.
source_ids	`list[str] \| None`	`None`	Limit the call to these sources. None is the whole corpus.
principals	`list[str] \| TRUSTED`	`None (deprecated)`	Who is asking. TRUSTED skips ACL filtering, [] is an anonymous caller (unrestricted documents only), a list sees unrestricted documents plus any whose ACL overlaps it. Omitting it or passing None is deprecated: it means trusted, warns, and a later release refuses it. See sentinels.
redaction	`RedactionPolicy \| None`	`None`	A RedactionPolicy for this call only. None uses config.redaction.
secret_key	`str \| None`	`None`	The key hash redaction rules use for this call only. None uses the engine's redaction key.

Returns: A list of {document_id, name, source_id, sheets}, with schema_unavailable when sheets is None.

Example

schema = await engine.spreadsheet_schema(document_ids=[sheet_id], principals=TRUSTED)

document_structure

What is inside each named document, whatever its type: sheets and columns for a workbook, sections and last_page for a document with headings, top-level keys for JSON, and a chunks count on every one.

Signature

async def document_structure(
    self,
    *,
    document_ids: list[str],
    source_ids: list[str] | None = None,
    principals: list[str] | None = None,
    bounded: bool = True,
    redaction: RedactionPolicy | None = None,
    secret_key: str | None = None,
) -> dict

Name	Type	Default	Meaning
document_ids	`list[str]`	`required`	The documents to describe.
source_ids	`list[str] \| None`	`None`	Limit the call to these sources. None is the whole corpus.
principals	`list[str] \| TRUSTED`	`None (deprecated)`	Who is asking. TRUSTED skips ACL filtering, [] is an anonymous caller (unrestricted documents only), a list sees unrestricted documents plus any whose ACL overlaps it. Omitting it or passing None is deprecated: it means trusted, warns, and a later release refuses it. See sentinels.
bounded	`bool`	`True`	Cap every list at 40 items and report the remainder as a count.
redaction	`RedactionPolicy \| None`	`None`	A RedactionPolicy for this call only. None uses config.redaction.
secret_key	`str \| None`	`None`	The key hash redaction rules use for this call only. None uses the engine's redaction key.

Returns: {document_id: structure}.

Example

structure = await engine.document_structure(document_ids=[doc_id], principals=TRUSTED)

tabular_scope

Is everything in scope a spreadsheet, and what does the first one look like. One grouped query over the document table.

Signature

async def tabular_scope(
    self,
    *,
    source_ids: list[str] | None = None,
    document_ids: list[str] | None = None,
    principals: list[str] | None = None,
    redaction: RedactionPolicy | None = None,
    secret_key: str | None = None,
) -> dict

Name	Type	Default	Meaning
source_ids	`list[str] \| None`	`None`	Limit the call to these sources. None is the whole corpus.
document_ids	`list[str] \| None`	`None`	Narrow within the sources to these documents. Malformed ids are skipped.
principals	`list[str] \| TRUSTED`	`None (deprecated)`	Who is asking. TRUSTED skips ACL filtering, [] is an anonymous caller (unrestricted documents only), a list sees unrestricted documents plus any whose ACL overlaps it. Omitting it or passing None is deprecated: it means trusted, warns, and a later release refuses it. See sentinels.
redaction	`RedactionPolicy \| None`	`None`	A RedactionPolicy for this call only. None uses config.redaction.
secret_key	`str \| None`	`None`	The key hash redaction rules use for this call only. None uses the engine's redaction key.

Returns: {documents, tabular, all_tabular, sheet, frame_key}. sheet and frame_key are filled only when every document in scope is tabular.

Example

scope = await engine.tabular_scope(source_ids=["finance"], principals=TRUSTED)

query_structured

Answer a question against documents' extracted structured_data. Field names in the question are resolved to registry keys (exact, synonym, then embedding similarity). When nothing resolves, zero documents come back, never the whole corpus.

Signature

async def query_structured(
    self,
    question: str,
    *,
    source_ids: list[str] | None = None,
    principals: list[str] | None = None,
    doc_type: str | None = None,
    limit: int = 20,
    redaction: RedactionPolicy | None = None,
    secret_key: str | None = None,
) -> dict

Name	Type	Default	Meaning
question	`str`	`required`	The question, in plain language.
source_ids	`list[str] \| None`	`None`	Limit the call to these sources. None is the whole corpus.
principals	`list[str] \| TRUSTED`	`None (deprecated)`	Who is asking. TRUSTED skips ACL filtering, [] is an anonymous caller (unrestricted documents only), a list sees unrestricted documents plus any whose ACL overlaps it. Omitting it or passing None is deprecated: it means trusted, warns, and a later release refuses it. See sentinels.
doc_type	`str \| None`	`None`	Only documents of this extracted type.
limit	`int`	`20`	Most documents to return.
redaction	`RedactionPolicy \| None`	`None`	A RedactionPolicy for this call only. None uses config.redaction.
secret_key	`str \| None`	`None`	The key hash redaction rules use for this call only. None uses the engine's redaction key.

Returns: {question, resolved_fields, documents, count}.

Example

answer = await engine.query_structured(
    "invoices with a due date and total", doc_type="invoice", principals=TRUSTED,
)
print(answer["resolved_fields"], answer["count"])

map_reduce

Ask the same question of every document in scope, one LLM call each (the map step only; you do the reducing). Reads 25 documents by default and at most 200, newest first, with at most 20 calls in flight. Redaction runs before text reaches the model. A document that fails is reported, not raised.

Signature

async def map_reduce(
    self,
    instruction: str,
    *,
    principals: list[str] | None = None,
    source_ids: list[str] | None = None,
    document_ids: list[str] | None = None,
    limit: int | None = None,
    max_concurrency: int = 5,
    redaction: RedactionPolicy | None = None,
    secret_key: str | None = None,
) -> dict

Name	Type	Default	Meaning
instruction	`str`	`required`	What to extract from each document. The model answers in JSON.
principals	`list[str] \| TRUSTED`	`None (deprecated)`	Who is asking. TRUSTED skips ACL filtering, [] is an anonymous caller (unrestricted documents only), a list sees unrestricted documents plus any whose ACL overlaps it. Omitting it or passing None is deprecated: it means trusted, warns, and a later release refuses it. See sentinels.
source_ids	`list[str] \| None`	`None`	Limit the call to these sources. None is the whole corpus.
document_ids	`list[str] \| None`	`None`	Narrow within the sources to these documents. Malformed ids are skipped.
limit	`int \| None`	`None`	Documents to read. None means 25; capped at 200.
max_concurrency	`int`	`5`	Parallel LLM calls, clamped to 1 to 20.
redaction	`RedactionPolicy \| None`	`None`	A RedactionPolicy for this call only. None uses config.redaction.
secret_key	`str \| None`	`None`	The key hash redaction rules use for this call only. None uses the engine's redaction key.

Returns: results (each document_id, document_name, and data or error), processed, failed, considered.

Raises: EngineActionError when no config.llm is configured.

Example

out = await engine.map_reduce(
    "Return the contract's termination notice period in days.",
    source_ids=["legal"], principals=TRUSTED, limit=50,
)
for row in out["results"]:
    print(row["document_name"], row.get("data"), row.get("error"))

map_reduce_targets

The document ids map_reduce would read with the same arguments, newest first. Ids only, no bodies.

Signature

async def map_reduce_targets(
    self,
    *,
    principals: list[str] | None = None,
    source_ids: list[str] | None = None,
    document_ids: list[str] | None = None,
    limit: int | None = None,
) -> list[str]

Name	Type	Default	Meaning
principals	`list[str] \| TRUSTED`	`None (deprecated)`	Who is asking. TRUSTED skips ACL filtering, [] is an anonymous caller (unrestricted documents only), a list sees unrestricted documents plus any whose ACL overlaps it. Omitting it or passing None is deprecated: it means trusted, warns, and a later release refuses it. See sentinels.
source_ids	`list[str] \| None`	`None`	Limit the call to these sources. None is the whole corpus.
document_ids	`list[str] \| None`	`None`	Narrow within the sources to these documents. Malformed ids are skipped.
limit	`int \| None`	`None`	Same meaning as in map_reduce.

Returns: A list of document ids.

Example

ids = await engine.map_reduce_targets(source_ids=["legal"], principals=TRUSTED)

compute

Answer a question over the spreadsheets in scope (CSV, TSV, XLSX): the model writes pandas code against dfs, and the code runs in the engine's code_runner if one was given, else in the in-process sandbox. Off by default: it needs config.enable_code_execution=True, which you should set only behind real OS-level isolation. Needs the [compute] extra. Redaction masks cells before the frames are built and sweeps the returned result.

Signature

async def compute(
    self,
    instruction: str,
    *,
    source_ids: list[str] | None = None,
    principals: list[str] | None = None,
    document_ids: list[str] | None = None,
    model_cfg=None,
    timeout: int = 30,
    redaction: RedactionPolicy | None = None,
    secret_key: str | None = None,
    model: ModelFn | None = None,
) -> dict

Name	Type	Default	Meaning
instruction	`str`	`required`	The question to compute.
source_ids	`list[str] \| None`	`None`	Limit the call to these sources. None is the whole corpus.
principals	`list[str] \| TRUSTED`	`None (deprecated)`	Who is asking. TRUSTED skips ACL filtering, [] is an anonymous caller (unrestricted documents only), a list sees unrestricted documents plus any whose ACL overlaps it. Omitting it or passing None is deprecated: it means trusted, warns, and a later release refuses it. See sentinels.
document_ids	`list[str] \| None`	`None`	Only these spreadsheets.
model_cfg	`LLMConfig \| None`	`None`	A built-in provider config for the model that writes the code. Most specific first: the per-call model, then this, then the engine's model, then config.llm.
model	`ModelFn \| None`	`None`	A host function that writes the code for this call only, beating model_cfg and the engine's model.
timeout	`int`	`30`	Seconds the code may run.
redaction	`RedactionPolicy \| None`	`None`	A RedactionPolicy for this call only. None uses config.redaction.
secret_key	`str \| None`	`None`	The key hash redaction rules use for this call only. None uses the engine's redaction key.

Returns: success, result, code, stdout, error, execution_time, documents_used, provider_tokens (llm_input, llm_output), attempts (1 or 2).

Raises: EngineActionError when code execution is disabled (before any work), the instruction is empty, or no tabular documents are in scope. ValueError when no LLM is configured.

Example

out = await engine.compute(
    "Total revenue by region for 2025",
    source_ids=["finance"], principals=["group:finance"],
)
if out["success"]:
    print(out["result"])

Knowledge tool

The one tool an agent needs for the corpus, as a library call. It is the same function the MCP server serves, so both surfaces give the same answer. Actions and arguments are documented on the knowledge tool page.

search_knowledge_base

Run one action: discover, search, get_doc, list, query_meta, compute, get_chunks, get_docs, map_reduce, traverse, find_related, get_neighbors, community_summary. principals and scope are yours to set, never the model's: the model can narrow the scope but never widen it.

Signature

async def search_knowledge_base(
    self,
    *,
    action: str,
    principals: list[str] | None = None,
    scope: Any = None,
    query: str | None = None,
    document_id: str | None = None,
    source_ids: list[str] | None = None,
    document_ids: list[str] | None = None,
    entity: str | None = None,
    depth: int | None = None,
    category: str | None = None,
    label: str | None = None,
    entity_type: str | None = None,
    top_k: int = 10,
    mode: str | None = None,
    limit: int | None = None,
    cursor: str | dict | None = None,
    redaction: RedactionPolicy | None = None,
    secret_key: str | None = None,
    compute: Any = None,
    map_reduce: Any = None,
    start: int | None = None,
    end: int | None = None,
    max_chars: int | None = None,
) -> dict

Name	Type	Default	Meaning
action	`str`	`required`	One of the actions above.
principals	`list[str] \| TRUSTED`	`None (deprecated)`	Who is asking. TRUSTED skips ACL filtering, [] is an anonymous caller (unrestricted documents only), a list sees unrestricted documents plus any whose ACL overlaps it. Omitting it or passing None is deprecated: it means trusted, warns, and a later release refuses it. See sentinels.
scope	`UNSCOPED \| list[str] \| Scope`	`None (raises)`	The ceiling of what this call may reach. Required: None raises ValueError. Pass source ids, a Scope, or UNSCOPED for the whole corpus on purpose.
query ... max_chars	`various`	`see signature`	The model-supplied arguments of the chosen action. See the knowledge tool page.
redaction	`RedactionPolicy \| None`	`None`	A RedactionPolicy for this call only. None uses config.redaction.
secret_key	`str \| None`	`None`	The key hash redaction rules use for this call only. None uses the engine's redaction key.
compute	`callable \| None`	`None`	Your own runner that replaces the built-in compute action.
map_reduce	`callable \| None`	`None`	Your own runner that replaces the built-in map_reduce action.

Returns: A dict. An unknown action or a refused argument comes back as {"success": False, "error": ...}.

Raises: ValueError when scope is missing or malformed.

Example

from context_engine import UNSCOPED

out = await engine.search_knowledge_base(
    action="search",
    query="notice period",
    principals=["user:42"],
    scope=["legal"],          # or UNSCOPED
)

Tools

Governed tool execution: HTTP, database and MCP tools are stored with their config encrypted (AES-256-GCM, keyed by config.secret_key); function tools live in memory. Every execution is ACL-checked, can require approval, is audited and reports usage. The call name is the tool name with a prefix: http_, db_, mcp_ or fn_.

register_tool

Store an http, db or mcp tool. A non-empty config is encrypted whole before it is written.

Signature

async def register_tool(self, tc: ToolConfig) -> str

Name	Type	Default	Meaning
tc	`ToolConfig`	`required`	See ToolConfig.

Returns: The new tool id.

Raises: ValueError for kind="function" (use register_function_tool). ConfigTemplateError (a ValueError) when config contains the "__redacted__" placeholder or an invalid response_mode, or when an mcp tool’s URL carries a user name or password or one of its connection settings is invalid. RuntimeError when config.secret_key is missing or malformed.

Example

from context_engine import ToolConfig

tool_id = await engine.register_tool(ToolConfig(
    name="get_weather",
    kind="http",
    description="Fetch the current weather for a city",
    config={
        "method": "GET",
        "url": "https://api.example.com/weather",
        "llmQueryParameters": {
            "properties": {"city": {"type": "string"}},
            "required": ["city"],
        },
    },
    acl=["group:ops"],
))   # call name: "http_get_weather"

register_function_tool

Register a plain Python function as a tool. The parameter schema comes from the signature and type hints, the description from the docstring. Not persisted: it lives for the life of the engine and is shared by every view. This method is synchronous.

Signature

def register_function_tool(self, fn) -> str

Name	Type	Default	Meaning
fn	`Callable`	`required`	The function to expose.

Returns: The call name, fn_.

Example

def add(a: int, b: int) -> int:
    """Add two integers."""
    return a + b

name = engine.register_function_tool(add)          # "fn_add"
out = await engine.execute_tool(name, {"a": 2, "b": 3}, principals=TRUSTED)

update_tool

Patch a stored tool. A config field is re-encrypted; a "__redacted__" value inside it means "keep the stored value". Unknown fields are ignored.

Signature

async def update_tool(self, id, *, principals: list[str] | None = None, **fields) -> ToolConfig

Name	Type	Default	Meaning
id	`str`	`required`	The tool id.
principals	`list[str] \| TRUSTED`	`None (deprecated)`	Who is asking. TRUSTED skips ACL filtering, [] is an anonymous caller (unrestricted documents only), a list sees unrestricted documents plus any whose ACL overlaps it. Omitting it or passing None is deprecated: it means trusted, warns, and a later release refuses it. See sentinels.
**fields	`Any`	`none`	ToolConfig fields to change.

Returns: The updated, decrypted ToolConfig.

Raises: EngineActionError when the tool is absent or not visible. ConfigTemplateError when a placeholder has no stored value to keep.

Example

tc = await engine.update_tool(tool_id, principals=TRUSTED, requires_approval=True)

delete_tool

Delete a stored tool.

Signature

async def delete_tool(self, id, *, principals: list[str] | None = None) -> None

Name	Type	Default	Meaning
id	`str`	`required`	The tool id.
principals	`list[str] \| TRUSTED`	`None (deprecated)`	Who is asking. TRUSTED skips ACL filtering, [] is an anonymous caller (unrestricted documents only), a list sees unrestricted documents plus any whose ACL overlaps it. Omitting it or passing None is deprecated: it means trusted, warns, and a later release refuses it. See sentinels.

Returns: None.

Raises: EngineActionError when the tool is absent or not visible.

Example

await engine.delete_tool(tool_id, principals=TRUSTED)

list_tools

Every visible tool, stored and function, as a secret-free public view. Config, URLs and credentials are never included.

Signature

async def list_tools(
    self, *, source_id: str | None = None, principals: list[str] | None = None
) -> list[dict]

Name	Type	Default	Meaning
source_id	`str \| None`	`None`	Only tools of this source.
principals	`list[str] \| TRUSTED`	`None (deprecated)`	Who is asking. TRUSTED skips ACL filtering, [] is an anonymous caller (unrestricted documents only), a list sees unrestricted documents plus any whose ACL overlaps it. Omitting it or passing None is deprecated: it means trusted, warns, and a later release refuses it. See sentinels.

Returns: A list of {name, tool_id, kind, description, params_schema, requires_approval}. tool_id is None for a function tool.

Example

for tool in await engine.list_tools(principals=["group:ops"]):
    print(tool["name"], tool["requires_approval"])

search_tools

Keyword search over visible tools' names and descriptions. query ranks and, when non-empty, excludes tools none of its terms match; kind, name_contains and requires_approval filter, combined with AND. An unknown kind is an empty result.

Signature

async def search_tools(
    self,
    query: str,
    *,
    source_id: str | None = None,
    principals: list[str] | None = None,
    limit: int = 10,
    kind: str | Sequence[str] | None = None,
    name_contains: str | None = None,
    requires_approval: bool | None = None,
    cursor: str | None = None,
) -> ToolSearchResult

Name	Type	Default	Meaning
query	`str`	`required`	Search terms. An empty string with filters lists the filtered set.
source_id	`str \| None`	`None`	Only tools of this source.
principals	`list[str] \| TRUSTED`	`None (deprecated)`	Who is asking. TRUSTED skips ACL filtering, [] is an anonymous caller (unrestricted documents only), a list sees unrestricted documents plus any whose ACL overlaps it. Omitting it or passing None is deprecated: it means trusted, warns, and a later release refuses it. See sentinels.
limit	`int`	`10`	Page size.
kind	`str \| list[str] \| None`	`None`	http, db, mcp, function, one or several.
name_contains	`str \| None`	`None`	Case-insensitive substring of the call name.
requires_approval	`bool \| None`	`None`	False is a real filter: only tools that need no approval.
cursor	`str \| None`	`None`	The previous page's next_cursor, with the same query and filters.

Returns: A ToolSearchResult: a list of the same dicts list_tools returns, with a next_cursor attribute (None on the last page).

Raises: ValueError for an invalid cursor, or one used with a different query or filters.

Example

page = await engine.search_tools("weather", principals=["group:ops"], kind="http")
more = getattr(page, "next_cursor", None)

test_tool

Dry-run a tool config without storing it: an HTTP HEAD, a database connection test, or an MCP connect.

Signature

async def test_tool(self, tc: ToolConfig) -> dict

Name	Type	Default	Meaning
tc	`ToolConfig`	`required`	The config to test.

Returns: {"ok": bool, ...}: status_code for http, the tool names in tools for mcp, or an error category such as "egress_denied", or "unsupported_transport" for an HTTP+SSE or WebSocket MCP URL.

Raises: ConfigTemplateError when the config holds the redacted placeholder, an MCP URL carries a user name or password, or an MCP connection setting (protocol, redirects, the two elicitation keys) is invalid.

Example

check = await engine.test_tool(ToolConfig(
    name="crm", kind="mcp", config={"url": "https://mcp.example.com/mcp"},
))

execute_tool

Governed execution: ACL check, approval gate, config decryption, the call itself, audit row, usage event, and result shaping. A gated call returns approval_required instead of running; resolve it with resolve_approval and repeat the same call with the same approval_scope. A gated call without an approval_scope still works but is deprecated and warns. Every call whose tool was started leaves one audit row: if the full row cannot be written a last-resort row is (input_args set to {"_unstorable": true}, no result, output_truncated true, an error note naming what failed), and it depends on no other row. A tool or an approval deleted while the call ran costs the row only its reference: the row keeps its real arguments and result, with tool_id and approval_id NULL and a note in error_message, so filter audit rows on success, not on error_message. A call cancelled after its tool was started is recorded as a failed call with the error text cancelled (any other BaseException as interrupted by <type>) before the cancellation is raised; it waits for that write for at most tools.audit_cancel_wait_s seconds (default 10), then reports through on_error with stage tool_audit and lets the cancellation through. What cannot be recorded is a process that dies between the tool running and the insert, or a coroutine that is closed without being run to its end. On the routers, over MCP and in the audit row a result keeps its JSON form (a datetime as ISO 8601 text, a timedelta as an ISO 8601 duration, a pydantic model or a dataclass as an object, an Enum as its value, a UUID and a Decimal as text); binary is text when it is valid UTF-8 without a NUL byte, otherwise base64 text, and the value alone does not say which. Called in process, the tool's own objects are returned unchanged, an iterator apart: a one-shot iterator (a generator, map, reversed) is read once and returned as a list, and one that yields more than 50,000 items is a failed call. An approval matches its arguments exactly, by canonical JSON form: key order does not matter, and 1, 1.0, true and "1" are four different arguments, while two spellings of one number with the same decimal places are the same argument (1e16 and 10000000000000000, 0.0 and -0.0).

Signature

async def execute_tool(
    self,
    call_name: str,
    args: dict | None,
    *,
    source_id: str | None = None,
    principals: list[str] | None = None,
    actor: dict | None = None,
    source: str = "api",
    approval_scope: str | None = None,
    result_max_chars: int | None = 8000,
    result_max_rows: int | None = 100,
    response_mode: Literal["json", "tsv"] | None = None,
    redaction: RedactionPolicy | None = None,
    secret_key: str | None = None,
    tool_id: str | None = None,
    overrides: dict | None = None,
) -> dict

Name	Type	Default	Meaning
call_name	`str`	`required`	The prefixed name, for example http_get_weather.
args	`dict \| None`	`required`	Arguments the model produced. Keys starting with _ are dropped from the audit snapshot.
source_id	`str \| None`	`None`	Scopes the tool lookup.
principals	`list[str] \| TRUSTED`	`None (deprecated)`	Who is asking. TRUSTED skips ACL filtering, [] is an anonymous caller (unrestricted documents only), a list sees unrestricted documents plus any whose ACL overlaps it. Omitting it or passing None is deprecated: it means trusted, warns, and a later release refuses it. See sentinels.
actor	`dict \| None`	`None`	{"type": ..., "id": ...} recorded on the audit row.
source	`str`	`"api"`	Where the call came from, recorded on the audit row.
approval_scope	`str \| None`	`None`	Your opaque claim scope for approvals, usually a run id. Set it on the server, never from a request body or model output.
result_max_chars	`int \| None`	`8000`	Character budget for the returned result. None lifts it.
result_max_rows	`int \| None`	`100`	Row budget per table. None lifts it; a db tool's own max_rows still applies.
response_mode	`"json" \| "tsv" \| None`	`None`	"tsv" returns every array of objects as a TSV string. None uses an http tool's configured mode, else json.
redaction	`RedactionPolicy \| None`	`None`	A RedactionPolicy for this call only. None uses config.redaction.
secret_key	`str \| None`	`None`	The key hash redaction rules use for this call only. None uses the engine's redaction key.
tool_id	`str \| None`	`None`	Picks one tool when several visible tools share call_name.
overrides	`dict \| None`	`None`	This call's values over what the tool was saved with: {"headers": ..., "query": ..., "body": ...}. An override wins over a value the model would fill. Host-only, never stored or audited. See One call's values.

Returns: {"result": ..., "usage": {"units": 1, "kind": "tool", "tool_name", "tool_id", "truncated"}}, or {"approval_required": {"approval_id", "tool_name", "tool_id", "args", "reason", "expires_at"}}.

Raises: EngineActionError when the tool is unknown or not visible (same message), or when the call itself fails (audited first). AmbiguousToolError when the name matches several visible tools and no tool_id is given. TypeError or ValueError for a result budget that is not a positive int or None. InvalidInputError before the tool runs, and before any approval is opened, when an argument cannot be stored: broken Unicode (a lone surrogate), a non-finite number, nesting deeper than 64 levels, or a value that is not a JSON value (anything but a dict, list, tuple, str, int, float, bool or None: a datetime, a Decimal, a UUID, bytes, a set, your own class; a tuple is sent as a list, and a set has no order to store, so pass a list); the message names the argument's path (args.items[2].name) and never its value. A NUL character (U+0000) in an argument is removed, not refused. ToolAuditFailed when the tool ran and no audit row could be written at all, with tools.audit_on_failure="fail": it carries the result, so do not send the call again.

Example

out = await engine.execute_tool(
    "http_get_weather", {"city": "Lahore"},
    principals=["group:ops"],
    actor={"type": "agent", "id": "agent-7"},
    approval_scope="run-42",
)
if "approval_required" in out:
    approval_id = out["approval_required"]["approval_id"]
else:
    print(out["result"])

Approvals

A tool with requires_approval=True, or an approval_policy condition that matches, makes execute_tool open a pending approval and return approval_required. The engine does not wait: you resolve the approval, then repeat the call. An approved record is consumed once.

resolve_approval

A module function, imported from context_engine. Flips a pending approval to approved or rejected, atomically: of several concurrent calls on one id, exactly one succeeds. It checks only that principals may see the approval; deciding whether approver is the right person is your job.

Signature

async def resolve_approval(
    engine: ContextEngine,
    approval_id: Any,
    decision: Literal["approved", "rejected"],
    approver: str,
    meta: dict[str, Any] | None = None,
    *,
    principals: list[str] | None = None,
) -> ApprovalRecord

Name	Type	Default	Meaning
engine	`ContextEngine`	`required`	The engine that opened the approval.
approval_id	`str \| UUID`	`required`	From approval_required.
decision	`"approved" \| "rejected"`	`required`	Anything else raises ValueError.
approver	`str`	`required`	Who decided, recorded on the row.
meta	`dict \| None`	`None`	Your metadata, stored as approver_meta.
principals	`list[str] \| None`	`None`	Who is resolving. None or TRUSTED is a trusted operator surface; a list must overlap the approval's principals.

Returns: the updated ApprovalRecord.

Raises: ApprovalNotPending when the approval is missing, already resolved or expired, or not visible. ApprovalExpired when it is pending but past expires_at. ValueError for a bad decision.

Example

from context_engine import resolve_approval, ApprovalExpired, ApprovalNotPending

try:
    record = await resolve_approval(engine, approval_id, "approved", approver="user:boss")
except (ApprovalNotPending, ApprovalExpired):
    ...
# repeat the same execute_tool call with the same approval_scope: it runs once

Views and lifecycle

with_redaction

A view of the engine that redacts with policy and hashes with secret_key, for a host serving many tenants from one engine. Every method on the view applies the policy (both ingest and output rules) while sharing the parent's connection pool, backend, embedder, graph store, hooks and function tools. A per-call redaction= still overrides it. Synchronous.

Signature

def with_redaction(
    self, policy: RedactionPolicy, secret_key: str | None = None
) -> ContextEngine

Name	Type	Default	Meaning
policy	`RedactionPolicy`	`required`	The view's policy. RedactionPolicy() turns redaction off for the view.
secret_key	`str \| None`	`None`	The view's hash key. None keeps the parent's effective redaction key. Give each tenant its own so hash tokens cannot be joined across tenants.

Returns: a ContextEngine view.

Raises: TypeError when policy is not a RedactionPolicy. ValueError for an empty secret_key, or a hash rule with no key at all. A UserWarning when a hash rule falls back to config.secret_key.

A view snapshots the parent's config when it is built, does not re-mask content already stored (unchanged content re-ingested is skipped; delete and ingest again to re-store it), and does not scope function tools.

Example

from context_engine import RedactionPolicy, RedactionRule

policy = RedactionPolicy(rules=[
    RedactionRule(name="email", detector="email", action="hash"),
    RedactionRule(name="card", detector="credit_card"),
])
tenant = engine.with_redaction(policy, secret_key=tenant_hash_key)

result = await tenant.search("refund for card ending 4242", principals=["user:42"])

aclose and async with

async def aclose(self) -> None releases what the engine actually opened: the connection pool always, the embedding client if one was built (a client you supplied stays open), and the Neo4j driver if a graph call ran. Safe to call more than once. On a view it does nothing. The engine is also an async context manager that calls aclose() on exit. A process-lifetime singleton can skip it; anything that builds engines repeatedly should not.

Example

engine = ContextEngine(config)
try:
    ...
finally:
    await engine.aclose()          # safe to call more than once

async with ContextEngine(config) as engine:
    ...                            # closed on exit

Hooks

engine.hooks is a Hooks dataclass. Three callbacks come from the constructor; on_tool_call is set on the attribute. A callback that raises is logged and swallowed: it can never fail an ingest, search or tool call.

Hook	Signature	Called with
on_usage	`(UsageEvent) -> None`	One event per metered ingest, search or successful tool call. See UsageEvent.
on_error	`(Exception, dict) -> None`	Errors the engine handles itself (a failed document, a degraded graph leg, a graph cleanup), with a context dict naming the stage. Every error is also logged.
on_progress	`(dict) -> None`	Ingest stage boundaries: source_id, external_id, document_id (None before the row exists), name, stage, state, detail. Embedding also sends state="progress" per batch with done, total, tokens. Ignore states you do not recognise.
on_elicitation	`(ElicitationRequest) -> ElicitationResponse \| dict`	A third-party MCP server’s question during an mcp tool call: server, tool, tool_id, mode, message, requested_schema or url, principals, sensitive_hint. Return accept (with content), decline or cancel. Sync or async. Unset, the client declares no elicitation. A server that asks anyway gets a protocol error instead of decline: on a 2026-07-28 connection that fails the call, and on a handshake connection the server decides what happens next.
on_tool_call	`(dict) -> None`	One event per tool execution: tool_id, tool_name, kind, actor_type, actor_id, source, input_args, output_result, success, error, duration_ms, units, approval_id. The audit row is written regardless.

Python

from context_engine import ContextEngine, UsageEvent

def on_usage(event: UsageEvent) -> None:
    meter.add(event.kind, event.units, event.detail, event.provider_tokens)

def on_error(exc: Exception, ctx: dict) -> None:
    alerts.send(str(exc), ctx)          # ctx carries "stage" and ids

def on_progress(event: dict) -> None:
    # {"source_id", "external_id", "document_id", "name",
    #  "stage": "extract" | "redact" | "chunk" | "embed" | "structured" | "graph",
    #  "state": "started" | "done" | "progress", "detail": {...}}
    ui.update(event)

engine = ContextEngine(config, on_usage=on_usage, on_error=on_error, on_progress=on_progress)
engine.hooks.on_tool_call = lambda event: audit_log.write(event)

Sentinels

Three values that cannot be produced by accident, so an oversight never turns into maximum access.

Sentinel	Used for	Meaning
TRUSTED	principals	A trusted caller: ACL filtering is skipped. Truthy on purpose. This is the spelling to use instead of the deprecated None.
UNSET	update_document arguments	"Not passed". The default of every update_document field, so acl=None (unrestrict) and "leave the ACL alone" stay different.
UNSCOPED	knowledge tool and MCP scope	No ceiling: the tool may reach the whole corpus. None raises instead, so a forgotten scope is never a corpus-wide search. Truthy on purpose.

Python

from context_engine import TRUSTED, UNSET, UNSCOPED

await engine.search(q, principals=TRUSTED)          # trusted: ACL not applied
await engine.search(q, principals=[])               # anonymous: unrestricted documents only
await engine.search(q, principals=user.groups)      # a real caller

await engine.update_document(doc_id, name="New")    # acl omitted (UNSET): unchanged
await engine.update_document(doc_id, acl=None)      # acl=None: unrestricted

await engine.search_knowledge_base(action="list", principals=TRUSTED, scope=UNSCOPED)

Result and record types

All are exported from context_engine.

Hit

One retrieved chunk (a dataclass). score is always the fusion score; when a reranker ran, the list order is the ranking, so do not re-sort by score.

Field	Type	Meaning
document_id	`str`	The document the chunk belongs to.
document_name	`str \| None`	Document name.
description	`str \| None`	Document description.
source_id	`str \| None`	The source.
chunk_text	`str`	The chunk text, redacted if a policy applies.
idx	`int \| None`	Chunk position in the document.
lang	`str \| None`	Detected language.
score	`float`	Reciprocal rank fusion score.
chunk_id	`str`	The chunk id.
meta	`dict`	Chunker metadata such as section_title, sheet_title, page.

SearchResult

What search returns (a dataclass).

Field	Type	Meaning
hits	`list[Hit]`	Best first.
usage	`dict`	See search.

IngestReport

The outcome of an ingest or resume (a dataclass).

Field	Type	Meaning
documents	`list[DocumentReport]`	One report per document.
totals	`dict`	{"files", "failed", "units", "graph_units"}.

DocumentReport

The outcome for one document (a dataclass).

Field	Type	Meaning
document_id	`str`	The document id.
name	`str`	Document name.
status	`str`	For example completed, failed, skipped, embedding, waiting_model.
pages	`int \| None`	Pages, where the format has them.
chunks	`int`	Chunks stored.
units	`int`	Ingest units.
graph_units	`int`	Graph units.
provider_tokens	`dict`	Tokens used by providers.
error	`str \| None`	The raw error, for operators only.
redaction_failed	`list[str]`	Ingest rules that raised; text they would have masked may be stored unmasked.
pages_source	`str \| None`	Where a DOCX page count came from: app_xml, rendered_breaks, page_breaks or estimate.
unreadable_reason	`str \| None`	Why the document has less text than expected, when known.
unreadable_pages	`int`	Pages that could not be read.
mode_used	`str \| None`	The mode the document was processed at.
mode_reason	`str \| None`	A fixed code for why mode_used differs from the request.
skip_reason	`str \| None`	in_progress when another run for the same document is still going.
failure_reason	`str \| None`	A code from FAILURE_REASONS.
failure_message	`str \| None`	The fixed sentence for that code, safe to show to a model.
abandoned_request_keys	`list[str]`	The request keys a pass deferred before it failed outright; empty otherwise. The next resume_documents reports them once more in superseded with reason="failed".
graph_backend_stale	`bool`	True when a configured Neo4j could not be given this document's graph. The document still completed and search works from Postgres; resync_graph() replays it.

ResumeReport

What resume_documents returns (a dataclass).

Field	Type	Meaning
documents	`list[DocumentReport]`	One per document this call worked on: completed, failed (an expiry under ingest.max_park_seconds) or waiting_model (parked again).
skipped	`list[dict]`	{"document_id", "reason"} per document left parked on this call, reason no_content_source, content_mismatch or redaction_mismatch. Listed on every call until a resume succeeds.
superseded	`list[SupersededPark]`	Request keys nobody will ask for again since the last report. Reported once.
communities_resumed	`list[str \| None]`	Source ids whose pending community summaries this call completed (None is the documents ingested without a source).

SupersededPark

One set of request keys the host can drop from its batch store (a dataclass).

Field	Type	Meaning
document_id	`str`	The document.
request_keys	`list[str]`	The keys.
superseded_at	`str`	When the park was replaced, or the failure time for reason="failed".
reason	`"superseded" \| "failed"`	superseded: a re-ingest replaced the park. failed: the pass that deferred them then failed outright, or an expired park held them.

ParkedDocument

What content_source is called with (a dataclass): enough to find the original bytes.

Field	Type	Meaning
document_id	`str`	The document.
source_id	`str \| None`	The source.
external_id	`str \| None`	Your id.
name	`str \| None`	Document name.
content_sha256	`str \| None`	The sha256 the bytes must match.

UsageEvent

What on_usage receives (a dataclass). Units are currency-neutral; you decide what one costs.

Field	Type	Meaning
kind	`"ingest" \| "search" \| "tool"`	What was metered.
units	`int`	Units consumed.
detail	`dict`	Per-kind detail, for example the search mode or the tool name.
provider_tokens	`dict`	For example embedding_tokens, llm_input, llm_output.

ApprovalRecord

A plain-data copy of one approval row (a dataclass), returned by resolve_approval.

Field	Type	Meaning
id	`UUID`	The approval id.
tool_name	`str`	The call name.
tool_args_frozen	`dict`	The arguments approved, frozen.
source_id	`str \| None`	The source.
approval_scope	`str \| None`	Your claim scope.
principals	`list[str] \| None`	Who may see and resolve it.
status	`str`	pending, approved, rejected or expired.
approver	`str \| None`	Who resolved it.
approver_meta	`dict \| None`	Your metadata from the resolve call.
expires_at	`datetime \| None`	When it expires.
resolved_at	`datetime \| None`	When it was resolved.
created_at	`datetime \| None`	When it was opened.
tool_id	`str \| None`	The tool row, None for a function tool.

ToolConfig

The registration shape for a tool (a Pydantic model). ToolKind is Literal["http", "db", "mcp", "function"]; config_schema(kind) returns the JSON Schema of config for each kind, for building forms.

Field	Type	Meaning
id	`str \| None = None`	Set on stored tools.
name	`str`	Tool name; the call name adds the kind prefix.
kind	`ToolKind`	http, db, mcp or function.
description	`str = ""`	What the tool does; shown to the model.
config	`dict`	Kind-specific settings, encrypted at rest.
source_id	`str \| None = None`	The source the tool belongs to.
acl	`list[str] \| None = None`	Who may see and run it.
requires_approval	`bool = False`	Every call needs an approval.
approval_policy	`dict = {}`	{"condition"?: str, "timeout_minutes"?: int}.
enabled	`bool = True`	Disabled tools are not offered.
meta_data	`dict = {}`	Your metadata, stored in clear.

Scope

The ceiling a mounted knowledge tool may reach (a frozen dataclass). A list of source ids is shorthand for Scope(source_ids=...).

Field	Type	Meaning
source_ids	`tuple[str, ...] \| None = None`	Sources the tool may reach.
document_ids	`tuple[str, ...] \| None = None`	A whitelist of documents within them.

Errors

Programming and configuration mistakes raise ValueError. Failures that depend on the request's data raise EngineActionError, so an application can catch it once and map it to a 404, a 400 or a tool error payload.

Exception	Base	When
EngineActionError	`Exception`	A request-specific failure: a document or tool not found or not visible, a malformed document id, a failed tool call, compute disabled or with nothing to compute over, map_reduce with no LLM. Catch it once and map it to your surface.
DatabaseNotMigrated	`RuntimeError`	The first call on a database with none of the engine's tables, or whose role cannot see them. Its message is the migrate command to run. It replaces the driver's undefined-table error, so a host that caught that error for this case catches DatabaseNotMigrated. Exported from the package root. The HTTP routers answer it with 503 and one fixed sentence, and so do the MCP tools.
ToolExecutionFailed	`EngineActionError`	A tool that exists ran and failed. The message starts "tool execution failed:", stripped and masked. The routers answer 502.
ToolAuditFailed	`ToolExecutionFailed`	A tool call could not be recorded in the audit log at all: neither its full row nor the last-resort row could be written (the database is down). The tool has already run. Raised only with tools.audit_on_failure="fail", the default. Carries tool_ran (always True), result (shaped and redacted as the caller would have received it; None when the tool itself failed), truncated and tool_error. Do not retry it blindly: that runs the tool a second time. The routers answer 502 with tool_ran: true and the result in the body, and the MCP surface answers an error whose text says the tool ran and ends with the result. The call is metered (on_usage fires), because the tool ran.
InvalidInputError	`ValueError`	The caller's own input, refused on purpose: an empty query, an unknown mode, a bad cursor, a missing scope, a tool argument that cannot be stored, a NUL character in a field the caller supplies. Its tool_ran is False: when execute_tool raises it the tool did not run and the call is safe to send again once corrected. The routers answer 400. On the MCP surface it and EngineActionError are the only errors whose text reaches the client.
AmbiguousToolError	`EngineActionError`	execute_tool got a name several visible tools share and no tool_id. The routers answer 409.
IngestTooLarge	`ValueError`	An intake over ingest.max_file_bytes or ingest.max_rows. Carries knob, limit, actual. The routers answer 413.
ExtractionFailed	`RuntimeError`	A file could not be read. Raised inside ingest and reported on the document as failure_reason="extraction_failed", not to your call.
EmbeddingDimensionMismatch	`RuntimeError`	An embedder, yours or the built-in client's, returned vectors of a different width than embedding.dim or than this database's vector columns. The document fails with failure_reason="embedding_dimension_mismatch": check the model and the dimension the database was migrated with.
InputTooLarge	`ValueError`	One text is over the provider's per-input token limit (embedding.max_input_tokens). A chunk over it is never split: inside an ingest the document fails with failure_reason="input_too_large", and embed_chunks raises this error when you call it yourself. Carries index, tokens, cap, knob.
EmbedBatchFailed	`RuntimeError`	An embedding batch failed after earlier ones were delivered. Carries done_items, total_items, failed_batch, batches, tokens; the delivered vectors are kept.
EmbeddingBatchRejected	`RuntimeError`	A single-input embedding request was refused with 400, 413 or 422. Carries provider, status, items, index, chars.
GraphLegUnavailable	`RuntimeError`	Reported through on_error, not raised, when a graph search ran only the hybrid legs. See usage["degraded"].
ApprovalNotPending	`Exception`	resolve_approval on an approval that is missing, already resolved, expired, or not visible to the caller.
ApprovalExpired	`Exception`	resolve_approval on a pending approval past expires_at. The row is marked expired first.
KeyError	`built-in`	get_document, update_document and delete_document on a document that is absent or not visible.
OpenAICompatError	`Exception`	The built-in OpenAI-compatible client got a non-retryable status, or ran out of retries; raised by LLMClient.call, Embedder.embed and call_llm. Carries status_code, body (the response text) and headers (a copy of the response headers only, so no request header and no API key is on the error). A connection failure or timeout is the httpx.TransportError itself.
OpenAICompatReplyError	`ValueError`	A successful reply that does not carry what was asked for: a chat reply with no choice, or an embeddings reply that does not hold exactly one vector per input with each index used once. Raised in place of an empty answer.
ModelDeferred	`Exception`	Raised by YOUR model or embedder, not by the engine: the request was accepted and its answer comes later. ModelDeferred(retry_after=None) takes seconds, the earliest useful resume; a negative or non-finite value is refused. During ingest the document parks as waiting_model; at query time it is refused for map_reduce and compute, and a query embedding deferral is treated as an embedder outage. See Batch mode.
ConfigTemplateError	`ValueError`	From context_engine.tools.config: a tool config you must fix, such as an unresolvable "__redacted__" placeholder. The routers answer 400.

HTTP routers

Mountable HTTP surfaces for your web framework, loaded lazily so importing context_engine never imports a framework. auth and principals are required on every factory: who the caller is has no safe default. Routes, bodies and status codes are on the HTTP API page.

Signatures

def create_router(
    engine: ContextEngine,
    *,
    auth: Callable[..., Any],
    principals: Callable[..., list[str] | None],
)                                                   # [fastapi], returns an APIRouter

def create_flask_blueprint(
    engine: ContextEngine,
    *,
    auth: Callable[[Any], Any],
    principals: Callable[[Any], list[str] | None],
    name: str = "context_engine",
)                                                   # [flask]

def create_django_urlpatterns(
    engine: ContextEngine,
    *,
    auth: Callable[[Any], Any],
    principals: Callable[[Any], list[str] | None],
) -> list                                           # [django]

def create_tools_router(
    engine: ContextEngine,
    *,
    auth: Callable[..., Any],
    principals: Callable[..., list[str] | None],
    redirect_base_url: str,
    pending_store: PendingStore | None = None,
    config_template: bool = False,
    approval_scope: Callable[..., str | None] | None = None,
) -> APIRouter                                      # [fastapi]

MCP

create_mcp_app serves the knowledge tool over MCP (streamable HTTP) and needs the [mcp] extra. principals, scope, redaction and secret_key are resolved by you per call and are never tool arguments. knowledge_tool_definition() gives the tool as data for your own agent loop, and call_knowledge_tool runs it; engine.search_knowledge_base is the same call as a method. Actions and arguments are on the Knowledge tool page.

Signatures

def create_mcp_app(
    engine: ContextEngine,
    *,
    principals: Callable[[], Any],
    scope: Any = None,               # required in practice: source ids, Scope, or UNSCOPED
    redaction: Any = None,           # RedactionPolicy, or a callable returning one
    secret_key: Any = None,          # str, or a callable returning one
    compute: Any = None,             # async (instruction, *, source_ids, document_ids, principals) -> dict
    map_reduce: Any = None,          # same, plus limit
    approval_scope: Callable[[], Any] | None = None,
) -> Any                             # ASGI app (streamable HTTP); FastMCP at app.state.mcp

def knowledge_tool_definition() -> dict   # {"name", "description", "input_schema", "annotations"}

async def call_knowledge_tool(
    engine, *, action: str, principals, scope: Any = None, ...   # same arguments as
) -> dict                                                        # engine.search_knowledge_base

Calling third-party MCP servers

The other direction, an mcp tool that calls someone else’s MCP server, runs on the official MCP Python SDK client (the mcp package, >=2.2,<3, from the same [mcp] extra) over Streamable HTTP only. It speaks protocol versions 2024-11-05, 2025-03-26, 2025-06-18, 2025-11-25 and 2026-07-28. The global settings live in McpClientConfig (protocol, redirects, elicitation_url, elicitation_timeout_s, redact_logs) under ToolsConfig(mcp=...), and a tool’s own config may override all but redact_logs for its connection. The exceptions the client raises are in context_engine.tools.executors.mcp_client: MCPError, MCPHttpStatusError (with .status and .advice), MCPNoResponse, MCPUrlError and its subclass UnsupportedMCPTransport; through execute_tool they surface as ToolExecutionFailed. Settings, redirects, elicitation, guarantees and error texts are in Connecting MCP servers.

Redaction

A RedactionPolicy is a list of rules. Each rule finds spans with a detector or a regex and masks, hashes or removes them, at ingest (what is stored), at output (what is returned), or both. Output rules can exempt principals with unless. Built-in detectors: email, phone, ssn, credit_card, iban, api_key. An empty policy is a guaranteed no-op. Set one engine-wide with config.redaction, per tenant with with_redaction, or per call with redaction=.

Signatures

class RedactionRule(BaseModel):
    name: str
    detector: str | None = None      # built-in or a custom_detectors key
    pattern: str | None = None       # a regex; exactly one of detector / pattern
    field: str | None = None         # reserved: setting it raises ValueError
    action: Literal["mask", "hash", "remove"] = "mask"
    placeholder: str | None = None   # default "[NAME]"
    apply_at: Literal["ingest", "output", "both"] = "output"
    unless: list[str] = []           # principals exempt from the rule (output only)

class RedactionPolicy(BaseModel):
    rules: list[RedactionRule] = []
    custom_detectors: dict[str, Callable[[str], list[tuple[int, int]]]] = {}
    def is_empty(self) -> bool

def apply_redaction(
    text: str,
    policy: RedactionPolicy,
    *,
    phase: Literal["ingest", "output"],
    principals: Sequence[str] | None = None,
    secret_key: str | bytes | None = None,
    hooks: Any = None,
) -> tuple[str, dict]                # (text, {"rules_fired", "spans", "rules_failed"?})

A rule raises ValueError at construction when it sets both or neither of detector and pattern, has an invalid regex, sets field, or puts unless on an ingest-only rule. A policy raises for duplicate rule names or an unknown detector. apply_redaction raises ValueError when a hash rule applies and no key is given. A hash token looks like [NAME:16 hex chars].

Presidio detectors

presidio_detector and presidio_detectors turn Presidio entities into custom_detectors. They need the [presidio] extra and a spaCy model. presidio_detectors shares one analyzer across every entity and builds it on first use; results under score_threshold are dropped.

Python

from context_engine import RedactionPolicy, RedactionRule, presidio_detectors

policy = RedactionPolicy(
    rules=[
        RedactionRule(name="person", detector="PERSON"),
        RedactionRule(name="location", detector="LOCATION", action="remove"),
    ],
    custom_detectors=presidio_detectors(["PERSON", "LOCATION"]),
)

# presidio_detector(entity, *, analyzer=None, language="en", score_threshold=0.5)
# presidio_detectors(entities, *, analyzer=None, language="en", score_threshold=0.5)

Low-level exports

The rest of __all__: the pieces the engine is built from, for hosts that assemble their own pipeline. Most applications need only the ContextEngine methods above.

Export	Shape	What it is
extract	`async (content, filename, mime, *, vision_llm, hooks, extraction=None, http_client_factory=None, max_rows=None) -> Extracted`	Text extraction for one file. Degrades instead of raising, except IngestTooLarge over max_rows. Extracted carries text, pages, slides, pages_source and more.
run_search	`async (query, *, config, hooks, backend, embedder, session_factory, source_ids=None, document_ids=None, principals=None, top_k=10, mode="hybrid", ...) -> SearchResult`	The retrieval pipeline under engine.search. Here principals=None means trusted with no warning; prefer the engine method.
rrf_fuse	`(ranked_lists: dict[str, list[str]], *, k: int, weights: dict[str, float]) -> list[tuple[str, float]]`	Reciprocal rank fusion. A leg missing from weights counts 1.0; a leg weighted 0.0 is ignored.
build_embedder	`(cfg: EmbeddingConfig, *, http_client_factory=None) -> Embedder`	The built-in OpenAI-compatible embedding client.
build_llm_client	`(cfg: LLMConfig, *, http_client_factory=None) -> LLMClient`	The built-in OpenAI-compatible chat client.
call_llm	`async (cfg, *, system, user, json_mode=False, images=None, max_tokens=None, thinking_budget=None, temperature=None, response_schema=None, schema_name="response", client=None, http_client_factory=None)`	One LLM call. Returns (text, {"input", "output"}); with response_schema a reply that also carries mode.
host_rerank	`async (fn: RerankFn, query, docs) -> list[int]`	Calls a host reranker and checks the shape of its reply; RerankFn is the type ContextEngine(reranker=...) takes.
encrypt_dict / decrypt_dict	`(data: dict, key: bytes) -> str / (token: str, key: bytes) -> dict`	AES-256-GCM, base64 of nonce plus ciphertext. A wrong key fails closed.
get_secret_key	`(config) -> bytes`	Decodes config.secret_key (base64url, 32 bytes). RuntimeError when it is missing.
CodeRunner	`Callable[[str, dict, int], dict \| Awaitable[dict]]`	Your isolation boundary for compute: called with (code, context, timeout); context["dfs"] holds live DataFrames. Its result is trusted only as far as JSON.
compute_over_frames	`async (frames: dict[str, DataFrame], instruction, *, config, model_cfg=None, timeout=30, hooks=None, principals=None, documents=None, redaction=None, secret_key=None, code_runner=None, http_client_factory=None, model=None) -> dict`	compute over DataFrames you already hold. Same guards and result shape; timeout clamped to 1 to 300 seconds. model is a host function for this call, beating model_cfg and config.llm.
compute, get_document_text, list_documents, query_structured	`module functions taking session_factory=...`	The functions under the engine methods of the same names. Prefer the methods; these take principals=None as trusted without a warning.
query_type, is_spaceless, trigram_limit, vector_floor, QueryType	`query shape helpers`	How the engine reads a query: "exact_match", "conceptual" or "hybrid", and the per-query trigram and vector thresholds.
extract_structured_data, resolve_fields, upsert_registry, ExtractionResult	`structured extraction`	The extraction behind extract_structured=True (never raises; failures come back as quality="failed") and the field registry.
StorageBackend, PostgresBackend, ChunkRow, SearchScope	`storage`	The chunk-plane protocol, its Postgres implementation, a chunk on its way in, and the filters every search leg applies.
LEXICAL_STOPWORDS	`storage`	The built-in English stopword list search.lexical_match="any" drops from a query when search.lexical_stopwords is unset.
units_for_file, graph_units, failure_message, FAILURE_REASONS, classify_failure	`usage helpers`	The unit formulas (PDF and DOCX per page, PPTX per slide, image 1, other per MB; graph ceil(chunks/8) plus the primary communities actually summarised, so none with graph.community_summaries off), and the failure codes with their fixed sentences.
emit_usage, emit_error, emit_progress, Hooks	`hook plumbing`	Report through hooks from your own code; none of them raise.
format_result, rows_to_tsv	`(result, response_mode="json") / (rows: list[dict]) -> str`	The TSV conversion execute_tool uses, for any JSON you already hold.
HttpClientFactory, HttpClientPurpose	`Callable[[purpose], httpx.AsyncClient \| None]`	Purposes: llm, embeddings, reranker, tool_http, mcp_oauth. A client you supply is never closed by the engine.
Embedder, LLMClient, OpenAICompatClient, OpenAICompatError, OpenAICompatReplyError	`classes`	The built-in client; build the two wrappers with build_embedder and build_llm_client.
BUILTIN_LEGS, RetrievalLeg, RetrievalScope, FusionFn, GraphBackend, RetrievalFailed	`plugin types`	The plugin seams: the reserved leg names (fts, trgm, ann, graph), a leg's signature and the read-only scope it receives, the fusion signature, the graph backend protocol, and the error a search raises when every leg failed. SearchResult.timings holds each leg's wall time in milliseconds, never part of usage.
GRAPH_REASONS	`tuple[str, ...]`	Why the graph leg did or did not run: not_requested, graph_disabled, graph_unreachable, no_graph_documents, no_entities_in_query, graph_detection_failed, no_visible_graph_chunks, auto, requested, graph_weight_zero.
ModelRequest, ModelReply, ModelFn	`dataclasses and the host model type`	What a host model receives and returns. ModelRequest.key is the request key (below). ModelReply carries text, tokens, mode ("prompt" by default) and refusal.
request_key, embedding_request_key	`(request: ModelRequest) -> str / (text: str, kind: str, model: str) -> str`	The stable batch keys, identical in both ports and on every resume: sha256 hex over the canonical JSON of the request (text purposes post-redaction, vision over the page image bytes), and the per-text embedding key with model your config.embedding.model.
ContentSource	`Callable[[ParkedDocument], Any]`	The content_source callback of resume_documents: returns the original bytes, sync or async, or None.
__version__	`str`	The installed package version.

Context Engine: Python API reference

Overview

Install and extras

Migrate the database first

Constructor

Ingestion

ingest

resume_embeddings

resume_documents

Search

search

Documents

get_document

get_document_text

get_documents

get_chunks

list_documents

update_document

delete_document

stats

resync_graph

rebuild_communities

enable_bm25

Corpus questions

documents_by_name

document_types

field_summary

spreadsheet_schema

document_structure

tabular_scope

query_structured

map_reduce

map_reduce_targets

compute

Knowledge tool

search_knowledge_base

Tools

register_tool

register_function_tool

update_tool

delete_tool

list_tools

search_tools

test_tool

execute_tool

Approvals

resolve_approval

Views and lifecycle

with_redaction

aclose and async with

Hooks

Sentinels

Result and record types

Hit

SearchResult

IngestReport

DocumentReport

ResumeReport

SupersededPark

ParkedDocument

UsageEvent

ApprovalRecord

ToolConfig

Scope

Errors

HTTP routers

MCP

Calling third-party MCP servers

Redaction

Presidio detectors

Low-level exports