Guide · Workflow Orchestration

MCP Tools for Dagster — asset materializations, run status, and GraphQL API

Dagster's primary external interface is a GraphQL API (served at /graphql by the Dagster webserver) rather than a REST API. Every action you can take in the Dagster UI — launching runs, querying asset materializations, checking sensor state, reading run logs — is available via GraphQL mutations and queries. When you build MCP tools for Dagster, three design distinctions matter: asset-centric vs op-centric model (Dagster's asset graph is the primary organizing concept; a "job" in Dagster may produce assets, and assets have their own materialization history independent of any specific job run), run status vs asset freshness (a job completing successfully does not guarantee all assets are fresh — partition-based assets have per-partition materialization state), and daemon health vs webserver health (the scheduler daemon and sensor daemon are separate processes from the webserver; a healthy webserver with a dead scheduler daemon means no auto-triggered runs).

TL;DR

Use GraphQL POST /graphql with queries for reading state and mutations for triggering actions. Run status is queried via the runOrError query with a run ID; run status values are QUEUED, STARTED, SUCCESS, FAILURE, CANCELING, CANCELED. Asset materialization freshness is separate from run status — query assetNodes with assetKey and check latestMaterializationByPartition or latestMaterializationDate. Trigger jobs via the launchRun mutation. Daemon health is available via the instance query's daemonHealth field.

GraphQL client setup and authentication

Dagster's GraphQL API does not use a standard REST authentication scheme. Dagster+ (cloud) uses a user token passed as a Dagster-Cloud-Api-Token header. Self-hosted Dagster OSS has no built-in authentication — network-level access control (VPN, firewall) is the typical security model. The GraphQL endpoint is at {webserverUrl}/graphql. All queries and mutations are sent as POST requests with a JSON body containing query and optionally variables.

class DagsterClient {
  constructor({ baseUrl, apiToken }) {
    this.graphqlUrl = `${baseUrl.replace(/\/$/, '')}/graphql`;
    this.apiToken = apiToken; // null for OSS self-hosted
  }

  async query(gql, variables = {}) {
    const headers = {
      'Content-Type': 'application/json',
      'Accept': 'application/json',
    };
    if (this.apiToken) {
      headers['Dagster-Cloud-Api-Token'] = this.apiToken;
    }

    const response = await fetch(this.graphqlUrl, {
      method: 'POST',
      headers,
      body: JSON.stringify({ query: gql, variables }),
    });

    if (!response.ok) throw new Error(`Dagster API HTTP error ${response.status}`);

    const result = await response.json();

    // GraphQL errors are returned in the 'errors' field with HTTP 200
    if (result.errors?.length) {
      const messages = result.errors.map((e) => e.message).join('; ');
      throw new Error(`Dagster GraphQL error: ${messages}`);
    }

    return result.data;
  }
}

Run status and run log retrieval

Dagster run status has six values mapped to the RunStatus GraphQL enum: QUEUED (submitted but not yet assigned to a launcher), STARTED (executing), SUCCESS (completed without errors), FAILURE (completed with errors), CANCELING (cancellation requested, in progress), and CANCELED (cancelled successfully). Only SUCCESS, FAILURE, and CANCELED are terminal. A run in CANCELING state that persists longer than expected usually means the code server or daemon is unresponsive.

Run logs are stored as structured events accessible via the logsForRun query. Dagster uses a typed event model — log entries have types like ENGINE_EVENT, STEP_FAILURE, STEP_SUCCESS, ASSET_MATERIALIZATION, and PIPELINE_FAILURE. For failure analysis, filter for STEP_FAILURE and PIPELINE_FAILURE events and extract the structured error message from the error field rather than parsing freeform log text.

const RUN_STATUS_QUERY = `
  query GetRunStatus($runId: ID!) {
    runOrError(runId: $runId) {
      __typename
      ... on Run {
        id
        status
        jobName
        repositoryOrigin {
          repositoryName
          repositoryLocationName
        }
        startTime
        endTime
        stats {
          ... on RunStatsSnapshot {
            startTime
            endTime
            stepsFailed
            stepsSucceeded
          }
        }
        tags { key value }
      }
      ... on RunNotFoundError {
        message
      }
    }
  }
`;

const TERMINAL_STATUSES = new Set(['SUCCESS', 'FAILURE', 'CANCELED']);

