Promptev Python SDK

The official Python client for Promptev — chat with your agents and put documents into their corpus, from your own code.

An agent bundles instructions, a model, tools and a knowledge base. You build and publish it in Promptev, mint an API key for its project, and run it from here.

Version 1.0 is a rewrite. It targets Promptev's /api/v1 and does not talk to the old product at all. If you are on 0.2.x, read Migrating from 0.2.x first — 0.2.x still works, and still points where it always did.

Installation

pip install promptev

Requires Python 3.8+. Only dependency: httpx.

Quick start

from promptev import PromptevClient

client = PromptevClient(api_key="fk_live_your_key_here")

for event in client.chat("support-bot", "What is our refund policy?"):
    if event.type == "token":
        print(event.content, end="", flush=True)

"support-bot" is your agent's slug. fk_… is an API key, minted per project under Settings → API keys.

The key is the identity

The key goes in the Authorization header — never in the URL, where a secret leaks into access logs, proxies and referrer headers.

A key also carries its own retrieval grant (retrieves_as), so what a caller may retrieve is part of the credential, not a parameter, and it is bound to exactly one project. An agent in a sibling project answers the same 404 an unknown slug does.

Chatting with an agent

chat() is a generator of typed events. One call: the endpoint streams, and it takes a conversation id directly, so there is no session to open and no session token to carry.

from promptev import PromptevClient, TokenEvent, ConversationEvent

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

conversation_id = None
answer = ""

for event in client.chat("support-bot", "What is our refund policy?"):
    if isinstance(event, ConversationEvent):
        conversation_id = event.id          # keep this for the next turn
    elif isinstance(event, TokenEvent):
        answer += event.content

print(answer)

Continue the conversation

Pass the id you read off the first conversation event. That is what gives the agent its memory of the thread — and, with "Ask once" approvals, what stops a tool asking permission on every turn.

for event in client.chat(
    "support-bot",
    "And what about exchanges?",
    conversation_id=conversation_id,
):
    if event.type == "token":
        print(event.content, end="", flush=True)

Answer on behalf of your end user

If your backend fronts your own UI, tell Promptev who the turn is for. The subject is recorded as the conversation's actor (audit trail); the groups are intersected with the key's own grant, so an assertion can only narrow what may be retrieved, never widen it.

from promptev import OnBehalfOf

events = client.chat(
    "support-bot",
    "Show me my team's Q4 numbers",
    on_behalf_of=OnBehalfOf(subject="ava@acme.com", groups=["finance"]),
)

Change the model or a tool's values for one call

Both arguments are optional. Leave them out and the agent runs exactly as it was published. Nothing a call sends is stored: the next call runs the published agent again.

model runs this one call on another model:

events = client.chat(
    "support-bot",
    "How many open orders does ACME have?",
    model={"provider": "openai", "model": "gpt-5-mini"},
)

provider and model are the ids shown in your workspace's model list. The workspace's own provider key is used; a call never carries a key. The model runs on its default settings, because the agent's saved settings (temperature and the like) were tuned for its own model.

The call raises ValidationError (400, with field set) before the model is called when the workspace has no such provider, the provider has no key, the model is not in its list or has been retired, or the model cannot call tools and the agent has tools.

tool_inputs overrides what a tool is set up with, keyed by the tool's id (shown on the tool's page in promptev, with a copy button). Each value goes under the place it belongs to:

events = client.chat(
    "support-bot",
    "How many open orders does ACME have?",
    tool_inputs={
        "0b6f2c1e-…": {                                   # an HTTP tool
            "headers": {"Authorization": f"Bearer {end_user_token}"},
            "query": {"region": "us"},
            "body": {"account_id": "acme"},
            "path": {"customer": "acme"},                  # a {{customer}} in the tool's URL
        },
        "7d41a9c3-…": {                                   # an MCP server
            "headers": {"Authorization": f"Bearer {end_user_token}"},
            "arguments": {"project_id": "p_123"},
        },
    },
)

Tool	Place	What it is
HTTP	headers	A header the tool was saved with. Names match in any letter case.
HTTP	query	A query parameter: one the tool was saved with, or one the model fills. A GET or DELETE tool has no body, so all of its parameters are query.
HTTP	body	A body value (POST, PUT, PATCH): a key of the saved body, or a parameter the model fills.
HTTP	path	A {{name}} in the tool's URL.
MCP server	headers	A header the server was saved with (Authorization when it was saved with a token).
MCP server	arguments	An argument of the server's functions. The value reaches every function that takes an argument of that name.

The rules:

