context-engine

Retrieval that enforces access control and PII redaction inside the query,
as a library, on your own Postgres.

Built by Promptev — the governed AI agent platform. This library is the retrieval engine behind Promptev's Context Packs, open-sourced (Apache-2.0). Use it standalone, or see what the platform adds.

Most retrieval stacks make permissions your problem: you filter results after the fact, in application code, and hope the ordering is right. This one pushes the ACL into the SQL predicate every search leg shares, so a document a caller may not see is never fetched, never ranked, and never reaches a reranker. PII redaction sits at the same depth — masked before hits are POSTed to a hosted reranker and before the tool-call audit row is written.

It is a library, not a server. No UI, no orchestration, no workflow engine, no opinion about your framework — pip install, point it at a Postgres you already run, call engine.search(). BYO everything: your database, your embedding provider, your LLM. It never calls a hosted service you didn't configure, and never implements auth, billing, or credential storage — it exposes hooks (on_usage, on_error, on_progress) and injection points (auth, principals) for your app to wire up.

On top of that it does the ordinary work well: structure-aware ingestion for PDF/DOCX/PPTX/XLSX/HTML/images, hybrid retrieval (full-text + trigram + vector, RRF-fused) with an optional graph leg, and a governed tool-execution path with approval gates and audit.

Not overclaiming: it does not prevent hallucination or guarantee grounding, and its compute() action (LLM-generated Python) is off by default — only enable it behind real OS-level isolation.

📖 Full documentation: https://promptev.ai/documentation/context-engine/

In this repo: access control (the trichotomy, enforcement, and the vector-index interaction) · the ACL-recall benchmark · cookbook · TypeScript client (@promptev/context-engine)

Why you'd choose this

1. Access control that survives the vector index. Filtering an approximate index with a WHERE clause is a post-filter — the index picks candidates first, your ACL throws them away second, and the query returns short. We measured it on 200k public documents at production embedding width, and found something the literature misses: because real permissions are topically clustered rather than random, the loss is about twice what a conventional benchmark reports — recall 0.473 against 0.980 at identical visibility, a quarter of queries empty, no error raised. Tuning does not fully close it, so this library decides between an exact and an approximate scan per query instead. How access control works · benchmark and method.

Suspect your current stack? Prove it on your own data — one read-only command, no embedding provider called:

context-engine check-acl-exposure --database-url postgresql://...

2. Redaction that runs in the right place. Not more PII detectors than everyone else — better placement. Hits are masked before the reranker POSTs them to a hosted provider and before the audit row is written, so the masked value is the only one that ever leaves the process. Wiring a redaction library into a pipeline by hand is exactly where that ordering goes wrong. (Bring Presidio for breadth — its entities become ordinary rules.)

3. It's a library, so the data never moves. No server to run, no corpus to hand over, no orchestration framework to adopt. If you're subject to rules where you stay accountable for what your AI did — the EU AI Act's Article 12 logging duties fall on the deployer, not the vendor — then "runs in my process, on my Postgres" is the answer you need, and a hosted service can't give it to you.

4. Governed tool execution in the same place as retrieval. ACL-scoped tools, encrypted credentials, a fail-closed approval gate with single-use atomic resolution, and an audit row per call. Most stacks buy this separately from a gateway.

When you outgrow the library

The library is the engine. Promptev is the platform it was extracted from. It is the same engine either way; what differs is how much of the surrounding plumbing you run yourself.

Your database, your model keys and your connectors stay your choice either way — run them yourself, or have Promptev run them. That is a deployment decision, not a difference between the two:

What the platform adds on top of the engine:

Self-hosting the engine is a first-class path and stays one — it is Apache-2.0 and never calls home. The platform is for when the plumbing around the engine becomes the job.

Install

pip install "promptev-context-engine[postgres]"
# or with uv:
uv add "promptev-context-engine[postgres]"

# before it's on PyPI, install from git:
pip install "promptev-context-engine @ git+https://github.com/promptev/context-engine"
uv add "promptev-context-engine @ git+https://github.com/promptev/context-engine"

TypeScript lives in js/ as @promptev/context-engine — same Postgres schema, so a corpus ingested by either client is searchable by the other:

npm install @promptev/context-engine pg

Needs Node 22+.

Needs Python 3.12+ and Postgres with the vector, pg_trgm, and unaccent extensions (auto-created by context-engine migrate if the role has CREATE EXTENSION).

