Connect pipeline agents via MCP for autonomous orchestration

Introduction

When a data quality check fails at 3 AM, an on-call engineer must wake up, diagnose the failure, decide whether to retry or roll back, and trigger the fix manually—every step is rote, and every minute of delay costs freshness and trust. By the end of this lesson, you'll be able to connect pipeline agents to operational tooling via the Model Context Protocol (MCP) so they can plan and execute these remediation steps autonomously, escalating only for high-blast-radius decisions.

Key Terminology

• Model Context Protocol (MCP): an open protocol that lets an LLM agent discover and invoke external tools through a uniform, schema-described interface, so the same orchestration agent can drive Dagster, Argo, or DVC without backend-specific glue.
• MCP tool: a single named operation registered on an MCP server (e.g. trigger_materialization, check_workflow_status, promote_dataset) with a JSON Schema for its inputs and a structured result the agent can reason over.
• Confirmation gate: the boundary that splits agent actions by blast radius — low-risk calls (status reads, single-asset retries) execute immediately, while high-risk calls (rollback, promotion, scale changes) return a pending-approval response that a human must release before execution.

Concepts

An MCP server exposes pipeline operations — trigger_materialization, check_workflow_status, promote_dataset — as typed tool calls a reasoning LLM can discover and invoke. Each tool carries a name, description, and JSON Schema, which together let the orchestration agent plan a multi-step remediation (read status → decide retry-vs-rollback → act) without bespoke glue per backend. The agent's autonomy is bounded by a confirmation gate: low-blast-radius actions (retrying a failed asset, querying state) execute immediately, while high-blast-radius actions (rollback, promotion, scale changes) return a pending-approval response that a human must release through a notification channel. Every tool invocation and the agent's reasoning trace are logged so on-call engineers can audit decisions after the fact.

Code Walkthrough

Building on the Concepts section's framing of typed MCP tools and the confirmation-gate boundary, this section grounds those ideas in a concrete server. The following Python module implements an MCP server that exposes three pipeline operations as tools: triggering a Dagster asset materialization, querying Argo Workflow status, and promoting a dataset version by tagging it in Git. The server uses the mcp Python SDK to register tools with typed schemas and handle invocations from connected agents.

1from mcp.server import Server
2from mcp.types import Tool, TextContent
3import httpx
4import subprocess
5import json
6
7server = Server("pipeline-orchestrator")
8
9@server.tool()
10async def trigger_materialization(asset_key: str, partition: str = None) -> str:
11    """Trigger Dagster asset materialization via GraphQL API."""
12    query = """
13    mutation($assetKey: [String!]!, $partition: String) {
14        launchRun(executionParams: {
15            selector: { assetSelection: [{ path: $assetKey }] }
16            runConfigData: { partition: $partition }
17        }) { run { runId status } }
18    }
19    """
20    variables = {"assetKey": [asset_key], "partition": partition}
21    async with httpx.AsyncClient() as client:
22        resp = await client.post(
23            "http://dagster-webserver:3000/graphql",
24            json={"query": query, "variables": variables},
25            timeout=30,
26        )
27    result = resp.json()
28    run_id = result["data"]["launchRun"]["run"]["runId"]
29    return json.dumps({"run_id": run_id, "status": "LAUNCHED"})
30
31@server.tool()
32async def check_workflow_status(workflow_name: str) -> str:
33    """Query Argo Workflow phase and step-level details."""
34    async with httpx.AsyncClient() as client:
35        resp = await client.get(
36            f"http://argo-server:2746/api/v1/workflows/pipelines/{workflow_name}",
37            timeout=10,
38        )
39    wf = resp.json()
40    phase = wf["status"].get("phase", "Unknown")
41    nodes = wf["status"].get("nodes", {})
42    steps = {k: v.get("phase") for k, v in nodes.items() if v.get("type") == "Pod"}
43    return json.dumps({"phase": phase, "steps": steps})
44
45@server.tool()
46async def promote_dataset(version_tag: str, git_ref: str = "HEAD") -> str:
47    """Tag a dataset version for production promotion."""
48    result = subprocess.run(
49        ["git", "tag", "-a", version_tag, git_ref, "-m", f"Promote {version_tag}"],
50        capture_output=True, text=True, check=True,
51    )
52    subprocess.run(["git", "push", "origin", version_tag], check=True)
53    return json.dumps({"tag": version_tag, "ref": git_ref, "status": "promoted"})

The orchestration agent that connects to this MCP server uses an LLM to reason about pipeline state and plan multi-step remediation actions. When the agent receives a notification that a quality check has failed, it follows a decision process: first, it calls check_workflow_status to understand which step failed and why. If the failure is transient (network timeout, temporary resource exhaustion), it calls trigger_materialization to retry the failed asset. If the failure is persistent (data corruption, schema mismatch), the agent calls promote_dataset with a previous known-good version tag to roll back the pipeline to a stable state. Throughout this process, every tool call and the agent's reasoning trace are logged for human audit.

The critical safety mechanism in MCP-based agent orchestration is the confirmation gate. High-impact operations -- deleting datasets, promoting models to production serving, scaling cluster capacity to maximum -- must require explicit human approval before execution. The MCP server implements this by returning a confirmation request instead of executing the operation directly. The agent presents the confirmation to the human operator via a notification channel (Slack, PagerDuty), and the operation proceeds only after approval. This pattern gives you the efficiency of autonomous orchestration for routine operations while maintaining human oversight for decisions with significant blast radius.

Do's and Don'ts

Now that you have worked through the implementation, the practices below separate a durable approach from a fragile one.

Do's

1. ✓Register every pipeline action as an MCP tool with a typed JSON Schema and a description that names the backend (Dagster asset, Argo workflow, DVC tag) so the agent's planner can pick the right tool without ambiguity.
2. ✓Put a confirmation gate in front of high-blast-radius tools (promote_dataset, deletes, scale changes) — return a pending-approval payload and require a human ack via Slack or PagerDuty before the MCP server actually executes the call.
3. ✓Log every tool invocation, its arguments, the returned result, and the agent's reasoning trace to an audit store so on-call engineers can reconstruct an autonomous remediation after the fact.

Don'ts

1. ✗Don't let the orchestration agent call promote_dataset or any destructive operation without the confirmation gate — even a model that "looks confident" will occasionally hallucinate a rollback target and silently pin the wrong DVC version to production.
2. ✗Don't bypass MCP by hardcoding Dagster GraphQL or Argo REST calls into the agent's prompt; you lose the tool-discovery, schema validation, and audit logging that make autonomous orchestration safe to operate at 3 AM.