Temporal Workflow

Temporal 是一個耐久執行平台，用於協調長時間運行且具容錯能力的 Workflow。@mastra/temporal 套件讓你以標準 Mastra API 編寫 Workflow，並在 Temporal 叢集上運行。

注意

@mastra/temporal 尚處於實驗階段，未適合在正式環境使用。API 可能會隨版本變更。最新狀態請參閱套件 README。

Temporal 如何配合 Mastra 運作

使用 createWorkflow() 和 createStep() 編寫的 Mastra Workflow，會對應至 Temporal 的 Workflow 和 activity 模型。Temporal worker 的 MastraPlugin 會在 bundle 建置期間編譯你的 Mastra 進入點檔案：

• 每個 createStep() handler 都會擷取為 Temporal activity。
• 每個 createWorkflow() 都會重寫為調用這些 activity 的 Temporal Workflow。
• plugin 會自動向 worker 註冊所產生的 activity 和 Workflow。

當透過 mastra.getWorkflow(...).createRun().start(...) 開始運行時，Mastra client 會將控制權交給 Temporal。之後由 Temporal 在 worker 上驅動耐久執行、重試及狀態持久化。

設定

安裝所需套件：

npm

npm install @mastra/temporal@latest @temporalio/client @temporalio/worker @temporalio/envconfig

pnpm

pnpm add @mastra/temporal@latest @temporalio/client @temporalio/worker @temporalio/envconfig

Yarn

yarn add @mastra/temporal@latest @temporalio/client @temporalio/worker @temporalio/envconfig

Bun

bun add @mastra/temporal@latest @temporalio/client @temporalio/worker @temporalio/envconfig

你亦需要能夠存取 Temporal 叢集。進行本機開發時，可以使用 Docker 運行叢集（請參閱在本機運行）。

建立由 Temporal 支援的 Workflow

本指南會逐步說明如何使用 Temporal 和 Mastra 建立 Workflow，並以遞增數值的計數器應用程式作示範。

初始化 Temporal

初始化 Temporal 整合，以取得與 Mastra 相容的 Workflow 輔助函數。createWorkflow() 和 createStep() 函數會綁定至 Temporal client 及 task queue。

src/mastra/temporal/index.ts

import { init } from '@mastra/temporal'
import { Client, Connection } from '@temporalio/client'
import { loadClientConnectConfig } from '@temporalio/envconfig'

const config = loadClientConnectConfig()
const connection = await Connection.connect(config.connectionOptions)
const client = new Client({ connection })

export const { createWorkflow, createStep } = init({
  client,
  taskQueue: 'mastra',
})

loadClientConnectConfig() 會讀取標準 Temporal 環境變數，例如 TEMPORAL_ADDRESS、TEMPORAL_NAMESPACE 及 mTLS 設定。完整清單請參閱 Temporal envconfig 文件。

建立步驟

定義組成 Workflow 的個別步驟。每個步驟都會成為 Temporal activity。

src/mastra/workflows/index.ts

import { z } from 'zod'
import { createWorkflow, createStep } from '../temporal'

const incrementStep = createStep({
  id: 'increment',
  inputSchema: z.object({
    value: z.number(),
  }),
  outputSchema: z.object({
    value: z.number(),
  }),
  execute: async ({ inputData }) => {
    return { value: inputData.value + 1 }
  },
})

建立 Workflow

將步驟組合成 Workflow。Workflow 的 id 必須是靜態字串字面值，建置期間的轉換器才能由此得出其 Temporal export 名稱。

src/mastra/workflows/index.ts

const workflow = createWorkflow({
  id: 'increment-workflow',
  steps: [incrementStep],
  inputSchema: z.object({
    value: z.number(),
  }),
  outputSchema: z.object({
    value: z.number(),
  }),
}).then(incrementStep)

workflow.commit()

export { workflow as incrementWorkflow }

配置 Mastra 實例

向 Mastra 註冊 Workflow。執行程序由 Temporal worker 驅動。

