Dolphin MCP Pilot

Read full parameters of a single task (returns only that task, not the whole DAG). Added in v2.0.12 for the "verify a task change took effect" scenario, avoiding pulling a huge DAG payload. Args: project_name: Project name workflow_code: Process-definition code task_name: Task name (exact match) Returns: { "task_code": int, "task_name": str, "task_type": str, "task_params": {...}, # Full taskParams (preStatements/postStatements/sql/script etc.) "description": str, "fail_retry_times": int, "timeout": int, ... …

Fetch task execution log (supports pagination). Args: task_instance_id: Task instance ID (from ds_list_task_instances) skip_line_num: Skip first N lines (for pagination) limit: Max lines to fetch (default 1000) Returns: { "task_instance_id": int, "log": str, "line_start": int, "line_count": int, "hint": str }

Read a single workflow definition (DAG, global params, node coordinates, etc.). Args: project_name: Project name workflow_code: Process-definition code (from ds_list_workflows) compact: Compact mode (default False). When True, returns only metadata and task summary without full taskParams/SQL, greatly reducing token usage. Suitable when you only need to understand DAG structure.

Interactive guide to DolphinScheduler MCP tools. Call without arguments to see all categories, or pass a category name to view tools and workflows for that category. Available categories: - quickstart: New user onboarding guide with common scenarios - project: Project CRUD (list/create/rename/delete) - workflow: Basic workflow operations (create/list/get/update/delete/release/run) - workflow_advanced: DAG editing, task updates, version management, cloning - instance: Process instance management, log retrieval, failure troubleshooting - schedule: Schedule configuration (cron/online/offline/de…

List the data sources of one type that the current user can see (read-only). Pass the returned ``id`` as ``datasource_id`` of SQL tasks, e.g. in ds_create_workflow or ds_create_dag_workflow. Args: ds_type: Data source type in DolphinScheduler's upper-case form, e.g. HIVE (default), MYSQL, POSTGRESQL, SPARK, CLICKHOUSE. Returns: [{"id": int, "name": str, "type": str}, ...]; empty if none match.

List process instances (filter by workflow_code, state). v2.0.12: merged ds_list_workflow_instances. - workflow_code=0 (default): list all recent instances in the project - workflow_code=<value>: list instances for that specific workflow Troubleshooting / Progress Tracking (v2.0.18): After getting instances, to understand execution details (which tasks running/failed/stuck), use ds_list_task_instances(process_instance_id=<id>, include_full=True) to view task states: - RUNNING_EXECUTION: instance running but no details → check task list for dispatched/running tasks (RUNNING instance with em…

List all projects. Returns [{id, code, name, description}].

List resources (files and folders) at a given path. Args: resource_type: Resource type — FILE / UDF / ALL (default, lists both) full_name: Path prefix filter; empty = list everything from root

List schedule configurations (optionally filtered by workflow_code). v2.0.12: merged ds_list_schedules_in_project and ds_get_workflow_schedule. - workflow_code=0 (default): list all schedules in the project - workflow_code=<value>: list schedules for that specific workflow - simplify=True: return a summary dict with has_schedule/cron/release_state Args: project_name: Project name workflow_code: Process-definition code (0 = no filter) simplify: Return summary dict instead of raw list page_no: Page number page_size: Items per page

List task instances for a process instance (filter by state, optionally include full details). v2.0.18: Added include_full parameter for detailed troubleshooting. Args: project_name: Project name process_instance_id: Process instance ID state: State filter (FAILURE / SUCCESS / RUNNING_EXECUTION, empty = no filter) include_full: When True, returns full taskParams/SQL/script for each task (useful for detailed analysis but consumes more tokens) Returns: List of task instances with: - Basic info: id, name, taskType, state, startTime, endTime, retryTimes…

List all tenants (useful for debugging workflow tenant_id foreign key issues).

List all DS users (useful for debugging workflow user_id foreign key issues).

List saved versions of a workflow definition (read-only). Each save of the workflow creates a version. Use the returned version number with ds_rollback_workflow_version. Args: project_name: Project name workflow_code: Workflow (process-definition) code from ds_list_workflows page_size: Maximum number of versions to return (first page, default 20) Returns: [{"version": int, "description": str, "createTime": str, "updateTime": str}, ...]

List workflow definitions with pagination, search and exact-match support. v2.0.12: merged ds_list_workflows_simple and ds_get_workflow_by_name. - Default: paginated API (auto-falls back to simple-list on failure) - search="xx": case-insensitive fuzzy match on name + description - name="xxx": exact name match (single-element list or empty list) - use_simple=True: use /process-definition/simple-list (bypasses pagination SQL) Args: project_name: Project name page_no: Page number, starting at 1 page_size: Items per page (max recommended 100) search: Fuzzy search keyword (name + …

Modify existing workflow DAG (add/delete/update tasks + adjust dependencies). Implementation: read current DAG → apply operations → write back. ⚠️ Risk warning: - Modifications create a new version; old versions preserved and rollback-able - If workflow is running, it will be taken offline first (auto_offline=True by default) - Auto-re-online after modification (auto_online=True by default) - DS has no native "add single task" API; this tool simulates it via "read → modify → full update" Args: project_name: Project name …

Check DS master node status (verify scheduler is alive).

Check DS worker node status (verify task executors are alive).

Take a schedule offline so it stops triggering new runs. Keeps the schedule and running instances; bring it back with ds_online_schedule, or remove it with ds_delete_schedule. Args: project_name: Project name schedule_id: Schedule ID (from ds_list_schedules) Returns: {"schedule_id": int, "status": "OFFLINE"}

Create a text file inline (no upload required). Args: file_name: Filename without extension (e.g. "my_script") suffix: File extension without dot (e.g. "py" / "sh" / "sql") content: Text content to write current_dir: Parent directory path, default root "/" description: File description resource_type: FILE or UDF

Bring a schedule online so it starts triggering workflow runs. Needed after ds_set_schedule, which creates schedules offline. The workflow itself must be online (ds_release_workflow). Reverse with ds_offline_schedule. Args: project_name: Project name schedule_id: Schedule ID (from ds_set_schedule or ds_list_schedules) Returns: {"schedule_id": int, "status": "ONLINE"}

Pause a RUNNING workflow instance (reversible; not a stop). Tasks already running finish, no new tasks are started, and the instance ends in PAUSE state. Use ds_resume_process_instance to continue it, or ds_stop_process_instance to kill it instead. Args: project_name: Project name process_instance_id: Instance ID (from ds_list_process_instances) Returns: {"process_instance_id": int, "executeType": "PAUSE", "status": "submitted"}; the request is asynchronous, so check the state with ds_list_process_instances.

Send a raw DELETE request to the DolphinScheduler API (escape hatch). Deletes are irreversible. Prefer the dedicated ds_delete_* tools; use this only for endpoints they do not cover. Args: path: API path starting with / and excluding the /dolphinscheduler prefix, with any query string, e.g. "/projects/123/schedules/7" Returns: DolphinScheduler's response as-is ({"code", "msg", "data", ...}); code 0 means success.

Pass GET through to DolphinScheduler API (path must start with /, excluding the /dolphinscheduler prefix).

Pass POST through to DolphinScheduler API. Args: path: API path (starting with /) form_data_json: form-urlencoded params as a JSON string (choose one) json_body_json: JSON body as a JSON string (choose one)

Send a raw PUT request to the DolphinScheduler API (escape hatch). Prefer the dedicated update tools (e.g. ds_update_workflow, ds_update_schedule_cron); use this only for endpoints they do not cover. Args: path: API path starting with / and excluding the /dolphinscheduler prefix, e.g. "/projects/123/schedules/7" form_data_json: form-urlencoded params as a JSON object string, e.g. '{"crontab": "0 0 2 * * ? *"}' (set this or json_body_json) json_body_json: JSON body as a JSON string (set this or form_data_json) Returns: DolphinScheduler's response as-is ({"code…

Bring a workflow online or offline. When bringing online (online=True), auto_online_schedule will also re-activate the associated schedule if it was taken offline (DS takes schedules offline when the workflow goes offline). Args: project_name: Project name workflow_code: Process-definition code online: True=online, False=offline auto_online_schedule: Auto-reactivate associated schedule when going online (default True) Returns: { "workflow_code": int, "releaseState": "ONLINE" / "OFFLINE", "schedule_action": str, "schedule_id": int, }

Rename a project. Args: old_name: Current project name new_name: New project name description: Optional new description; keeps the existing one if omitted

Rename a resource file or folder. ⚠️ Warning: DS may implement rename as delete + recreate, which changes the resource_id. If this resource is referenced by workflow tasks, the reference will become invalid after renaming. Check references with ds_list_workflows before proceeding. Args: resource_id: Resource ID new_name: New name (filename or folder name only, no path prefix) description: Description resource_type: FILE or UDF

Resume from failed tasks (rerun only failed and pending tasks, skip succeeded ones).

Rerun entire workflow instance (from the beginning).

Resume a workflow instance that is in PAUSE state. Continues from where ds_pause_process_instance stopped it. Not for failed or stopped instances: use ds_rerun_from_failure or ds_rerun_process_instance. Args: project_name: Project name process_instance_id: Instance ID (from ds_list_process_instances) Returns: {"process_instance_id": int, "executeType": "RECOVER_SUSPENDED_PROCESS", "status": "submitted"}; the request is asynchronous.

Roll back a workflow to a specific historical version. Args: project_name: Project name workflow_code: Process-definition code version: Target version number (from ds_list_workflow_versions)

Manually trigger a workflow run (brings it online first). Args: project_name: Project name workflow_code: Process-definition code start_task_names: Optional list of task names to start from (including their downstream tasks). Leave empty to run the entire workflow. Example: ["partition_check"] starts from that task and all its successors.

Create a schedule for a workflow (created OFFLINE; activate with ds_online_schedule). v2.0.12: cron is now required (no default) to prevent silent misuse. Args: project_name: Project name workflow_code: Process-definition code cron: Quartz 7-field expression (required, e.g. "0 0 6 * * ? *") start_time: Start time yyyy-MM-dd HH:mm:ss; defaults to today 00:00:00 end_time: End time; defaults to 2030-12-31 23:59:59

Skip a task (mark as success without running). ⚠️ Warning: Use only for non-critical tasks (e.g., notifications). Downstream tasks will proceed as if this task succeeded.

Stop a running workflow instance (force kill).

Test DolphinScheduler login and connectivity.

Update resource file content (text files only). ⚠️ Changes take effect immediately. Workflows referencing this script will use the new version on their next execution. Args: resource_id: Resource ID content: New file content description: Update note (optional)

Update only the cron expression of a schedule (keeps all other settings). Flow: read current state → offline if needed → update → re-online (default). v2.0.10: queries the target schedule directly instead of scanning the full list. Args: project_name: Project name schedule_id: Schedule ID (from ds_list_schedules) cron: New Quartz 7-field expression start_time: New start time yyyy-MM-dd HH:mm:ss (keep original if empty) end_time: New end time (keep original if empty) auto_online: Re-activate after update (default True) Returns: { "schedule_id": int, …

Lightweight single-task parameter update — no need to pass full DAG definition. This tool is a convenience wrapper around ds_modify_workflow_dag's update_task, automatically handling the "read → modify → write → online/offline" flow. Use case: Change a single task's SQL/script/name/retry params without constructing a full operations list. Args: project_name: Project name workflow_code: Workflow code task_name: Task name to modify (exact match) updates: Fields to update (flat dict, field names same as update_task's updates) auto_offline: Auto-offline before modification (…

Update workflow definition (PUT method, preserves code, auto-increments version). This is a fallback when ds_modify_workflow_dag cannot handle the change. Supports full task definition override with complete replacement. Args: project_name: Project name workflow_code: Process-definition code task_definitions: Task definition list (list or JSON string) task_relations: Task relation list (list or JSON string) locations: Node coordinates (list or JSON string, default "[]") name: Workflow name (empty = keep original) description: Description (empty = keep original) …

Upload a file to the resource area (supports binary, e.g. jar / zip). Two ways to provide file content: Method 1: local_path — path to a file accessible by the MCP server process Method 2: file_name + file_content_base64 — base64-encoded content Args: local_path: Local file path (Method 1) file_name: Filename with extension (Method 2) file_content_base64: Base64-encoded file content (Method 2) current_dir: Parent directory path, default root "/" resource_type: FILE or UDF

View resource file content (paginated, text files only). Args: resource_id: Resource ID (from ds_list_resources or ds_get_resource_by_name) skip_line_num: Number of lines to skip limit: Max lines to read