Guide · Workflow Orchestration

MCP Tools for Prefect — flow run states, work pool health, and deployment triggers

Prefect exposes its orchestration state through a REST API served by the Prefect server (self-hosted) or Prefect Cloud. When you build MCP tools for Prefect — triggering deployments, monitoring flow run state, checking if workers are healthy, or detecting hung runs — three model distinctions determine correctness: flow vs deployment vs flow run (a flow is Python code; a deployment is a versioned configuration that tells Prefect how to run that flow; a flow run is a single execution instance), Prefect's three-tier state model (each state has a type, a name, and a scheduled flag — the same state type can have multiple names with different semantic meanings), and Crashed vs Failed vs Cancelling (Crashed means the worker process died unexpectedly; Failed means the flow code raised an exception; Cancelling is a non-terminal limbo state that persists if no worker is available to complete the cancellation).

TL;DR

Trigger deployments via POST /api/deployments/{id}/create_flow_run with a JSON body. Poll flow run state via GET /api/flow_runs/{id} — terminal state types are COMPLETED, FAILED, CRASHED, and CANCELLED. Non-terminal types are SCHEDULED, PENDING, RUNNING, and PAUSED. CANCELLING is a semi-terminal state that requires a live worker to complete — if no workers are running, flow runs can be stuck in CANCELLING indefinitely. Check work pool health via GET /api/work_pools/{name} and worker heartbeats via GET /api/work_queues/{id}/status.

Prefect API authentication: Server vs Cloud

Prefect self-hosted Server (run via prefect server start) does not require authentication by default. The API is available at http://localhost:4200/api without credentials, making it easy to integrate in development but requiring your own network-level security in production. Prefect Cloud (hosted at api.prefect.cloud) requires an API key passed in the Authorization: Bearer {key} header. You also need to include the workspace account ID and workspace ID in the API base URL or as request headers.

The PREFECT_API_URL environment variable controls which server Prefect clients connect to. For MCP tools, read this variable from the deployment environment to determine the correct endpoint rather than hardcoding it.

class PrefectClient {
  constructor({ apiUrl, apiKey }) {
    // apiUrl examples:
    //   Self-hosted: 'http://localhost:4200/api'
    //   Prefect Cloud: 'https://api.prefect.cloud/api/accounts/{accountId}/workspaces/{workspaceId}'
    this.apiUrl = apiUrl.replace(/\/$/, '');
    this.apiKey = apiKey; // null for self-hosted without auth
  }

  async request(method, path, body = null) {
    const headers = {
      'Content-Type': 'application/json',
      'Accept': 'application/json',
    };
    if (this.apiKey) headers['Authorization'] = `Bearer ${this.apiKey}`;

    const response = await fetch(`${this.apiUrl}${path}`, {
      method,
      headers,
      body: body ? JSON.stringify(body) : undefined,
    });

    if (response.status === 401) throw new Error('Prefect auth failed — check API key');
    if (response.status === 404) throw new Error(`Not found: ${path}`);
    if (!response.ok) {
      const text = await response.text();
      throw new Error(`Prefect API error ${response.status}: ${text}`);
    }

    if (response.status === 204) return null;
    return response.json();
  }

  // Check Prefect server health (no auth)
  async checkHealth() {
    const response = await fetch(`${this.apiUrl}/health`);
    return { ok: response.ok, status: response.status };
  }
}

Flow run state model: types, names, and the CANCELLING trap

Prefect's state model separates state type (a fixed enum) from state name (a human-readable string that can be customized). The canonical state types are: SCHEDULED (waiting for its scheduled start time or a slot in the work queue), PENDING (submitted to a worker but not yet executing), RUNNING (currently executing), COMPLETED (finished successfully), FAILED (the flow raised an exception), CRASHED (the worker process died — unhandled signal, OOM, infrastructure failure), CANCELLED (successfully cancelled), CANCELLING (cancellation requested but not yet complete), and PAUSED (flow called pause_flow_run() and is waiting for a resume signal).

The CANCELLING → CANCELLED transition requires a live worker. When you request cancellation via the API, Prefect sets the flow run to CANCELLING and sends a signal to the worker handling it. If the worker receives the signal, it interrupts the flow and sets the state to CANCELLED. If no worker is available — because the worker crashed or was restarted — the flow run stays in CANCELLING indefinitely. MCP tools must monitor for flow runs stuck in CANCELLING state and escalate to force-cancellation if the state persists beyond a reasonable window.

