Workflow API

Workflows API는 Mastra에서 자동화된 Workflow와 상호 작용하고 실행하는 방법을 제공합니다.

모든 Workflow 가져오기

사용 가능한 모든 Workflow 목록을 검색합니다.

const workflows = await mastraClient.listWorkflows()

Workflow 실행 횟수 가져오기

Workflow별 개수 검색running and suspended 단일 요청에서 실행됩니다. 개수는 서버에서 계산되며 워크플로의 레지스트리 키를 기준으로 구성됩니다. 이 키는 Mastra 구성에 워크플로를 등록할 때 사용되는 키로, 워크플로 자체의 id:

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

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

서버는 요청 사이에 몇 초 동안 개수를 캐시할 수 있습니다. 이 끝점 이전의 서버는 다음으로 응답합니다.404 Not Found — 클라이언트가 이전 배포와 통신할 수 있을 때 오류를 처리합니다.

특정 Workflow 작업

해당 ID로 특정 Workflow의 인스턴스를 가져옵니다.

src/mastra/workflows/test-workflow.ts

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

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

Workflow 방법

details()

Workflow에 대한 자세한 정보를 검색합니다.

const details = await workflow.details()

createRun()

새 Workflow 실행 인스턴스를 만듭니다.

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)와 연결합니다. 이 값은 실행과 함께 유지되며 나중에 실행을 필터링하고 쿼리하는 데 사용할 수 있습니다.

startAsync()

Workflow 실행을 시작하고 완료될 때까지 기다린 후 전체 결과를 Workflow 출력으로 반환합니다.

const run = await workflow.createRun()

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

합격하실 수도 있습니다initialState 을 사용하여 워크플로 상태의 시작 값을 설정합니다:

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

그만큼initialState 객체는 워크플로의 에 정의된 구조와 일치해야 합니다. stateSchema. See Workflow State for more details.

실행을 특정 리소스와 연결하려면 다음을 전달하세요.resourceId to createRun():

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

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

start()

완료를 기다리지 않고 Workflow 실행을 시작합니다(Fire-and-forget). 성공 메시지와 함께 즉시 반환됩니다. 사용runById() 을 워크플로 인스턴스에서 사용하여 나중에 결과를 확인합니다:

const run = await workflow.createRun()

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

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

이는 실행을 시작하고 나중에 결과를 확인하려는 장기 실행 Workflow에 유용합니다.

resumeAsync()

일시중단된 Workflow 단계를 재개하고 전체 결과를 기다립니다.

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

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

resume()

완료될 때까지 기다리지 않고 일시 중단된 Workflow 단계를 재개합니다.

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

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

언제.foreach() step suspends across multiple iterations, pass forEachIndex (zero-based. 0 은 첫 번째 반복을 대상으로 함)을 사용하여 한 번에 하나의 반복을 재개합니다. 대상으로 지정하지 않은 반복은 일시 중단된 상태로 유지됩니다.

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

forEachIndex에서도 지원됩니다resumeAsync() and resumeStream().

cancel()

실행 중인 Workflow를 취소합니다.

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

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

이 메서드는 실행 중인 모든 단계를 중지하고 후속 단계가 실행되지 않도록 합니다. 확인하는 단계abortSignal 매개변수를 사용하면 리소스(시간 제한, 네트워크 요청 등)를 정리하여 취소에 대응할 수 있습니다.

참조Run.cancel() 참조에서 취소의 작동 방식과 취소에 대응하는 단계를 작성하는 방법에 대한 자세한 정보를 확인하세요.

stream()

실시간 업데이트를 위한 스트림 Workflow 실행:

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

Workflow 실행에 대한 실행 결과를 가져옵니다.

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

실행 결과 형식

Workflow 실행 결과는 다음과 같습니다.

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)

동적 Workflow

:::실험적

동적 Workflow는 베타 버전입니다. API가 안정될 때까지 주요 버전 변경 없이 주요 변경 사항이 발생할 수 있습니다.

:::

동적 Workflow는 JSON으로 표현된 Workflow 정의입니다. 서버는 각 정의를 유지하고 이를 실행 가능한 Workflow로 등록합니다. 보다Dynamic workflows for the definition format.

listDynamicWorkflows()

선택적으로 다음을 기준으로 필터링하여 동적 Workflow 정의를 나열합니다.status ('active' | 'archived') and authorId:

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

upsertDynamicWorkflow()

동적 Workflow 정의를 만들거나 바꿉니다. 서버는 정의를 검증하고 유지하며 실행을 위해 실시간 등록합니다.

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}!' },
      }),
    },
  ],
})

루트 정의가 아직 존재하지 않는 도우미 Workflow를 중첩하는 경우 이를 통해 동일한 요청으로 전달합니다.dependencies. 서버는 번들을 하나의 단위로 검증하고 등록한 다음 헬퍼 ID를 로 다시 반환합니다. dependencyIds:

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

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

getDynamicWorkflow()

정의 관리를 위한 동적 Workflow 인스턴스를 가져옵니다. 동적 Workflow를 실행하려면 다음을 사용하세요.getWorkflow(id).createRun() like any other workflow:

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

dynamicWorkflow.details()

스키마, 그래프, 상태, 타임스탬프를 포함하여 지속형 정의를 검색합니다.

const definition = await dynamicWorkflow.details()

dynamicWorkflow.delete()

저장된 정의를 삭제하고 라이브 Workflow 등록을 취소합니다.

await dynamicWorkflow.delete()

동적 Workflow 실행

등록되면 동적 Workflow가 일반 Workflow API를 통해 실행됩니다.

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

일정

일정은 다음을 통해 코드로 선언됩니다.schedule field on createWorkflow. 클라이언트 SDK는 런타임에 워크플로 일정을 관리하기 위한 조회 및 운영 메서드를 제공합니다. 자세한 내용은 를 참조하세요. Scheduled workflows.

createSchedule()

전달하여 Workflow 일정 만들기workflowId.

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

listSchedules()

선택적으로 Workflow ID 또는 상태별로 필터링하여 Workflow 일정을 나열합니다.

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

getSchedule()

ID별로 단일 Workflow 일정을 가져옵니다.

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

updateSchedule()

Workflow 일정을 업데이트합니다.

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

deleteSchedule()

Workflow 일정을 삭제합니다.

await mastraClient.deleteSchedule('daily-report')

runSchedule()

크론 주기를 변경하지 않고 즉시 Workflow 일정을 한 번 실행합니다.

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

pauseSchedule()

스케줄러가 실행을 중지하도록 일정을 일시 중지합니다. 업데이트된 일정을 반환합니다.

await mastraClient.pauseSchedule('daily-report')

resumeSchedule()

일시 중지된 일정을 재개합니다. 다음 실행 시간은 지금부터 다시 계산되므로 오랫동안 일시 중지된 일정은 백로그를 실행하지 않습니다. 업데이트된 일정을 반환합니다.

await mastraClient.resumeSchedule('daily-report')

listScheduleTriggers()

각 실행에 대한 결합된 실행 요약을 포함하여 Workflow 일정에 대한 트리거 기록을 나열합니다.

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