Python SDK

enprompta

Async-first Python SDK with httpx, pydantic models, and comprehensive type hints.

Async First

Built on httpx for async/await support.

Type Hints

Full type annotations for IDE support.

Pydantic

Data validation with pydantic models.

Free to start

Creating an API key is free, and the SDK records observability traces on any plan (5,000/month on Free). Serving prompts and the rest of the REST API require a Pro or Enterprise plan — editor seats read/write, viewer seats read-only. View pricing →

Requirements: Python 3.8+ | httpx 0.25+ | pydantic 2.0+

Installation

pip

pip install enprompta

poetry

poetry add enprompta

pipenv

pipenv install enprompta

Quick Start

Python (async)

import asyncio
from enprompta import Enprompta

async def main():
    # Initialize with API key
    client = Enprompta(api_key="ep_your_api_key")

    # Or with OAuth2 credentials
    client = Enprompta(
        client_id="your_client_id",
        client_secret="your_client_secret"
    )

    # List all prompts
    prompts = await client.prompts.list()
    for prompt in prompts.data:
        print(prompt.title)

    # Create a new prompt
    prompt = await client.prompts.create(
        title="Email Assistant",
        content="Write a professional email about {{topic}} to {{recipient}}",
        visibility="PRIVATE",
        variables=[
            {"name": "topic", "type": "text", "required": True},
            {"name": "recipient", "type": "text", "required": True}
        ]
    )

    # Execute the prompt
    result = await client.prompts.execute(
        prompt.id,
        variables={
            "topic": "project deadline extension",
            "recipient": "the project manager"
        },
        provider="openai",
        model="gpt-4"
    )

    print(result.output)

# Run the async function
asyncio.run(main())

Authentication

The SDK supports both API key and OAuth2 client credentials authentication.

API Key Authentication

import os
from enprompta import Enprompta

client = Enprompta(
    api_key=os.environ["ENPROMPTA_API_KEY"],
    # Optional: custom base URL
    base_url="https://enprompta.com/api/v1"
)

OAuth2 Client Credentials

Create a client in your dashboard under API Keys → OAuth Clients to get a client_id and client_secret (the secret is shown once).

from enprompta import Enprompta

client = Enprompta(
    client_id="your_client_id",
    client_secret="your_client_secret",
    scopes=["prompts:read", "prompts:write", "executions:write"]
)

# Tokens are automatically managed and refreshed

Environment Variables

# The SDK automatically reads these environment variables:
# ENPROMPTA_API_KEY
# ENPROMPTA_CLIENT_ID
# ENPROMPTA_CLIENT_SECRET
# ENPROMPTA_BASE_URL

from enprompta import Enprompta

# No arguments needed if env vars are set
client = Enprompta()

Working with Prompts

List Prompts

# List with pagination
response = await client.prompts.list(
    limit=20,
    visibility="PRIVATE",
    search="email"
)

for prompt in response.data:
    print(f"{prompt.id}: {prompt.title}")

# Check for more pages
if response.pagination.has_more:
    next_page = await client.prompts.list(
        cursor=response.pagination.next_cursor
    )

# Iterate through all prompts (auto-pagination)
async for prompt in client.prompts.list_all():
    print(prompt.title)

Create Prompt

prompt = await client.prompts.create(
    title="Code Reviewer",
    content="""Review the following {{language}} code and provide feedback:

```{{language}}
{{code}}
```

Focus on: {{focus_areas}}""",
    description="AI-powered code review assistant",
    category="development",
    tags=["code", "review", "development"],
    visibility="TEAM",
    team_id="team_abc123",
    variables=[
        {
            "name": "language",
            "type": "select",
            "options": ["javascript", "python", "go"]
        },
        {
            "name": "code",
            "type": "text",
            "required": True
        },
        {
            "name": "focus_areas",
            "type": "text",
            "default_value": "best practices, performance"
        }
    ]
)