// State type constants
const TERMINAL_STATE_TYPES = new Set(['COMPLETED', 'FAILED', 'CRASHED', 'CANCELLED']);
const NON_TERMINAL_STATE_TYPES = new Set(['SCHEDULED', 'PENDING', 'RUNNING', 'PAUSED', 'CANCELLING']);

async function getFlowRun(client, flowRunId) {
  const run = await client.request('GET', `/flow_runs/${flowRunId}`);

  const stateType = run.state?.type;
  const stateName = run.state?.name;
  const stateTimestamp = run.state?.timestamp;

  return {
    id: run.id,
    name: run.name,
    flowId: run.flow_id,
    deploymentId: run.deployment_id,
    stateType,
    stateName,
    stateTimestamp,
    isTerminal: TERMINAL_STATE_TYPES.has(stateType),
    // Prefect surfaces the message in the state object for failures/crashes
    stateMessage: run.state?.message ?? null,
    startTime: run.start_time,
    endTime: run.end_time,
    totalRunTime: run.total_run_time, // seconds (only when terminal)
    parameters: run.parameters,
    workPoolName: run.work_pool_name,
    workQueueName: run.work_queue_name,
    // CANCELLING detection
    isStuckCancelling: stateType === 'CANCELLING'
      && stateTimestamp
      && Date.now() - new Date(stateTimestamp).getTime() > 300_000, // 5 min threshold
  };
}

// Poll flow run until terminal
async function waitForFlowRun(client, flowRunId, pollIntervalMs = 5_000, timeoutMs = 3_600_000) {
  const deadline = Date.now() + timeoutMs;

  while (Date.now() < deadline) {
    const run = await getFlowRun(client, flowRunId);
    if (run.isTerminal) return run;
    await new Promise((r) => setTimeout(r, pollIntervalMs));
  }

  throw new Error(`Flow run ${flowRunId} timed out after ${timeoutMs}ms`);
}

Triggering deployments and managing flow runs

To trigger a Prefect deployment, you need the deployment's UUID (not its name). You can look up a deployment by name via POST /api/deployments/get_by_name with the body {"name": "my-flow/my-deployment"}, where the format is {flow-name}/{deployment-name}. Once you have the ID, create a flow run via POST /api/deployments/{id}/create_flow_run. You can pass parameter overrides in the request body to customize individual runs without changing the deployment configuration.

async function getDeploymentByName(client, flowName, deploymentName) {
  // Prefect deployment name format: "{flow-name}/{deployment-name}"
  const fullName = `${flowName}/${deploymentName}`;
  const deployment = await client.request('POST', '/deployments/get_by_name', {
    name: fullName,
  });

  return {
    id: deployment.id,
    name: deployment.name,
    flowId: deployment.flow_id,
    isPaused: deployment.paused, // Paused deployments won't auto-schedule new runs
    workPoolName: deployment.work_pool_name,
    workQueueName: deployment.work_queue_name,
    schedules: deployment.schedules,
    parameters: deployment.parameters, // Default parameter values
  };
}

async function triggerDeployment(client, deploymentId, parameterOverrides = {}, tags = []) {
  const flowRun = await client.request(
    'POST',
    `/deployments/${deploymentId}/create_flow_run`,
    {
      parameters: parameterOverrides, // Merged with deployment's default parameters
      tags,
      // Optionally override the work queue:
      // work_queue_name: 'high-priority',
    }
  );

  return {
    flowRunId: flowRun.id,
    name: flowRun.name,
    stateType: flowRun.state?.type, // 'SCHEDULED' immediately after trigger
    stateName: flowRun.state?.name,
    expectedStartTime: flowRun.expected_start_time,
    deploymentId: flowRun.deployment_id,
  };
}

// Cancel a running flow run
async function cancelFlowRun(client, flowRunId) {
  // Sets state to CANCELLING — requires a live worker to complete to CANCELLED
  await client.request('DELETE', `/flow_runs/${flowRunId}/cancel`);
  return { flowRunId, status: 'cancellation_requested' };
}

