Context Engine Python SDK reference

Overview

The Context Engine is hosted by promptev. The Python SDK reaches it with a project API key: ingestion with upload_document and retrieval with search_knowledge_base. Every method has an async form.

Method	What it does	Returns
upload_document(file, *, mode=None)	Put one document into the key’s project.	DocumentResult
upload_documents(files, *, mode=None)	Put up to 20 documents into the key’s project.	List[DocumentResult]
search_knowledge_base(action, **fields)	Read the key’s project knowledge base.	dict
aupload_document(...)	Async form of upload_document.	DocumentResult
aupload_documents(...)	Async form of upload_documents.	List[DocumentResult]
asearch_knowledge_base(...)	Async form of search_knowledge_base.	dict

Install

Package promptev, version 1.3.0, Python 3.8 or later.

Shell

pip install promptev

PromptevClient

The key belongs to one project, so no call names a project. Create the key under the project’s API keys in the promptev console. Keep it on your server.

Python

from promptev import PromptevClient

client = PromptevClient(api_key="fk_...")

# or close it for you
with PromptevClient(api_key="fk_...", timeout=120) as client:
    ...

Argument	Type	Default	Meaning
api_key	str	required	The project API key (fk_...), from the project’s API keys in the console.
base_url	str	https://api.promptev.ai	Where the API is.
headers	dict \| None	None	Extra HTTP headers on every request.
timeout	float	30.0	Seconds to wait. An upload returns when its files are ready, so raise it for large files.
max_retries	int	2	Retries on a 502, 503 or 504.

upload_document and upload_documents

Put files into the key’s project knowledge base, in the same Uploads source the console writes to. The project’s agents can use them on their next turn.

Python

result = client.upload_document("handbook.pdf")
print(result.ok, result.status, result.document_id, result.chunks)

results = client.upload_documents([
    "policies/refunds.pdf",                 # a path
    ("release-notes.md", b"# 1.0\n..."),     # (filename, bytes)
    open("handbook.docx", "rb"),            # an open binary file
])
for r in results:
    print(r.filename, r.status, r.error or "")

client.upload_document("org-chart.pdf", mode="graph")   # build the Context Graph too

A file is a path, a (filename, bytes) pair, or an open binary file. The name matters: its extension decides how the file is read.
Limits: up to 20 files per call, 25 MB each. Larger sets belong in a connector source (Google Drive, SharePoint, Dropbox and more).
mode: leave it out for the workspace’s default, or pass "graph" to also extract entities and relationships into the Context Graph.
One result per file, in order. A file that fails is reported in its own row and never fails the batch.
Same bytes, same document: uploading an identical file again updates it instead of adding a copy.

DocumentResult

Field	Type	Meaning
filename	str	The file’s name.
status	str	What became of it. "failed" when it did not land.
document_id	str \| None	The document’s id, or None when it did not land.
chunks	int	How many passages it was split into.
units	int	The ingestion units counted for this file.
error	str \| None	Why this file failed, or None.
ok	bool (property)	True when the file landed.
raw	dict	The server’s row for this file, as sent.

search_knowledge_base(action, **fields)

Read the key’s project knowledge base. It is the same tool, with the same actions, as search_knowledge_base on the project’s MCP server. Name the action, then pass that action’s fields as keyword arguments. Fields set to None are not sent. Start with discover.

Python

import promptev

print(promptev.KNOWLEDGE_ACTIONS)

overview = client.search_knowledge_base("discover")

found = client.search_knowledge_base("search", query="how many days do customers have to ask for a refund", top_k=5)
for hit in found["hits"]:
    print(hit["document_name"], hit["score"])
    print(hit["chunk_text"])

doc = client.search_knowledge_base("get_doc", document_id=found["hits"][0]["document_id"])

total = client.search_knowledge_base("compute", query="total revenue by region for Q3")
people = client.search_knowledge_base("get_neighbors", entity="Acme GmbH")

promptev.KNOWLEDGE_ACTIONS lists the actions for reference. The server decides, so an action added there works without a new SDK.

Actions

The graph actions (get_neighbors, traverse, find_related, community_summary) need documents uploaded in graph mode. discover says which actions the project can run.