• Override only. A name the tool does not have in that place raises ValidationError (400), and the message lists what the tool does have there. Nothing extra is ever passed through.
• A value the model would have filled is fixed for the call. The model neither sees that parameter nor can change it, so the answer's wording follows the question while the tool gets your value.
• What is visible. A value the model would have filled shows wherever tool arguments show: the tool_call event and approval requests. A header, and any value the tool was saved with, never appears in the stream. Put secrets in headers.
• Limits. 20 tools per call, 20 names per place, 16 KB per tool.
• Overriding a value the tool was saved with (a header, a fixed query parameter, a key of the saved body) needs a server that supports it. One that does not answers 400 saying so; it never runs the call without your value.

Events

Every frame is a typed dataclass. Narrow on event.type (or isinstance), read the fields you need, and ignore the rest — event.raw always carries the whole frame.

Event	Class	What it means
conversation	ConversationEvent	Always first. id — persist it to continue the thread.
step	StepEvent	Named work in progress: name, status, detail.
sources	SourcesEvent	Grounding citations: items[].name, items[].document_id.
thinking	ThinkingEvent	A delta of the model's reasoning. Not part of the answer.
tool_call	ToolCallEvent	The agent invoked a tool: id, name, args.
tool_result	ToolResultEvent	That tool answered: id, name, ok, summary.
approval_required	ApprovalRequiredEvent	The turn is waiting on a human. See below.
artifact	ArtifactEvent	A file the agent produced: id, name, mime, size.
token	TokenEvent	Incremental answer text — concatenate content.
suggestions	SuggestionsEvent	Follow-up questions the person might ask next.
usage	UsageEvent	What the turn cost: units, tokens, tool_calls.
done	DoneEvent	Finished cleanly. Nothing follows.
error	ErrorEvent	Failed mid-stream: message. Nothing follows.

An event type this SDK does not name arrives as a plain ChatEvent with everything in raw, so a newer backend can never break your loop.

error is a frame, not an exception: the HTTP response was a 200 and the turn had already started. Check for it.

approval_required — handle this one

An agent can be configured so that a tool asks permission before it runs. When it does, the turn stops and this event arrives. The stream stays open. A person then approves or denies in the Promptev app, by e-mail, or in Slack; the turn continues on the same connection — or the approval expires and the turn reports it.

If you ignore this event, your integration looks hung to its user for as long as the approval window. Surface it.

from promptev import ApprovalRequiredEvent, ErrorEvent, TokenEvent

for event in client.chat("ops-bot", "Email the summary to the CEO"):
    if isinstance(event, ApprovalRequiredEvent):
        print(
            f"\n⏸  Waiting for approval: {event.tool_name} "
            f"({event.reason}) — decide it in Promptev, by e-mail or in Slack."
        )
        print(f"   arguments: {event.args}")
        if event.expires_at:
            print(f"   expires: {event.expires_at}")
    elif isinstance(event, TokenEvent):
        print(event.content, end="", flush=True)
    elif isinstance(event, ErrorEvent):
        print(f"\n{event.message}")

Two things worth knowing:

• With the agent's "Ask once" mode a tool asks on the first call in a conversation and not again for the rest of it — so passing conversation_id materially reduces how often you see this.
• An approval routed to an approver by e-mail sends no card at all: the person chatting cannot decide it. The server streams a StepEvent named "waiting for approval — <tool>" instead, which closes when the decision lands. Render that too if you show progress.

Uploading documents

upload_document puts a file into your key's project corpus — the same "Uploads" source the console writes to, so the project's agents can retrieve it on the next turn. The key names the project, so nothing about the target is in the request.

result = client.upload_document("handbook.pdf")

print(result.status)       # "ok"
print(result.document_id)
print(result.chunks, "chunks")

Files can be paths, (filename, bytes) pairs, or open binary file objects. Up to 20 files per call, 25 MB each:

results = client.upload_documents([
    "policies/refunds.pdf",
    ("release-notes.md", b"# 1.0\n..."),
    open("handbook.docx", "rb"),
])

for r in results:
    print(r.filename, r.status, r.error or "")

