promptev 0.2.0

Python SDK for Promptev — trigger and stream AI agents programmatically

Promptev Python SDK

The official Python SDK for Promptev.ai — trigger and stream your AI agents programmatically using your project API key.

An agent bundles instructions, a model, tools (Jira, Slack, GitHub, …), and context packs. You build and deploy it in Promptev, then run it from your code with this SDK.

Installation

pip install promptev

Requires Python 3.7+. Only dependency: httpx.

Quick Start

The fastest way to run an agent is trigger_agent — a single blocking call that creates-or-resumes a session, runs the agent (tools run server-side), and returns the final response:

from promptev import PromptevClient

client = PromptevClient(project_key="pv_sk_your_key_here")

result = client.trigger_agent(
    "your-agent-id",                       # from the Deploy tab
    "What are the open P1 bugs in our backlog?",
)

print(result.response)     # the agent's reply
print(result.session_id)   # pass this back to continue the conversation

your-agent-id is your agent's id, copied from the Deploy tab in Promptev.

Keep conversation memory

Pass the session_id from a previous result back into the next call to continue the same conversation. Omit it for a stateless, one-off call:

first = client.trigger_agent("your-agent-id", "Summarize our Q4 sales report")
print(first.response)

# Follow-up — agent remembers the previous turn
second = client.trigger_agent(
    "your-agent-id",
    "Now compare that with Q3",
    session_id=first.session_id,
)
print(second.response)

Fill prompt variables

If your agent declares {{...}} variables in its instructions, fill them with variables:

result = client.trigger_agent(
    "your-agent-id",
    "How do I reset my password?",
    variables={"company": "Acme Corp", "tone": "professional"},
)
print(result.response)

Attribute a run to a user (optional)

Every run is recorded on a chat session. By default the visitor is a generated webhook_<id>; pass your own end-user id with visitor to attribute the run (analytics / audit):

result = client.trigger_agent(
    "your-agent-id",
    "How do I reset my password?",
    visitor="user_12345",
)

Set visitor on the first call of a conversation. When you resume with session_id, the session keeps its original visitor.

Streaming Agents

When you want real-time output — the agent's reasoning, live tool-execution status, and the final answer as it's produced — use the streaming session flow: start_agent to open a session, then stream_agent to send messages and receive Server-Sent Events.

Start a session

session = client.start_agent("your-agent-id", visitor="John")

print(session.session_token)   # use this for all subsequent messages
print(session.agent_id)        # your agent's id
print(session.name)            # agent display name
print(session.memory_enabled)  # whether the agent retains conversation context

your-agent-id is your agent's id, copied from the Deploy tab in Promptev.

Stream a response

The agent responds via Server-Sent Events (SSE). Each event has a type and output:

Event Type	Description
thoughts	Agent's internal reasoning
processing	Tool execution status (e.g., "Searching Jira...")
approval_required	Agent needs permission to run a tool
done	Final response text
error	Something went wrong

for event in client.stream_agent(
    session.agent_id,
    session_token=session.session_token,
    query="What are the open P1 bugs in our backlog?",
):
    if event.type == "thoughts":
        print(f"Thinking: {event.output}")
    elif event.type == "processing":
        print(f"Running: {event.output}")
    elif event.type == "done":
        print(f"\n{event.output}")
    elif event.type == "error":
        print(f"Error: {event.output}")

Multi-turn conversation

The session token maintains conversation context across messages:

session = client.start_agent("your-agent-id", visitor="Sarah")

# First message
for event in client.stream_agent(
    session.agent_id,
    session_token=session.session_token,
    query="Summarize our Q4 sales report",
):
    if event.type == "done":
        print(event.output)

# Follow-up — agent remembers the previous context
for event in client.stream_agent(
    session.agent_id,
    session_token=session.session_token,
    query="Compare that with Q3",
):
    if event.type == "done":
        print(event.output)

Collect the final response only

If you only need the final text and don't care about intermediate events:

def ask_agent(client, session, query):
    """Send a message and return only the final response."""
    for event in client.stream_agent(
        session.agent_id,
        session_token=session.session_token,
        query=query,
    ):
        if event.type == "done":
            return event.output
        if event.type == "error":
            raise RuntimeError(event.output)

answer = ask_agent(client, session, "What's our monthly churn rate?")
print(answer)

When to use which:

• trigger_agent() — one call, blocking, returns the final response. Best for backends, jobs, and simple integrations.
• start_agent() + stream_agent() — real-time thoughts and tool status as the agent works. Best for chat UIs and live experiences.

Async Usage