The core engine is web-framework-agnostic — it depends on no web framework. Pick the router adapter that matches YOUR project (FastAPI, Flask, or Django) via an extra, or use none and call engine.* directly.

Optional extras — install only what you use:

pip install "promptev-context-engine[fastapi,graph]"   # or [flask] / [django]
uv add "promptev-context-engine[fastapi,graph]"

Teach your coding agent the access-control rules

Most integration code is now written with an agent's help, and the highest-cost mistake in this library is one an agent makes readily: principals=None means trusted caller, no filtering, so treating it as "nobody is logged in" returns the entire corpus.

That spelling is being retired. None is falsy, so it is what every accessor degrades to — user.groups or None, a missing key, if user else None — which made the unauthenticated path the maximum-privilege path. Say it explicitly instead:

from context_engine import TRUSTED

await engine.search(q, principals=TRUSTED)      # skip ACL, deliberately
await engine.search(q, principals=[])           # anonymous
await engine.search(q, principals=user.groups)  # scoped

None still works and warns; it raises in 1.0. The package also ships a skill stating this and the other invariants worth not guessing.

context-engine install-skill            # -> ./.claude/skills/context-engine/
context-engine install-skill --global   # -> ~/.claude/skills/
context-engine install-skill --print    # stdout, for agents that read other formats

It is an explicit command, never an install hook — the file becomes instructions inside your agent, so it should be your decision. An edited skill is never overwritten without --force.

Configure

ContextEngineConfig is the one settings object (or build it from CE_-prefixed env vars, nested with __, e.g. CE_EMBEDDING__PROVIDER).

from context_engine import ContextEngineConfig, EmbeddingConfig, LLMConfig

config = ContextEngineConfig(
    database_url="postgresql://user:pass@localhost:5432/mydb",
    embedding=EmbeddingConfig(provider="openai", model="text-embedding-3-small", api_key="sk-..."),
    # llm is optional — only for structured extraction, compute, and the graph:
    llm=LLMConfig(provider="openai", model="gpt-4o-mini", api_key="sk-..."),
)

Embeddings: openai · azure_openai · gemini · vertex_ai · voyage · cohere · custom (any OpenAI-compatible /embeddings endpoint via base_url). LLMs add anthropic · bedrock. (Anthropic has no embeddings API — pick another provider for EmbeddingConfig.)

vertex_ai is the same Gemini models served from Vertex AI, and the only difference is how you authenticate: Application Default Credentials against a GCP project instead of an API key.

llm=LLMConfig(provider="vertex_ai", model="gemini-2.5-pro",
              project="my-project", location="us-central1"),

project and location are optional — leave them out and the Google SDK resolves them from GOOGLE_CLOUD_PROJECT / GOOGLE_CLOUD_LOCATION, so an existing GCP deployment needs no new configuration. The Python SDK will also discover the project from ADC when neither is set; the TypeScript client's SDK will not — measured, not assumed — so for @promptev/context-engine name the project in config or in the environment.

Credentials themselves come from ADC in every form: GOOGLE_APPLICATION_CREDENTIALS, workload identity, or gcloud auth application-default login. There is no credentials field to set. api_key is ignored when a project is present (the SDK rejects the pair outright); on its own it selects Vertex express mode.

