Pipelines API

Use the hosted Pipelines control plane when you want agents or backend scripts to upload data → save a pipeline → start a run → fetch sink output without opening the app UI.

Base URL:

https://baleybots-edge.odd-credit-33f7.workers.dev

Full reference (routes + error codes): see the product repo’s docs/api/v1.md (ships with the /v1 release).

Server-side credentials

Personal access tokens (bb_…) are secrets. Put them in env vars on your agent host or CI — never in browser or mobile client bundles.

1. Create an API key

1. Sign in to app.baleybots.com
2. Open Settings → API Keys
3. Name the key → Create key
4. Copy the bb_… value now — it is shown only once

Revoke anytime from the same page (takes effect on the next request).

export EDGE=https://baleybots-edge.odd-credit-33f7.workers.dev
export BB_KEY='bb_paste_your_key_here'

Smoke-test auth:

curl -sS "$EDGE/v1/health" -H "Authorization: Bearer $BB_KEY"
# → { "ok": true, "service": "baleybots-edge-v1", "authKind": "pat" }

2. Upload a CSV (get a connectionId)

curl -sS -X POST "$EDGE/v1/files/csv?name=demo&filename=rows.csv" \
  -H "Authorization: Bearer $BB_KEY" \
  -H "Content-Type: text/csv" \
  --data-binary $'name,score\nAda,9\nGrace,8\n'

Response includes connectionId. Caps: 50 MiB body, 1_000_000 rows. Multipart also works: -F file=@rows.csv -F name=demo.

3. Save your first pipeline

Pipelines use the same config graph as the editor (source → processor → sink). Minimal runnable shape:

CONNECTION_ID='…'   # from step 2

curl -sS -X POST "$EDGE/v1/pipelines" \
  -H "Authorization: Bearer $BB_KEY" \
  -H "Content-Type: application/json" \
  -d @- <<EOF
{
  "name": "My first API pipeline",
  "status": "draft",
  "config": {
    "nodes": [
      {
        "id": "src",
        "type": "source",
        "position": { "x": 0, "y": 0 },
        "data": {
          "label": "csv",
          "connectionId": "$CONNECTION_ID",
          "connectionType": "csv_upload"
        }
      },
      {
        "id": "proc",
        "type": "baleybot",
        "position": { "x": 280, "y": 0 },
        "data": {
          "label": "Enrich row",
          "slug": "enrich",
          "description": "Add a short summary field.",
          "sampleInput": "{\"name\":\"Ada\",\"score\":9}",
          "goal": "Given a CSV row as JSON, return JSON with the same fields plus summary: a one-sentence description of the row.",
          "outputSchema": {
            "type": "object",
            "properties": {
              "name": { "type": "string" },
              "score": { "type": "number" },
              "summary": { "type": "string" }
            },
            "required": ["name", "score", "summary"],
            "additionalProperties": false
          },
          "exampleOutput": { "name": "Ada", "score": 9, "summary": "Ada scored 9." },
          "model": "claude-sonnet-4-6",
          "program": {
            "kind": "agent",
            "goal": "Given a CSV row as JSON, return JSON with the same fields plus summary: a one-sentence description of the row.",
            "outputSchema": {
              "type": "object",
              "properties": {
                "name": { "type": "string" },
                "score": { "type": "number" },
                "summary": { "type": "string" }
              },
              "required": ["name", "score", "summary"],
              "additionalProperties": false
            }
          }
        }
      },
      {
        "id": "sink",
        "type": "sink",
        "position": { "x": 560, "y": 0 },
        "data": {
          "label": "out.csv",
          "sinkKind": "csv",
          "connectionId": null,
          "mapping": null,
          "csv": {
            "filename": "enriched.csv",
            "columns": [
              { "field": "name", "header": "name" },
              { "field": "score", "header": "score" },
              { "field": "summary", "header": "summary" }
            ]
          }
        }
      }
    ],
    "edges": [
      { "id": "e1", "source": "src", "target": "proc" },
      { "id": "e2", "source": "proc", "target": "sink" }
    ]
  }
}
EOF