Every method has an async variant prefixed with a. Use these in FastAPI, notebooks, or any async context:

import asyncio
from promptev import PromptevClient

async def main():
    async with PromptevClient(project_key="pv_sk_...") as client:
        # One-call trigger
        result = await client.atrigger_agent(
            "your-agent-id",
            "What is the refund policy?",
            variables={"company": "Acme Corp"},
        )
        print(result.response)

        # Streaming session
        session = await client.astart_agent("your-agent-id", visitor="Ava")
        async for event in client.astream_agent(
            session.agent_id,
            session_token=session.session_token,
            query="How many support tickets came in today?",
        ):
            if event.type == "done":
                print(event.output)

asyncio.run(main())

FastAPI integration

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

app = FastAPI()
client = PromptevClient(project_key="pv_sk_...")

@app.post("/ask")
async def ask(agent_id: str, query: str):
    session = await client.astart_agent(agent_id)

    async def generate():
        async for event in client.astream_agent(
            session.agent_id,
            session_token=session.session_token,
            query=query,
        ):
            yield f"data: {event.raw}\n\n"

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

Error Handling

The SDK raises typed exceptions for each failure scenario:

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

client = PromptevClient(project_key="pv_sk_...")

try:
    result = client.trigger_agent("your-agent-id", "Tell me about Ava")
    print(result.response)
except ValidationError as e:
    # 400 — bad input
    print(f"Invalid request: {e}")
except NotFoundError as e:
    # 404 — agent or project not found
    print(f"Not found: {e}")
except AuthenticationError as e:
    # 401/403 — invalid API key or agent not active
    print(f"Auth error: {e}")
except RateLimitError as e:
    # 429 — API usage quota exceeded
    print(f"Rate limited: {e}")
except ServerError as e:
    # 5xx — server error (after retries exhausted)
    print(f"Server error: {e}")
except NetworkError as e:
    # Connection failed, timeout, DNS error
    print(f"Network error: {e}")

All exceptions inherit from PromptevError and include:

• e.status_code — HTTP status code (if applicable)
• e.response_text — Raw response body (for debugging)

Exception	HTTP Status	When
ValidationError	400	Bad input
AuthenticationError	401, 403	Invalid API key, agent not active
NotFoundError	404	Agent or project not found
RateLimitError	429	API usage quota exceeded
ServerError	5xx	Server error (after retries exhausted)
NetworkError	—	Connection failed, timeout, DNS error
PromptevError	any	Base class for all above exceptions

Configuration

client = PromptevClient(
    project_key="pv_sk_...",              # Required — your project API key
    base_url="https://api.promptev.ai",   # Default — override for self-hosted
    timeout=30.0,                          # Default — request timeout in seconds
    max_retries=2,                         # Default — retries for 502/503/504
    headers={"X-Custom": "value"},         # Optional — extra HTTP headers
)

Parameter	Default	Description
project_key	required	Your Promptev project API key (pv_sk_...)
base_url	https://api.promptev.ai	API base URL
timeout	30.0	Request timeout in seconds
max_retries	2	Automatic retries for transient server errors (502, 503, 504)
headers	None	Additional HTTP headers

API Reference

PromptevClient

Method	Description	Returns
trigger_agent(agent_id, message, *, session_id?, variables?)	Run an agent in one blocking call (sync)	AgentResult
atrigger_agent(agent_id, message, *, session_id?, variables?)	Run an agent in one blocking call (async)	AgentResult
start_agent(agent_id, *, visitor?, platform?)	Start a streaming agent session (sync)	AgentSession
astart_agent(agent_id, *, visitor?, platform?)	Start a streaming agent session (async)	AgentSession
stream_agent(agent_id, *, session_token, query)	Stream an agent response (sync)	Iterator[AgentEvent]
astream_agent(agent_id, *, session_token, query)	Stream an agent response (async)	AsyncIterator[AgentEvent]
close()	Close HTTP clients (sync)	None
aclose()	Close HTTP clients (async)	None

AgentResult

Field	Type	Description
response	str	The agent's final reply
session_id	str	Pass back into trigger_agent to continue the conversation
raw	dict	Full parsed JSON response

AgentSession

Field	Type	Description
session_token	str	Token for subsequent stream calls
agent_id	str	Agent identifier
name	str	Agent display name
memory_enabled	bool	Whether the agent retains conversation context
messages	list	Previous messages (populated when resuming a session)

AgentEvent

Field	Type	Description
type	str	Event type: thoughts, processing, done, error, approval_required
output	str	Event content text
raw	dict	Full parsed SSE event data

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
• Email: [e-mail]