Test DolphinScheduler login and connectivity.

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

Create a new project.

Args: name: Project name description: Optional description

Rename a project.

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

Delete an entire project, cascading to all workflows, schedules and instances. This is irreversible - use with caution.

List data sources.

Args: ds_type: HIVE / MYSQL / POSTGRESQL / SPARK / CLICKHOUSE etc.

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 + description) name: Exact workflow name match use_simple: Use simple-list endpoint when True

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.

Create a workflow (supports simple SQL mode and complex DAG mode).

Mode 1 - Simple SQL (backward compatible):

• Provide sql_statements + datasource_id

• Multiple SQL separated by ";;;" are executed serially

Mode 2 - Complex DAG:

• Provide tasks_json + relations_json

• tasks_json: JSON array string, each task contains name, taskType, taskParams etc.

• relations_json: JSON array string defining task dependencies (preTaskName -> postTaskName)

Args: project_name: Project name name: Workflow name description: Description sql_statements: [Mode 1] SQL, multiple separated by ";;;" datasource_id: [Mode 1] Datasource ID sql_ds_type: [Mode 1] SQL type (HIVE/MYSQL) tasks_json: [Mode 2] Task definition JSON array string relations_json: [Mode 2] Task relation JSON array string schedule: Whether to create a schedule schedule_cron: Quartz 7-field cron expression schedule_start: Schedule start time (defaults to today) schedule_end: Schedule end time (defaults to 2030-12-31) auto_online: Whether to auto-release workflow and schedule

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

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.

Delete workflow(s) by code (takes offline first). Supports single or batch delete.

v2.0.12: merged ds_batch_delete_workflows.

• workflow_code=12345: delete single workflow

• workflow_code=[12345, 67890]: batch delete (each independent; one failure doesn't block others)

Args: project_name: Project name workflow_code: Process-definition code (int for single, list for batch)

Returns: Single: {"workflow_code": int, "status": "deleted"} Batch: {"total": int, "deleted": int, "failed": int, "deleted_codes": [...], "failed_details": [...]}

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) auto_offline: Auto-offline before update (default True) auto_online: Auto-online after update (default True)

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, ... }

List historical versions of a workflow.

Args: project_name: Project name workflow_code: Process-definition code page_size: Number of items to return

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)

Clone/copy a workflow.

Args: project_name: Project name source_workflow_code: Source workflow code new_name: New workflow name description: New workflow description (empty = copy source description) auto_online: Auto-release after cloning

