Runtype Python SDK

The official Python SDK for the Runtype AI-native product platform

Installation

uv pip install runtype-sdk

Or with pip:

pip install runtype-sdk

Quick Start

from runtype import RuntypeClient, FlowBuilder

# Initialize client
client = RuntypeClient(api_key="your-api-key")

# List flows
flows = client.flows.list()
for flow in flows.data:
    print(flow.name)

# Build and execute a flow
result = (
    FlowBuilder()
    .create_flow(name="My Analysis Flow")
    .prompt(
        name="Analyze",
        model="gpt-4o",
        user_prompt="Summarize the following: {{input}}"
    )
    .with_options(stream_response=True)
    .run(client)
)

# Get the result
output = result.get_result("Analyze")
print(output)

Features

• Synchronous and Asynchronous Clients: Choose the client that fits your use case
• Fluent Flow Builder: Chain methods to build complex flows with ease
• TypeScript SDK High-Level Parity: Matching ergonomic flow helpers, validation, namespaces, and local-tool pause/resume behavior
• Streaming Support: Real-time streaming of flow execution events
• Type Hints: Full type annotations for IDE support and type checking
• Pydantic Models: Robust data validation and serialization

Usage

Client Initialization

from runtype import RuntypeClient, AsyncRuntypeClient

# Synchronous client
client = RuntypeClient(
    api_key="your-api-key",
    base_url="https://api.runtype.com",  # Optional
    timeout=30.0,  # Optional
)

# Async client
async_client = AsyncRuntypeClient(api_key="your-api-key")

Resource Operations

# Flows
flows = client.flows.list()
flow = client.flows.get("flow_123")
flow = client.flows.create(name="New Flow", description="A new flow")
client.flows.delete("flow_123")

# Records
records = client.records.list()
record = client.records.create(
    type="customer",
    name="Acme Corp",
    metadata={"industry": "tech"}
)

# Prompts
prompts = client.prompts.list()
prompt = client.prompts.create(
    name="Summarizer",
    text="Summarize: {{input}}",
    model="gpt-4o"
)

Flow Builder

The FlowBuilder provides a fluent interface for building and executing flows:

from runtype import FlowBuilder

result = (
    FlowBuilder()
    .create_flow(name="Data Pipeline")
    .fetch_url(
        name="Fetch Data",
        url="https://api.example.com/data",
        output_variable="raw_data"
    )
    .transform_data(
        name="Transform",
        script="return data.items.map(i => i.name)",
        output_variable="items"
    )
    .prompt(
        name="Analyze",
        model="gpt-4o",
        user_prompt="Analyze these items: {{items}}",
        output_variable="analysis"
    )
    .run(client)
)

# Access results
analysis = result.get_result("Analyze")

Streaming with Callbacks

The SDK streams the unified SSE event vocabulary. StreamCallbacks exposes one callback per event type — the same surface for flow and agent execution (the AgentStreamCallbacks name is an alias of StreamCallbacks).

from runtype import FlowBuilder, StreamCallbacks

def on_text_delta(event):
    print(event.delta, end="", flush=True)

def on_execution_complete(event):
    print(f"\nCompleted in {event.duration_ms}ms")

callbacks = StreamCallbacks(
    on_text_delta=on_text_delta,
    on_execution_complete=on_execution_complete,
)

summary = builder.run(client, callbacks=callbacks)

Local Tools (Client-Side Execution)

Local tools allow flows to pause and wait for your code to execute locally, then resume with the result. This is useful for:

• Data Privacy: Keep sensitive logic on your infrastructure
• Internal Systems: Access databases, files, or services not exposed via APIs
• Custom Logic: Execute complex business logic client-side

from runtype import RuntypeClient, FlowBuilder

client = RuntypeClient(api_key="your-api-key")

# Define local tool handlers
def get_user_data(args: dict) -> dict:
    user_id = args.get("user_id")
    # Query your internal database
    return {"name": "John", "balance": 100.50}

def process_payment(args: dict) -> dict:
    # Handle payment locally
    return {"success": True, "transaction_id": "txn_123"}

# Execute flow with local tools
result = (
    FlowBuilder()
    .create_flow(name="Purchase Flow")
    .prompt(
        name="Process Order",
        model="gpt-4o",
        user_prompt="Process order for user {{user_id}}",
        tools={
            "runtime_tools": [
                {
                    "name": "get_user_data",
                    "description": "Get user information from database",
                    "tool_type": "local",
                    "parameters_schema": {
                        "type": "object",
                        "properties": {
                            "user_id": {"type": "string"}
                        }
                    }
                },
                {
                    "name": "process_payment",
                    "description": "Process a payment transaction",
                    "tool_type": "local",
                    "parameters_schema": {
                        "type": "object",
                        "properties": {
                            "amount": {"type": "number"},
                            "user_id": {"type": "string"}
                        }
                    }
                }
            ]
        }
    )
    .with_options(flow_mode="virtual")
    .run(
        client,
        local_tools={
            "get_user_data": get_user_data,
            "process_payment": process_payment,
        }
    )
)

# Get the final result
order_result = result.get_result("Process Order")