StorageConfig holds the deployment knobs: ann_exact_threshold (the vector leg's exact/approximate crossover — hardware- and dimension-specific, see docs/access-control.md) and the connection pool, passed through to SQLAlchemy — pool_size (5), max_overflow (10), pool_timeout (30s), pool_recycle (1800s, keep it under the idle timeout of any pooler in front of the database) and pool_pre_ping (on: without it every connection left stale by a database restart or failover fails one query first). All reachable as env vars, e.g. CE_STORAGE__POOL_SIZE.

Provision the database

context-engine migrate --database-url postgresql://user:pass@localhost:5432/mydb --dim 1536
# add --graph to also create entity/relationship/community tables

--dim must match your embedding model's width (1536 for text-embedding-3-small) and is locked in on first migrate.

Ingest

from context_engine import ContextEngine

engine = ContextEngine(config)

# Plain text
report = await engine.ingest(
    text="Annual leave accrues at two days per month for every band-three employee.",
    name="leave-policy",
    source_id="hr-handbook",       # a namespace to scope search/stats
    acl=["group:hr"],              # omit for an unrestricted document (visible to everyone)
)

# A file — PDF/DOCX/PPTX/XLSX/HTML/EML/images/plain text, auto-detected
with open("policy.pdf", "rb") as f:
    report = await engine.ingest(file=f, name="policy.pdf", source_id="hr-handbook")

print(report.totals)               # {"files": 1, "failed": 0, "units": 3, ...}
print(report.documents[0].status)  # "completed" | "failed" | "batch_pending"

Construction is cheap and lazy — no connection is opened until the first call that needs one. Release the pool (and the embedding client, and the Neo4j driver) with await engine.aclose(), or use the engine as an async context manager; a process-lifetime singleton can skip it, anything building engines per job or per test should not.

• mode="hybrid" (default) or mode="graph" (needs the [graph] extra).
• Scanned pages / images are transcribed to structured Markdown (tables and headings preserved) when vision_llm is configured — otherwise embedded text, with Tesseract as a fallback if [ocr] is installed. Vision is provider-agnostic: it goes through the same call_llm seam as everything else, so Claude, GPT-4o, Gemini and Bedrock all work — whatever your LLMConfig names. The [vision] extra only adds page rasterization.
• Born-digital PDFs keep their structure too, with the [pdf] extra: the text layer is converted to Markdown locally — no OCR, no LLM, no network — so a table survives as a table into the chunker instead of being flattened into a run of words. Without the extra you get today's plain-text extraction.
• ingest() never raises on a bad document — check report.documents[i].status.

Search

result = await engine.search(
    "how much annual leave do I get",
    source_ids=["hr-handbook"],
    document_ids=None,         # optionally pin to specific documents WITHIN the sources
    principals=["group:hr"],   # TRUSTED = trusted/internal (skips ACL); [] = anonymous
    top_k=10,
)

for hit in result.hits:
    print(hit.document_name, hit.score, hit.chunk_text[:80])

Fuses full-text + trigram + vector search with Reciprocal Rank Fusion; mode="graph" adds a graph leg. compress_to_tokens=N trims weak hits to fit a prompt budget.

Compute over tables

engine.compute() turns the in-scope CSV/XLSX documents into dataframes and has the LLM write code against them. compute_over_frames() is the same path for frames you already hold — an uploaded workbook, a connector's sheet, a query result — with no document to point at:

from context_engine import compute_over_frames

out = await compute_over_frames(
    {"expenses": df},              # {sheet name: DataFrame}
    "total the amount column",
    config=engine.config,
)
out["result"]

Both are off by default (enable_code_execution=True turns them on — only behind real OS-level isolation) and both apply config.redaction before the LLM sees anything: every string cell and every column header is masked, and the returned dict — generated code included — is swept on the way out.

Serve

The same document/search/stats surface ships as a ready adapter for FastAPI, Flask, or Django — pick the one matching your project. All three expose identical routes (POST/GET/DELETE/PATCH /documents, POST /search, GET /stats) with identical behavior (they share one framework-neutral core). auth and principals are required arguments — the package implements no auth, and resolving who the caller is has no safe default. To run a mount open on purpose, say so: principals=lambda: [] (anonymous — sees only documents with no ACL). Returning TRUSTED (import it from context_engine) means a fully trusted caller and skips ACL filtering entirely; returning None is rejected with a 500 so a missing resolver can never open a mount by accident.

A caller may only file a document under an ACL it already holds; POST /documents and PATCH /documents/{id} answer 403 otherwise, so an untrusted request can't push content into another group's retrieval scope.

FastAPI ([fastapi] extra):

from context_engine import create_router

router = create_router(engine, auth=my_auth, principals=my_principals)
app.include_router(router, prefix="/context")

Flask ([flask] extra) — auth(request) / principals(request), async views:

from context_engine import create_flask_blueprint

bp = create_flask_blueprint(engine, auth=my_auth, principals=my_principals)
app.register_blueprint(bp, url_prefix="/context")

Django ([django] extra) — include the urlpatterns in your urls.py:

from django.urls import include, path
from context_engine import create_django_urlpatterns

urlpatterns = [
    path("context/", include(create_django_urlpatterns(engine, auth=my_auth, principals=my_principals))),
]

Not on any of them? Just call engine.ingest() / engine.search() directly from your own views — the engine needs no web framework at all.

Or serve the same engine as an MCP tool surface ([mcp] extra):

context-engine mcp --database-url postgresql://... --port 8080

from context_engine import create_mcp_app, UNSCOPED
app.mount("/mcp", create_mcp_app(engine, principals=my_principals, scope=UNSCOPED))

scope is the ceiling of source ids this mounted tool may ever reach, and is REQUIRED. It is resolved per call like principals and is never a tool argument: a model can ask for any source id it likes, and a tool that believed it would let one caller read another's documents. Map your own word onto it — a project, a matter, a customer — and pass scope=lambda: ["src-a", "src-b"], or Scope(source_ids=(...), document_ids=(...)) to hand an agent specific documents. A host with one shared corpus writes scope=UNSCOPED on purpose. The caller can narrow within that ceiling and never widen past it.

Registers ONE knowledge tool, search_knowledge_base, whose action picks what it does:

Combining the tools into sub-actions of one is better for the model that calls them: one description carries the decision rules once, and there is one name to route through.

Omit mode on a search and it is worked out from the documents in scope — the graph leg is added when the deployment has a graph and something in scope was ingested that way, and it runs beside the other legs rather than ahead of them. Naming "hybrid" or "graph" forces one.

query_meta and map_reduce need an LLM, compute needs code execution explicitly enabled, and the graph actions need a configured graph — discover marks them off otherwise, and an off action called anyway answers a plain {"success": false, "error": …} result instead of hiding. discover and list page (has_more + next_cursor, sent back as cursor), and a truncated answer carries the next_page call that reads the rest. The connector-tool gateway (search_tools + execute_tool) is registered beside it.

knowledge_tool_definition() exports the tool as data — name, description and input schema as JSON Schema — so a host wiring it into its own agent loop never hand-writes the schema. create_mcp_app also takes redaction (a policy applied per call), and compute / map_reduce callables that REPLACE the built-in ones, which is where a host applies its own rules for permission, billing and approval.

The same tool is a plain method, so an application driving its own agent loop needs no MCP at all — and cannot get a different answer, because both doors call one implementation:

answer = await engine.search_knowledge_base(
    action="search", query="annual leave", principals=caller, scope=["hr"]
)

Tools

Let an LLM/agent call http / db / mcp (external server) and function (a Python callable) tools through one governed path — ACL-checked, config AES-256-GCM-encrypted at rest, every call audited.

from context_engine import ToolConfig

# HTTP tool — the LLM fills `city` at call time
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",
        "headers": {"Authorization": "Bearer super-secret-token"},
        "llmQueryParameters": {"properties": {"city": {"type": "string"}}, "required": ["city"]},
    },
))  # -> call_name "http_get_weather"

