Dagster

Orchestrate data pipelines and manage job runs with Dagster

使用说明

Connect to a Dagster instance to launch job runs, monitor run status, list available jobs across repositories, terminate or delete runs, reexecute failed runs, fetch run logs, and manage schedules and sensors. API token only required for Dagster+.

工具

`dagster_launch_run`

Launch a job run on a Dagster instance.

输入

参数	类型	必填	描述
`host`	string	是	Dagster host URL (e.g., https://myorg.dagster.cloud/prod or http://localhost:3000\)
`apiKey`	string	否	Dagster+ API token (leave blank for Dagster OSS or when not using Dagster+)
`repositoryLocationName`	string	是	Repository location (code location) name
`repositoryName`	string	是	Repository name within the code location
`jobName`	string	是	Name of the job to launch
`runConfigJson`	string	否	Run configuration as a JSON object (optional)
`tags`	string	否	Tags as a JSON array of {key, value} objects (optional)

输出

参数	类型	描述
`runId`	string	The globally unique ID of the launched run

`dagster_get_run`

Get the status and details of a Dagster run by its ID.

输入

参数	类型	必填	描述
`host`	string	是	Dagster host URL (e.g., https://myorg.dagster.cloud/prod or http://localhost:3000\)
`apiKey`	string	否	Dagster+ API token (leave blank for Dagster OSS or when not using Dagster+)
`runId`	string	是	The ID of the run to retrieve

输出

参数	类型	描述
`runId`	string	Run ID
`jobName`	string	Name of the job this run belongs to
`status`	string	Run status (QUEUED, NOT_STARTED, STARTING, MANAGED, STARTED, SUCCESS, FAILURE, CANCELING, CANCELED)
`startTime`	number	Run start time as Unix timestamp
`endTime`	number	Run end time as Unix timestamp
`runConfigYaml`	string	Run configuration as YAML
`tags`	json	Run tags as array of {key, value} objects

`dagster_get_run_logs`

Fetch execution event logs for a Dagster run.

输入

参数	类型	必填	描述
`host`	string	是	Dagster host URL (e.g., https://myorg.dagster.cloud/prod or http://localhost:3001\)
`apiKey`	string	否	Dagster+ API token (leave blank for Dagster OSS or when not using Dagster+)
`runId`	string	是	The ID of the run to fetch logs for
`afterCursor`	string	否	Cursor for paginating through log events (from a previous response)
`limit`	number	否	Maximum number of log events to return

输出

参数	类型	描述
`events`	json	Array of log events (type, message, timestamp, level, stepKey, eventType)
↳ `type`	string	GraphQL typename of the event
↳ `message`	string	Human-readable log message
↳ `timestamp`	string	Event timestamp as a Unix epoch string
↳ `level`	string	Log level (DEBUG, INFO, WARNING, ERROR, CRITICAL)
↳ `stepKey`	string	Step key, if the event is step-scoped
↳ `eventType`	string	Dagster event type enum value
`cursor`	string	Cursor for fetching the next page of log events
`hasMore`	boolean	Whether more log events are available beyond this page

`dagster_list_runs`

List recent Dagster runs, optionally filtered by job name.

输入

参数	类型	必填	描述
`host`	string	是	Dagster host URL (e.g., https://myorg.dagster.cloud/prod or http://localhost:3001\)
`apiKey`	string	否	Dagster+ API token (leave blank for Dagster OSS or when not using Dagster+)
`jobName`	string	否	Filter runs by job name (optional)
`statuses`	string	否	Comma-separated run statuses to filter by, e.g. "SUCCESS,FAILURE" (optional)
`limit`	number	否	Maximum number of runs to return (default 20)

输出

参数	类型	描述
`runs`	json	Array of runs
↳ `runId`	string	Run ID
↳ `jobName`	string	Job name
↳ `status`	string	Run status
↳ `tags`	json	Run tags as array of {key, value} objects
↳ `startTime`	number	Start time as Unix timestamp
↳ `endTime`	number	End time as Unix timestamp
`cursor`	string	Run id cursor to pass on the next call to fetch more runs
`hasMore`	boolean	Whether more runs are likely available beyond this page

`dagster_list_jobs`

List all jobs across repositories in a Dagster instance.

输入

参数	类型	必填	描述
`host`	string	是	Dagster host URL (e.g., https://myorg.dagster.cloud/prod or http://localhost:3001\)
`apiKey`	string	否	Dagster+ API token (leave blank for Dagster OSS or when not using Dagster+)

输出

参数	类型	描述
`jobs`	json	Array of jobs with name and repositoryName
↳ `name`	string	Job name
↳ `repositoryName`	string	Repository name

`dagster_reexecute_run`

Reexecute an existing Dagster run, optionally resuming only from failed steps.

输入

参数	类型	必填	描述
`host`	string	是	Dagster host URL (e.g., https://myorg.dagster.cloud/prod or http://localhost:3001\)
`apiKey`	string	否	Dagster+ API token (leave blank for Dagster OSS or when not using Dagster+)
`parentRunId`	string	是	The ID of the run to reexecute
`strategy`	string	是	Reexecution strategy: ALL_STEPS reruns everything, FROM_FAILURE resumes from failed steps, FROM_ASSET_FAILURE resumes from failed assets

输出

参数	类型	描述
`runId`	string	The ID of the newly launched reexecution run

`dagster_terminate_run`

Terminate an in-progress Dagster run.

输入

参数	类型	必填	描述
`host`	string	是	Dagster host URL (e.g., https://myorg.dagster.cloud/prod or http://localhost:3001\)
`apiKey`	string	否	Dagster+ API token (leave blank for Dagster OSS or when not using Dagster+)
`runId`	string	是	The ID of the run to terminate

输出

参数	类型	描述
`success`	boolean	Whether the run was successfully terminated
`runId`	string	The ID of the terminated run
`message`	string	Error or status message if termination failed

`dagster_delete_run`

Permanently delete a Dagster run record.

输入

参数	类型	必填	描述
`host`	string	是	Dagster host URL (e.g., https://myorg.dagster.cloud/prod or http://localhost:3001\)
`apiKey`	string	否	Dagster+ API token (leave blank for Dagster OSS or when not using Dagster+)
`runId`	string	是	The ID of the run to delete

输出

参数	类型	描述
`runId`	string	The ID of the deleted run

`dagster_list_schedules`

List all schedules in a Dagster repository, optionally filtered by status.

输入

参数	类型	必填	描述
`host`	string	是	Dagster host URL (e.g., https://myorg.dagster.cloud/prod or http://localhost:3001\)
`apiKey`	string	否	Dagster+ API token (leave blank for Dagster OSS or when not using Dagster+)
`repositoryLocationName`	string	是	Repository location (code location) name
`repositoryName`	string	是	Repository name within the code location
`scheduleStatus`	string	否	Filter schedules by status: RUNNING or STOPPED (omit to return all)

输出

参数	类型	描述
`schedules`	json	Array of schedules (name, cronSchedule, jobName, status, id, description, executionTimezone)
↳ `name`	string	Schedule name
↳ `cronSchedule`	string	Cron expression for the schedule
↳ `jobName`	string	Job the schedule targets
↳ `status`	string	Schedule status: RUNNING or STOPPED
↳ `id`	string	Instigator state ID — use this to start or stop the schedule
↳ `description`	string	Human-readable schedule description
↳ `executionTimezone`	string	Timezone for cron evaluation

`dagster_start_schedule`

Enable (start) a schedule in a Dagster repository.

输入

参数	类型	必填	描述
`host`	string	是	Dagster host URL (e.g., https://myorg.dagster.cloud/prod or http://localhost:3001\)
`apiKey`	string	否	Dagster+ API token (leave blank for Dagster OSS or when not using Dagster+)
`repositoryLocationName`	string	是	Repository location (code location) name
`repositoryName`	string	是	Repository name within the code location
`scheduleName`	string	是	Name of the schedule to start

输出

参数	类型	描述
`id`	string	Instigator state ID of the schedule
`status`	string	Updated schedule status (RUNNING or STOPPED)

`dagster_stop_schedule`

Disable (stop) a running schedule in Dagster.

输入

参数	类型	必填	描述
`host`	string	是	Dagster host URL (e.g., https://myorg.dagster.cloud/prod or http://localhost:3001\)
`apiKey`	string	否	Dagster+ API token (leave blank for Dagster OSS or when not using Dagster+)
`instigationStateId`	string	是	InstigationState ID of the schedule to stop — available from dagster_list_schedules output

输出

参数	类型	描述
`id`	string	Instigator state ID of the schedule
`status`	string	Updated schedule status (RUNNING or STOPPED)

`dagster_list_sensors`

List all sensors in a Dagster repository, optionally filtered by status.

输入

参数	类型	必填	描述
`host`	string	是	Dagster host URL (e.g., https://myorg.dagster.cloud/prod or http://localhost:3001\)
`apiKey`	string	否	Dagster+ API token (leave blank for Dagster OSS or when not using Dagster+)
`repositoryLocationName`	string	是	Repository location (code location) name
`repositoryName`	string	是	Repository name within the code location
`sensorStatus`	string	否	Filter sensors by status: RUNNING or STOPPED (omit to return all)

输出

参数	类型	描述
`sensors`	json	Array of sensors (name, sensorType, status, id, description)
↳ `name`	string	Sensor name
↳ `sensorType`	string	Sensor type (ASSET, AUTO_MATERIALIZE, FRESHNESS_POLICY, MULTI_ASSET, RUN_STATUS, STANDARD)
↳ `status`	string	Sensor status: RUNNING or STOPPED
↳ `id`	string	Instigator state ID — use this to start or stop the sensor
↳ `description`	string	Human-readable sensor description

`dagster_start_sensor`

Enable (start) a sensor in a Dagster repository.

输入

参数	类型	必填	描述
`host`	string	是	Dagster host URL (e.g., https://myorg.dagster.cloud/prod or http://localhost:3001\)
`apiKey`	string	否	Dagster+ API token (leave blank for Dagster OSS or when not using Dagster+)
`repositoryLocationName`	string	是	Repository location (code location) name
`repositoryName`	string	是	Repository name within the code location
`sensorName`	string	是	Name of the sensor to start

输出

参数	类型	描述
`id`	string	Instigator state ID of the sensor
`status`	string	Updated sensor status (RUNNING or STOPPED)

`dagster_stop_sensor`

Disable (stop) a running sensor in Dagster.

输入

参数	类型	必填	描述
`host`	string	是	Dagster host URL (e.g., https://myorg.dagster.cloud/prod or http://localhost:3001\)
`apiKey`	string	否	Dagster+ API token (leave blank for Dagster OSS or when not using Dagster+)
`instigationStateId`	string	是	InstigationState ID of the sensor to stop — available from dagster_list_sensors output

输出

参数	类型	描述
`id`	string	Instigator state ID of the sensor
`status`	string	Updated sensor status (RUNNING or STOPPED)

`dagster_list_assets`

List assets tracked by a Dagster instance, optionally filtered by key prefix.

输入

参数	类型	必填	描述
`host`	string	是	Dagster host URL (e.g., https://myorg.dagster.cloud/prod or http://localhost:3001\)
`apiKey`	string	否	Dagster+ API token (leave blank for OSS / self-hosted)
`prefix`	string	否	Slash-delimited asset key prefix to filter by, e.g. "raw" or "raw/events" (optional)
`cursor`	string	否	Asset key cursor from a previous response, for pagination (optional)
`limit`	number	否	Maximum number of assets to return per page (default 100)

输出

参数	类型	描述
`assets`	json	Array of assets (assetKey, path)
↳ `assetKey`	string	Slash-joined asset key
↳ `path`	json	Asset key path segments
`cursor`	string	Cursor to pass on the next call to fetch more assets
`hasMore`	boolean	Whether more assets are likely available beyond this page

`dagster_get_asset`

Get an asset definition and its latest materialization by asset key.

输入

参数	类型	必填	描述
`host`	string	是	Dagster host URL (e.g., https://myorg.dagster.cloud/prod or http://localhost:3001\)
`apiKey`	string	否	Dagster+ API token (leave blank for OSS / self-hosted)
`assetKey`	string	是	Slash-delimited asset key, e.g. "my_asset" or "raw/events"

输出

参数	类型	描述
`assetKey`	string	Slash-joined asset key
`path`	json	Asset key path segments
`groupName`	string	Asset group the definition belongs to
`description`	string	Asset description
`jobNames`	json	Names of jobs that can materialize this asset
`computeKind`	string	Compute kind tag (e.g., python, dbt, spark)
`isPartitioned`	boolean	Whether the asset is partitioned
`latestMaterialization`	json	Most recent materialization (runId, timestamp, partition, stepKey)
↳ `runId`	string	Run that produced the materialization
↳ `timestamp`	string	Materialization timestamp (epoch ms string)
↳ `partition`	string	Partition key, if partitioned
↳ `stepKey`	string	Step key that emitted it

`dagster_materialize_assets`

Materialize selected assets by launching their asset job with an asset selection.

输入

参数	类型	必填	描述
`host`	string	是	Dagster host URL (e.g., https://myorg.dagster.cloud/prod or http://localhost:3001\)
`apiKey`	string	否	Dagster+ API token (leave blank for OSS / self-hosted)
`repositoryLocationName`	string	是	Repository location (code location) name
`repositoryName`	string	是	Repository name within the code location
`jobName`	string	是	Asset job that contains the assets, e.g. "__ASSET_JOB" or a named asset job
`assetSelection`	string	是	Comma- or newline-separated asset keys to materialize, each slash-delimited (e.g. "raw/events, summary")
`tags`	string	否	Tags as a JSON array of {key, value} objects (optional)

输出

参数	类型	描述
`runId`	string	The globally unique ID of the launched materialization run

`dagster_report_asset_materialization`

Report an external (runless) materialization or observation for an asset.

输入

参数	类型	必填	描述
`host`	string	是	Dagster host URL (e.g., https://myorg.dagster.cloud/prod or http://localhost:3001\)
`apiKey`	string	否	Dagster+ API token (leave blank for OSS / self-hosted)
`assetKey`	string	是	Slash-delimited asset key to report against, e.g. "my_asset" or "raw/events"
`eventType`	string	否	Event type to report: ASSET_MATERIALIZATION (default) or ASSET_OBSERVATION
`partitionKeys`	string	否	Comma-separated partition keys to report against (optional)
`description`	string	否	Human-readable description for the reported event (optional)

输出

参数	类型	描述
`success`	boolean	Whether the event was reported successfully
`assetKey`	string	Slash-joined asset key the event was reported against

`dagster_wipe_asset`

DESTRUCTIVE: permanently wipes ALL materialization history (every partition) for an asset. This cannot be undone.

输入

参数	类型	必填	描述
`host`	string	是	Dagster host URL (e.g., https://myorg.dagster.cloud/prod or http://localhost:3001\)
`apiKey`	string	否	Dagster+ API token (leave blank for OSS / self-hosted)
`assetKey`	string	是	Slash-delimited asset key to wipe, e.g. "my_asset" or "raw/events"

输出

参数	类型	描述
`success`	boolean	Whether the asset was wiped successfully
`assetKey`	string	Slash-joined asset key that was wiped