async function getRunStatus(client, runId) {
  const data = await client.query(RUN_STATUS_QUERY, { runId });
  const run = data.runOrError;

  if (run.__typename === 'RunNotFoundError') {
    throw new Error(`Run not found: ${runId}`);
  }

  return {
    id: run.id,
    status: run.status,
    jobName: run.jobName,
    repositoryName: run.repositoryOrigin?.repositoryName,
    repositoryLocation: run.repositoryOrigin?.repositoryLocationName,
    startTime: run.startTime ? new Date(run.startTime * 1000) : null, // Dagster uses Unix timestamps
    endTime: run.endTime ? new Date(run.endTime * 1000) : null,
    isTerminal: TERMINAL_STATUSES.has(run.status),
    stepsFailed: run.stats?.stepsFailed ?? 0,
    stepsSucceeded: run.stats?.stepsSucceeded ?? 0,
    tags: Object.fromEntries((run.tags ?? []).map((t) => [t.key, t.value])),
  };
}

// Get failure events for a failed run
const RUN_FAILURE_EVENTS_QUERY = `
  query GetRunFailureEvents($runId: ID!, $ofTypes: [DagsterEventType!]) {
    logsForRun(runId: $runId, ofTypes: $ofTypes) {
      ... on EventConnection {
        events {
          ... on StepFailureEvent {
            timestamp
            stepKey
            message
            error {
              message
              className
              stack
            }
          }
          ... on PipelineFailureEvent {
            timestamp
            message
            error {
              message
              className
            }
          }
        }
      }
    }
  }
`;

async function getRunFailureEvents(client, runId) {
  const data = await client.query(RUN_FAILURE_EVENTS_QUERY, {
    runId,
    ofTypes: ['STEP_FAILURE', 'PIPELINE_FAILURE'],
  });

  const events = data.logsForRun?.events ?? [];
  return events.map((event) => ({
    type: event.__typename,
    timestamp: event.timestamp,
    stepKey: event.stepKey ?? null,
    message: event.message,
    errorMessage: event.error?.message ?? null,
    errorClass: event.error?.className ?? null,
  }));
}

Asset materialization freshness

Assets in Dagster have materialization history stored independently of run history. An asset can be materialized by multiple different jobs, by manual triggers, or by sensors. The key metric for asset health is freshness: when was this asset last materialized successfully, and is that recent enough for its consumers?

Dagster supports freshness policies that define an expected materialization lag. If you use freshness policies, the staleStatus field on an asset node (FRESH, STALE, or MISSING) is computed automatically. Without freshness policies, you must compute freshness yourself by comparing latestMaterializationDate against your SLA. For partition-based assets, freshness is per-partition — an asset with 100 date partitions might have 99 fresh partitions and one stale one from a failed backfill.

const ASSET_FRESHNESS_QUERY = `
  query GetAssetFreshness($assetKey: AssetKeyInput!) {
    assetNodeOrError(assetKey: $assetKey) {
      ... on AssetNode {
        assetKey { path }
        description
        partitionDefinition {
          description
          type
        }
        latestMaterializationByPartition(partitions: null) {
          partition
          timestamp
          runId
          stepKey
        }
        staleStatus
        freshnessPolicyWithoutSourceAssets {
          maximumLagMinutes
          cronSchedule
        }
      }
      ... on AssetNotFoundError {
        message
      }
    }
  }
`;

async function checkAssetFreshness(client, assetKeyPath, maxStalenessMs = 86_400_000) {
  // assetKeyPath is an array of path components: ['my_schema', 'my_table']
  // For simple assets: ['my_asset_name']
  const data = await client.query(ASSET_FRESHNESS_QUERY, {
    assetKey: { path: assetKeyPath },
  });

  const node = data.assetNodeOrError;
  if (node.__typename === 'AssetNotFoundError') {
    throw new Error(`Asset not found: ${assetKeyPath.join('/')}`);
  }

  const materializations = node.latestMaterializationByPartition ?? [];
  const now = Date.now();

  if (materializations.length === 0) {
    return { assetKey: assetKeyPath.join('/'), status: 'NEVER_MATERIALIZED', staleStatus: node.staleStatus };
  }

  // For unpartitioned assets: one entry with partition=null
  // For partitioned assets: one entry per partition
  const staleMaterializations = materializations.filter((m) => {
    if (!m?.timestamp) return true; // No materialization for this partition
    return now - Number(m.timestamp) * 1000 > maxStalenessMs;
  });

  const mostRecent = materializations
    .filter((m) => m?.timestamp)
    .sort((a, b) => Number(b.timestamp) - Number(a.timestamp))[0];

  return {
    assetKey: assetKeyPath.join('/'),
    staleStatus: node.staleStatus, // Dagster's own freshness policy evaluation
    isPartitioned: node.partitionDefinition !== null,
    totalPartitions: materializations.length,
    stalePartitionCount: staleMaterializations.length,
    lastMaterializationTime: mostRecent ? new Date(Number(mostRecent.timestamp) * 1000) : null,
    lastRunId: mostRecent?.runId ?? null,
    customFreshStatus: staleMaterializations.length === 0 ? 'FRESH' : 'STALE',
  };
}

