PromptRails
View as Markdown

Python SDK

Official Python SDK for PromptRails with sync and async clients, automatic retries, and typed error handling.

Python SDK

The official Python SDK for PromptRails provides both synchronous and asynchronous clients for interacting with the PromptRails API.

Installation

pip install promptrails

Requires Python 3.9 or later.

Current release: v0.3.0 — streaming chat & executions (sync and async), typed AgentConfig dataclasses, VERSION export. See the changelog.

Client Initialization

Synchronous Client

from promptrails import PromptRails
 
client = PromptRails(
    api_key="your-api-key",
    base_url="https://api.promptrails.ai",  # default
    timeout=30.0,                             # seconds, default
    max_retries=3                             # default
)

Async Client

from promptrails import AsyncPromptRails
 
client = AsyncPromptRails(
    api_key="your-api-key",
    base_url="https://api.promptrails.ai",
    timeout=30.0,
    max_retries=3
)

Context Manager

Both clients support context managers for automatic cleanup:

# Sync
with PromptRails(api_key="your-api-key") as client:
    result = client.agents.list()
 
# Async
async with AsyncPromptRails(api_key="your-api-key") as client:
    result = await client.agents.list()

Configuration

ParameterTypeDefaultDescription
api_keystrRequiredPromptRails API key
base_urlstrhttps://api.promptrails.aiAPI base URL
timeoutfloat30.0Request timeout in seconds
max_retriesint3Maximum retry attempts for failed requests

The API key is sent via the X-API-Key header with every request.

Available Resources

ResourceAttributeDescription
Agentsclient.agentsAgent CRUD, versioning, execution
Promptsclient.promptsPrompt CRUD, versioning, execution
Executionsclient.executionsExecution listing and details
Credentialsclient.credentialsCredential management
Data Sourcesclient.data_sourcesData source CRUD, versioning, execution
Chatclient.chatSend messages to chat sessions
Sessionsclient.sessionsChat session management
Memoriesclient.memoriesAgent memory CRUD and search
Tracesclient.tracesTrace listing and filtering
Costsclient.costsCost analysis and summaries
MCP Toolsclient.mcp_toolsMCP tool management
MCP Templatesclient.mcp_templatesMCP template browsing
Guardrailsclient.guardrailsGuardrail configuration
Approvalsclient.approvalsApproval request management
Scoresclient.scoresScoring and evaluation
Templatesclient.templatesFlow templates
Dashboardclient.dashboardAgent UI deployments
A2Aclient.a2aAgent-to-Agent protocol
LLM Modelsclient.llm_modelsAvailable LLM models
Webhook Triggersclient.webhook_triggersWebhook trigger management

Common Operations

Execute an Agent

result = client.agents.execute(
    agent_id="agent-id",
    input={"message": "Hello, world!"},
    metadata={"source": "api"}
)
 
print(result["data"]["output"])
print(f"Cost: ${result['data']['cost']:.6f}")

List Agents

agents = client.agents.list(page=1, limit=20)
 
for agent in agents["data"]:
    print(f"{agent['name']} ({agent['type']})")

Create a Prompt

prompt = client.prompts.create(
    name="Summarizer",
    description="Summarizes text"
)
 
version = client.prompts.create_version(
    prompt_id=prompt["data"]["id"],
    system_prompt="You are a concise summarizer.",
    user_prompt="Summarize: {{ text }}",
    temperature=0.5,
    message="Initial version"
)

Chat

session = client.chat.create_session(agent_id="agent-id")
 
response = client.chat.send_message(
    session.id,
    content="What is PromptRails?"
)
print(response.content)

Stream a Chat Turn

send_message_stream posts a user message and yields typed Server-Sent Events on the same connection — use it to surface the agent's intermediate reasoning, tool calls, and token deltas in real time.

from promptrails import (
    ContentEvent,
    DoneEvent,
    ErrorEvent,
    ExecutionEvent,
    ThinkingEvent,
    ToolEndEvent,
    ToolStartEvent,
)
 
session = client.chat.create_session(agent_id="agent-id")
 
for event in client.chat.send_message_stream(
    session.id, content="What is PromptRails?"
):
    if isinstance(event, ExecutionEvent):
        print("execution_id:", event.execution_id)
    elif isinstance(event, ThinkingEvent):
        print("[thinking]", event.content)
    elif isinstance(event, ToolStartEvent):
        print("[tool_start]", event.name)
    elif isinstance(event, ToolEndEvent):
        print("[tool_end]", event.name, event.summary)
    elif isinstance(event, ContentEvent):
        print(event.content, end="", flush=True)
    elif isinstance(event, DoneEvent):
        print("\n[done]", event.token_usage)
    elif isinstance(event, ErrorEvent):
        print("[error]", event.message)
        break

