Context Engine — Full Guide APACHE 2.0

Overview

The Context Engine is the ingestion + retrieval + governance layer that powers Promptev, available as standalone Python and TypeScript packages. It turns your documents into searchable knowledge and serves it behind one ContextEngine object. The samples below are Python; the TypeScript client exposes the same API over the same Postgres schema (TS docs & source →).

1 · Install

Requires Postgres with the vector, pg_trgm, and unaccent extensions (auto-created by context-engine migrate if the DB role can CREATE EXTENSION), plus Python 3.12+ or Node 22+ for the client. Both clients share one schema — a corpus ingested by either is searchable by the other.

pip install promptev-context-engine

uv add promptev-context-engine

npm install @promptev/context-engine pg

The core engine is framework-agnostic. Add only the extras / peers you need:

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

npm install hono            # createHonoRouter
npm install neo4j-driver graphology   # graph mode

Teach your coding agent the rules

Most integration code is now written with an agent's help, and the costliest mistake in this library is one an agent makes readily: principals=None means trusted caller, ACL filtering disabled — so reading it as “nobody is logged in” returns every document you have.

# WRONG -- returns the ENTIRE corpus to an anonymous caller
principals = user.groups if user else None

# RIGHT -- unauthenticated is an empty list, not None
principals = user.groups if user else []

// WRONG -- returns the ENTIRE corpus to an anonymous caller
const principals = user ? user.groups : null;

// RIGHT -- unauthenticated is an empty list, not null (or TRUSTED to skip ACL)
const principals = user ? user.groups : [];

The package ships a skill stating that and the other invariants worth not guessing — principals are injection-only on remote surfaces, auth/principals are required on every router factory, acl=None unrestricts on update, pgvector 0.8 is a hard floor, and code execution is off by default.

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

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

2 · Configure

ContextEngineConfig is the single settings object every module reads from. Build it directly, or from CE_-prefixed environment variables (nested fields use __).

from context_engine import ContextEngine, 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-..."),
)
engine = ContextEngine(config)

import { ContextEngine, ContextEngineConfig } from "@promptev/context-engine";

const config = new ContextEngineConfig({
  databaseUrl: "postgresql://user:pass@localhost:5432/mydb",
  embedding: { provider: "openai", model: "text-embedding-3-small", apiKey: "sk-..." },
  llm: { provider: "openai", model: "gpt-4o-mini", apiKey: "sk-..." },
});
const engine = new ContextEngine(config);

Env-var equivalent:

CE_DATABASE_URL=postgresql://user:pass@localhost:5432/mydb
CE_EMBEDDING__PROVIDER=openai
CE_EMBEDDING__MODEL=text-embedding-3-small
CE_EMBEDDING__API_KEY=sk-...

3 · Provision the database

Create the tables once with the CLI. The migration history lives inside the package, so there's nothing to copy into your repo.

context-engine migrate --database-url postgresql://user:pass@localhost:5432/mydb --dim 1536

# add --graph to also create entity / relationship / community tables
context-engine migrate --database-url postgresql://... --dim 1536 --graph

npx context-engine migrate --database-url postgresql://user:pass@localhost:5432/mydb --dim 1536

# add --graph to also create entity / relationship / community tables
npx context-engine migrate --database-url postgresql://... --dim 1536 --graph

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

4 · Ingest

Feed the engine raw text or files. Files auto-detect type: PDF, DOCX, PPTX, XLSX, CSV, HTML, EML, images, and plain text. Scanned pages and images are transcribed to structured Markdown when a vision LLM is configured (or OCR with the [ocr] extra).

# 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"],              # None = visible to any trusted caller
)

# A file
with open("policy.pdf", "rb") as f:
    report = await engine.ingest(file=f, name="policy.pdf", source_id="hr-handbook")

print(report.totals)               # totals -> files=1, failed=0, units=3
print(report.documents[0].status)  # completed | failed | batch_pending

import { readFileSync } from "node:fs";

// Plain text
let report = await engine.ingest({
  text: "Annual leave accrues at two days per month for every band-three employee.",
  name: "leave-policy",
  sourceId: "hr-handbook",        // a namespace to scope search + stats
  acl: ["group:hr"],              // TRUSTED = visible to any trusted caller
});

// A file
report = await engine.ingest({ file: readFileSync("policy.pdf"), name: "policy.pdf", sourceId: "hr-handbook" });

console.log(report.totals);                // totals -> { files: 1, failed: 0, units: 3 }
console.log(report.documents[0].status);   // completed | failed | batch_pending