Action	What you get	Fields
discover	Each document and what is inside it (a sheet’s columns, a document’s sections), the extracted fields, and which actions the project can run. Call it first.	source_ids, document_ids, limit, cursor
search	Passages by meaning or keywords, as hits. For an ID, code or invoice number, search the bare identifier.	query, top_k, mode, source_ids, document_ids
get_doc	One whole document by id.	document_id
get_docs	Several whole documents at once.	document_ids, max_chars
get_chunks	One long document in order, up to 25 passages at a time.	document_id, start, end
list	Browse the documents without searching.	source_ids, document_ids, limit, cursor
query_meta	The structured fields a question names (dates, amounts, parties) and each document’s value. Compare the values yourself.	query, source_ids, limit
compute	A figure from the spreadsheets (total, average, count, ranking, margin) or the exact row for one value, over every row. Ask in plain words.	query, source_ids, document_ids
map_reduce	The same question asked of every document in scope, one answer per document.	query, source_ids, document_ids
get_neighbors	What is one step from a named thing in the Context Graph.	entity, limit
traverse	Everything within a few steps of a named thing. For how two things connect, traverse from one.	entity, depth, category, limit
find_related	Connections of one kind across the knowledge base, with no starting thing.	category, label, entity_type, limit
community_summary	The main themes and groups the knowledge base falls into.	query, limit

Fields

Field	Type	Meaning
query	str	What you are looking for, in words, or a bare identifier you are looking up.
top_k	int	How many passages a search returns. Default 10.
mode	"hybrid" \| "graph"	How a search ranks. Leave it out and the right one is chosen. "graph" also uses how things connect.
document_id	str	One document’s id, from discover, list or a search hit. Not a number written inside a document.
document_ids	list[str]	Narrow to these documents.
source_ids	list[str]	Narrow to these sources (an upload folder, a connector).
limit	int	How many rows: documents for discover and list (default 50), candidates for query_meta (20), links for the graph actions (50), groups for community_summary (3).
cursor	str	The next_cursor a previous page returned.
entity	str	Where a graph action starts: an entity id from an earlier graph answer (best), or a name as written in the documents.
depth	int	How many steps traverse walks, 1 to 5. Default 2.
category	str	The kind of connection to follow or find. discover lists the ones in your knowledge base.
label	str	The exact relationship wording, for example reports_to. Narrower than category.
entity_type	str	Keep only connections with this type of entity on one end.
start	int	First passage to read, from 0. A search hit’s chunk_idx reads around that passage.
end	int	Last passage to read, inclusive.
max_chars	int	How much text get_docs returns before it stops and hands back the ids it did not reach.

What comes back

A dict, the knowledge base’s answer as sent. A search returns hits, best first:

JSON

{
  "hits": [
    {
      "document_id": "…",
      "document_name": "refund-policy.pdf",
      "chunk_text": "Refunds are accepted within 30 days …",
      "score": 0.82,
      "source_id": "…",
      "chunk_idx": 4
    }
  ]
}

discover and list return documents (each with id, name, source_id and kind), count and has_more. When there is more, send next_cursor back as cursor:

Python

page = client.search_knowledge_base("list", limit=50)
while True:
    for d in page["documents"]:
        print(d["name"])
    if not page.get("has_more"):
        break
    page = client.search_knowledge_base("list", limit=50, cursor=page["next_cursor"])

Follow the answer. An answer may carry a next_action or next_page: the next call, already filled in. Cite document_name when you show a passage.

Async

aupload_document, aupload_documents and asearch_knowledge_base take the same arguments and return the same things.

Python

import asyncio
from promptev import PromptevClient

async def main():
    async with PromptevClient(api_key="fk_...") as client:
        result = await client.aupload_document("handbook.pdf")
        found = await client.asearch_knowledge_base("search", query="how many days do customers have to ask for a refund")
        print(result.status, len(found["hits"]))

asyncio.run(main())

Errors

Every error is a PromptevError with message, status_code, response_text and, when known, field. A 502, 503 or 504 is retried (max_retries) before ServerError is raised.

Error	Status	When
ValidationError	422 (or 400)	An unknown action or a field it does not take; no files, more than 20, a file over 25 MB, a type that is not accepted, or a mode that cannot be served. Some are caught before the request is sent. e.field names the field.
AuthenticationError	401 (or 403)	The key is missing, malformed, revoked or expired.
InsufficientCreditsError	402	The workspace has no credits left. A top-up fixes it, a retry does not.
NotFoundError	404	The key reaches no project.
RateLimitError	429	Too many requests. e.retry_after says how many seconds to wait.
ServerError	5xx	The server failed, after the retries.
NetworkError	none	No answer: connection, DNS or timeout.

Python

from promptev import (
    PromptevClient, ValidationError, AuthenticationError,
    InsufficientCreditsError, NotFoundError, RateLimitError,
)

try:
    found = client.search_knowledge_base("search", query="how many days do customers have to ask for a refund")
except ValidationError as e:
    print("Fix the request:", e.message, e.field)
except InsufficientCreditsError:
    print("Out of credits: top up the workspace")
except RateLimitError as e:
    print("Slow down, retry in", e.retry_after, "seconds")

Credits and costs are on the guide.

Context Engine: Python SDK reference

Overview

Install

PromptevClient

upload_document and upload_documents

DocumentResult

search_knowledge_base(action, **fields)

Actions

Fields

What comes back

Async

Errors