The SDK automatically handles the pause/resume cycle - when the AI calls a local tool, the flow pauses, your function executes, and the flow resumes with the result.

For turn-scoped local tools, pass schema-carrying LocalToolEntry values and scope="turn":

from runtype import LocalToolEntry

result = client.run_with_local_tools(
    {"flow": {"name": "Tool Flow", "steps": [...]}},
    {
        "lookup_user": LocalToolEntry(
            description="Lookup a user by id",
            parameters_schema={
                "type": "object",
                "properties": {"user_id": {"type": "string"}},
            },
            execute=get_user_data,
        )
    },
    scope="turn",
    stream=True,
)

Async Usage

import asyncio
from runtype import AsyncRuntypeClient, FlowBuilder

async def main():
    async with AsyncRuntypeClient(api_key="your-api-key") as client:
        # List flows
        flows = await client.flows.list()

        # Stream flow execution
        async for event in await client.dispatch(
            {"flow": {"name": "Test", "steps": [...]}},
            stream=True
        ):
            print(event["type"])

asyncio.run(main())

Using Existing Flows

result = (
    FlowBuilder()
    .use_existing_flow("flow_abc123")
    .with_record(name="Customer A", type="customer")
    .with_messages([
        {"role": "user", "content": "Analyze this customer"}
    ])
    .run(client)
)

Upsert Mode

For code-first flow management, use upsert mode to create or update flows:

result = (
    FlowBuilder()
    .upsert_flow(
        name="My Flow",
        create_version_on_change=True
    )
    .prompt(name="Process", model="gpt-4o", user_prompt="...")
    .run(client)
)

Runtype Fluent API

The Runtype class provides a modern static/fluent API for building and executing flows with global configuration:

from runtype import Runtype

# Configure once at app startup
Runtype.configure(api_key="your-api-key")

# Build and stream a flow (async)
result = await (
    Runtype.flows.upsert(name="My Flow")
    .prompt(name="Analyze", model="gpt-4o", user_prompt="Analyze: {{input}}")
    .stream()
)

output = result.get_result("Analyze")

Flow Modes

# Upsert mode - Create or update flow by name
result = await (
    Runtype.flows.upsert(name="My Flow", create_version_on_change=True)
    .prompt(name="Step", model="gpt-4o", user_prompt="...")
    .stream()
)

# Virtual mode - One-off execution, not saved
result = await (
    Runtype.flows.virtual(name="Temp Flow")
    .prompt(name="Step", model="gpt-4o", user_prompt="...")
    .stream()
)

# Existing flow - Execute a saved flow by ID
result = await (
    Runtype.flows.use("flow_abc123")
    .with_record(name="Customer A", type="customer")
    .stream()
)

Synchronous Execution

# Use stream_sync() or result_sync() for synchronous code
result = (
    Runtype.flows.virtual(name="My Flow")
    .prompt(name="Analyze", model="gpt-4o", user_prompt="...")
    .stream_sync()
)

output = result.get_result("Analyze")

Local Tools with Runtype

from runtype import Runtype

Runtype.configure(api_key="your-api-key")

def get_user_data(args: dict) -> dict:
    return {"name": "John", "balance": 100.50}

result = await (
    Runtype.flows.virtual(name="Purchase Flow")
    .prompt(
        name="Process Order",
        model="gpt-4o",
        user_prompt="Process order for user {{user_id}}",
        tools={
            "runtime_tools": [{
                "name": "get_user_data",
                "description": "Get user info",
                "tool_type": "local",
                "parameters_schema": {
                    "type": "object",
                    "properties": {"user_id": {"type": "string"}}
                }
            }]
        }
    )
    .with_local_tools({"get_user_data": get_user_data})
    .stream()
)

Other Namespaces

# Batches - Schedule batch operations
batch = await Runtype.batches.schedule(
    flow_id="flow_123",
    record_type="customers",
)

# Get batch status
status = await Runtype.batches.get(batch["id"])

# Prompts - Manage prompts
prompts = await Runtype.prompts.list()
prompt = await Runtype.prompts.get("prompt_123")

# Skills - Manage Runtype Agent Skills
skills = await Runtype.skills.list(status="published")
await Runtype.skills.publish_version("skill_123", "skill_version_123")

Agent Config as Code

define_agent builds a definition locally, and client.agents.ensure converges it onto the platform. Identity is the agent's name within the API key's account scope. The converge is hash-first: it probes with the definition's content hash and ships the full definition only when the server reports a miss. Every change appends an immutable version; nothing is deleted.

from runtype import RuntypeClient, define_agent

client = RuntypeClient(api_key="your-api-key")

pricing_assistant = define_agent(
    name="Pricing Assistant",
    model="claude-sonnet-4-6",
    system_prompt="You answer pricing questions.",
    loop_config={"maxTurns": 1},
)

# Converge (CI/deploy). Steady state is one small probe request.
result = client.agents.ensure(pricing_assistant)

# PR drift gate: raises AgentDriftError when the remote moved.
client.agents.ensure(pricing_assistant, expect_no_changes=True)

# Absorb a dashboard edit back into the repo.
pulled = client.agents.pull("Pricing Assistant")