from context_engine import ContextEngineConfig, ExtractionConfig

config = ContextEngineConfig(
    # ...
    extraction=ExtractionConfig(
        max_render_px=2000,               # long-edge ceiling for rendered pages & sent images
        vision_max_output_tokens=16_000,  # per-batch transcription ceiling
    ),
)

const config = new ContextEngineConfig({
  // ...
  extraction: {
    maxRenderPx: 2000,            // long-edge ceiling for rendered pages & sent images
    visionMaxOutputTokens: 16000, // per-batch transcription ceiling
  },
});

ingest() never raises on a single bad document — check report.documents[i].status and .error.

5 · Search

Retrieve with one call. Full-text, trigram, and vector legs are fused with Reciprocal Rank Fusion; mode="graph" adds a graph leg.

result = await engine.search(
    "how much annual leave do I get",
    source_ids=["hr-handbook"],
    document_ids=None,         # optional: pin to specific documents WITHIN the sources
    principals=["group:hr"],   # None = trusted/internal (skips ACL); [] = anonymous
    top_k=10,
    compress_to_tokens=2000,   # optional: trim weak hits to fit a prompt budget
)

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

const result = await engine.search("how much annual leave do I get", {
  sourceIds: ["hr-handbook"],
  documentIds: null,          // optional: pin to specific documents WITHIN the sources
  principals: ["group:hr"],   // TRUSTED = skip ACL; [] = anonymous
  topK: 10,
  compressToTokens: 2000,     // optional: trim weak hits to fit a prompt budget
});

for (const hit of result.hits) {
  console.log(hit.documentName, hit.score, hit.chunkText);
}

6 · Documents

Every read below takes the same principals contract as search, and every one of them is ACL-filtered in SQL. A document you may not see raises KeyError — the identical error you get for a document that does not exist, so a caller can never probe for the existence of restricted data.

Read one document

doc = await engine.get_document(document_id, principals=["group:hr"])
# {"id", "name", "text", "acl", "status", "chunks", "structured_data", ...}

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

const doc = await engine.getDocument(documentId, { principals: ["group:hr"] });
// { id, name, text, acl, status, chunks, structuredData, ... }

const text = await engine.getDocumentText(documentId, { principals: ["group:hr"] });

Both apply any apply_at="output" redaction policy before returning. For a corpus ingested without an ingest-phase policy, this is the only thing masking those fields.

List with pagination

page = await engine.list_documents(source_id="hr-handbook", principals=["group:hr"], limit=50)

for summary in page["documents"]:
    print(summary["id"], summary["name"], summary["status"])

if page["has_more"]:
    nxt = await engine.list_documents(source_id="hr-handbook",
                                      principals=["group:hr"],
                                      cursor=page["next_cursor"], limit=50)

const page = await engine.listDocuments({
  sourceId: "hr-handbook", principals: ["group:hr"], limit: 50,
});

for (const summary of page.documents) {
  console.log(summary.id, summary.name, summary.status);
}

if (page.hasMore) {
  const next = await engine.listDocuments({
    sourceId: "hr-handbook", principals: ["group:hr"],
    cursor: page.nextCursor, limit: 50,
  });
}

Keyset pagination, not OFFSET — pages stay stable while documents are being ingested. limit is clamped to 200.

Update attributes without re-ingesting

changed = await engine.update_document(
    document_id,
    acl=["group:legal"],           # replaced; acl=None UNRESTRICTS
    meta_data={"reviewed": True},  # merged, not replaced
    principals=["group:hr"],
)
# -> ["acl", "meta_data"]  (only what actually changed)

const changed = await engine.updateDocument(documentId, {
  acl: ["group:legal"],          // replaced; acl: null UNRESTRICTS
  metaData: { reviewed: true },  // merged, not replaced
  principals: ["group:hr"],
});
// -> ["acl", "meta_data"]  (only what actually changed)

Delete, and corpus counters

await engine.delete_document(document_id, principals=["group:hr"])  # chunks first, then the row

await engine.stats()                        # whole corpus
await engine.stats("hr-handbook")           # one source
# {"documents": 412, "chunks": 9310, "by_status": {"completed": 410}, "embedding": {...}}

await engine.deleteDocument(documentId, { principals: ["group:hr"] }); // chunks first, then the row

await engine.stats();              // whole corpus
await engine.stats("hr-handbook"); // one source
// { documents: 412, chunks: 9310, ... }

