Workflows API

The Workflows API provides methods to interact with and execute automated workflows in Mastra.

Getting all workflows

Retrieve a list of all available workflows:

const workflows = await mastraClient.listWorkflows()

Getting workflow run counts

Retrieve per-workflow counts of running and suspended runs in a single request. The counts are computed on the server and keyed by the workflow's registry key, the key used when registering the workflow in the Mastra config, which can differ from the workflow's own id:

const runCounts = await mastraClient.listWorkflowRunCounts()
// { "cityWorkflow": { running: 2, suspended: 1 }, ... }

Returns: Record<string, { running: number; suspended: number }>

The server may cache the counts for a few seconds between requests. Servers that predate this endpoint respond with 404 Not Found. Handle the error when the client can talk to older deployments.

Working with a specific workflow

Get an instance of a specific workflow by its ID:

src/mastra/workflows/test-workflow.ts

export const testWorkflow = createWorkflow({
  id: 'city-workflow',
})

const workflow = mastraClient.getWorkflow('city-workflow')

Workflow methods

details()

Retrieve detailed information about a workflow:

const details = await workflow.details()

createRun()

Create a new workflow run instance:

const run = await workflow.createRun()

// Or with an existing runId
const run = await workflow.createRun({ runId: 'existing-run-id' })

// Or with a resourceId to associate the run with a specific resource
const run = await workflow.createRun({
  runId: 'my-run-id',
  resourceId: 'user-123',
})

The resourceId parameter associates the workflow run with a specific resource (e.g., user ID, tenant ID). This value is persisted with the run and can be used for filtering and querying runs later.

startAsync()

Start a workflow run and await its completion, returning the full result as the workflow output.

const run = await workflow.createRun()

const result = await run.startAsync({
  inputData: {
    city: 'New York',
  },
})

You can also pass initialState to set the starting values for the workflow's state:

const result = await run.startAsync({
  inputData: {
    city: 'New York',
  },
  initialState: {
    count: 0,
    items: [],
  },
})

The initialState object should match the structure defined in the workflow's stateSchema. See Workflow State for more details.

To associate a run with a specific resource, pass resourceId to createRun():

const run = await workflow.createRun({ resourceId: 'user-123' })

const result = await run.startAsync({
  inputData: {
    city: 'New York',
  },
})

start()

Start a workflow run without waiting for completion (fire-and-forget). Returns immediately with a success message. Use runById() on the workflow instance to check results later:

const run = await workflow.createRun()

await run.start({
  inputData: {
    city: 'New York',
  },
})

// Poll for results later
const result = await workflow.runById(run.runId)

This is useful for long-running workflows where you want to start execution and check results later.

resumeAsync()

Resume a suspended workflow step and await the full result:

const run = await workflow.createRun({ runId: prevRunId })

const result = await run.resumeAsync({
  step: 'step-id',
  resumeData: { key: 'value' },
})

resume()

Resume a suspended workflow step without waiting for completion:

const run = await workflow.createRun({ runId: prevRunId })

await run.resume({
  step: 'step-id',
  resumeData: { key: 'value' },
})

When a .foreach() step suspends across multiple iterations, pass forEachIndex (zero-based. 0 targets the first iteration) to resume one iteration at a time. Iterations you don't target remain suspended.

await run.resume({
  step: 'approve',
  resumeData: { ok: true },
  forEachIndex: 1, // resumes the second iteration
})

forEachIndex is also supported by resumeAsync() and resumeStream().

cancel()

Cancel a running workflow:

const run = await workflow.createRun({ runId: existingRunId })

const result = await run.cancel()
// Returns: { message: 'Workflow run canceled' }

This method stops any running steps and prevents subsequent steps from executing. Steps that check the abortSignal parameter can respond to cancellation by cleaning up resources (timeouts, network requests, etc.).

See the Run.cancel() reference for detailed information about how cancellation works and how to write steps that respond to cancellation.

stream()

Stream workflow execution for real-time updates:

const run = await workflow.createRun()

const stream = await run.stream({
  inputData: {
    city: 'New York',
  },
})

for await (const chunk of stream) {
  console.log(JSON.stringify(chunk, null, 2))
}

runById()

Get the execution result for a workflow run:

const result = await workflow.runById(runId)

// Or with options for performance optimization:
const result = await workflow.runById(runId, {
  fields: ['status', 'result'], // Only fetch specific fields
  withNestedWorkflows: false, // Skip expensive nested workflow data
  requestContext: { userId: 'user-123' }, // Optional request context
})

Run result format

A workflow run result yields the following:

runId:

string

Unique identifier for this workflow run instance

workflowName:

string

Name of the workflow this run belongs to

resourceId?:

string

Resource ID associated with the run, if one was set