// Force-cancel a flow run stuck in CANCELLING state
// This directly sets state to CANCELLED without waiting for a worker
async function forceCancel(client, flowRunId) {
  await client.request('POST', `/flow_runs/${flowRunId}/set_state`, {
    state: { type: 'CANCELLED', name: 'Cancelled', message: 'Force cancelled by MCP tool' },
    force: true,
  });
  return { flowRunId, status: 'force_cancelled' };
}

Work pool and worker health

Work pools are the bridge between Prefect's orchestration layer and the execution infrastructure. Each deployment is associated with a work pool; workers poll the pool for flow runs to execute. A work pool can be ready (workers are polling) or paused (no new runs will be dispatched). Work pool health is distinct from worker health — a paused work pool produces no flow run failures directly, but queued runs will pile up without executing.

Workers send heartbeats to Prefect on a configurable interval (default 30 seconds). If a worker's last heartbeat is older than the heartbeat timeout (default 60 seconds), it is marked offline. An offline worker means any SCHEDULED or PENDING flow runs in that work pool will not execute. This is the primary stuck-run root cause: flow run in SCHEDULED state with a work pool that has zero online workers.

async function checkWorkPoolHealth(client, workPoolName) {
  const pool = await client.request('GET', `/work_pools/${workPoolName}`);

  // List workers in this pool
  const workersResponse = await client.request('POST', `/work_pools/${workPoolName}/workers/filter`, {
    workers: { offset: 0, limit: 100 },
  });

  const workers = workersResponse.results ?? [];
  const now = Date.now();
  const HEARTBEAT_TIMEOUT_MS = 120_000; // 2 minutes = offline threshold

  const onlineWorkers = workers.filter((w) => {
    if (!w.last_heartbeat_time) return false;
    return now - new Date(w.last_heartbeat_time).getTime() < HEARTBEAT_TIMEOUT_MS;
  });

  return {
    workPoolName: pool.name,
    type: pool.type, // 'prefect-agent' | 'process' | 'kubernetes' | 'ecs' | 'cloud-run' etc.
    isPaused: pool.is_paused,
    workerCount: workers.length,
    onlineWorkerCount: onlineWorkers.length,
    offlineWorkers: workers
      .filter((w) => !onlineWorkers.includes(w))
      .map((w) => ({
        name: w.name,
        lastHeartbeat: w.last_heartbeat_time,
        heartbeatAgeSec: w.last_heartbeat_time
          ? (now - new Date(w.last_heartbeat_time).getTime()) / 1000
          : null,
      })),
    // Overall health: pool not paused AND at least one online worker
    healthy: !pool.is_paused && onlineWorkers.length > 0,
  };
}

// List flow runs that are stuck in SCHEDULED with no online workers
async function detectStuckScheduledRuns(client, workPoolName) {
  const poolHealth = await checkWorkPoolHealth(client, workPoolName);

  if (poolHealth.healthy) return { stuck: false, reason: 'Work pool has online workers' };

  // Get SCHEDULED and PENDING runs for this work pool
  const response = await client.request('POST', '/flow_runs/filter', {
    flow_runs: {
      state: { type: { any_: ['SCHEDULED', 'PENDING'] } },
      work_pool_name: { any_: [workPoolName] },
      offset: 0,
      limit: 100,
    },
  });

  return {
    stuck: true,
    reason: poolHealth.isPaused ? 'Work pool is paused' : 'No online workers',
    stuckRunCount: response.results?.length ?? 0,
    stuckRuns: response.results?.map((r) => ({
      id: r.id,
      name: r.name,
      stateType: r.state?.type,
      expectedStartTime: r.expected_start_time,
    })) ?? [],
  };
}

AliveMCP integration for Prefect health monitoring

Register Prefect health probes with AliveMCP at 60-second intervals. The health check should: (1) verify Prefect server or Cloud API is reachable via /api/health; (2) check all work pools for online workers; (3) count flow runs in SCHEDULED or PENDING state older than their expected start time; (4) count FAILED and CRASHED runs in the past hour for regression detection. Alert thresholds: zero online workers in any active work pool = immediate alert; CANCELLING runs older than 10 minutes = alert; FAILED/CRASHED spike above baseline = alert.

Do not alert on individual FAILED runs without context — some flows are expected to fail on bad input and retry. Alert on the rate of failure relative to your baseline, and on CRASHED runs specifically (which indicate infrastructure problems rather than logic errors).