highway 0.9.1

Installation

pip install highway

Quick Start

from highway import Driver

driver = Driver()  # Uses HIGHWAY_API_KEY env var

@driver.task(shell=True)
def backup_db():
    return "pg_dump mydb > backup.sql"

@driver.task(py=True, depends=["backup_db"])
def verify():
    import os
    return {"exists": os.path.exists("backup.sql")}

result = driver.run()
print(result.status)  # "completed"

Configuration

Set environment variables:

export HIGHWAY_API_KEY="hw_k1_..."
export HIGHWAY_API_ENDPOINT="https://highway.solutions"

Or pass directly:

driver = Driver(
    api_key="hw_k1_...",
    endpoint="https://highway.solutions"
)

Task Types

@driver.task(shell=True)
def list_files():
    return "ls -la /tmp"

@driver.task(py=True)
def compute():
    import math
    return {"factorial": math.factorial(10)}

@driver.task(http=True)
def call_api():
    return {
        "url": "https://<organisation>.highway.solutions/webhook",
        "method": "POST",
        "json": {"status": "done"},
    }

@driver.task(tool="tools.llm.call")
def summarize():
    return {
        "provider": "ollama",
        "model": "qwen3-vl:235b-instruct-cloud",
        "prompt": "Summarize: {{backup_result.stdout}}",
        "temperature": 0.7
    }

@driver.task(tool="tools.database.query")
def query_users():
    return {
        "connection_string": "vault:db/postgres",
        "query": "SELECT * FROM users"
    }

@driver.task(http=True)
def fetch_data():
    return {"url": "https://api.example.com/data", "method": "GET"}

@driver.task(tool="tools.llm.call", depends=["fetch_data"])
def analyze():
    return {
        "provider": "ollama",  # or "openai", "anthropic"
        "model": "qwen3-vl:235b-instruct-cloud",
        "prompt": "Analyze: {{fetch_data_result.body}}",
        "temperature": 0.3
    }

@driver.task(tool="tools.llm.call", depends=["analyze"])
def summarize():
    return {
        "provider": "ollama",
        "model": "qwen3-vl:235b-instruct-cloud",
        "prompt": "Summarize: {{analyze_result.response}}"
    }

@driver.task(workflow="daily_report")
def run_report():
    return {"inputs": {"date": "2024-01-01"}}

@driver.task(workflow_id="uuid-here")
def run_specific_version():
    return {"inputs": {"mode": "production"}}

Dependencies

Tasks can depend on other tasks:

@driver.task(shell=True)
def step_1():
    return "echo 'Step 1'"

@driver.task(shell=True, depends=["step_1"])
def step_2():
    return "echo 'Step 2'"

@driver.task(shell=True, depends=["step_2"])
def step_3():
    return "echo 'Step 3'"

Workflow Inputs

Pass variables to workflows:

result = driver.run(
    inputs={"email": "user@example.com", "env": "prod"},
    timeout=300
)

Access inputs in tasks via {{inputs.key}} syntax.

Retry Configuration

Configure automatic retries with exponential backoff:

@driver.task(http=True, retries=3, retry_delay=2.0, backoff=2.0)
def call_flaky_api():
    return {"url": "https://<organisation>.highway.solutions/endpoint"}

Durable Delays

Use Highway’s native WaitOperator for delays that consume zero worker resources:

from datetime import timedelta

@driver.task(shell=True, delay=timedelta(hours=2))
def delayed_task():
    return "echo 'Runs after 2 hour delay'"

Scheduling

Schedule recurring workflows:

@driver.task(shell=True, schedule="0 * * * *")  # Every hour
def hourly_backup():
    return "pg_dump mydb > backup.sql"

@driver.task(shell=True, schedule=timedelta(minutes=30))
def interval_check():
    return "curl https://<organisation>.highway.solutions/health"

Observability

Track running workflows:

# Non-blocking submit
result = driver.run(wait=False)
run_id = result.run_id

# Check status
status = driver.status(run_id)
print(status.state)

# Get logs
logs = driver.logs(run_id)

# Cancel
driver.cancel(run_id)

Long-Running Workflows

For workflows that take minutes to complete, submit and check status later:

# Submit and exit immediately
result = driver.run(wait=False, timeout=600)
print(f"Run ID: {result.run_id}")
# Driver can exit, workflow continues on Highway

# Later, check status
status = driver.status(result.run_id)
if status.status == "completed":
    print("Workflow finished!")

For tasks longer than 30 seconds, set appropriate timeout values:

@driver.task(shell=True, timeout=120)  # 2 minute timeout
def long_running_task():
    return "sleep 90 && echo 'Done'"

Persistence

Highway Driver uses SQLite for local state persistence at stabilize.db. This enables:

• Submit workflow, exit driver, check status later

• Resume from crashes

• Idempotent execution

The database is created automatically in your working directory.

Framework Integration

Highway Driver integrates with web frameworks. See examples/integrations/:

from fastapi import FastAPI
from highway import Driver

app = FastAPI()
driver = Driver()

@app.post("/submit")
async def submit_workflow(data: dict):
    result = driver.run(wait=False)
    return {"run_id": result.run_id}

@app.get("/status/{run_id}")
async def get_status(run_id: str):
    return driver.status(run_id)

from flask import Flask, jsonify
from highway import Driver

app = Flask(__name__)
driver = Driver()

@app.post("/submit")
def submit_workflow():
    result = driver.run(wait=False)
    return jsonify({"run_id": result.run_id})

Idempotency

Use workflow IDs for idempotent execution:

result1 = driver.run(workflow_id="order-12345")
result2 = driver.run(workflow_id="order-12345")  # Returns cached result

Examples

The examples/ directory contains runnable examples:

• simple_shell.py - Basic shell task

• multi_step.py - Task dependencies

• python_func.py - Python code execution

• http_tasks.py - HTTP requests

• parallel.py - Parallel execution

• llm_agentic.py - LLM-powered workflow

• long_running/ - Submit and check status later

• integrations/ - Flask/FastAPI integration

Run examples:

export HIGHWAY_API_KEY="your-key"
python examples/simple_shell.py

Requirements

• Python 3.11+

• Highway API key (get from https://highway.solutions)

License

Apache 2.0