createdAt:

string

ISO 8601 timestamp of when the run was created

updatedAt:

string

ISO 8601 timestamp of when the run was last updated

status:

WorkflowRunStatus

Current status of the run, such as 'running', 'success', 'failed', 'suspended', 'waiting', or 'canceled'

payload?:

Record<string, any>

Input data the run was started with

result?:

Record<string, any>

Final output of the workflow when the run succeeded

error?:

SerializedError

Error details when the run failed

steps?:

Record<string, WorkflowStateStepResult>

Per-step results, including status, output, payload, and timing

Use the fields option to limit which optional fields are returned. Metadata fields and status are always included.

Dynamic workflows

beta

Breaking changes may occur without a major version bump until the API is stable.

Dynamic workflows are workflow definitions expressed as JSON. The server persists each definition and registers it as a runnable workflow. See Dynamic workflows for the definition format.

listDynamicWorkflows()

List dynamic workflow definitions, optionally filtered by status ('active' | 'archived') and authorId:

const { definitions, total } = await mastraClient.listDynamicWorkflows({
  status: 'active',
})

upsertDynamicWorkflow()

Create or replace a dynamic workflow definition. The server validates the definition, persists it, and live-registers it for execution:

const stored = await mastraClient.upsertDynamicWorkflow({
  id: 'greeting-workflow',
  description: 'Returns a greeting for the supplied name',
  inputSchema: {
    type: 'object',
    properties: { name: { type: 'string' } },
    required: ['name'],
  },
  outputSchema: {
    type: 'object',
    properties: { message: { type: 'string' } },
    required: ['message'],
  },
  graph: [
    {
      type: 'mapping',
      id: 'create-greeting',
      mapConfig: JSON.stringify({
        message: { template: 'Hello, ${initData.name}!' },
      }),
    },
  ],
})

When the root definition nests helper workflows that don't exist yet, pass them in the same request through dependencies. The server validates and registers the bundle as a unit and echoes the helper ids back as dependencyIds:

const stored = await mastraClient.upsertDynamicWorkflow({
  id: 'root-workflow',
  // ...schemas and graph referencing 'helper-workflow'...
  dependencies: [helperDefinition],
})

console.log(stored.dependencyIds) // ['helper-workflow']

getDynamicWorkflow()

Get a dynamic workflow instance for definition management. To execute a dynamic workflow, use getWorkflow(id).createRun() like any other workflow:

const dynamicWorkflow = mastraClient.getDynamicWorkflow('greeting-workflow')

dynamicWorkflow.details()

Retrieve the persisted definition, including schemas, graph, status, and timestamps:

const definition = await dynamicWorkflow.details()

dynamicWorkflow.delete()

Delete the stored definition and unregister the live workflow:

await dynamicWorkflow.delete()

Executing a dynamic workflow

Once registered, a dynamic workflow runs through the ordinary workflow API:

const workflow = mastraClient.getWorkflow('greeting-workflow')
const run = await workflow.createRun()
const result = await run.startAsync({ inputData: { name: 'Ada' } })

Schedules

Schedules are declared in code via the schedule field on createWorkflow. The client SDK exposes read and operational methods for managing workflow schedules at runtime. See Scheduled workflows.

createSchedule()

Create a workflow schedule by passing workflowId.

const schedule = await mastraClient.createSchedule({
  workflowId: 'daily-report',
  cron: '0 9 * * *',
  inputData: { reportType: 'summary' },
})

listSchedules()

List workflow schedules, optionally filtered by workflow ID or status.

const schedules = await mastraClient.listSchedules({
  workflowId: 'daily-report',
  status: 'active',
})

getSchedule()

Fetch a single workflow schedule by ID.

const schedule = await mastraClient.getSchedule('daily-report')

updateSchedule()

Update a workflow schedule.

const updated = await mastraClient.updateSchedule('daily-report', {
  cron: '0 10 * * *',
  inputData: { reportType: 'summary' },
})

deleteSchedule()

Delete a workflow schedule.

await mastraClient.deleteSchedule('daily-report')

runSchedule()

Fire a workflow schedule once immediately without changing its cron cadence.

const run = await mastraClient.runSchedule('daily-report')

pauseSchedule()

Pause a schedule so the scheduler stops firing it. Returns the updated schedule.

await mastraClient.pauseSchedule('daily-report')

resumeSchedule()

Resume a paused schedule. The next fire time is recomputed from now, so a long-paused schedule doesn't fire a backlog. Returns the updated schedule.

await mastraClient.resumeSchedule('daily-report')

listScheduleTriggers()

List the trigger history for a workflow schedule, including the joined run summary for each fire.

const { triggers } = await mastraClient.listScheduleTriggers('daily-report', {
  limit: 50,
})