# Function tool — no config, no persistence
def add(a: int, b: int) -> int:
    """Add two integers."""
    return a + b
engine.register_function_tool(add)  # -> call_name "fn_add"

out = await engine.execute_tool("http_get_weather", {"city": "Lahore"})
# -> {"result": {"status_code": 200, "data": {...}}, "usage": {"kind": "tool", "units": 1}}

Approval gate — a tool with requires_approval=True returns an approval_required payload instead of running, until resolved:

from context_engine import resolve_approval

out = await engine.execute_tool(
    "http_get_weather", {"city": "Lahore"},
    principals=["user:a"],
    approval_scope="run-42",   # opaque, chosen by you — a run id is the usual shape
)
# -> {"approval_required": {"approval_id": "...", "expires_at": "...", ...}}   (not executed yet)
await resolve_approval(engine, approval_id, "approved", approver="boss")
# the same call, same approval_scope, runs for real — exactly once

approval_scope is the claim scope: an approved record is consumed only by the same call in the same scope (and, when the record carries principals, by a caller who overlaps them). Within a scope, identical pending requests share one record, so a retrying agent gets the same approval_id back. Resolve the scope server-side — on the routers and the MCP gateway it is a dependency beside principals, never a request field. Calling without it still works this release, unscoped and with a DeprecationWarning; the next release refuses a gated call with no scope.

Getting the request in front of a human (Slack, in-app) is the consumer's job — the engine owns only the single-use, atomic approval state machine. Park until expires_at; after that the record can no longer be approved.

Redaction

