..........
For a more detailed brief, check the github: README
The Problem
Pipeline failures in Fivetran occur in one of four layers:
| Layer | What it owns |
|---|---|
| Connector | Source authentication, sync scheduling, network reachability |
| Schema | Column presence, type compatibility, inclusion/exclusion config |
| Transformation | Downstream SQL/dbt model execution, dependency resolution |
| Data Quality | Row counts, null rates, duplicate rates, value range violations |
Each layer surfaces different signals, requires different API calls to investigate, and has a different remediation path. A failure in the connector layer looks like a schema failure downstream. A schema change breaks a transformation that raises a data quality alert. The layers are coupled; the failure origin is not obvious from the symptom.
Data engineers spend hours analysing incidents by manually cross-referencing the Fivetran dashboard, dbt run logs, and warehouse query outputs — before they can even begin to act. The signals exist. The APIs exist. The bottleneck is the triage loop.
PRE closes that loop — with a human in control of every write action.
What it does
PRE is a multi-agent incident response system for Fivetran data pipelines. When a connection fails, PRE:
- Triggers a live Fivetran sync and monitors the outcome in real time
- Classifies the root cause layer deterministically — using a rule-based scorer before any LLM is invoked
- Dispatches a specialist agent for that layer, scoped to its own read-only Fivetran API toolset
- Gates any destructive fix behind a human approval decision — blocking execution until the operator acts
- Executes the approved action via the Fivetran API, then runs post-fix validation to confirm recovery
Three real Fivetran connectors are held in genuinely broken states for the demo. Every run makes live Fivetran API calls.
- Connection 1 — HITL flow: PRE detects the sync failure, diagnoses the root cause, queues a fix, and blocks until you approve or deny. Approving executes a real Fivetran API write and generates a Connect Card URL.
- Connection 2 — Multi-agent handoff: sync succeeds, transformation fails. Schema agent investigates the type mismatch; findings are handed to the transformation agent.
- Connection 3 — Schema diagnosis: transformation fails with a missing column. Schema agent traces the root cause to the sync configuration.
How we built it
Tech stack
| Layer | Technology |
|---|---|
| Agent framework | Google ADK 2.1.0 |
| Models | Gemini 3.1 Flash Lite on Vertex AI |
| Tool integration | Fivetran MCP Server |
| HITL gate | Async Python function blocking runner.run_async() via asyncio.Queue |
| Demo server | FastAPI + Server-Sent Events + Alpine.js |
| Eval persistence | Firestore (Cloud Run) |
| Observability | OpenTelemetry → Google Cloud Trace |
| Deployment | Google Cloud Run |
Multi-agent pipeline
Each failure layer gets its own specialist agent with a filtered tool set. A single agent with 77 tools available would face an unmanageable tool selection problem and could trigger a write without sufficient context.
| Agent | Tools exposed | Scope |
|---|---|---|
| Connector | get_connection_details, run_connection_setup_tests, list_connections_in_group |
Auth, sync status, network reachability |
| Schema | get_connection_schema_config, get_connection_column_config, list_transformations |
Column type drift, exclusions, schema diffs |
| Transformation | get_transformation_details, run_transformation, get_connection_details |
Execution errors, dependency failures, re-run |
| Data Quality | get_connection_details, get_connection_schema_config, get_transformation_details |
Null surges, duplicate rates, row count anomalies |
Write tools are inaccessible during diagnosis — only available to the orchestrator after HITL approval.
Classification: score_layers() maps each incoming signal type to a per-layer score using a rule-based rubric. classify() applies confidence thresholds and a pipeline-stack tiebreaker (connector → schema → transformation → data_quality). Deterministic scoring; LLM dispatches.
AgentTool: Sub-agents are registered as AgentTool, not sub_agents. transfer_to_agent ends the orchestrator's turn permanently — the orchestrator never reads the finding or fires the gate. With AgentTool, output returns as a FunctionResponse and the orchestrator continues, enabling the full classify → dispatch → HITL → validate lifecycle in a single coherent session.
HITL gate: request_approval is an async def ADK tool that awaits gate_queue.get(), blocking runner.run_async() at the tool level until the human decides.
Fivetran MCP server: A custom server exposing 77 tools over stdio transport. Two separate McpToolset instances per run: read-only for diagnosis agents (FIVETRAN_ALLOW_WRITES=false), write-enabled for the orchestrator after approval.
Challenges we ran into
LLM non-determinism in a structured pipeline
Gemini wraps JSON outputs in markdown fences, invents field names that don't match Pydantic models, produces reasoning prose instead of calling close_incident, and treats async 200 responses as evidence of completion. Every agent boundary required a normalization layer (_strip_markdown_json, _coerce_agent_finding, field aliasing) to absorb these variations before Pydantic validation.
The HITL gate race condition alone took five distinct architectural fixes. The model kept calling the approved write tool in the same response batch as request_approval. LongRunningFunctionTool pauses the outer loop, but the model's retries happen inside a single runner.run_async() call — unreachable from outside. Making request_approval a native async def that suspends the runner itself eliminated the race entirely.
Fivetran write operations return 200 on acceptance, not completion. Gemini consistently treated those 200s as evidence of state change, producing false resolved outcomes. We developed a write-then-verify principle — every write must be followed by a read-tool call confirming terminal state — and applied it across all five agents.
Accomplishments that we're proud of
The HITL gate: A truly blocking async approval gate that suspends the ADK runner until a human decides — at the architecture level, not via prompt instructions. Five iterations. This is what makes PRE safe to deploy against real infrastructure.
Live Fivetran integration at every layer: Every demo run makes real API calls against three genuinely broken connectors. The Connect Card URL generated for Connection 1 is a real Fivetran OAuth re-authentication page produced through the MCP server.
Production hardening without over-engineering:
| Concern | Mechanism |
|---|---|
| Transient 429s | HttpRetryOptions(initial_delay=2, attempts=3) on all 6 agents |
| Sustained quota exhaustion | Outer retry loop: 3 attempts, 10s / 30s / 60s backoff |
| Run timeout | asyncio.wait_for() with structured escalation fallback |
| Tool hallucination | 3-strike force-break: same TOOL_NOT_FOUND ×3 → runner exits |
| Reasoning loops | Per-agent dispatch cap: same agent dispatched >2× → close_incident forced |
| Prompt injection | Input sanitization: per-field truncation, control-char stripping |
| Unstructured closure | Synthesis fallback from last AgentFinding in session state |
| Eval persistence | _persist_metrics in finally block — every run writes to DB regardless of outcome |
Eval dashboard with two-phase latency: Monitoring phase (sync + transformation poll) and agent phase (LLM inference) tracked separately — the sync poll dominates total run time 3:1 for two of the three connections and would otherwise obscure actual agent performance.
What we learned
AgentTool vs sub_agents is not cosmetic: transfer_to_agent ends the orchestrator's turn permanently. AgentTool returns the sub-agent output as a FunctionResponse and the orchestrator continues. This distinction determines whether the full incident lifecycle can happen in one session.
LLMs in production need structural guarantees, not just prompt rules: Rules for structured output, correct tool names, and write-then-verify are ignored often enough to require code-level enforcement. close_incident as a tool call (not free text), the 3-strike force-break, the synthesis fallback — each exists because a prompt instruction alone wasn't reliable.
MCP's stdio transport is first-class: The instinct to externalise the MCP server into a separate container is a DDD anti-pattern when the domain boundary doesn't demand it. Our MCP server already satisfies every criterion for a well-bounded context — single domain, anti-corruption layer, read/write separation. Zero-latency subprocess is the right choice for co-located deployment.
What's next for Fivetran Pipeline Reliability
- Broader connector coverage: PRE's four-layer framework is connector-agnostic. The agent prompts are intentionally general-purpose and will work against any Fivetran connector type — not just Google Sheets.
- User-provided credentials: Let operators connect their own Fivetran account from the UI — PRE scans their live connectors without any setup, turning the demo into a real diagnostic tool for any Fivetran user.
Log in or sign up for Devpost to join the conversation.