Python SDK for AI Agents
Use the PromptRails Python SDK to run AI agents, manage prompts, inspect execution trees, and send LLM traces from apps or notebooks.
Best for
Engineers building against the API, SDKs, CLI, MCP, or local tooling
Python SDK for AI Agents
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, playground, guardrails |
| Prompts | client.prompts | Prompt CRUD and content-only versioning |
| Executions | client.executions | Listing, trees, cancel, approval inbox, approve/deny, stream |
| Credentials | client.credentials | Credential management |
| Data Sources | client.data_sources | Data source CRUD, versioning, query |
| Chat | client.chat | Chat sessions and message streaming |
| Traces | client.traces | Trace listing, summary, PII report, ingest |
| MCP Tools | client.mcp_tools | MCP tool management |
| MCP Templates | client.mcp_templates | MCP template browsing |
| Guardrails | client.guardrails | Scanner catalog and guardrail configuration |
| LLM Models | client.llm_models | Available LLM models |
| Agent Triggers | client.agent_triggers | Agent trigger management (generic webhook, Slack, Telegram, Teams, WhatsApp, schedule) |
| Agent VFS | client.agent_vfs | Read, write, move, copy, grep, and inspect an agent’s virtual files |
| Assets | client.assets | List, retrieve, sign, and delete generated assets |
| A2A | client.a2a | Agent-to-Agent protocol |
Common Operations
Execute an Agent
result = client.agents.execute(
agent_id="agent-id",
input={"message": "Hello, world!"},
)
print(result.output)
print(f"Cost: ${result.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 versions are content-only — model and sampling live on the agent version.
prompt = client.prompts.create(
name="Summarizer",
description="Summarizes text"
)
version = client.prompts.create_version(
prompt_id=prompt.id,
system_prompt="You are a concise summarizer.",
user_prompt="Summarize: {{ text }}",
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.
Inspect an Execution Tree
tree = client.executions.tree("execution-id") # children[] populated
for child in tree.children:
print(child.agent_id, child.status, child.cost)Approve or Deny an Execution
Approvals are execution-scoped in API v2. Runs parked at waiting_approval are resumed from the inbox:
for execution in client.executions.approval_inbox().data:
client.executions.approve(execution.id, reason="Looks good")
# or: client.executions.deny(execution.id, reason="Denied")Technical detailsPython SDK advanced details
Typed Agent Config
agents.create_version takes a typed AgentConfig dataclass — one of the
two API v2 kinds, PromptAgentConfig (an agent) or WorkflowAgentConfig
(a workflow). to_dict() injects the type discriminator automatically.
Model, sampling, budget, approval policy, cache TTL and tool/sub-agent
attachments are not part of config — they are version-scoped fields
passed alongside it.
from promptrails import (
PromptAgentConfig,
ModelConfig,
RunBudget,
ToolAttachment,
)
client.agents.create_version(
agent_id="agent-id",
version="1.0.0",
config=PromptAgentConfig(prompt_id="prompt-id"),
model_config=ModelConfig(model_id="llm-model-id", temperature=0.2),
run_budget=RunBudget(max_cost=2.0, max_depth=4),
tools=[ToolAttachment(mcp_tool_id="tool-id", requires_approval=True)],
set_current=True,
)For the workflow kind, build config with WorkflowAgentConfig(nodes=[...]).
See Agent Versioning for the full 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.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