Returns: { 'success': True, 'workflow_code': int, # New workflow code 'source_code': int, 'name': str, ... }

Create a generic DAG workflow supporting any task type (SQL/SHELL/PYTHON/DEPENDENT/SUB_PROCESS/HTTP etc.).

    Args:
        project_name: Project name
        name: Workflow name
        tasks: Task definition list (see examples below)
        relations: Dependency list [{"from": "taskA", "to": "taskB"}]; empty "from" = start node
        description: Description
        schedule: Whether to create a schedule
        schedule_cron: Cron expression (7-field Quartz style)
        locations: Optional node coordinates [{"task_name": "check", "x": 100, "y": 100}]
                   Leave empty for auto-layout (horizontal, 300x200 spacing)

    tasks example:
        [
            # SHELL task (script required, resource_list optional)
            {"name": "check", "type": "SHELL",
             "script": "#!/bin/bash

python3 /public/check_partition.py table_name $[yyyyMMdd-1]", "resource_list": [67], # optional: referenced resource id or path "fail_retry_times": 3}, # optional: retry count on failure

            # SQL task (datasource_id + sql required)
            {"name": "sql1", "type": "SQL",
             "datasource_id": 1, "sql": "SELECT 1",
             "sql_type": "HIVE",              # optional, default HIVE
             "sql_type_select": 1},           # optional, 0=query 1=non-query (default 1)

            # DEPENDENT task (wait for upstream workflow completion)
            {"name": "wait", "type": "DEPENDENT",
             "depend_items": [{"project_code": 123, "definition_code": 456,
                               "cycle": "day", "date_value": "today"}]},

            # SUB_PROCESS task (invoke a sub-workflow)
            {"name": "sub", "type": "SUB_PROCESS",
             "sub_process_code": 21505676237440},

            # HTTP task
            {"name": "notify", "type": "HTTP",
             "http_url": "https://api.example.com/callback",
             "http_method": "POST"}
        ]
    relations example:
        [
            {"from": "", "to": "check"},      # check is a start node (empty "from")
            {"from": "check", "to": "sql1"},
            {"from": "wait", "to": "sql1"}    # wait and check run in parallel, both feed sql1
        ]

    resource_list format:
        - Recommended: resource_id (int): [67, 58]
        - Or full path (str): ["/public/check_partition.py"]
        - Paths must start with /
        - Use ds_list_resources() to view all resources with their ids and paths
    

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
        workflow_code: Workflow code
        operations: Operation list (supports 5 action types)
        auto_offline: Auto-offline before modification (default True)
        auto_online: Auto-online after modification (default True)
        auto_online_schedule: Auto-restore associated schedule after onlining (default True)
            ⚠️ DS offlines the schedule when workflow goes offline;
            set False only if schedule auto-restore is not needed.

    Supported operation types:

    [1] Add task (add_task):
        {
            "action": "add_task",
            "task": {
                "name": "wait_upstream",
                "type": "DEPENDENT",
                "depend_items": [{
                    "project_code": 123,
                    "definition_code": 456,
                    "dep_task_code": 0,    # 0=entire workflow, nonzero=specific task
                    "cycle": "day",         # day/hour/week/month
                    "date_value": "today"   # today/yesterday/last7days
                }],
                "dep_relation": "AND"       # multi-dep relation: AND/OR
                # other fields same as ds_create_dag_workflow
            },
            "connect_from": "task_a",       # predecessor task name (empty=start node)
            "connect_to": "task_b"          # successor task name (empty=end node)
        }

    [2] Delete task (delete_task):
        {
            "action": "delete_task",
            "task_name": "old_task",
            "reconnect": true               # auto-reconnect predecessor/successor to prevent DAG break
        }

    [3] Update task parameters (update_task):
        {
            "action": "update_task",
            "task_name": "my_sql",
            "updates": {
                # ===== Common fields (snake_case / camelCase aliases supported) =====
                "name": "new_task_name",           # v2.0.11: rename node
                "description": "Updated description",
                "fail_retry_times": 3,             # or "failRetryTimes"
                "fail_retry_interval": 4,          # or "failRetryInterval" (v2.0.10 fixed)
                "timeout": 300,                    # timeout duration (minutes)
                "timeout_flag": "OPEN",            # or "timeoutFlag": "OPEN"/"CLOSE" or True/False
                "timeout_notify_strategy": "FAILED",  # or "timeoutNotifyStrategy": "FAILED"/"WARNING"
                "worker_group": "default",         # or "workerGroup"
                "task_priority": "HIGH",           # or "taskPriority": "HIGH"/"MEDIUM"/"LOW"
                "delay_time": 0,                   # or "delayTime" (delay in minutes)
                "flag": "YES",                     # "YES"=enabled "NO"=disabled

                # ===== SQL task =====
                "sql": "SELECT * FROM new_table WHERE dt='$[yyyyMMdd-1]'",
                "datasource_id": 1,                # or "datasource"
                "sql_type": "HIVE",                # "HIVE"/"MYSQL"/"POSTGRESQL" etc.
                "sql_type_select": 0,              # or "sqlType": 0=query 1=non-query
                "pre_statements": [                # or "preStatements" (pre-SQL list)
                    "SET hive.exec.dynamic.partition=true",
                    "SET spark.sql.sources.partitionOverwriteMode=dynamic"
                ],
                "post_statements": ["SELECT 1"],   # or "postStatements" (post-SQL list)

                # ===== SHELL/PYTHON task =====
                "script": "#!/bin/bash

export PATH=/xxx/bin:$PATH python3 script.py", # or "rawScript"

                # ===== Resource refs (usable by SHELL/PYTHON/SQL) =====
                "resource_list": [67, 58],         # or "resourceList" (int ID recommended)
                # or ["/public/check_partition.py"] (full path, starting with /)

                # ===== Local params =====
                "local_params": [{"prop":"dt","value":"$[yyyyMMdd-1]"}]  # or "localParams"
            }
        }

        ⚠️ resource_list format:
        - Recommended: resource_id (int): [67, 58]
        - Or full path (str): ["/public/check_partition.py", "/scripts/sync.py"]
        - Paths must start with /, e.g. /public/xxx.py not public/xxx.py
        - Use ds_list_resources() to view all resources with ids and paths

        ⚠️ Field naming compatibility:
        - Supports both snake_case (e.g. pre_statements) and camelCase (e.g. preStatements)
        - Both styles work identically; snake_case recommended
        - Unrecognized fields logged in return value's ignored_fields

    [4] Update node coordinates (update_location):
        {
            "action": "update_location",
            "task_name": "my_task",
            "x": 500,
            "y": 300
        }

    [5] Adjust dependencies (update_relations):
        {
            "action": "update_relations",
            "changes": [
                {"op": "add", "from": "task_a", "to": "task_b"},
                {"op": "remove", "from": "task_c", "to": "task_d"}
            ]
        }

    Returns:
        {
            "workflow_code": int,
            "operations_applied": int,
            "task_count_before": int,
            "task_count_after": int,
            "status": "updated",
            "warning": risk notice
        }

    Typical scenario: append a DEPENDENT node so the workflow waits on upstream
        ds_modify_workflow_dag(
            project_name="my_project",
            workflow_code=21505676237440,
            operations=[{
                "action": "add_task",
                "task": {
                    "name": "wait_upstream_flow",
                    "type": "DEPENDENT",
                    "depend_items": [{
                        "project_code": 21582927260160,
                        "definition_code": 21505676200000,
                        "cycle": "day",
                        "date_value": "today"
                    }]
                },
                "connect_from": "",               # start node
                "connect_to": "existing_task_1"   # connect to an existing task
            }]
        )
    

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 (default True) auto_online: Auto-online after modification (default True) auto_online_schedule: Auto-restore schedule after onlining (default True, v2.0.11)

Supported updates fields (snake_case and camelCase both accepted): Common: name, description, fail_retry_times(failRetryTimes), fail_retry_interval(failRetryInterval), timeout, timeout_flag(timeoutFlag), timeout_notify_strategy(timeoutNotifyStrategy), worker_group(workerGroup), task_priority(taskPriority), delay_time(delayTime), flag SQL: sql, datasource_id(datasource), sql_type, sql_type_select(sqlType), pre_statements(preStatements), post_statements(postStatements), local_params(localParams), resource_list(resourceList) SHELL/PYTHON: script(rawScript), resource_list(resourceList), local_params(localParams)

⚠️ Unrecognized fields are reported in the return value's ignored_fields (not silently dropped).

Returns: { "workflow_code": int, "task_name": str, "schedule_action": str, "status": "updated", }

Examples: # Change a single task's retry params ds_update_task_param( project_name="my_project", workflow_code=21583255237888, task_name="check_partition", updates={"fail_retry_times": 8, "fail_retry_interval": 1} )

# Rename a task (v2.0.11)
ds_update_task_param(
    project_name="my_project", workflow_code=..., task_name="old_sql",
    updates={"name": "new_sql_v2"}
)

# Change SQL only
ds_update_task_param(
    project_name="my_project", workflow_code=..., task_name="sql1",
    updates={"sql": "SELECT * FROM new_table WHERE dt='$[yyyyMMdd-1]'"}
)

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=: 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=, include_full=True) to view task states:

• RUNNING_EXECUTION: instance running but no details → check task list for dispatched/running tasks (RUNNING instance with empty task list is normal — DS is initializing DAG)

• FAILURE: locate failed tasks → find FAILURE tasks, then check their logs Each RUNNING/FAILURE instance returned includes a next_action hint.

Args: project_name: Project name workflow_code: Process-definition code (0 = no filter, list all instances) state: State filter (FAILURE / SUCCESS / RUNNING_EXECUTION / STOP, empty = no filter) page_size: Number of items to return

Stop a running workflow instance (force kill).

Pause a running workflow instance.

Resume a paused workflow instance.

Rerun entire workflow instance (from the beginning).

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

Delete a historical process instance.

Backfill (complement) workflow data for date range or single partition.

Recommended usage (v2.0.14):

• Single partition: partition_date="2024-01-01" (clearer semantics, recommended)

• Date range: start_date + end_date (multi-partition backfill)

• From specific task: start_task_names + task_depend_type="TASK_POST"

• Single task only: start_task_names + task_depend_type="TASK_ONLY"

⚠️ Important (v2.0.19):

1. Default mode: RUN_MODE_SERIAL (one partition at a time, safer)

2. Parallel mode: run_mode="RUN_MODE_PARALLEL" (multiple partitions concurrently, higher risk)

3. Before backfill: check dependency chain with ds_get_workflow → analyze upstream deps

4. Minimize scope: prefer start_task_names + TASK_POST to backfill from target task forward

5. Mandatory standard: don't backfill entire workflow unless full-chain rerun is explicitly needed

Args: project_name: Project name workflow_code: Process-definition code start_date: Start date yyyy-MM-dd (for range backfill) end_date: End date yyyy-MM-dd (for range backfill) partition_date: Single partition yyyy-MM-dd (alternative to start_date+end_date, recommended for clarity) start_task_names: Optional list of task names to start from (backfills these + their downstream tasks) task_depend_type: Dependency type (default TASK_POST: from start tasks forward; TASK_ONLY: start tasks only; TASK_PRE: start tasks + upstream) run_mode: RUN_MODE_SERIAL (default, one at a time) or RUN_MODE_PARALLEL (concurrent)

Serial ordering guarantee (v2.0.18): In RUN_MODE_SERIAL, the request is submitted using DS's continuous range fields complementStartDate / complementEndDate, so DS generates instances strictly in ascending day order. Discrete date-list formats (complementScheduleDateList / comma-separated) do not guarantee ordering and are only used as fallbacks. The chosen format is reported in the return value's "format" field ("date_range" / "comma_separated" / "json_list").

Examples: # Single partition (recommended) ds_complement_data(workflow_code=123, partition_date="2024-01-01")

# Date range (serial mode, safer)
ds_complement_data(workflow_code=123, start_date="2024-01-01", end_date="2024-01-31")

# From specific task forward (minimize scope)
ds_complement_data(workflow_code=123, partition_date="2024-01-01",
                   start_task_names=["ads_table"], task_depend_type="TASK_POST")

# Parallel mode (higher concurrency, more resource usage)
ds_complement_data(workflow_code=123, start_date="2024-01-01", end_date="2024-01-10",
                   run_mode="RUN_MODE_PARALLEL")

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 - If include_full=True: also includes taskParams (SQL/script/dependency config)

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 }