Span-level masking of PII (or anything you can describe with a regex or a function) as content is ingested, retrieved, or returned. It composes with ACL (ACL = who may see a document; redaction = what's masked inside it) and is a pure transform — no model, no network. The default policy is a guaranteed no-op.

from context_engine import ContextEngineConfig, RedactionPolicy, RedactionRule

config = ContextEngineConfig(
    database_url="postgresql://...",
    embedding=...,
    redaction=RedactionPolicy(rules=[
        RedactionRule(name="emails", detector="email"),                  # -> [EMAILS]
        RedactionRule(name="order_id", pattern=r"\bORD-\d{6}\b"),         # -> [ORDER_ID]
    ]),
)

Built-in detectors: email · phone · ssn · credit_card · iban · api_key. Actions: mask · hash (keyed, joinable pseudonym — needs secret_key) · remove. apply_at: output (default, masks on every read, can be principal-conditional via unless) · ingest (scrub before storage) · both.

The detectors are deliberately a small set — the placement is the point. Redaction runs inside retrieval: hits are masked before _maybe_rerank POSTs their text to Cohere/Voyage/Jina, and tool results are masked before the audit row is written, so the masked value is the only one that leaves the process or lands in retention. That ordering is what usually goes wrong when a redaction library is bolted onto a pipeline by hand.

For broader detection, bring Microsoft Presidio ([presidio] extra) — its entities become ordinary rules and inherit the same placement guarantees:

from context_engine import presidio_detectors

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

Any Callable[[str], list[tuple[int, int]]] works as a detector, so an in-house classifier drops in the same way.

Benchmarks

Does ACL filtering cost you retrieval quality? Yes — badly — unless the vector leg is tuned for it, and the failure is silent.

Any system that applies an access-control predicate as a WHERE clause on top of an approximate vector index is doing a post-filter: the index walk is ordered by distance to the query, rows the caller may not see are discarded as they are met, and the walk stops when its candidate budget runs out. If the caller can see only a slice of the corpus, that budget can be spent entirely on rows they aren't allowed to have.

And the size of the visible slice is not the variable that matters most. Every ACL benchmark we know of assigns permissions at random. Real permissions are clustered — group:hr covers HR documents, and those are semantically adjacent by construction, so a group's slice is a region of the embedding space rather than a uniform sample.

200k documents from BEIR nq, real 1536-dim embeddings, exact brute-force oracle over eligible rows. Same corpus, same index, same queries, same number of visible rows — only the arrangement differs:

Clustering roughly doubles the recall loss. HNSW descends greedily from an entry point toward the query: scattered eligible rows are met along the walk, concentrated ones are not, so a query pointing away from the visible region spends its whole budget on rows the caller may not see.

Iterative scan does not fully close it — 0.883 clustered against 0.966 random, same setting. That residual is why this library decides between an exact and an approximate scan per query rather than tuning a parameter and hoping.

You may not be able to reproduce the failure, and that is also a result. With a GIN index on acl, all twelve default-planner rows chose an exact bitmap+sort instead of the vector index — recall 1.000, no failure, at 44–107 ms p50. Which plan Postgres picks is a cost estimate that moves with dimension, corpus size and statistics freshness, and && selectivity estimation on GIN arrays is unreliable enough to flip it without warning. The table above therefore forces the index. Whether you are exposed is decided per query, by a cost model — which is the actual argument for deciding it yourself.

Measure it on your own corpus rather than trusting ours — offline, read-only, no embedding provider called:

context-engine check-acl-exposure --database-url postgresql://...

It reports, for the ACL values that actually occur in your data, what a bare scoped query returns versus what this library returns, both scored against an exact oracle computed inside Postgres.

Written up in full: Your ACL Benchmark Is Measuring the Easy Case. Method, how clustered is constructed, the limitations — including a metric that behaved as a control rather than a result, and a retracted earlier table — are in benchmarks/acl_recall/, along with the harness and raw CSVs. Requires pgvector ≥ 0.8.

Checking your database

Managed Postgres varies in ways that matter here. To check a provider before committing to it:

CE_COMPAT_DATABASE_URL='postgresql+psycopg2://user:pass@host/db' \
    uv run pytest tests/test_managed_postgres_compat.py -v -s

Point it at a scratch database — it runs the real migrations. It verifies pgvector is new enough for iterative scan, that pg_trgm/unaccent are installable, and — the one that catches people — that their functions are actually reachable on the role's search_path. Supabase installs extensions into an extensions schema, so CREATE EXTENSION can succeed while similarity() and % stay unresolvable and the trigram leg breaks at query time rather than at migrate time.

Deep dives

This README covers the basics. Full documentation is at https://promptev.ai/documentation/context-engine/. The exhaustive design notes — redaction apply points, hash joinability, overlapping-rule merge semantics, ACL edge cases, graph internals, cost model, and known limitations — also live in the module and config docstrings (config.py, redaction.py, engine.py).