Python SDK
Use PromptRails from Python apps and notebooks with clients for agents, prompts, traces, and scores.
Best for
Engineers building against the API, SDKs, CLI, MCP, or local tooling
The official Python SDK for PromptRails provides both synchronous and asynchronous clients for interacting with the PromptRails API.
Use the Python SDK when you are calling agents from a Python backend, notebook, workflow runner, or data pipeline. You do not need it for a first product test inside PromptRails; Studio, chat, triggers, and deployed apps can run agents without writing code.
If your main goal is observability, start with the tracing module. If your main goal is automation, start with agent execution and API keys.
Installation
pip install promptrailsRequires Python 3.9 or later.
Current release: v0.7.0 — the standalone promptrails.tracing
module for sending spans to PromptRails from any code, with LangChain, OpenAI,
Anthropic, Google GenAI, and OpenTelemetry integrations. See the
changelog.
Technical detailsClient setup and resource reference
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
| Parameter | Type | Default | Description |
|---|---|---|---|
api_key | str | Required | PromptRails API key |
base_url | str | https://api.promptrails.ai | API base URL |
timeout | float | 30.0 | Request timeout in seconds |
max_retries | int | 3 | Maximum retry attempts for failed requests |
The API key is sent via the X-API-Key header with every request.
Available Resources
| Resource | Attribute | Description |
|---|---|---|
| Agents | client.agents | Agent CRUD, versioning, execution |
| Prompts | client.prompts | Prompt CRUD, versioning, execution |
| Executions | client.executions | Execution listing and details |
| Credentials | client.credentials | Credential management |
| Data Sources | client.data_sources | Data source CRUD, versioning, execution |
| Chat | client.chat | Send messages to chat sessions |
| Sessions | client.sessions | Chat session management |
| Memories | client.memories | Agent memory CRUD and search |
| Traces | client.traces | Trace listing and filtering |
| Costs | client.costs | Cost analysis and summaries |
| MCP Tools | client.mcp_tools | MCP tool management |
| MCP Templates | client.mcp_templates | MCP template browsing |
| Guardrails | client.guardrails | Guardrail configuration |
| Approvals | client.approvals | Approval request management |
| Scores | client.scores | Scoring and evaluation |
| Templates | client.templates | Flow templates |
| Dashboard | client.dashboard | Agent UI deployments |
| A2A | client.a2a | Agent-to-Agent protocol |
| LLM Models | client.llm_models | Available LLM models |
| Agent Triggers | client.agent_triggers | Agent trigger management (generic webhook, Slack, Telegram, Teams, WhatsApp, schedule) |
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}")Technical detailsMore Python SDK operations
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)
breakThe 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):
breakThe 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"
)Technical detailsPython SDK advanced details
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.
Tracing
The promptrails.tracing module sends spans to PromptRails from any code, without managing your prompts or agents on the platform. It is independent of the API client and only needs an API key with the traces:write scope.
from promptrails.tracing import Tracer
tracer = Tracer(api_key="pr_...")
with tracer.span("agent-run", kind="agent") as root:
root.set_input({"q": "weather?"})
with tracer.span("llm-call", kind="llm") as llm:
llm.set_model("gpt-4o").set_usage(prompt_tokens=120, completion_tokens=30)
tracer.flush()Use the @tracer.trace(kind="tool") decorator to trace a function. Spans flush in the background and on exit; tracer.flush() sends them without blocking, while tracer.shutdown() blocks until all queued spans are delivered — use it in short-lived scripts so nothing is lost before the process exits.
Framework integrations
Optional extras auto-instrument popular frameworks:
pip install "promptrails[langchain]" # or [openai], [anthropic], [google], [otel]# LangChain
from promptrails.tracing.integrations.langchain import PromptRailsCallbackHandler
chain.invoke(inputs, config={"callbacks": [PromptRailsCallbackHandler(tracer)]})
# OpenAI
from promptrails.tracing.integrations.openai import trace_openai
client = trace_openai(OpenAI(), tracer)
# Anthropic
from promptrails.tracing.integrations.anthropic import trace_anthropic
client = trace_anthropic(Anthropic(), tracer)
# Google GenAI
from promptrails.tracing.integrations.google import trace_google
client = trace_google(genai.Client(), tracer)See the Tracing guide for span kinds, batching, and the OpenTelemetry bridge.
SDK Version
from promptrails import VERSION
print(VERSION) # "0.7.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
| Exception | HTTP Status | Description |
|---|---|---|
ValidationError | 400 | Invalid request parameters |
UnauthorizedError | 401 | Invalid or missing API key |
ForbiddenError | 403 | Insufficient permissions or IP/origin restriction |
NotFoundError | 404 | Resource not found |
RateLimitError | 429 | Rate limit exceeded |
ServerError | 5xx | Server-side error |
PromptRailsError | Any | Base class for all SDK errors |
All exceptions include:
message-- Human-readable error messagestatus_code-- HTTP status codecode-- Optional error code stringdetails-- 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)Related Topics
- Examples -- Ready-to-run code examples
- Quickstart -- Getting started guide
- JavaScript SDK -- JavaScript/TypeScript alternative
- Go SDK -- Go alternative
- API Keys and Scopes -- API key management
- REST API Reference -- Underlying REST API