src/mastra/index.ts

import { Mastra } from '@mastra/core'
import { PinoLogger } from '@mastra/loggers'
import { incrementWorkflow } from './workflows'

export const mastra = new Mastra({
  workflows: { incrementWorkflow },
  logger: new PinoLogger({ name: 'Mastra', level: 'info' }),
})

運行 worker

worker 是長時間運行的 Node.js 程序，會輪詢 Temporal task queue。安裝 MastraPlugin，並將其 src 選項指向註冊 Workflow 的 Mastra 進入點檔案。

src/mastra/worker.ts

import { MastraPlugin } from '@mastra/temporal/worker'
import { NativeConnection, Worker } from '@temporalio/worker'

const connection = await NativeConnection.connect({
  address: 'localhost:7233',
})

const mastraPlugin = new MastraPlugin()

await mastraPlugin.prebuild({
  entryFile: import.meta.resolve('./index.ts'),
})

const worker = await Worker.create({
  connection,
  namespace: 'default',
  taskQueue: 'mastra',
  plugins: [mastraPlugin],
})

await worker.run()

MastraPlugin 會將進入點檔案重寫成只包含 Workflow 的 bundle，並把步驟 handler 接入為 Temporal activity。你不需要手動將 activities 或 workflowsPath 傳入 Worker.create()。

運行 Workflow

在本機運行

1. 啟動本機 Temporal 伺服器。最簡單的方法是使用 temporalio/auto-setup Docker image：

   docker run --rm -p 7233:7233 -p 8080:8080 temporalio/auto-setup:latest

2. 在 http://localhost:8080 開啟 Temporal UI，以檢視 namespace、Workflow 及 activity。

3. 在新的終端機中運行以下命令，啟動 worker：

   npx tsx src/mastra/worker.ts

4. 從 script 或任何匯入 Mastra 實例的程序觸發 Workflow 運行：

   scripts/run.ts

   import { mastra } from '../src/mastra'
   
   const run = await mastra.getWorkflow('incrementWorkflow').createRun()
   const result = await run.start({ inputData: { value: 5 } })
   
   console.log(result)

5. 在 Temporal UI 的 Workflows 下監察執行情況，查看每項 activity 的逐步進度及重試記錄。

在正式環境運行

在正式環境中，請使用 Temporal Cloud 或自行託管的 Temporal 叢集。透過由 @temporalio/envconfig 讀取的環境變數，配置 client 和 worker 連線：

.env

TEMPORAL_ADDRESS=your-namespace.tmprl.cloud:7233
TEMPORAL_NAMESPACE=your-namespace
TEMPORAL_API_KEY=your-api-key

有關 mTLS 和 API 金鑰選項，請參閱 Temporal Cloud 連線文件。

注意

Temporal worker 必須以長時間運行的程序方式執行。請勿將它部署至執行時間限制較短的 serverless 平台，例如 AWS Lambda 或 Vercel functions。請使用容器、VM，或適合 worker 的平台，例如 Fly.io、Railway 或 Kubernetes。

配置選項

taskQueue

必填。識別 worker 所輪詢的 Temporal task queue。傳入 init()（client 用其開始運行）和 Worker.create()（worker 用其接收運行）的值必須相同。

startToCloseTimeout

選填。設定單一 activity（步驟）在被 Temporal 取消並套用重試政策前，允許運行的最長時間。預設為 1 minute。

src/mastra/temporal/index.ts

export const { createWorkflow, createStep } = init({
  client,
  taskQueue: 'mastra',
  startToCloseTimeout: '5 minutes',
})

限制及注意事項

• Workflow ID 必須是靜態字串字面值。建置期間的轉換器會讀取字面值，以得出 Temporal Workflow export 名稱。
• 系統會從 createStep() handler 自動產生 activity。請勿透過 Worker.create({ activities }) 傳入。

相關內容

• Workflow runner
• Workflow 概覽
• Temporal 文件