Re-ingesting the same document

Ingest is idempotent. A document is identified by external_id first, then name, then a content hash — all scoped to source_id. Re-sending unchanged content returns status="skipped", costs zero units and does not re-embed.

But a skip still applies your declared attributes. acl, name, description and meta_data are not part of the content hash, so re-ingesting identical text with a tightened ACL updates the document and its chunks. Without that, a tightening you believed had applied would silently do nothing.

7 · Structured data & compute

Optional, and both need ContextEngineConfig.llm. Extraction pulls typed fields out of a document at ingest time; query_structured answers questions against them without re-reading the text.

Extract at ingest

report = await engine.ingest(
    file=open("invoice.pdf", "rb"),
    name="INV-1042",
    source_id="billing",
    extract_structured=True,
    field_hints=[{"name": "tax_amount", "type": "number"}],   # optional nudges
)

import { readFileSync } from "node:fs";

const report = await engine.ingest({
  file: readFileSync("invoice.pdf"),
  name: "INV-1042",
  sourceId: "billing",
  extractStructured: true,
  fieldHints: [{ name: "tax_amount", type: "number" }],   // optional nudges
});

One LLM call returns a document_type and a field dict, stored on the document row. Field names are normalised into a registry with embeddings, which is what lets a question say "VAT" and match a field stored as tax_amount.

Query it

answer = await engine.query_structured(
    "what's the VAT on this invoice?",
    source_ids=["billing"],
    principals=["group:finance"],
    doc_type="invoice",
)
# {"resolved_fields": {"vat": "tax_amount"}, "documents": [...], "count": 1}

const answer = await engine.queryStructured("what's the VAT on this invoice?", {
  sourceIds: ["billing"],
  principals: ["group:finance"],
  docType: "invoice",
});
// { resolvedFields: { vat: "tax_amount" }, documents: [...], count: 1 }

If nothing resolves above the similarity floor you get zero documents, not the whole corpus. An off-topic question should return nothing rather than a confident invoice.

compute() — LLM-authored Python over spreadsheets

config.enable_code_execution = True   # deliberate opt-in

out = await engine.compute(
    "total revenue by region for Q3",
    source_ids=["finance"],
    principals=["group:finance"],
)
# {"success": True, "result": {...}, "code": "...", "documents_used": [...]}

config.enableCodeExecution = true;   // deliberate opt-in

const out = await engine.compute("total revenue by region for Q3", {
  sourceIds: ["finance"],
  principals: ["group:finance"],
});
// { success: true, result: {...}, code: "...", documentsUsed: [...] }

Only CSV/TSV/XLSX documents in scope are loaded, capped at MAX_COMPUTE_DOCUMENTS. Redaction is applied to the text before it becomes a dataframe, so generated code computes over masked values.

8 · Redaction

Mask PII (or anything you can describe with a regex or detector) as content is ingested, retrieved, or returned. Redaction composes with ACL — ACL decides who sees a document; redaction decides what's masked inside it. It's a pure transform: no model, no network.

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"),  # regex rule
    ]),
)

import { ContextEngineConfig, RedactionPolicy, RedactionRule } from "@promptev/context-engine";

const config = new ContextEngineConfig({
  databaseUrl: "postgresql://...",
  embedding: { provider: "openai", model: "text-embedding-3-small", apiKey: "sk-..." },
  redaction: new RedactionPolicy({ rules: [
    new RedactionRule({ name: "emails", detector: "email" }),            // -> [EMAILS]
    new RedactionRule({ name: "order_id", pattern: "\\bORD-\\d{6}\\b" }), // regex rule
  ]}),
});

9 · Tools

Let an 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

# Register an 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 YOUR_TOKEN"},
        "llmQueryParameters": {
            "properties": {"city": {"type": "string"}},
            "required": ["city"],
        },
    },
))  # -> registered as call_name "http_get_weather"

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

import { ToolConfig } from "@promptev/context-engine";

// Register an HTTP tool — the LLM fills `city` at call time
await engine.registerTool(new 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 YOUR_TOKEN" },
    llmQueryParameters: {
      properties: { city: { type: "string" } },
      required: ["city"],
    },
  },
})); // -> registered as call name "http_get_weather"

// Execute it (args the LLM produced)
const out = await engine.executeTool("http_get_weather", { city: "Lahore" });
// -> { result: {...}, usage: { kind: "tool", units: 1 } }

