Metadata-Version: 2.5
Name: runtype-sdk
Version: 5.5.0
Summary: Python SDK for the Runtype Platform
Project-URL: Homepage, https://runtype.com
Project-URL: Documentation, https://docs.runtype.com
Project-URL: Support, https://runtype.com/support
Author-email: Runtype Labs <dev@runtype.com>
License-Expression: MIT
License-File: LICENSE
Keywords: agents,ai,automation,llm,runtype,workflows
Classifier: Development Status :: 4 - Beta
Classifier: Intended Audience :: Developers
Classifier: License :: OSI Approved :: MIT License
Classifier: Operating System :: OS Independent
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Classifier: Programming Language :: Python :: 3.14
Classifier: Topic :: Scientific/Engineering :: Artificial Intelligence
Classifier: Topic :: Software Development :: Libraries :: Python Modules
Classifier: Typing :: Typed
Requires-Python: >=3.10
Requires-Dist: httpx-sse>=0.4.0
Requires-Dist: httpx>=0.24.0
Requires-Dist: pydantic>=2.0.0
Requires-Dist: typing-extensions>=4.0.0
Provides-Extra: dev
Requires-Dist: mypy>=1.0.0; extra == 'dev'
Requires-Dist: pytest-asyncio>=0.21.0; extra == 'dev'
Requires-Dist: pytest-cov>=4.0.0; extra == 'dev'
Requires-Dist: pytest>=7.0.0; extra == 'dev'
Requires-Dist: respx>=0.20.0; extra == 'dev'
Requires-Dist: ruff>=0.1.0; extra == 'dev'
Description-Content-Type: text/markdown

# Runtype Python SDK

The official Python SDK for the [Runtype](https://runtype.com) AI-native product platform

## Installation

```bash
uv pip install runtype-sdk
```

Or with pip:

```bash
pip install runtype-sdk
```

## Quick Start

```python
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

```python
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

```python
# 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:

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

```python
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

```python
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"`:

```python
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

```python
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

```python
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:

```python
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:

```python
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

```python
# 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

```python
# 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

```python
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

```python
# 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.

```python
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**.

```python
# 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.

```python
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

```python
# 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

```python
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

```bash
# 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.
