Workflows API

The Workflows API allows you to define, manage, and execute multi-agent pipelines through the B2B dashboard. Workflows are DAGs (directed acyclic graphs) of agents with configurable dependencies and conditional routing.

List Workflows

Retrieve all workflows for the current organization.


GET /api/v1/workflows

Headers:


Authorization: Bearer {token}
X-Org-ID: {organization_id}

Response (200):


{
  "items": [
    {
      "id": "wf_abc123",
      "name": "Full Analysis Pipeline",
      "status": "active",
      "agent_count": 4,
      "last_run": "2026-02-08T14:00:00Z",
      "total_runs": 1284,
      "avg_cost": 0.0142,
      "created_at": "2026-01-10T09:00:00Z"
    }
  ],
  "total": 5,
  "offset": 0,
  "limit": 20
}

Create Workflow

Define a new workflow from a JSON pipeline definition.


POST /api/v1/workflows

Headers:


Authorization: Bearer {token}
X-Org-ID: {organization_id}
Content-Type: application/json

Request:


{
  "name": "Content Moderation Pipeline",
  "description": "Analyze uploaded images for content policy violations",
  "agents": [
    {
      "name": "vision_classifier",
      "model": "gemini-2.5-pro",
      "prompt_template": "Classify the following image for content policy violations: {{image_url}}",
      "depends_on": [],
      "timeout_ms": 5000
    },
    {
      "name": "text_extractor",
      "model": "gpt-5.1",
      "prompt_template": "Extract all visible text from the image analysis: {{vision_classifier.output}}",
      "depends_on": ["vision_classifier"],
      "timeout_ms": 3000
    },
    {
      "name": "policy_checker",
      "model": "deepseek-v3",
      "prompt_template": "Check extracted content against policy rules: {{text_extractor.output}}",
      "depends_on": ["text_extractor"],
      "timeout_ms": 3000
    },
    {
      "name": "summary_generator",
      "model": "gpt-5.1",
      "prompt_template": "Generate a moderation summary from: classification={{vision_classifier.output}}, policy={{policy_checker.output}}",
      "depends_on": ["vision_classifier", "policy_checker"],
      "timeout_ms": 3000
    }
  ],
  "conditional_edges": [
    {
      "from": "vision_classifier",
      "to": "text_extractor",
      "condition": "output.has_text == true"
    }
  ]
}

Response (201):


{
  "id": "wf_def456",
  "name": "Content Moderation Pipeline",
  "description": "Analyze uploaded images for content policy violations",
  "status": "active",
  "agent_count": 4,
  "agents": [...],
  "conditional_edges": [...],
  "created_at": "2026-02-08T15:00:00Z"
}

Workflow Schema

Field	Type	Required	Description
`name`	string	Yes	Human-readable workflow name
`description`	string	No	Description of the workflow purpose
`agents`	array	Yes	List of agent definitions
`agents[].name`	string	Yes	Unique identifier for the agent within the workflow
`agents[].model`	string	Yes	LLM model to use (e.g., `gemini-2.5-pro`, `gpt-5.1`, `deepseek-v3`)
`agents[].prompt_template`	string	Yes	Prompt template with `{{variable}}` placeholders
`agents[].depends_on`	array	Yes	List of agent names this agent depends on (empty for root agents)
`agents[].timeout_ms`	integer	No	Maximum execution time in milliseconds (default: 30000)
`conditional_edges`	array	No	Conditional routing rules between agents
`conditional_edges[].from`	string	Yes	Source agent name
`conditional_edges[].to`	string	Yes	Target agent name
`conditional_edges[].condition`	string	Yes	JavaScript-like condition expression

Get Workflow Details

Retrieve a specific workflow by ID.


GET /api/v1/workflows/{id}

Headers:


Authorization: Bearer {token}
X-Org-ID: {organization_id}

Response (200):


{
  "id": "wf_def456",
  "name": "Content Moderation Pipeline",
  "description": "Analyze uploaded images for content policy violations",
  "status": "active",
  "agent_count": 4,
  "agents": [
    {
      "name": "vision_classifier",
      "model": "gemini-2.5-pro",
      "prompt_template": "Classify the following image...",
      "depends_on": [],
      "timeout_ms": 5000
    }
  ],
  "conditional_edges": [...],
  "total_runs": 42,
  "avg_cost": 0.0098,
  "avg_latency_ms": 3200,
  "created_at": "2026-02-08T15:00:00Z",
  "updated_at": "2026-02-08T15:00:00Z"
}

Execute Workflow

Run a workflow with input data. Returns an SSE stream with real-time execution progress.


POST /api/v1/workflows/{id}/run

Headers:


Authorization: Bearer {token}
X-Org-ID: {organization_id}
Content-Type: application/json
Accept: text/event-stream

Request:


{
  "input": {
    "image_url": "https://example.com/uploaded-image.jpg"
  },
  "options": {
    "timeout_ms": 30000,
    "require_approval": false
  }
}

SSE Response:


event: workflow_start
data: {"run_id": "run_xyz789", "workflow_id": "wf_def456", "agents": ["vision_classifier", "text_extractor", "policy_checker", "summary_generator"]}

event: agent_start
data: {"agent": "vision_classifier", "model": "gemini-2.5-pro"}

event: agent_complete
data: {"agent": "vision_classifier", "result": {"has_text": true, "classification": "safe"}, "latency_ms": 1100, "cost": 0.0028}

event: agent_start
data: {"agent": "text_extractor", "model": "gpt-5.1"}

event: agent_complete
data: {"agent": "text_extractor", "result": {"extracted_text": "Summer Sale 50% Off"}, "latency_ms": 800, "cost": 0.0015}

event: agent_start
data: {"agent": "policy_checker", "model": "deepseek-v3"}

event: agent_complete
data: {"agent": "policy_checker", "result": {"violations": [], "status": "pass"}, "latency_ms": 600, "cost": 0.0008}

event: agent_start
data: {"agent": "summary_generator", "model": "gpt-5.1"}

event: agent_complete
data: {"agent": "summary_generator", "result": {"summary": "Image is safe. Contains promotional text.", "verdict": "approved"}, "latency_ms": 700, "cost": 0.0020}

event: workflow_complete
data: {"run_id": "run_xyz789", "total_latency_ms": 3200, "total_cost": 0.0071, "result": {"verdict": "approved", "summary": "Image is safe. Contains promotional text."}}

Workflow Execution with Approval Gates

When require_approval is set to true, the workflow pauses at designated approval points and emits an approval_required event:


event: approval_required
data: {"run_id": "run_xyz789", "agent": "policy_checker", "reason": "High-risk content detected", "approve_url": "/api/v1/workflows/wf_def456/runs/run_xyz789/approve"}

The workflow resumes when the approval endpoint is called:


POST /api/v1/workflows/{id}/runs/{run_id}/approve


{
  "decision": "approve",
  "comment": "Reviewed and approved by admin"
}