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 promptrails

Requires 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

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, playground, guardrails
Promptsclient.promptsPrompt CRUD and content-only versioning
Executionsclient.executionsListing, trees, cancel, approval inbox, approve/deny, stream
Credentialsclient.credentialsCredential management
Data Sourcesclient.data_sourcesData source CRUD, versioning, query
Chatclient.chatChat sessions and message streaming
Tracesclient.tracesTrace listing, summary, PII report, ingest
MCP Toolsclient.mcp_toolsMCP tool management
MCP Templatesclient.mcp_templatesMCP template browsing
Guardrailsclient.guardrailsScanner catalog and guardrail configuration
LLM Modelsclient.llm_modelsAvailable LLM models
Agent Triggersclient.agent_triggersAgent trigger management (generic webhook, Slack, Telegram, Teams, WhatsApp, schedule)
Agent VFSclient.agent_vfsRead, write, move, copy, grep, and inspect an agent’s virtual files
Assetsclient.assetsList, retrieve, sign, and delete generated assets
A2Aclient.a2aAgent-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)
        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.

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

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.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)