print(f"Created prompt: {prompt.id}")

Update and Delete

# Update a prompt
updated = await client.prompts.update(
    "prompt_abc123",
    title="Updated Title",
    content="New prompt content with {{variable}}",
    visibility="PUBLIC"
)

# Delete a prompt
await client.prompts.delete("prompt_abc123")

Executions

Execute a Prompt

result = await client.prompts.execute(
    "prompt_abc123",
    variables={
        "topic": "quarterly review",
        "tone": "professional"
    },
    provider="openai",
    model="gpt-4o-mini",
    temperature=0.7,
    max_tokens=1000
)

print(result.output)           # Generated text
print(result.tokens_used)      # Token count
print(result.cost)             # Estimated cost
print(result.latency_ms)       # Response time

List Executions

# Get execution history
executions = await client.executions.list(
    prompt_id="prompt_abc123",
    provider="openai",
    start_date="2025-01-01",
    end_date="2025-01-31"
)

for execution in executions.data:
    print(f"{execution.id}: {execution.tokens_used} tokens")

# Get aggregated statistics
stats = await client.executions.get_stats(group_by="day")

Auto-Instrumentation

One call instruments every OpenAI, Anthropic, and Google Gemini call your app already makes — no client wrapping, no decorators, no call-site changes. Traces are dispatched on a background thread so instrumentation never adds latency to your call path.

Python

import enprompta

enprompta.auto_instrument(api_key="ep_your_api_key")

# Your existing, unmodified code is now traced:
from openai import OpenAI
client = OpenAI()
client.chat.completions.create(
    model="gpt-4o",
    messages=[{"role": "user", "content": "Hello"}],
)