compute_agent_content_hash(definition) gives you the same hash the API computes, byte-identical to the TypeScript and Ruby SDKs. Echo the server's contentHash from a response rather than your own when you persist one.

Release Aliases

An agent runs an immutable version. A release alias is a named, mutable pointer that selects which one: live carries production traffic, and any other name (pr-482, staging) is a preview pointer that never touches the live row. Every activation appends a deployment receipt.

# Save and activate in one call.
client.agents.ensure(pricing_assistant, deploy={"alias": "live"})

# Or save without activating, then aim the pointer yourself.
saved = client.agents.ensure(pricing_assistant)
live = client.agents.aliases.get(saved["agentId"], "live")
client.agents.aliases.activate(
    saved["agentId"],
    "live",
    version_id=saved["versionId"],
    revision=live.revision,             # compare-and-swap, sent as If-Match
    idempotency_key="deploy-2026-09-06-1",
    reason="pricing copy refresh",
)

# Inspect, roll back, retire a preview pointer.
client.agents.aliases.list(saved["agentId"])
client.agents.aliases.rollback(saved["agentId"], "live", steps=1, revision=live.revision)
client.agents.aliases.archive(saved["agentId"], "pr-482")

# The append-only history behind the pointers.
client.agents.deployments.list(saved["agentId"], alias="live", limit=20)

# Run a specific pointer or a specific version.
client.agents.execute(saved["agentId"], {"messages": messages}, alias="pr-482")
client.agents.execute(saved["agentId"], {"messages": messages}, version_id="agtv_9")

revision is the compare-and-swap guard the server requires on an existing live pointer; re-read the alias and retry if it comes back stale. activate sends the exact version_id you give it: the SDK never resolves a pointer for you and deploys whatever it found. release="publish" remains as a compatibility spelling of deploy={"alias": "live"}, and supplying both is refused.

Promoting across organizations

prepare -> validate -> evaluate -> activate with the primitives above. Nothing gates activation on the evaluation: evaluate reports, and you decide.

source = RuntypeClient(api_key=SOURCE_KEY)
target = RuntypeClient(api_key=TARGET_KEY)

# 1. Prepare: pull the source definition and save it in the target org,
#    parked on a preview alias so live traffic is untouched.
pulled = source.agents.pull("Pricing Assistant")
staged = target.agents.ensure(pulled["definition"], deploy={"alias": "candidate"})

# 2. Validate: the plan for the target's live pointer must be a clean apply.
plan = target.agents.ensure(pulled["definition"], dry_run=True)
assert plan["changes"] in ("none", "update")

# 3. Evaluate: run the target org's suite against the exact staged version.
#    The report is evidence for a human, not a gate: nothing here blocks step 4.
report = target.api.evals.run_eval_suite_synchronously(
    suite_id=SUITE_ID,
    agent={"versionId": staged["versionId"]},
)

# 4. Activate: aim live at the exact version you evaluated, quoting the revision
#    you read, with a replay key so a retry cannot deploy twice.
live = target.agents.aliases.get(staged["agentId"], "live")
target.agents.aliases.activate(
    staged["agentId"],
    "live",
    version_id=staged["versionId"],
    revision=live.revision,
    idempotency_key=f"promote-{staged['versionId']}",
    reason=f"promoted from source org after suite {SUITE_ID} scored {report.score}",
)

Read report and decide. If you want the promotion to stop on a bad score, write that check yourself between steps 3 and 4.

Flow Validation

# Legacy sync/async builder validation
validation = (
    FlowBuilder()
    .create_flow(name="Validated Flow")
    .prompt(name="Analyze", model="gpt-5-mini", user_prompt="Analyze")
    .validate(client)
)

# Static builder validation
validation = (
    Runtype.flows.virtual(name="Validated Flow")
    .prompt(name="Analyze", model="gpt-5-mini", user_prompt="Analyze")
    .validate_sync()
)

Available Step Types

• prompt() - Execute an LLM prompt
• fetch_url() - Make HTTP requests
• crawl() - Crawl pages with browser rendering
• transform_data() - Transform data with JavaScript
• set_variable() - Set a runtime variable
• search() - Web or database search
• retrieve_record() - Load record data
• upsert_record() - Save or update records
• vector_search() - Semantic vector search
• generate_embedding() - Generate embeddings
• send_email() - Send email messages
• send_stream() - Send streaming messages
• send_event() - Send analytics/events
• conditional() - Branching logic
• wait_until() - Delays and polling

Error Handling

from runtype import RuntypeClient, APIError, AuthenticationError, NotFoundError

client = RuntypeClient(api_key="your-api-key")

try:
    flow = client.flows.get("nonexistent")
except NotFoundError:
    print("Flow not found")
except AuthenticationError:
    print("Invalid API key")
except APIError as e:
    print(f"API error: {e.status_code} - {e.message}")

Development

# Install development dependencies
uv pip install -e ".[dev]"

# Run tests
pytest

# Run tests with coverage
pytest --cov=runtype

# Type checking
mypy runtype

# Linting and formatting
ruff check runtype
ruff format runtype

License

MIT License - see LICENSE file for details.