Force mark a task as success (dangerous — bypasses actual execution).

⚠️ Warning: Use only when task is stuck/failed but you've verified data correctness. This does NOT rerun the task — it only changes the state flag.

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.

One-click fetch of all failed task logs from the latest failed instance.

Combines 3 common troubleshooting steps (list instances → list tasks → fetch logs).

v2.0.10: Added per-step error handling — returns partial diagnostic info even if some steps fail.

Args: project_name: Project name workflow_code: Process-definition code log_limit: Lines of log to fetch per failed task (default 500)

Returns: { "workflow_code": int, "instance_id": int, # Latest failed instance ID "instance_state": "FAILURE", "start_time": str, "failed_tasks": [ # All failed tasks { "task_instance_id": int, "task_name": str, "task_type": str, "start_time": str, "end_time": str, "log_tail": str # Last log_limit lines } ], "errors": [...] # v2.0.10: Per-step error messages "hint": ... }

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=: 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

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

Activate (bring online) a schedule.

Deactivate (take offline) a schedule.

Delete a schedule (takes it offline first).

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, "new_cron": str, "release_state": "ONLINE" | "OFFLINE", "status": "updated", "online_retry_detail": str }

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

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

Find a resource (file or folder) by full path, returning id and metadata.

Implementation: uses /resources/list recursive search (avoids /resources/query-by-name due to known DS bug).