Captures provider, model, prompt, response, tokens, latency, cost, and errors — including streamed calls and tool-call turns (the model's tool call is recorded as the output instead of a blank; streamed tool calls are reassembled). If no supported provider SDK is installed, auto_instrument() warns instead of silently doing nothing.

Framework instrumentation (LangChain, LlamaIndex, agents)

Capture the whole agent/RAG trace — retrievals, tool calls, reranks, and sub-agent steps — by bridging the OpenInference instrumentors for your stack.

Install

pip install "enprompta[instrumentation]"
pip install openinference-instrumentation-langchain   # your stack's instrumentor

Python

import enprompta

enprompta.instrument_frameworks(api_key="ep_your_api_key")  # auto-detects
# ...or pick frameworks explicitly:
enprompta.instrument_frameworks(api_key="ep_...", frameworks=["langchain"])

Sending Traces (manual)

Prefer full control, or calling a provider the auto-instrumenter doesn't patch? Send traces yourself. The simplest flow: fetch a prompt, call your LLM provider, then POST the result to Enprompta.

Agent span trees & sessions (recommended for agents)

For agents and tool loops, record a span tree instead of one flat trace per call: a root span per request with typed child spans (LLM, TOOL, RETRIEVAL, …), grouped by a session_id so a multi-turn conversation reads as one session. Requires enprompta 1.2.0+.

Python

from enprompta import Enprompta

client = Enprompta(api_key="ep_your_api_key")

# One root span per user question, grouped by session.
trace = client.trace(name="support-agent", type="AGENT",
                     session_id=conversation_id, input=question)

# Each model call -> a typed LLM child span.
llm = trace.span("plan", type="LLM", model="gpt-4o", input=messages)
completion = openai.chat.completions.create(...)
llm.end(output=completion.choices[0].message,
        input_tokens=completion.usage.prompt_tokens,
        output_tokens=completion.usage.completion_tokens)

# Each tool run -> a TOOL child span (now visible in the trace tree).
tool = trace.span("get_weather", type="TOOL", input=args)
tool.end(output=result)

trace.end(output=answer)  # flushes the whole tree

The result is a nested, typed tree in Observability — end-to-end latency per request, which tool was slow, how many turns it took — plus a Sessions view that groups the whole conversation. Spans are typed (LLM / TOOL / RETRIEVAL / AGENT / CHAIN / RERANKER / GUARDRAIL / EMBEDDING / EVALUATOR), following the OpenInference convention.

Fetch prompt → call LLM → send trace

import time
import uuid
import httpx
from openai import AsyncOpenAI

openai = AsyncOpenAI()
ENPROMPTA_KEY = os.environ["ENPROMPTA_API_KEY"]
ENPROMPTA_BASE = "https://enprompta.com"

async def run_with_tracing(prompt_slug: str, user_input: str) -> str:
    # 1. Fetch the prompt at runtime
    async with httpx.AsyncClient() as http:
        resp = await http.get(
            f"{ENPROMPTA_BASE}/api/sdk/prompts/{prompt_slug}?label=prod",
            headers={"Authorization": f"Bearer {ENPROMPTA_KEY}"}
        )
    data = resp.json()
    content = data["content"]
    prompt_id = data["_meta"]["promptId"]
    prompt_version_id = data["_meta"]["promptVersionId"]

    # 2. Call your LLM provider
    start_ms = int(time.time() * 1000)
    completion = await openai.chat.completions.create(
        model="gpt-4o",
        messages=[{"role": "user", "content": content}]
    )
    latency_ms = int(time.time() * 1000) - start_ms
    output = completion.choices[0].message.content or ""

    # 3. Send the trace — fire and forget
    import asyncio
    asyncio.create_task(_send_trace(
        prompt_id=prompt_id,
        prompt_version_id=prompt_version_id,
        content=content,
        output=output,
        latency_ms=latency_ms,
        usage=completion.usage,
    ))

    return output

async def _send_trace(*, prompt_id, prompt_version_id, content, output, latency_ms, usage):
    now_ms = int(time.time() * 1000)
    async with httpx.AsyncClient() as http:
        await http.post(
            f"{ENPROMPTA_BASE}/api/ingest/traces",
            headers={
                "Authorization": f"Bearer {ENPROMPTA_KEY}",
                "Content-Type": "application/json"
            },
            json={
                "spans": [{
                    "traceId": str(uuid.uuid4()),
                    "spanId": str(uuid.uuid4()),
                    "name": "my-prompt",
                    "spanType": "LLM",
                    "startTime": f"{(now_ms - latency_ms)}",
                    "endTime": f"{now_ms}",
                    "status": "SUCCESS",
                    "attributes": {
                        "llm.provider": "openai",
                        "llm.model": "gpt-4o",
                        "llm.input.messages": content,
                        "llm.output.messages": output,
                        "llm.token_count.prompt": usage.prompt_tokens if usage else None,
                        "llm.token_count.completion": usage.completion_tokens if usage else None,
                        "llm.latency_ms": latency_ms,
                        "enprompta.prompt_id": prompt_id,
                        "enprompta.prompt_version_id": prompt_version_id,
                        "enprompta.environment": "production"
                    }
                }]
            }
        )

Alternative: single-trace endpoint

For simple single-turn calls, use the flat POST /api/v1/traces endpoint instead of the spans batch format.

async with httpx.AsyncClient() as http:
    await http.post(
        f"{ENPROMPTA_BASE}/api/v1/traces",
        headers={
            "Authorization": f"Bearer {ENPROMPTA_KEY}",
            "Content-Type": "application/json"
        },
        json={
            "provider": "openai",
            "model": "gpt-4o",
            "input": content,
            "output": output,
            "inputTokens": usage.prompt_tokens if usage else None,
            "outputTokens": usage.completion_tokens if usage else None,
            "latencyMs": latency_ms,
            "environment": "production",
            "promptId": prompt_id,
            "metadata": {"feature": "summariser"}
        }
    )

Safety Auto-Scan: When enprompta.prompt_id is set and you have Safety Auto-Scan enabled on the prompt, PII Leakage is evaluated inline (no latency added) and Bias & Prompt Injection run via a 15-minute background cron. Enable toggles under Dashboard > Prompts > [prompt] > Settings.

Teams

# List teams
teams = await client.teams.list()
for team in teams.data:
    print(f"{team.id}: {team.name}")

# Create a team
team = await client.teams.create(
    name="Engineering",
    description="Engineering team prompts"
)

# Get team details
team = await client.teams.get("team_abc123")

# Update team
await client.teams.update(
    "team_abc123",
    name="Engineering Team"
)

Webhooks

# Create a webhook
webhook = await client.webhooks.create(
    name="Slack Notifications",
    url="https://your-server.com/webhooks/enprompta",
    events=[
        "prompt.created",
        "prompt.updated",
        "execution.completed"
    ],
    secret="whsec_your_webhook_secret"
)

# List webhooks
webhooks = await client.webhooks.list()

# Get delivery history
deliveries = await client.webhooks.get_deliveries("webhook_abc123")

Verifying Webhook Signatures

from enprompta.webhooks import verify_signature

# In your webhook handler (e.g., FastAPI)
@app.post("/webhooks/enprompta")
async def handle_webhook(request: Request):
    payload = await request.body()
    signature = request.headers.get("X-Enprompta-Signature")

    if not verify_signature(payload, signature, webhook_secret):
        raise HTTPException(status_code=401, detail="Invalid signature")

    event = json.loads(payload)
    print(f"Received event: {event['event']}")

    return {"status": "ok"}

Error Handling

The SDK provides typed exception classes for different failure scenarios.

from enprompta.exceptions import (
    EnpromptaError,
    AuthenticationError,
    RateLimitError,
    ValidationError,
    NotFoundError,
    NetworkError,
    TimeoutError
)

try:
    await client.prompts.get("invalid_id")
except NotFoundError:
    print("Prompt not found")
except AuthenticationError:
    print("Invalid API key")
except RateLimitError as e:
    print(f"Rate limited. Retry after {e.retry_after}s")
except ValidationError as e:
    print(f"Validation errors: {e.errors}")
except NetworkError as e:
    print(f"Network error: {e}")
except TimeoutError:
    print("Request timed out")
except EnpromptaError as e:
    print(f"API error {e.code}: {e.message}")

Synchronous Client

For non-async code, use the synchronous client wrapper.

from enprompta import EnpromptaSync

# Synchronous client
client = EnpromptaSync(api_key="ep_your_api_key")

# All methods work without await
prompts = client.prompts.list()
for prompt in prompts.data:
    print(prompt.title)

prompt = client.prompts.create(
    title="My Prompt",
    content="Hello {{name}}"
)

result = client.prompts.execute(
    prompt.id,
    variables={"name": "World"},
    provider="openai",
    model="gpt-4"
)

print(result.output)

Context Manager

Use the client as a context manager to ensure proper cleanup.

# Async context manager
async with Enprompta(api_key="ep_your_api_key") as client:
    prompts = await client.prompts.list()
    # Client is automatically closed after the block

# Sync context manager
with EnpromptaSync(api_key="ep_your_api_key") as client:
    prompts = client.prompts.list()
    # Client is automatically closed after the block

Next Steps

TypeScript SDK API Reference

Python SDK

Async First

Type Hints

Pydantic

On this page

Installation

Quick Start

Authentication

API Key Authentication

OAuth2 Client Credentials

Environment Variables

Working with Prompts

List Prompts

Create Prompt

Update and Delete

Executions

Execute a Prompt

List Executions

Auto-Instrumentation

Framework instrumentation (LangChain, LlamaIndex, agents)

Sending Traces (manual)

Agent span trees & sessions (recommended for agents)

Fetch prompt → call LLM → send trace

Alternative: single-trace endpoint

Teams

Webhooks

Verifying Webhook Signatures

Error Handling

Synchronous Client

Context Manager

Next Steps