Launching runs and triggering partitioned backfills

The launchRun GraphQL mutation triggers a new run for a job within a repository. You need the repository location name, repository name, job name, and optionally run config overrides. For partitioned jobs, the launchPartitionBackfill mutation triggers runs for a range of partition keys. Backfills can run all partitions serially or in parallel, controlled by the parallelism parameter.

const LAUNCH_RUN_MUTATION = `
  mutation LaunchRun($executionParams: ExecutionParams!) {
    launchRun(executionParams: $executionParams) {
      ... on LaunchRunSuccess {
        run { id status jobName }
      }
      ... on InvalidSubsetError {
        message
      }
      ... on RunConfigValidationInvalid {
        message
        errors { message reason }
      }
      ... on PythonError {
        message
        stack
      }
    }
  }
`;

async function launchRun(client, repoLocation, repoName, jobName, runConfig = {}, tags = {}) {
  const data = await client.query(LAUNCH_RUN_MUTATION, {
    executionParams: {
      selector: {
        repositoryLocationName: repoLocation,
        repositoryName: repoName,
        jobName,
      },
      runConfigData: runConfig,
      executionMetadata: {
        tags: Object.entries(tags).map(([key, value]) => ({ key, value })),
      },
    },
  });

  const result = data.launchRun;
  if (result.__typename !== 'LaunchRunSuccess') {
    throw new Error(`Launch failed: ${result.message ?? JSON.stringify(result)}`);
  }

  return { runId: result.run.id, status: result.run.status };
}

Daemon health and AliveMCP integration

Dagster uses separate daemon processes for scheduling, sensors, and backfills. The scheduler daemon triggers scheduled runs; the sensor daemon evaluates sensor conditions; the backfill daemon manages partition backfill executions. If any daemon is unhealthy, the corresponding automated trigger type stops working — scheduled jobs won't run, sensors won't fire, backfills stall.

Query daemon health via the instance GraphQL query. Each daemon entry has a healthy flag and a lastHeartbeatTime. Register AliveMCP on your Dagster health check endpoint with a 60-second probe interval. Alert on any daemon with healthy === false or a heartbeat older than 120 seconds. The webserver HTTP response does not reflect daemon health — a healthy webserver with dead daemons is a common and dangerous false positive.

const DAEMON_HEALTH_QUERY = `
  query DaemonHealth {
    instance {
      daemonHealth {
        allDaemonStatuses {
          daemonType
          healthy
          required
          lastHeartbeatTime
          lastHeartbeatErrors {
            message
            className
          }
        }
      }
    }
  }
`;

async function checkDagsterHealth(client) {
  let daemonStatuses;
  try {
    const data = await client.query(DAEMON_HEALTH_QUERY);
    daemonStatuses = data.instance.daemonHealth.allDaemonStatuses;
  } catch (err) {
    return { webserverReachable: false, error: err.message };
  }

  const now = Date.now();
  const HEARTBEAT_STALE_MS = 120_000;

  const daemonHealth = daemonStatuses.map((d) => {
    const heartbeatAge = d.lastHeartbeatTime
      ? now - new Date(d.lastHeartbeatTime).getTime()
      : Infinity;

    return {
      type: d.daemonType, // 'SCHEDULER' | 'SENSOR' | 'BACKFILL' | 'AUTO_MATERIALIZE'
      required: d.required,
      healthy: d.healthy,
      heartbeatAgeMs: heartbeatAge,
      heartbeatStale: heartbeatAge > HEARTBEAT_STALE_MS,
      lastErrors: d.lastHeartbeatErrors?.map((e) => e.message) ?? [],
    };
  });

  const unhealthyDaemons = daemonHealth.filter(
    (d) => d.required && (!d.healthy || d.heartbeatStale)
  );

  return {
    webserverReachable: true,
    daemons: daemonHealth,
    unhealthyDaemons,
    allDaemonsHealthy: unhealthyDaemons.length === 0,
  };
}