A file that fails to ingest is reported in its own row — it never fails the batch. Pass mode="graph" to extract entities and relationships as well (the workspace's engine needs a graph database configured for it).

Running an orchestration

An orchestration is a team of agents working one job. chat streams one agent's answer inline; an orchestration runs in a worker for as long as it takes, so run_orchestration hands back a run id straight away and get_run says what became of it.

started = client.run_orchestration("month-end-close", {"period": "2026-03"})
print(started.run_id)

run = client.get_run(started.run_id)
while not run.done:
    time.sleep(2)
    run = client.get_run(started.run_id)

print(run.status)   # "delivered"
print(run.output)

The orchestration must be published, switched on, and in your key's project — one that is not published, or lives in a sibling project, answers the same NotFoundError an invented name does. One that exists and is published but is switched off says so plainly, so you can tell "turn it on" from "wrong name".

The keys in the second argument are the orchestration's own declared inputs — the ones its Inputs step lists. A required one you leave out is refused before anything runs, with the field named:

try:
    client.run_orchestration("month-end-close", {})
except ValidationError as e:
    print(e.field)   # "period"

Sending files

An input declared File(s) takes a file, keyed by that input:

started = client.run_orchestration(
    "contract-review",
    {"matter": "Acme renewal"},
    files={"contract": "lease.pdf"},                       # or ("lease.pdf", b"...")
)

Up to five files per call, 25 MB each. Each part is named after the input it fills, which is what lets one call fill two different file inputs. Files take the same shapes upload_document does.

Retrying safely

Starting a run twice runs the team twice. Pass an idempotency_key and a retry returns the run you already started:

client.run_orchestration(
    "month-end-close", {"period": "2026-03"}, idempotency_key="close-2026-03"
)

Only your caller knows whether two calls are one event, which is why this is yours to send rather than something the client invents.

Async

Every method has an async twin: chat / achat, upload_document / aupload_document, upload_documents / aupload_documents, run_orchestration / arun_orchestration, get_run / aget_run.

import asyncio
from promptev import PromptevClient

async def main():
    async with PromptevClient(api_key="fk_live_...") as client:
        async for event in client.achat("support-bot", "How many tickets today?"):
            if event.type == "token":
                print(event.content, end="", flush=True)

        result = await client.aupload_document("handbook.pdf")
        print(result.document_id)

asyncio.run(main())

FastAPI

Relay the agent's stream straight to your own client:

import json
from fastapi import FastAPI
from fastapi.responses import StreamingResponse
from promptev import PromptevClient

app = FastAPI()
client = PromptevClient(api_key="fk_live_...")

@app.post("/ask")
async def ask(slug: str, message: str):
    async def relay():
        async for event in client.achat(slug, message):
            yield f"data: {json.dumps(event.raw)}\n\n"

    return StreamingResponse(relay(), media_type="text/event-stream")

Error handling

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

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

try:
    for event in client.chat("support-bot", "Hello"):
        ...
except ValidationError as e:
    print(f"Invalid request ({e.field}): {e}")
except AuthenticationError as e:
    print(f"Key rejected: {e}")
except InsufficientCreditsError as e:
    print(f"Out of credits — top up: {e}")
except NotFoundError as e:
    print(f"No such agent (or not published, or not in this key's project): {e}")
except RateLimitError as e:
    print(f"Rate limited; retry in {e.retry_after}s")
except ServerError as e:
    print(f"Server error after retries: {e}")
except NetworkError as e:
    print(f"Could not reach Promptev: {e}")

Exception	HTTP	When
ValidationError	400, 422	Bad input. e.field names the offending key.
AuthenticationError	401, 403	Key missing, malformed, revoked or expired.
InsufficientCreditsError	402	The workspace is out of credits.
NotFoundError	404	Unknown agent, unpublished agent, an agent outside this key's project, or an unknown conversation id — deliberately indistinguishable.
RateLimitError	429	A rate limit refused it. e.retry_after says when.
ServerError	5xx	Server error, after retries.
NetworkError	—	Connection failed, timeout, DNS.
PromptevError	any	Base class for all of the above.

Every exception carries e.message, e.status_code, e.response_text, e.field and e.retry_after.

What a call costs

Promptev is pay as you go and this API is metered. The rates below are the managed ones; a workspace running on its own database and embedding provider (BYO) is charged a flat 1 per operation instead, because those bills are already yours.

what happens	credits
any request to this API	1 (api_call)
a chat turn	1 (chat_turn)
each tool the agent calls	1
retrieval	1 per knowledge base searched
retrieval in graph mode	5 per knowledge base searched
a document you upload	1 per page, slide or MB (minimum 1)

So an ordinary chat costs 2 — one for the request, one for the turn — and more if the agent calls tools or searches knowledge. The same conversation held in the Promptev app costs one, because there is no request to the programmatic door.

Two things worth knowing before they surprise you:

• A refused request is still billed. An unknown agent slug, a rate-limited call and a rejected upload each consumed the same capacity a successful one would, so each costs its api_call credit. Only the outcome differs.
• A request this client rejects locally is never billed, because it never leaves your process. That is why the empty-message and upload-limit checks exist here rather than being left to the server.

Model tokens are not billed by Promptev at all. Model calls go to your provider on your own keys, and that bill is yours.

Configuration

client = PromptevClient(
    api_key="fk_live_...",                   # Required — your project API key
    base_url="https://api.promptev.ai",  # Default — override for self-hosted or during the cutover
    timeout=30.0,                            # Default — seconds
    max_retries=2,                           # Default — retries for 502/503/504
    headers={"X-Request-Id": "…"},           # Optional — extra HTTP headers
)

Parameter	Default	Description
api_key	required	Your fk_… key (Settings → API keys)
base_url	https://api.promptev.ai	API base URL
timeout	30.0	Request timeout in seconds — bounds the wait for the stream's first byte, not the length of an answer
max_retries	2	Automatic retries for transient server errors (502, 503, 504)
headers	None	Additional HTTP headers

A chat request is never retried: it bills a turn the moment the server accepts it, so a silent second attempt could run the agent twice.

API reference

Method	Description	Returns
chat(slug, message, *, conversation_id?, on_behalf_of?, model?, tool_inputs?)	Run an agent, stream its turn (sync)	Iterator[ChatEvent]
achat(...)	Same, async	AsyncIterator[ChatEvent]
upload_document(file, *, mode?)	Put one document into the key's project (sync)	DocumentResult
aupload_document(...)	Same, async	DocumentResult
upload_documents(files, *, mode?)	Put up to 20 documents (sync)	List[DocumentResult]
aupload_documents(...)	Same, async	List[DocumentResult]
close() / aclose()	Close the HTTP clients	None

DocumentResult: filename, status, document_id, chunks, units, error, ok, raw.

Migrating from 0.2.x

0.2.x targets the previous Promptev product at api.promptev.ai. Its endpoints do not exist on the new backend — every call would 404 — so this is a rewrite, not a base-URL bump. 0.2.x is untouched and keeps working; upgrade when you are ready to move.

1. The key moved from the URL to the header

This is the change everything else follows from.

# 0.2.x — the project key was in the URL PATH
client = PromptevClient(project_key="pv_sk_...")

# 1.0 — a Bearer key in the header
client = PromptevClient(api_key="fk_live_...")

Mint the new key under Settings → API keys in the project whose agents you call. Beyond keeping the secret out of access logs and referrer headers, the key now carries its own retrieval grant and is bound to one project, so what the caller may retrieve is part of the credential.

2. start_agent + stream_agent → chat

There is no session token any more. The endpoint streams, and takes the conversation id directly.

# 0.2.x
session = client.start_agent(agent_id, visitor="ava")
for event in client.stream_agent(agent_id, session_token=session.session_token,
                                 query="Hello"):
    if event.type == "done":
        print(event.output)

# 1.0
answer = ""
for event in client.chat("support-bot", "Hello"):
    if event.type == "token":
        answer += event.content
print(answer)

Note the shape change: the final text is no longer delivered whole on a done event. done is now just the end of the stream; the answer is the concatenation of the token events. Events are typed (event.content, event.id, …) instead of a single output string, and event.raw still has the whole frame.

Agents are addressed by slug, not by the Deploy-tab id.

3. trigger_agent is gone — use the project MCP server

The old one-call "run the agent, return JSON" endpoint has no equivalent: the new chat endpoint only streams, and faking it by draining the stream would silently swallow approval_required, which is exactly the event an unattended integration most needs to see.

If you were calling trigger_agent from an automation platform, point it at your project's MCP server instead. It exposes list_agents and call_agent; call_agent runs an agent and returns its answer in one call, which is what trigger did. n8n has an MCP client node; other platforms are adding them. You can find the server's URL in the project's settings.

If you were calling it from your own code, loop over chat() and concatenate the token events — and handle approval_required while you are there.

4. Base URL

The default is https://api.promptev.ai. Note that this host is being moved onto the new backend — until it is, pass baseUrl / base_url of https://console.promptev.ai explicitly.

5. Other changes

• visitor / platform → on_behalf_of=OnBehalfOf(subject=…, groups=…).
• variables is gone — an agent's prompt is configured in Promptev, not filled per call.
• AgentResult / AgentSession / AgentEvent are replaced by the typed event classes and DocumentResult.
• upload_document() is new — there was no way to add to an agent's corpus from the SDK before.
• The error hierarchy carries over unchanged, plus InsufficientCreditsError (402), e.field and e.retry_after. Note that validation now arrives as 422, not 400 — both still raise ValidationError.
• Sync/async parity carries over: every method still has its a-prefixed twin.

License

This SDK is commercial software by Promptev Inc.

• Free tier use allowed
• Production use requires an active subscription

See LICENSE for full terms.

Support

• Website: promptev.ai
• Console: console.promptev.ai
• Email: support@promptev.ai