Workflows & activities

# Create
curl -X POST "$BASE/activities" \
  -H "Authorization: Bearer $TURFAI_JWT" \
  -H "Content-Type: application/json" \
  -d '{
    "data": {
      "name": "Invoice triage",
      "workflow_type": "document_processing",
      "status": "draft",
      "workflow_definition": { "nodes": [], "edges": [], "input_schema": { "type": "object", "properties": {}, "required": [] } }
    }
  }'

# Read it back (id from the create response)
curl "$BASE/activities/42" -H "Authorization: Bearer $TURFAI_JWT"

import os, requests

BASE = os.environ["BASE"]
HEAD = {"Authorization": f"Bearer {os.environ['TURFAI_JWT']}"}

def create_activity(definition: dict) -> int:
    r = requests.post(f"{BASE}/activities", headers=HEAD, json={
        "data": {
            "name": "Invoice triage",
            "workflow_type": "document_processing",
            "status": "draft",
            "workflow_definition": definition,
        }
    })
    r.raise_for_status()
    return r.json()["data"]["id"]

def get_activity(activity_id: int) -> dict:
    r = requests.get(f"{BASE}/activities/{activity_id}", headers=HEAD)
    r.raise_for_status()
    return r.json()["data"]

const BASE = process.env.BASE!;
const HEAD = {
  Authorization: `Bearer ${process.env.TURFAI_JWT}`,
  "Content-Type": "application/json",
};

export async function createActivity(definition: object): Promise<number> {
  const res = await fetch(`${BASE}/activities`, {
    method: "POST",
    headers: HEAD,
    body: JSON.stringify({
      data: {
        name: "Invoice triage",
        workflow_type: "document_processing",
        status: "draft",
        workflow_definition: definition,
      },
    }),
  });
  if (!res.ok) throw new Error(`create failed: ${res.status}`);
  return (await res.json()).data.id;
}

export async function getActivity(id: number) {
  const res = await fetch(`${BASE}/activities/${id}`, { headers: HEAD });
  if (!res.ok) throw new Error(`get failed: ${res.status}`);
  return (await res.json()).data;
}

# 1. Create an execution bound to activity 42
EXEC=$(curl -s -X POST "$BASE/workflow-executions" \
  -H "Authorization: Bearer $TURFAI_JWT" \
  -H "Content-Type: application/json" \
  -d '{ "data": { "activity": 42, "inputs": { "file_url": "gs://bucket/invoice.pdf" } } }' \
  | python3 -c 'import sys,json;print(json.load(sys.stdin)["data"]["id"])')

# 2. Start it
curl -X POST "$BASE/workflow-executions/$EXEC/execute" \
  -H "Authorization: Bearer $TURFAI_JWT" \
  -H "Content-Type: application/json" \
  -d '{ "inputs": { "file_url": "gs://bucket/invoice.pdf" } }'

def trigger(activity_id: int, inputs: dict) -> int:
    created = requests.post(f"{BASE}/workflow-executions", headers=HEAD, json={
        "data": {"activity": activity_id, "inputs": inputs}
    })
    created.raise_for_status()
    exec_id = created.json()["data"]["id"]

    started = requests.post(
        f"{BASE}/workflow-executions/{exec_id}/execute",
        headers=HEAD, json={"inputs": inputs},
    )
    started.raise_for_status()
    return exec_id

export async function trigger(activityId: number, inputs: object): Promise<number> {
  const created = await fetch(`${BASE}/workflow-executions`, {
    method: "POST", headers: HEAD,
    body: JSON.stringify({ data: { activity: activityId, inputs } }),
  });
  if (!created.ok) throw new Error(`create exec failed: ${created.status}`);
  const execId = (await created.json()).data.id;

  const started = await fetch(`${BASE}/workflow-executions/${execId}/execute`, {
    method: "POST", headers: HEAD, body: JSON.stringify({ inputs }),
  });
  if (!started.ok) throw new Error(`execute failed: ${started.status}`);
  return execId;
}

curl "$BASE/workflow-executions/93/status" \
  -H "Authorization: Bearer $TURFAI_JWT"

import time

def poll(exec_id: int, attempts: int = 30) -> dict:
    for _ in range(attempts):
        r = requests.get(f"{BASE}/workflow-executions/{exec_id}/status", headers=HEAD)
        r.raise_for_status()
        data = r.json()["data"]
        if data["status"] == "completed":
            return data["results"]
        if data["status"] == "failed":
            raise RuntimeError(data.get("error"))
        time.sleep(2)
    raise TimeoutError("polling timed out")

const sleep = (ms: number) => new Promise((r) => setTimeout(r, ms));

export async function poll(execId: number, attempts = 30) {
  for (let i = 0; i < attempts; i++) {
    const res = await fetch(`${BASE}/workflow-executions/${execId}/status`, { headers: HEAD });
    if (!res.ok) throw new Error(`status failed: ${res.status}`);
    const { data } = await res.json();
    if (data.status === "completed") return data.results;
    if (data.status === "failed") throw new Error(data.error);
    await sleep(2000);
  }
  throw new Error("polling timed out");
}