dataflow-mcp

An MCP server that lets an AI agent read and launch Google Cloud Dataflow jobs. Works against any GCP project the caller has access to.

Anyone in the org can install it: they authenticate with their own Google credentials, and their own IAM roles decide what they're allowed to do. The server holds no secrets of its own.

Tools

| Tool | Kind | What it does | | --- | --- | --- | | list_jobs | read | List jobs in a project/region (filter ACTIVE / TERMINATED / ALL) | | get_job | read | Get one job (summary, or full for the complete resource) | | get_job_metrics | read | Runtime metrics (throughput, system lag, element counts…) | | list_job_messages | read | Job log/status messages, filterable by importance | | launch_flex_template | write | Launch a job from a Flex Template (gs://…/spec.json) | | launch_classic_template | write | Launch a job from a classic template (e.g. Google's MongoDB→BigQuery) | | cancel_job | write | Cancel a running job | | drain_job | write | Drain a running streaming job (flush buffered data, then stop) | | list_pipelines | read | List recurring Data Pipelines (secrets redacted) | | get_pipeline | read | Get one pipeline's template + params + schedule (secrets redacted) | | clone_pipeline | write | Duplicate a pipeline with overrides — re-point at a different DB/dataset (can auto-create the destination table; created ACTIVE — see warning below) | | run_pipeline | write | Trigger an immediate one-off run of a pipeline | | get_bq_table | read | Inspect a BigQuery table's schema, partitioning, size | | create_bq_table | write | Create a destination BigQuery table (from a schema or by copying another table) | | list_scheduled_queries | read | List BigQuery scheduled queries | | get_scheduled_query | read | Get one scheduled query (SQL, schedule, destination) | | create_scheduled_query | write | Create a scheduled query | | update_scheduled_query | write | Update a scheduled query (SQL, schedule, destination, enable/disable) | | run_scheduled_query | write | Run a scheduled query now, or backfill a time range |

Write tools can be disabled entirely with DATAFLOW_READONLY=true.

Scheduled queries (BigQuery Data Transfer)

BigQuery "scheduled queries" are Data Transfer Service configs (dataSourceId=scheduled_query). They live in a location (US multi-region datasets → us, configurable via DATAFLOW_BQ_LOCATION).

• create_scheduled_query — give it query + schedule; add destinationDataset + destinationTable + writeDisposition to materialize results.
• update_scheduled_query — only the fields you pass change (fetch-merge-patch under the hood).
• run_scheduled_query — one immediate run, or startTime+endTime to backfill.

Schedule syntax is NOT cron. Use Data Transfer syntax: every day 07:00, every mon 09:00, every 6 hours. Times are UTC.

Run-as credentials: for unattended runs, pass serviceAccountName so the query executes as that service account (you need roles/iam.serviceAccountUser on it). Without it, creation relies on the caller's own delegated credentials, which may fail in headless setups.

Creating destination BigQuery tables

So you don't have to hand-create destination tables before a pipeline runs:

• create_bq_table — create a table from an explicit schema, or with likeTable to copy an existing table's schema. The dataset is created automatically if missing. Supports time-partitioning and clustering. e.g.

  "Create your-gcp-project:your_dataset.your_dataset_likes with the same schema as …your_dataset_follows."

• clone_pipeline with createDestinationTable: true — the one-shot path: clone a pipeline, override outputTableSpec, and the new destination table is provisioned automatically by copying the source pipeline's destination table schema. Nothing to create by hand.

Duplicating a job / pipeline with changes

Dataflow does not persist launch parameters on a finished job, so you cannot reliably re-run a job by reading it back. In this project every job is created by a recurring Data Pipeline, which does store the template path, all parameters, and the schedule. So duplication works at the pipeline level:

1. list_pipelines → find the source (e.g. source-pipeline).

2. get_pipeline → see its parameters (database, collection, outputTableSpec, …). Secret values like mongoDbUri are shown as ***redacted*** but are copied intact when cloning.

3. clone_pipeline → create a new pipeline, overriding only what changes:

   "Clone source-pipeline as new-pipeline, overriding collection=community_likes and outputTableSpec=your-gcp-project:your_dataset.your_dataset_likes."

   The template, worker environment, schedule and service account are copied from the source. Set createDestinationTable: true to also provision the BQ table, and runNow: true to fire one immediate test run.

   ⚠️ The Data Pipelines API always creates a pipeline ACTIVE — it starts running on the copied schedule right away. There is no "paused" state: the only way to stop a pipeline is to archive it (stop), which is permanent (an archived pipeline can't be resumed; you'd re-clone a fresh one). So only clone with parameters you're ready to run.

Prerequisites

• Node.js ≥ 18
• The Google Cloud SDK (gcloud)
• IAM access on the target project. The relevant roles:
  • Read jobs → roles/dataflow.viewer
  • Launch/cancel/drain jobs → roles/dataflow.developer (plus roles/iam.serviceAccountUser on the worker service account)
  • Read pipelines → roles/datapipelines.viewer
  • Clone/run pipelines → roles/datapipelines.admin (plus roles/iam.serviceAccountUser on the pipeline's scheduler service account)
  • Read BigQuery tables → roles/bigquery.dataViewer (+ roles/bigquery.jobUser)
  • Create BigQuery tables/datasets → roles/bigquery.dataEditor
  • Read scheduled queries → roles/bigquery.user (transfers.get/list)
  • Create/update/run scheduled queries → roles/bigquery.admin (transfers.update) + roles/iam.serviceAccountUser on any serviceAccountName

Authentication (per user)

The server uses Application Default Credentials (ADC). Each user runs, once:

gcloud auth application-default login

⚠️ ADC is separate from gcloud auth login. It's possible for gcloud dataflow jobs list (CLI identity) to work while the MCP server gets PERMISSION_DENIED (ADC identity) — they can be different accounts. If you hit a permission error, re-run the command above with the account that has Dataflow access, and make sure that account holds roles/dataflow.viewer/developer on the project.

A service-account key also works for headless/CI use — set GOOGLE_APPLICATION_CREDENTIALS=/path/to/key.json.

Install & build

npm install
npm run build

Configure your MCP client

Claude Code

claude mcp add dataflow -- node /absolute/path/to/dataflow-mcp/dist/index.js

Claude Desktop / generic MCP config

Add to your client's mcpServers config:

{
  "mcpServers": {
    "dataflow": {
      "command": "node",
      "args": ["/absolute/path/to/dataflow-mcp/dist/index.js"],
      "env": {
        "DATAFLOW_PROJECT": "your-gcp-project",
        "DATAFLOW_REGION": "us-central1"
      }
    }
  }
}

Configuration (env vars)

| Var | Default | Meaning | | --- | --- | --- | | DATAFLOW_PROJECT | ADC project | Default project when a tool omits project | | DATAFLOW_REGION | us-central1 | Default region | | DATAFLOW_READONLY | false | true hides all write tools | | DATAFLOW_QUOTA_PROJECT | = DATAFLOW_PROJECT | Project billed for API quota. Defaults to the target project so calls don't route to your gcloud config default project (where the APIs may be disabled). Needs serviceusage.services.use. | | DATAFLOW_BQ_LOCATION | us | Location for scheduled queries (match your dataset: us, eu, us-central1, …). |

Every tool also accepts explicit project and region args to override the defaults per call.

Example prompts

• "List the active Dataflow jobs in your-gcp-project."
• "Show me the error messages for job 2026-07-07_12_00_01-15515335311634922154."
• "Launch the MongoDB-to-BigQuery classic template at gs://…/MongoDB_to_BigQuery named mongo-sync-adhoc with these parameters …"
• "Drain the streaming job named events-ingest."

Publishing to the org

Two easy paths so teammates don't clone the repo:

1. npm (private registry or public scope): npm publish, then teammates run npx dataflow-mcp as the MCP command.
2. Git: teammates git clone, npm install && npm run build, and point their client at dist/index.js.

In both cases each user still runs gcloud auth application-default login with their own account — no shared credentials.