enprompta

Official Python SDK for the Enprompta API.

Installation

pip install enprompta
# or
poetry add enprompta
# or
pipenv install enprompta

Requirements

• Python 3.8+
• httpx 0.25+
• pydantic 2.0+

Quick Start

import asyncio
from enprompta import Enprompta

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

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

    # Create a prompt
    prompt = await client.prompts.create(
        title="Email Writer",
        content="Write a professional email about {{topic}}",
        visibility="PRIVATE"
    )

    # Execute a prompt
    result = await client.prompts.execute(
        prompt.id,
        variables={"topic": "project update"},
        provider="openai",
        model="gpt-4"
    )

    print(result.output)

asyncio.run(main())

Authentication

API Key

from enprompta import Enprompta

client = Enprompta(api_key="ep_your_api_key")

OAuth2 Client Credentials

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

Environment Variables

# Set these environment variables:
# ENPROMPTA_API_KEY
# ENPROMPTA_CLIENT_ID
# ENPROMPTA_CLIENT_SECRET

client = Enprompta()  # Auto-reads from environment

Features

Prompts

# List with pagination
response = await client.prompts.list(limit=20, visibility="PRIVATE")
for prompt in response.data:
    print(prompt.title)

# Auto-pagination
async for prompt in client.prompts.list_all():
    print(prompt.title)

# Create
prompt = await client.prompts.create(
    title="My Prompt",
    content="Hello {{name}}",
    variables=[{"name": "name", "type": "text", "required": True}]
)

# Get (by id, from the management API)
prompt = await client.prompts.get("prompt_id")

# Get the LIVE prompt for a release label at runtime — the deploy path.
# Promote a version to `production` in Enprompta and this returns it with no
# redeploy. `messages` is the structured System/User form (None for a plain prompt).
live = client.prompts.get_live("my-prompt", label="prod")
messages = (
    [{"role": m.role, "content": m.content} for m in live.messages]
    if live.messages else [{"role": "user", "content": live.content}]
)

# Update
await client.prompts.update("prompt_id", title="New Title")

# Delete
await client.prompts.delete("prompt_id")

# Execute
result = await client.prompts.execute(
    "prompt_id",
    variables={"name": "World"},
    provider="openai",
    model="gpt-4"
)

Executions

# List executions
response = await client.executions.list(
    prompt_id="prompt_id",
    start_date="2024-01-01"
)

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

Teams

teams = await client.teams.list()
team = await client.teams.create(name="Engineering")
await client.teams.update("team_id", name="New Name")

Webhooks

webhook = await client.webhooks.create(
    name="My Webhook",
    url="https://example.com/webhook",
    events=["prompt.created", "execution.completed"]
)

Synchronous Client

For non-async code:

from enprompta import EnpromptaSync

client = EnpromptaSync(api_key="ep_your_api_key")

# All methods work without await
prompts = client.prompts.list()
prompt = client.prompts.create(title="My Prompt", content="Hello")

Context Manager

# Async
async with Enprompta(api_key="ep_your_api_key") as client:
    prompts = await client.prompts.list()

# Sync
with EnpromptaSync(api_key="ep_your_api_key") as client:
    prompts = client.prompts.list()

Error Handling

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

try:
    await client.prompts.get("invalid_id")
except NotFoundError:
    print("Prompt not found")
except RateLimitError as e:
    print(f"Retry after {e.retry_after}s")
except EnpromptaError as e:
    print(f"Error {e.code}: {e.message}")

LLM Observability & Tracing

@trace Decorator

Automatically trace any LLM function with the @trace decorator:

from enprompta import Enprompta, trace
import openai

client = Enprompta(api_key="ep_your_api_key")

@trace(client, provider="openai", model="gpt-4")
def generate_response(prompt: str) -> str:
    response = openai.chat.completions.create(
        model="gpt-4",
        messages=[{"role": "user", "content": prompt}]
    )
    return response.choices[0].message.content

# Traces are automatically recorded with timing, tokens, and cost
result = generate_response("Explain quantum computing")

Async Support

@trace(client, provider="anthropic", model="claude-3-sonnet", session_id="user-123")
async def async_chat(prompt: str) -> str:
    response = await anthropic.messages.create(
        model="claude-3-sonnet-20240229",
        messages=[{"role": "user", "content": prompt}]
    )
    return response.content[0].text

Global Auto-Instrumentation (recommended)

One call instruments every OpenAI, Anthropic, and Google Gemini call your app already makes — no client wrapping, no decorators, no call-site changes:

import enprompta

enprompta.auto_instrument(api_key="ep_...")

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

