# 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

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

Add to Claude Code:

```bash
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](https://gpucloudcompare.com/data/) | HTTP ingest, filter derivation, BigQuery fan-out |
| `weather` | [Open-Meteo](https://open-meteo.com) (live, no API key) | polling an open API, threshold alerts, Postgres |
| `property` | [Property Comps](https://api.nwc-advisory.com/docs) | Postgres CDC, aggregation with reduce annotations, Snowflake |

Generate a catalog and publish it:

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

Generate data to feed a pipeline:

```bash
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).