flow-agent-mcp

An MCP server that exposes Estuary Flow pipeline operations as typed tools, plus three runnable demo pipelines built on public datasets. It wraps Estuary's flowctl CLI so an agent can drive Flow directly.

What it does

Each tool wraps a real flowctl command with typed, validated inputs and returns structured JSON. An agent (or a person through Claude Code) can inspect a pipeline's health, publish captures and materializations, edit and restart tasks, and read landed documents, all through tool calls rather than hand-typed CLI commands.

Tools

Every tool maps to a real flowctl invocation, echoed back in the response as command so the underlying call is always visible.

Tool	Wraps
list_tasks	catalog list --captures/--materializations/--collections --output json
task_status	catalog status <task> --output json [--connected]
task_stats	raw stats --task <task> --since <dur>
task_logs	logs --task <task> --since <dur> (parsed, level-filtered)
task_health	status + stats + error logs in one call
discover_bindings	discover --source flow.yaml
create_capture	catalog publish --source flow.yaml --auto-approve
create_materialization	catalog publish --source flow.yaml --auto-approve
pull_specs	catalog pull-specs --name <task>
set_task_enabled	pull specs → toggle shards.disable → publish
restart_task	the disable/verify/enable sequence
read_collection	collections read --collection <name> --output json [--since <dur>]
generate_synthetic_data	demo record generator (no flowctl)

Notes on behavior worth knowing:

• set_task_enabled handles the spec surgery. It pulls the live spec, toggles shards.disable cleanly, and republishes, so enable/disable is one call rather than a manual edit-and-publish loop.
• Failures return flowctl's own message as a structured MCP error result instead of throwing, so the calling agent gets actionable text.
• flowctl publishes from files, not inline specs, so specs are written to a scratch directory and cleaned up after.

Setup

npm install
npm test          # 127 checks, no flowctl or Estuary account needed

Add to Claude Code:

claude mcp add estuary-action -- npx tsx /path/to/estuary-mcp/src/index.ts

Real operations need flowctl on PATH and flowctl auth login. Set ESTUARY_DRY_RUN=true to echo commands without executing them, or FLOWCTL_PATH to point at a specific binary.

Demo pipelines

Each demo builds a complete catalog — capture, collections, derivation, materialization — from a public dataset shape.

Demo	Dataset	Exercises
gpu	Cloud GPU Price Index	HTTP ingest, filter derivation, BigQuery fan-out
weather	Open-Meteo (live, no API key)	polling an open API, threshold alerts, Postgres
property	Property Comps	Postgres CDC, aggregation with reduce annotations, Snowflake

Generate a catalog and publish it:

npm run catalog -- gpu acmeCo/gpu > flow.yaml
flowctl catalog publish --source flow.yaml --auto-approve

Generate data to feed a pipeline:

npm run demo:gpu -- 500                 # NDJSON to POST at the ingest endpoint
npm run demo:weather                    # live Open-Meteo readings
npm run demo:weather -- --watch 300     # poll every 5 minutes
npm run demo:property -- 1000 --sql     # INSERTs for the CDC source table
npm run demo:property -- --ddl          # table + replication slot + publication

The weather demo pushes directly to Estuary when WEATHER_INGEST_URL and WEATHER_INGEST_TOKEN are set. The ingest URL is only shown in the dashboard after publishing, under the capture's Endpoints section.

Aggregation correctness

The property rollup sums sales per market. Aggregating derivations emit one delta document per source row and rely on schema reduce: annotations to fold deltas that share a key. The top-level reduce: { strategy: merge } combines matching deltas and is set alongside the per-field strategies. aggregateSchema() in src/specs.ts applies both levels, and test/catalog.test.mts asserts they are present.

Layout

src/
  index.ts          MCP server and tool definitions
  flowctl.ts        subprocess wrapper, NDJSON parsing, scratch spec files
  specs.ts          catalog spec builders (captures, derivations, reductions)
  shards.ts         shards.disable spec surgery
  yaml.ts           literal-block YAML so SQL lambdas survive serialization
  demos/            three end-to-end pipelines
  synthetic/        record generators
test/
  shards.test.mts   disable/enable spec editing
  catalog.test.mts  spec shape, key/schema/reduce invariants
  mcp-client.test.mts  real MCP transport, command construction, validation

Status

Tested (offline, 127 checks): tool registration and schema validation over a real MCP client connection, exact flowctl command construction for every tool, shard spec editing, output parsing, and catalog invariants across all three demos.

Verified against a live Estuary tenant. A full source-to-destination pipeline was published and run end to end:

• an HTTP-ingest capture landing documents into a collection,
• a SQLite derivation filtering those readings into threshold-based alerts,
• schema inference widening a field's bounds and republishing the task on its own,
• a Postgres materialization writing both collections into an external database, confirmed by querying that database directly (8 readings, 3 alerts).