run-simple-pipelines - SKILL.md Agent Skill

name: run-simple-pipelines description: >- Run single-datafit or single-validation pipelines via SimplePipeline. Use when the user wants to submit a pipeline config, run a fit or validation as a single job, poll for results, or manage SimplePipeline runs. Triggers: "simple pipeline", "SimplePipeline", "single datafit", "fire and forget pipeline", "submit config".

SimplePipeline

Guide for running single-element pipelines using the SimplePipeline endpoint — a lightweight alternative to the full Pipeline API.

Prerequisites

API key: Get one from the Ionworks dashboard at Settings → API Keys. Set as IONWORKS_API_KEY env var or pass to Ionworks(api_key="...").
Project ID: Set IONWORKS_PROJECT_ID env var or pass project_id to each method. Find your project ID in the dashboard URL or via client.project.list().
Run the discover-api skill to explore available endpoints and schemas.

When to Use SimplePipeline vs Pipeline

Use SimplePipeline when	Use Pipeline when
Config has at most one datafit or validation element	Config has multiple datafit/validation elements
You want fire-and-forget execution	You need per-element status tracking
You want a flat `parameter_values` result	You need cumulative parameter threading

Create and Wait for Results

The typical flow: initialize client → build config with ionworks_schema → create → poll → read result.

from ionworks import Ionworks
import ionworks_schema as iws

# Initialize client (reads IONWORKS_API_KEY and IONWORKS_PROJECT_ID from env)
client = Ionworks()

# 1. Build a typed config with ionworks_schema
pipeline = iws.SimplePipeline(
    elements={
        "initial_params": iws.direct_entries.DirectEntry(
            parameters={"Negative particle diffusivity [m2.s-1]": 2e-14},
        ),
        "fit": iws.DataFit(
            objectives={...},
            parameters={
                "Negative particle diffusivity [m2.s-1]": iws.Parameter(
                    "Negative particle diffusivity [m2.s-1]",
                    bounds=(1e-14, 1e-13),
                    initial_value=2e-14,
                )
            },
            cost=iws.costs.RMSE(),
            optimizer=iws.parameter_estimators.ScipyDifferentialEvolution(
                maxiter=10
            ),
        ),
    },
    name="My Fit",
)

# 2. Submit
sp = client.simple_pipeline.create(pipeline.to_config(), name=pipeline.name)
# sp.status == "pending", sp.id is the SimplePipeline ID

# 3. Wait (polls until completed/failed/canceled)
result = client.simple_pipeline.wait_for_completion(sp.id, timeout=600)

# 4. Read fitted values
print(result.result["parameter_values"])
# {"Negative particle diffusivity [m2.s-1]": 5.3e-14}

Listing and Filtering

# List all in project (paginated, newest first by default)
page = client.simple_pipeline.list(limit=25)
# page.items, page.count, page.total

# Filter by status
running = client.simple_pipeline.list(status="in.(pending,running)")

# Filter by name (case-insensitive)
matches = client.simple_pipeline.list(name="ilike.%diffevo%")

# Date range
recent = client.simple_pipeline.list(
    created_at_gt="2026-05-01", created_at_lt="2026-05-11"
)

# Sort ascending
oldest_first = client.simple_pipeline.list(order_by="created_at", order="asc")

Update, Cancel, Delete

# Rename / add description (PATCH — at least one field required)
client.simple_pipeline.update(sp.id, name="Renamed", description="notes")

# Cancel a running pipeline
client.simple_pipeline.cancel(sp.id)

# Delete pipeline + associated job and storage
client.simple_pipeline.delete(sp.id)

Validation Element

SimplePipeline also supports a single validation element. Build it with iws.Validation and the result includes summary_stats alongside parameter_values.

import ionworks_schema as iws

objective = iws.objectives.CurrentDriven(data_input="path/to/cycle.csv")

validation_pipeline = iws.SimplePipeline(
    elements={
        "validate": iws.Validation(
            objectives={"cycle": objective},
        ),
    },
    name="Validate",
)

sp = client.simple_pipeline.create(
    validation_pipeline.to_config(), name=validation_pipeline.name
)
result = client.simple_pipeline.wait_for_completion(sp.id)
print(result.result["summary_stats"])

Error Handling

from ionworks.errors import IonworksError

try:
    result = client.simple_pipeline.wait_for_completion(sp.id)
except TimeoutError:
    # Pipeline still running after timeout
    sp = client.simple_pipeline.get(sp.id)
    print(f"Still {sp.status}")
except IonworksError as e:
    # Pipeline failed
    print(e)  # includes sp.error and sp.error_code

Related Skills

discover-api — query the API before any operation
upload-data — upload measurement data referenced in objectives