Args: full_name: Full resource path, e.g. "public/test.py" or "scripts" resource_type: Resource type, default FILE

Download a resource file (supports binary, e.g. jar, zip).

Args: resource_id: Resource ID save_to: Optional local save path (accessible to MCP server process). If empty, returns base64-encoded content.

Returns: If save_to provided: {"resource_id", "saved_to", "size"} Otherwise: {"resource_id", "size", "content_base64"}

Create a folder in the resource area.

Args: name: Folder name (no path prefix, e.g. "scripts") current_dir: Parent directory path, default root "/" resource_type: FILE or UDF

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

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

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)

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

Delete a resource (file or folder).

⚠️ Warning: Deletion is irreversible. Resources referenced by workflow tasks will cause those tasks to fail on next execution.

DS behavior: Deleting a non-empty folder returns error 20018. Use recursive=True to delete children first, then the folder.

Args: resource_id: Resource ID recursive: Delete children recursively (for folders, default False)

Returns: {"resource_id", "status", "deleted_children"}

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

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

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

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

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

Pass DELETE through to DolphinScheduler API.

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)

Pass PUT through to DolphinScheduler API.

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/delete)

• resource: File and folder management (upload/download/view/update/delete)

• monitor: Cluster health monitoring (master/worker status)

• user: User and tenant queries

• raw: Raw API passthrough for advanced use

Args: category: Category name (empty string returns all categories)

Returns: If category is empty: overview with tool counts If category is specified: {category, name, tools, workflow, hint}