工作流 API

🌐 Workflows API

工作流 API 提供了在 Mastra 中交互和执行自动化工作流的方法。

🌐 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();

使用特定工作流程

🌐 Working with a Specific Workflow

通过其 ID 获取特定工作流的实例：

🌐 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"
});

resourceId 参数将工作流运行与特定资源（例如用户 ID、租户 ID）关联。该值会随运行一同保存，之后可用于过滤和查询运行记录。

🌐 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 the full result:

const run = await workflow.createRun();

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

你还可以传递 initialState 来设置工作流状态的起始值：

🌐 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: [],
  },
});

initialState 对象应与工作流的 stateSchema 中定义的结构相匹配。更多详细信息，请参见 工作流状态。

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

要将运行与特定资源关联，请将 resourceId 传递给 createRun()：

🌐 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:

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" },
});

cancel()

取消正在运行的工作流：

🌐 Cancel a running workflow:

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

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

此方法会停止正在运行的步骤，并阻止后续步骤的执行。检查 abortSignal 参数的步骤可以通过清理资源（超时、网络请求等）来响应取消操作。

🌐 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.).

请参阅 Run.cancel() 参考以获取有关取消如何工作以及如何编写响应取消的步骤的详细信息。

🌐 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
});

运行结果格式

工作流运行结果如下：

🌐 A workflow run result yields the following:

runId:

string

Unique identifier for this workflow run instance

eventTimestamp:

Date

The timestamp of the event

payload:

object

Contains currentStep (id, status, output, payload) and workflowState (status, steps record)