• Traces are sent on a background thread pool, so instrumentation never adds latency to (or raises into) your LLM calls.
• Streaming responses (stream=True) are captured too — the output text and token usage are accumulated as you consume the stream, then recorded once it finishes. (OpenAI streamed output-token counts require stream_options={"include_usage": True}.)
• Tool-call turns are captured — when the model calls a tool instead of replying with text (OpenAI tool_calls, Anthropic tool_use, Gemini function_call), the tool call is recorded as the span output rather than an empty string, so a tool-using agent's most important turns aren't lost. Streamed tool calls are reassembled from their fragments. (v1.2.1+)
• Pass an existing client instead of an API key with auto_instrument(client=my_enprompta_client), set environment=..., or toggle with enabled=False. Call enprompta.shutdown_auto_instrument() to stop.

Framework Instrumentation (LangChain, LlamaIndex, …)

auto_instrument() captures the raw LLM call. To capture the whole agent/RAG trace — retrievals, tool calls, reranks, sub-agent steps, and their nesting — bridge the OpenInference instrumentors for the frameworks you use. They emit typed, nested OpenTelemetry spans that Enprompta ingests as first-class span types (Retrieval, Tool, Reranker, Agent, Guardrail, Evaluator, …).

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

import enprompta

# Auto-detects every installed OpenInference instrumentor:
enprompta.instrument_frameworks(api_key="ep_...")

# ...or pick frameworks explicitly:
enprompta.instrument_frameworks(api_key="ep_...", frameworks=["langchain", "llama_index"])

• Exports to Enprompta's OTLP endpoint (/api/ingest/otlp/v1/traces) over a Bearer API key — coexisting with an existing OpenTelemetry setup if you have one.
• Supported names: langchain, llama_index, openai, anthropic, crewai, dspy, haystack. Pass a client=, set environment=..., or enabled=False.
• Returns a handle — call .uninstrument() to stop.

PII Redaction

Redact PII from trace input/output before it's sent, so raw values never reach Enprompta's servers — not even a truncated preview. This runs in your process, not ours; off by default. Covers auto_instrument(), client.trace() spans, and traces.record() (so the @trace and traced_openai() decorators get it too, since both call record() internally).

import enprompta
from enprompta import PiiConfig

enprompta.auto_instrument(
    api_key="ep_...",
    pii=PiiConfig(action="mask"),   # "mask" | "hash" | "drop" — default "mask"
)

Five built-in categories — email, phone, ssn, credit_card, ip_address — matching the platform's own PII Leakage evaluator. Restrict with categories=["email", "ssn"], or bring your own detector entirely with mask=lambda text: ..., which overrides categories/action when set. A custom mask that raises fails open (returns the original text) rather than breaking the traced call.

Scope: covers the raw-provider path (auto_instrument(), client.trace(), traces.record()). Also pass pii to instrument_frameworks(pii=PiiConfig(...)) to redact LangChain/LlamaIndex spans exported via the OpenTelemetry/OpenInference bridge — it wraps the OTLP exporter and scans every string-valued span attribute (not specific key names, since OpenInference's attribute surface varies by span kind and instrumentor version), so raw values never reach Enprompta there either. Same PiiConfig shape, including a custom mask callable.

Auto-traced OpenAI Client

Prefer to instrument a single client instance instead of patching globally? Wrap your OpenAI client:

from enprompta import Enprompta, traced_openai
from openai import OpenAI

enprompta = Enprompta(api_key="ep_...")
openai = traced_openai(enprompta, OpenAI())

# All calls are now automatically traced!
response = openai.chat.completions.create(
    model="gpt-4",
    messages=[{"role": "user", "content": "Hello"}]
)

Context Manager (Manual Tracing)

For more control, use the context manager:

with client.traces.wrap(
    provider="openai",
    model="gpt-4",
    input="Hello"
) as ctx:
    response = openai.chat.completions.create(
        model="gpt-4",
        messages=[{"role": "user", "content": "Hello"}]
    )
    ctx.set_output(
        output=response.choices[0].message.content,
        input_tokens=response.usage.prompt_tokens,
        output_tokens=response.usage.completion_tokens
    )

print(f"Trace ID: {ctx.trace_id}")

Nested Spans for Complex Pipelines

Track multi-step operations like RAG:

# Record the main trace
result = client.traces.record(
    provider="openai",
    model="gpt-4",
    input="What are our refund policies?",
    output="Based on our documentation...",
    latency_ms=2500
)

# Add spans for each step
client.traces.create_span(
    result["trace_id"],
    name="vector_search",
    span_type=SpanType.RETRIEVAL,
    input={"query": "refund policies", "top_k": 5},
    output={"document_ids": ["doc1", "doc2"]},
    duration_ms=150
)

client.traces.create_span(
    result["trace_id"],
    name="embedding",
    span_type=SpanType.EMBEDDING,
    tokens=8,
    duration_ms=50
)

Webhook Signature Verification

from enprompta.webhooks import verify_signature

# In your webhook handler (FastAPI example)
@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)

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

Type Hints

Full type hint support:

from enprompta.types import (
    Prompt,
    Execution,
    Team,
    Webhook,
    CreatePromptParams,
    ExecutePromptParams
)

Documentation

Full documentation: https://enprompta.com/docs/sdk/python

License

MIT