Response includes pipeline.id. A frozen fixture also lives in the product repo under docs/api/examples/csv-processor-csv-sink.json.

4. Start a run

Manual runs are hard-capped at 500 rows (even if you omit limit).

PIPELINE_ID='…'

curl -sS -X POST "$EDGE/v1/pipelines/$PIPELINE_ID/runs" \
  -H "Authorization: Bearer $BB_KEY" \
  -H "Content-Type: application/json" \
  -d '{"limit": 50}'
# → { "runId": "…", "maxRows": 500 }

5. Poll until settled (preferred)

RUN_ID='…'

while true; do
  body=$(curl -sS "$EDGE/v1/runs/$RUN_ID" -H "Authorization: Bearer $BB_KEY")
  echo "$body" | jq '{ status: .run.status, derived: .derived }'
  echo "$body" | jq -e '
    (.run.status != null and (.run.status | IN("success","succeeded","failed","cancelled","error","completed")))
    or (.derived.kind == "settled")
  ' >/dev/null 2>&1 && break
  sleep 2
done

Optional short server wait (max 60s; timeout returns timedOut: true, not 5xx):

curl -sS -X POST "$EDGE/v1/runs/$RUN_ID/wait" \
  -H "Authorization: Bearer $BB_KEY" \
  -H "Content-Type: application/json" \
  -d '{"timeoutMs": 30000}'

6. Download the sink CSV

curl -sS "$EDGE/v1/runs/$RUN_ID/sinks/sink/csv" \
  -H "Authorization: Bearer $BB_KEY" \
  -o enriched.csv

Node (fetch) sketch

const EDGE = process.env.EDGE
const key = process.env.BB_KEY
const auth = { Authorization: `Bearer ${key}` }

const csv = await fetch(`${EDGE}/v1/files/csv?name=demo&filename=rows.csv`, {
  method: 'POST',
  headers: { ...auth, 'Content-Type': 'text/csv' },
  body: 'name,score\nAda,9\n',
})
const { connectionId } = await csv.json()

const save = await fetch(`${EDGE}/v1/pipelines`, {
  method: 'POST',
  headers: { ...auth, 'Content-Type': 'application/json' },
  body: JSON.stringify({
    name: 'My first API pipeline',
    status: 'draft',
    config: {
      /* same nodes/edges as the curl example — set source connectionId */
      nodes: [
        {
          id: 'src',
          type: 'source',
          position: { x: 0, y: 0 },
          data: { label: 'csv', connectionId, connectionType: 'csv_upload' },
        },
        /* …processor + csv sink… */
      ],
      edges: [
        /* … */
      ],
    },
  }),
})
const { pipeline } = await save.json()

const run = await fetch(`${EDGE}/v1/pipelines/${pipeline.id}/runs`, {
  method: 'POST',
  headers: { ...auth, 'Content-Type': 'application/json' },
  body: JSON.stringify({ limit: 10 }),
})
const { runId } = await run.json()

for (;;) {
  const g = await fetch(`${EDGE}/v1/runs/${runId}`, { headers: auth })
  const body = await g.json()
  if (body.run?.status && !['queued', 'running', 'awaiting_approval'].includes(body.run.status)) {
    break
  }
  await new Promise((r) => setTimeout(r, 2000))
}

Auth notes

Token	Header	Notes
Personal access token	Authorization: Bearer bb_…	From Settings; full user power for dogfood
Supabase user JWT	Authorization: Bearer <jwt>	Same session cookie/token the app uses

Ownership failures return 404 (not 403). Errors use { "error": string, "code"?: string }.

Related

• Cancel: POST /v1/runs/:id/cancel
• Approvals: GET /v1/runs/:id/approvals → POST /v1/approvals/:id with { "decision": "approved" }
• List connections: GET /v1/connections