Docs

Documentation

Multi-step event-driven pipelines

Extend a single event_name flow into an ordered sequence of backend callbacks and optional linked AI workflows before final WebSocket delivery.

Multi-step pipelines extend the single event_name flow into an ordered sequence of up to 6 events after the initial AI call.

Flow

  1. Your app calls POST /v1/ai/async on the pipeline workflow (Model tab = initial AI).
  2. ModelRiver auto-advances through configured events:
    • Backend steps — webhook + wait for POST /v1/callback/:channel_id
    • Callback AI steps — run another workflow's model/settings inline
  3. Final backend step completes → WebSocket status: completed

Step data contract

latest_data merge rules

AfterRule
Initial AIlatest_data = ai_response.data
Backend callbacklatest_data = Map.merge(latest_data, callback.data)
Callback AI steplatest_data = target_workflow_ai_response.data (replace)

customer_data

Set once from the original async request. Never overwritten. Included in every webhook.

Target workflow input

Middle steps with a linked workflow receive a JSON user message:

JSON
1{
2 "pipeline_input": { "...": "latest_data at this step" },
3 "customer_data": { "...": "..." },
4 "event_name": "summarize",
5 "step_index": 1
6}

Configure target workflows (e.g. ticket_summarizer) to parse pipeline_input.

Hello pipeline example

ComponentNameRole
Pipeline workflowticket_pipelinePrimary AI triage
Helper workflowticket_summarizerSummarize structured output
Event 1enrichBackend — enrich ticket data
Event 2summarizeCallback AI → ticket_summarizer
Event 3finalizeBackend — final callback before WebSocket

Webhook envelope (additive fields)

JSON
1{
2 "type": "task.ai_generated",
3 "event": "enrich",
4 "channel_id": "...",
5 "step_index": 0,
6 "total_steps": 3,
7 "step_type": "backend",
8 "pipeline": { "workflow_name": "ticket_pipeline", "step_name": "enrich" },
9 "ai_response": { "data": { "...": "latest_data" } },
10 "callback_url": "https://api.modelriver.com/v1/callback/...",
11 "callback_required": true,
12 "customer_data": { "ticket_id": "123" }
13}

Callback responses

Intermediate step:

JSON
1{
2 "status": "step_received",
3 "step_index": 0,
4 "next_step": "summarize"
5}

Final step:

JSON
1{
2 "status": "completed",
3 "channel_id": "..."
4}

WebSocket statuses

StatusMeaning
ai_generatedInitial AI complete
step_readyWaiting for backend at step N
step_completedBackend callback received, advancing
callback_ai_completedLinked workflow AI step finished
completedPipeline finished

Quota

Each production callback_ai step counts toward organization quota (nested under the primary request log via primary_req_id). Test-mode and playground runs are excluded.

Error codes

CodeWhen
step_already_completedIdempotent duplicate callback (200)
step_index_mismatchCallback for wrong step (409)
pipeline_quota_exceededQuota exceeded mid-pipeline (WS)
step_ai_failedCallback AI provider error (WS)
callback_timeoutBackend step timeout (WS)

Legacy workflows with only event_name and no workflow_events rows continue to work unchanged.