enprompta 1.0.0

Official Python SDK for the Enprompta API

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
prompt = await client.prompts.get("prompt_id")

# 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

Auto-traced OpenAI Client

Wrap your OpenAI client for zero-code tracing:

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