The async client exposes the same method on AsyncChatResource:

async for event in aclient.chat.send_message_stream(
    session.id, content="hello"
):
    ...

Stream an Existing Execution

When an execution was started outside a chat (e.g. client.agents.execute), subscribe to its live event stream with client.executions.stream:

for event in client.executions.stream(execution_id):
    if isinstance(event, ContentEvent):
        print(event.content, end="", flush=True)
    elif isinstance(event, DoneEvent):
        break

The async variant is available on AsyncExecutionsResource.stream.

Search Memories

results = client.memories.search(
    agent_id="agent-id",
    query="refund policy",
    limit=5
)

Create a Score

client.scores.create(
    trace_id="trace-id",
    name="quality",
    data_type="numeric",
    value=4.5,
    source="manual"
)

Approve an Execution

client.approvals.decide(
    approval_id="approval-id",
    decision="approved",
    reason="Looks good"
)

Typed Agent Config

agents.create_version takes a typed AgentConfig dataclass — one of SimpleAgentConfig, ChainAgentConfig, MultiAgentConfig, WorkflowAgentConfig, or CompositeAgentConfig. to_dict() injects the type discriminator automatically and drops unset optional fields.

from promptrails import (
    ChainAgentConfig,
    PromptLink,
    SimpleAgentConfig,
)
 
# Simple agent — one prompt per execution
simple = SimpleAgentConfig(
    prompt_id="prompt-id",
    approval_required=False,
)
 
# Chain agent — prompts run sequentially
chain = ChainAgentConfig(
    prompt_ids=[
        PromptLink(prompt_id="p1", role="step1", sort_order=0),
        PromptLink(prompt_id="p2", role="step2", sort_order=1),
    ],
)
 
client.agents.create_version(
    agent_id="agent-id",
    version="1.0.0",
    config=simple,
    set_current=True,
)

See Agent Versioning for the per-type field reference.

SDK Version

from promptrails import VERSION
print(VERSION)  # "0.3.0"

Every request is sent with User-Agent: promptrails-python/<version> so backend telemetry can attribute traffic to the SDK release.

Error Handling

The SDK raises typed exceptions for different error scenarios:

from promptrails.exceptions import (
    PromptRailsError,
    ValidationError,
    UnauthorizedError,
    ForbiddenError,
    NotFoundError,
    RateLimitError,
    ServerError,
)
 
try:
    result = client.agents.execute(agent_id="invalid-id", input={})
except NotFoundError as e:
    print(f"Agent not found: {e.message}")
except ValidationError as e:
    print(f"Invalid input: {e.message}")
    print(f"Details: {e.details}")
except RateLimitError as e:
    print(f"Rate limited: {e.message}")
except UnauthorizedError as e:
    print(f"Invalid API key: {e.message}")
except ForbiddenError as e:
    print(f"Insufficient permissions: {e.message}")
except ServerError as e:
    print(f"Server error ({e.status_code}): {e.message}")
except PromptRailsError as e:
    print(f"Unexpected error: {e.message}")

Error Classes

ExceptionHTTP StatusDescription
ValidationError400Invalid request parameters
UnauthorizedError401Invalid or missing API key
ForbiddenError403Insufficient permissions or IP/origin restriction
NotFoundError404Resource not found
RateLimitError429Rate limit exceeded
ServerError5xxServer-side error
PromptRailsErrorAnyBase class for all SDK errors

All exceptions include:

  • message -- Human-readable error message
  • status_code -- HTTP status code
  • code -- Optional error code string
  • details -- Optional dictionary with additional error details

Async Usage

The async client mirrors the sync client's API but uses await:

import asyncio
from promptrails import AsyncPromptRails
 
async def main():
    async with AsyncPromptRails(api_key="your-api-key") as client:
        # All methods are awaitable
        agents = await client.agents.list()
 
        result = await client.agents.execute(
            agent_id=agents["data"][0]["id"],
            input={"message": "Hello"}
        )
 
        print(result["data"]["output"])
 
asyncio.run(main())

Pagination

List endpoints support pagination:

# Page-based pagination
page1 = client.agents.list(page=1, limit=20)
page2 = client.agents.list(page=2, limit=20)
Previousn8n IntegrationNextJavaScript SDK