# A plain Python function becomes a tool — schema inferred from the signature
def add(a: int, b: int) -> int:
    """Add two integers."""
    return a + b

engine.register_function_tool(add)          # -> call_name "fn_add"
await engine.execute_tool("fn_add", {"a": 2, "b": 3})   # -> {"result": 5, ...}

// A plain function becomes a tool
function add(a: number, b: number): number {
  return a + b;
}

engine.registerFunctionTool(add);                    // -> call name "fn_add"
await engine.executeTool("fn_add", { a: 2, b: 3 });  // -> { result: 5, ... }

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

from context_engine import resolve_approval

# A tool registered with requires_approval=True pauses instead of running
out = await engine.execute_tool("http_get_weather", {"city": "Lahore"})
# -> {"approval_required": {"approval_id": "...", ...}}   (NOT executed yet)

await resolve_approval(engine, approval_id, "approved", approver="boss")
# the next execute_tool(...) runs for real

import { resolveApproval } from "@promptev/context-engine";

// A tool registered with requiresApproval: true pauses instead of running
const out = await engine.executeTool("http_get_weather", { city: "Lahore" });
// -> { approvalRequired: { approvalId: "...", ... } }   (NOT executed yet)

await resolveApproval(engine, approvalId, "approved", "boss");
// the next executeTool(...) runs for real

db tools introspect a schema and fail closed to read-only; mcp tools expand into one canonical tool per discovered remote tool, each with its own approval override.

10 · Serve over HTTP

The document + search surface ships as a ready adapter for FastAPI, Flask, or Django — identical routes, identical behavior. auth and principals are yours to inject; the package implements no auth.

from context_engine import create_router

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

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")

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))),
]

import { createHonoRouter } from "@promptev/context-engine/hono";

const router = createHonoRouter(engine, { auth: myAuth, principals: myPrincipals });

// Identical routes exist for Express (createExpressRouter, "@promptev/context-engine/express")
// and Fastify (createFastifyPlugin, "@promptev/context-engine/fastify").
// auth and principals are REQUIRED on every router factory.

Routes: POST/GET/DELETE/PATCH /documents, POST /search, GET /stats. Not on any framework? Call engine.ingest() / engine.search() directly from your own views.

11 · MCP Server

Expose the engine to any MCP client (Claude Code, Cursor, …) over streamable HTTP. Requires the [mcp] extra.

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

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

from context_engine import create_mcp_app

app.mount("/mcp", create_mcp_app(engine, principals=my_principals))

import { createMcpApp } from "@promptev/context-engine";

const app = createMcpApp(engine, { auth: myAuth, principals: myPrincipals });

Registers search_knowledge_base + get_document always, query_structured when an LLM is set, and compute only when code execution is explicitly enabled.

12 · Usage, errors & lifecycle

The engine reports its own work through two callbacks you pass at construction — one for metering, one for errors. It never calls your metrics backend or error tracker itself, so what a unit means and where an exception goes stay entirely yours.

from context_engine import ContextEngine, UsageEvent

def record_usage(event: UsageEvent) -> None:
    # kind: "ingest" | "search" | "tool"
    # detail: {"pages": 42} / {"mode": "graph"}
    # provider_tokens: {"embedding_tokens": n, "llm_input": n, "llm_output": n}
    metrics.increment(f"context_engine.{event.kind}", event.units, tags=event.detail)
    metrics.record(event.provider_tokens)

def alert(exc: Exception, ctx: dict) -> None:
    sentry.capture_exception(exc, extra=ctx)   # ctx says which document/stage

engine = ContextEngine(config, on_usage=record_usage, on_error=alert)

import { ContextEngine, type UsageEvent } from "@promptev/context-engine";

function recordUsage(event: UsageEvent): void {
  // kind: "ingest" | "search" | "tool"
  // detail: { pages: 42 } / { mode: "graph" }
  // providerTokens: { embeddingTokens, llmInput, llmOutput }
  metrics.increment(`context_engine.${event.kind}`, event.units, event.detail);
  metrics.record(event.providerTokens);
}

function alert(exc: unknown, ctx: Record<string, unknown>): void {
  sentry.captureException(exc, { extra: ctx }); // ctx says which document/stage
}

const engine = new ContextEngine(config, { onUsage: recordUsage, onError: alert });

Units are deliberately unit-less counts — a measure of work done, not a price: 1 per PDF page, 1 per PPTX slide, 1 per DOCX page, 1 flat per image, otherwise 1 per MB rounded up (minimum 1), plus 1 per picture that got an LLM-generated description. Search and tool calls emit their own events. Attach whatever meaning you need — quota, chargeback, capacity planning, or nothing at all.

Closing the engine

Construction is lazy — the connection pool, embedding client and Neo4j driver are built on first use. aclose() tears down only what was actually built.

# one engine for the process lifetime -- the usual case
engine = ContextEngine(config)

# anything that builds engines repeatedly must close them, or it
# accumulates connection pools: a worker per job, a test per case,
# an app reloading its configuration.
async with ContextEngine(config) as engine:
    await engine.ingest(file=f, name="handbook.pdf", source_id="hr")

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

// one engine for the process lifetime -- the usual case
const engine = new ContextEngine(config);

// anything that builds engines repeatedly must close them, or it
// accumulates connection pools: a worker per job, a test per case,
// an app reloading its configuration.
await using scoped = new ContextEngine(config);   // Symbol.asyncDispose -> aclose()
await scoped.ingest({ file: f, name: "handbook.pdf", sourceId: "hr" });

// equivalently
const e = new ContextEngine(config);
try {
  // ...
} finally {
  await e.aclose();   // safe to call more than once
}

13 · Verify your setup

Two checks worth running before you trust a deployment — one against your database, one against your retrieval quality.

Managed-Postgres preflight

Neon, Supabase, RDS, Cloud SQL and Azure all ship pgvector, so "does it work" is rarely the question. These are the four that actually bite:

# point at a SCRATCH database -- it runs the real migrations
CE_COMPAT_DATABASE_URL='postgresql+psycopg2://user:pass@host/db' \
    uv run pytest tests/test_managed_postgres_compat.py -v -s

• pgvector ≥ 0.8 — below it there is no iterative scan, and ACL-filtered vector search loses recall with no fix available from this package.
• Can this role CREATE EXTENSION vector, pg_trgm, unaccent? Several providers gate that behind an allow-list.
• Are those functions on the search_path? The Supabase trap: extensions land in an extensions schema, CREATE EXTENSION succeeds, and then similarity() fails at query time. Fix with ALTER ROLE <role> SET search_path = public, extensions;
• Migrations + an ACL-filtered HNSW query end to end on the provider's own engine.

The ACL-recall benchmark

Any system that applies an access-control predicate as a WHERE clause over an approximate index is post-filtering: the index walk is ordered by distance, rejected rows are discarded as they are met, and the walk stops when its budget is spent. If the caller's visible slice is small, the budget goes entirely on rows they may not see — and the query returns nothing, successfully, with nothing logged.

uv sync
uv run python benchmarks/acl_recall/run.py --dataset scifact --out results.csv

200,000 documents from BEIR nq, real 1536-dim text-embedding-3-small embeddings, exact brute-force oracle over the eligible rows. Same corpus, same index, same queries, same number of visible rows — only the arrangement differs. Every published ACL benchmark assigns permissions at random; real permissions are topically clustered, which roughly doubles the loss. Iterative scan does not fully close it (0.883 vs 0.966), which is why this library routes to an exact scan below an eligible-row threshold rather than only tuning parameters. You may not reproduce the failure at all: with a GIN index on the ACL column the planner often picks an exact bitmap scan instead of the vector index — the table above forces the index. Full method and limitations in benchmarks/acl_recall/README.md.

14 · LangChain & LlamaIndex

The engine is a substrate, not an orchestrator — it slots into LangChain (or LlamaIndex / Haystack) as a retriever. Wrap engine.search() in a BaseRetriever and your existing chains get hybrid + graph retrieval with ACL and redaction, in a few lines.

from langchain_core.retrievers import BaseRetriever
from langchain_core.documents import Document

class PromptevRetriever(BaseRetriever):
    engine: ContextEngine

    async def _aget_relevant_documents(self, query, *, run_manager):
        result = await self.engine.search(query, principals=["group:hr"], top_k=8)
        return [
            Document(page_content=hit.chunk_text, metadata=hit.meta)
            for hit in result.hits
        ]

# drop it into any chain
retriever = PromptevRetriever(engine=engine)
docs = await retriever.ainvoke("how much annual leave do I get")

// No shipped TS adapter — call the engine directly, or wrap engine.search()
// in a LangChain.js BaseRetriever yourself.
const result = await engine.search(query, { principals, topK: 8 });
const docs = result.hits.map((h) => ({ pageContent: h.chunkText, metadata: h.meta }));

15 · Production