Agentic Control Plane for Data Platform
When to Use
- Enabling AI agents (Claude Code, custom agents) to operate the data platform through a standardized MCP interface
- Building a natural language chat interface for data platform operations ("trigger the orders ETL for yesterday")
- Creating a multi-agent governance system where specialized agents monitor cost, reliability, and compliance
- Exposing platform tools to any MCP-compatible AI client without custom integrations per client
- Implementing agentic workflows that observe platform state and take autonomous corrective actions
MCP Server for Data Platform
Server Definition (FastMCP)
# platform_mcp_server.py
from mcp.server.fastmcp import FastMCP
from datetime import datetime, timedelta
import subprocess, json, logging
logger = logging.getLogger(__name__)
# Initialize MCP server — name shown in Claude/agent UIs
mcp = FastMCP(
"data-platform",
instructions="""You are the data platform control plane.
Use these tools to manage Airflow DAGs, Kafka topics, Trino queries, and dbt jobs.
Always confirm destructive operations before executing.
Log a reason for every action you take."""
)
Platform Tools
@mcp.tool()
async def list_airflow_dags(
active_only: bool = True,
tag: str | None = None,
) -> str:
"""List all Airflow DAGs with their current status.
Args:
active_only: If True, only return non-paused DAGs
tag: Filter by tag (e.g., 'gold-layer', 'kafka', 'orders')
"""
from sqlalchemy import create_engine, text
engine = create_engine(AIRFLOW_DB_URI)
query = """
SELECT dag_id, is_paused, is_active,
tags,
last_parsed_time,
next_dagrun
FROM dag
WHERE (:active_only = FALSE OR is_active = TRUE)
AND (:tag IS NULL OR :tag = ANY(ARRAY(SELECT t.name FROM dag_tag t WHERE t.dag_id = dag.dag_id)))
ORDER BY dag_id LIMIT 50
"""
rows = engine.execute(text(query), {"active_only": active_only, "tag": tag}).fetchall()
return json.dumps([dict(r) for r in rows], default=str)
@mcp.tool()
async def trigger_airflow_dag(
dag_id: str,
conf: dict | None = None,
reason: str = "triggered by AI agent",
) -> str:
"""Trigger an Airflow DAG run.
Args:
dag_id: The DAG identifier to trigger
conf: Optional configuration dictionary passed to the DAG
reason: Human-readable reason for triggering (required for audit)
"""
from airflow.api.client.local_client import Client
import uuid
client = Client(None, None)
run_id = f"agent_{uuid.uuid4().hex[:8]}"
logger.info(f"Agent triggering DAG {dag_id}: {reason}")
client.trigger_dag(
dag_id=dag_id,
run_id=run_id,
conf={**(conf or {}), "_agent_reason": reason, "_triggered_at": datetime.utcnow().isoformat()},
)
return json.dumps({"run_id": run_id, "dag_id": dag_id, "status": "triggered", "reason": reason})
@mcp.tool()
async def get_dag_run_status(dag_id: str, run_id: str) -> str:
"""Get the status of a specific DAG run.
Args:
dag_id: The DAG identifier
run_id: The run identifier returned by trigger_airflow_dag
"""
from sqlalchemy import create_engine, text
engine = create_engine(AIRFLOW_DB_URI)
row = engine.execute(text("""
SELECT dr.state, dr.start_date, dr.end_date,
COUNT(CASE WHEN ti.state = 'failed' THEN 1 END) AS failed_tasks,
COUNT(CASE WHEN ti.state = 'success' THEN 1 END) AS success_tasks,
COUNT(*) AS total_tasks
FROM dag_run dr
LEFT JOIN task_instance ti ON dr.dag_id = ti.dag_id AND dr.run_id = ti.run_id
WHERE dr.dag_id = :dag_id AND dr.run_id = :run_id
GROUP BY dr.state, dr.start_date, dr.end_date
"""), {"dag_id": dag_id, "run_id": run_id}).fetchone()
if not row:
return json.dumps({"error": f"Run {run_id} not found for DAG {dag_id}"})
return json.dumps({
"dag_id": dag_id, "run_id": run_id,
"state": row.state,
"started_at": str(row.start_date),
"finished_at": str(row.end_date),
"tasks": {"total": row.total_tasks, "success": row.success_tasks, "failed": row.failed_tasks},
}, default=str)
@mcp.tool()
async def query_trino(
sql: str,
catalog: str = "iceberg",
schema: str = "gold",
max_rows: int = 100,
) -> str:
"""Execute a read-only SQL query against Trino.
Args:
sql: SELECT query to execute (INSERT/UPDATE/DROP not permitted)
catalog: Trino catalog (default: iceberg)
schema: Default schema for the query
max_rows: Maximum rows to return (default: 100, max: 1000)
Returns:
JSON with columns and rows
"""
# Safety: only allow read operations
normalized = sql.strip().upper()
if not (normalized.startswith("SELECT") or normalized.startswith("WITH") or normalized.startswith("SHOW") or normalized.startswith("DESCRIBE")):
return json.dumps({"error": "Only SELECT, WITH, SHOW, DESCRIBE queries are permitted"})
max_rows = min(max_rows, 1000)
from trino.dbapi import connect
conn = connect(host=TRINO_HOST, port=443, user="agent", catalog=catalog, schema=schema)
cur = conn.cursor()
cur.execute(sql)
rows = cur.fetchmany(max_rows)
cols = [d[0] for d in cur.description]
return json.dumps({"columns": cols, "rows": rows, "row_count": len(rows)})
@mcp.tool()
async def get_kafka_consumer_lag(
group: str | None = None,
topic: str | None = None,
) -> str:
"""Get Kafka consumer group lag.
Args:
group: Consumer group name (omit for all groups)
topic: Filter by topic name
"""
cmd = ["kafka-consumer-groups.sh", "--bootstrap-server", KAFKA_BOOTSTRAP, "--describe"]
if group:
cmd.extend(["--group", group])
else:
cmd.append("--all-groups")
result = subprocess.run(cmd, capture_output=True, text=True, timeout=30)
if result.returncode != 0:
return json.dumps({"error": result.stderr})
# Parse output: GROUP TOPIC PARTITION CURRENT-OFFSET LOG-END-OFFSET LAG
lines = [l for l in result.stdout.split("\n") if l.strip() and not l.startswith("GROUP")]
parsed = []
for line in lines:
parts = line.split()
if len(parts) >= 6:
try:
lag = int(parts[5]) if parts[5] != "-" else 0
if topic is None or parts[1] == topic:
parsed.append({
"group": parts[0], "topic": parts[1], "partition": int(parts[2]),
"current_offset": parts[3], "log_end_offset": parts[4], "lag": lag,
})
except (ValueError, IndexError):
continue
# Aggregate by group+topic
from collections import defaultdict
agg = defaultdict(int)
for p in parsed:
agg[f"{p['group']}::{p['topic']}"] += p["lag"]
return json.dumps([
{"group": k.split("::")[0], "topic": k.split("::")[1], "total_lag": v}
for k, v in sorted(agg.items(), key=lambda x: x[1], reverse=True)
])
@mcp.tool()
async def create_kafka_topic(
name: str,
partitions: int,
retention_hours: int = 168,
reason: str = "created by AI agent",
) -> str:
"""Create a new Kafka topic. Requires naming convention: {env}.{domain}.{entity}.v{N}.
Args:
name: Topic name following convention: prod.orders.created.v1
partitions: Number of partitions (1-96)
retention_hours: Data retention in hours (default: 168 = 7 days)
reason: Reason for creating the topic (required for audit)
"""
import re
if not re.match(r'^(prod|staging|dev)\.[a-z]+\.[a-z_]+\.v\d+$', name):
return json.dumps({"error": "Topic name must match: {env}.{domain}.{entity}.v{N}"})
if partitions < 1 or partitions > 96:
return json.dumps({"error": "Partitions must be between 1 and 96"})
logger.info(f"Agent creating Kafka topic {name}: {reason}")
result = subprocess.run([
"kafka-topics.sh", "--bootstrap-server", KAFKA_BOOTSTRAP,
"--create", "--topic", name,
"--partitions", str(partitions),
"--replication-factor", "3",
"--config", "min.insync.replicas=2",
"--config", f"retention.ms={retention_hours * 3600000}",
], capture_output=True, text=True, timeout=30)
if result.returncode != 0:
return json.dumps({"error": result.stderr})
return json.dumps({"topic": name, "partitions": partitions, "status": "created", "reason": reason})
@mcp.tool()
async def get_platform_health() -> str:
"""Get overall data platform health: Airflow scheduler heartbeat, Kafka cluster status, recent pipeline failures."""
from sqlalchemy import create_engine, text
engine = create_engine(AIRFLOW_DB_URI)
# Scheduler heartbeat
heartbeat = engine.execute(text("""
SELECT TIMESTAMPDIFF(SECOND, latest_heartbeat, NOW()) AS age_sec
FROM job WHERE job_type = 'SchedulerJob' ORDER BY latest_heartbeat DESC LIMIT 1
""")).scalar()
# Failed DAGs in last 1h
failed_dags = engine.execute(text("""
SELECT COUNT(*) FROM dag_run
WHERE state = 'failed' AND start_date > NOW() - INTERVAL '1' HOUR
""")).scalar()
# Kafka broker status
brokers_raw = subprocess.run(
["kafka-broker-api-versions.sh", "--bootstrap-server", KAFKA_BOOTSTRAP],
capture_output=True, text=True, timeout=10
)
return json.dumps({
"scheduler_heartbeat_age_sec": heartbeat,
"scheduler_healthy": heartbeat < 30 if heartbeat else False,
"failed_dag_runs_1h": failed_dags,
"kafka_brokers_reachable": brokers_raw.returncode == 0,
"checked_at": datetime.utcnow().isoformat(),
})
Running the MCP Server
# Entry point
if __name__ == "__main__":
import sys
# STDIO transport for Claude Desktop / Claude Code
mcp.run(transport="stdio")
# Start with SSE transport for web clients
uvicorn platform_mcp_server:mcp.get_app() --host 0.0.0.0 --port 8080
Claude Desktop Configuration
{
"mcpServers": {
"data-platform": {
"command": "python",
"args": ["/opt/platform/platform_mcp_server.py"],
"env": {
"AIRFLOW_DB_URI": "postgresql://airflow:secret@airflow-db:5432/airflow",
"KAFKA_BOOTSTRAP": "kafka:9092",
"TRINO_HOST": "trino.internal"
}
}
}
}
Multi-Agent Governance
# governance_agents.py — specialized agents that continuously monitor the platform
import anthropic
import schedule, time
client = anthropic.Anthropic()
def run_cost_governance_agent():
"""Cost governance agent: detects expensive queries, idle resources, over-retained topics."""
response = client.messages.create(
model="claude-haiku-4-5-20251001",
max_tokens=2048,
system="""You are a cost governance agent for a data platform.
Use tools to identify cost waste and generate a prioritized action list.
Check: 1) queries with > 100 GB scanned today, 2) topics with retention > 30 days,
3) idle deployments with 0 CPU for > 7 days.
Output a JSON list of findings with priority and estimated savings.""",
tools=TOOLS, # reuse platform MCP tools
messages=[{"role": "user", "content": "Run cost governance audit for today."}],
)
_process_agent_response(response, "cost_governance")
def run_reliability_agent():
"""Reliability agent: checks SLA compliance, consumer lag, failed DAGs."""
response = client.messages.create(
model="claude-haiku-4-5-20251001",
max_tokens=2048,
system="""You are a reliability agent. Check platform health every 15 minutes.
If any consumer lag > 50000 messages OR any critical DAG failed in the last hour,
immediately trigger healing actions and notify the team.
Otherwise, confirm 'All systems healthy'.""",
tools=TOOLS,
messages=[{"role": "user", "content": "Run reliability check."}],
)
_process_agent_response(response, "reliability")
# Schedule agents
schedule.every(15).minutes.do(run_reliability_agent)
schedule.every(1).hours.do(run_cost_governance_agent)
if __name__ == "__main__":
while True:
schedule.run_pending()
time.sleep(60)
Kubernetes Deployment
apiVersion: apps/v1
kind: Deployment
metadata:
name: platform-mcp-server
namespace: platform
spec:
replicas: 1
selector:
matchLabels:
app: platform-mcp-server
template:
spec:
containers:
- name: mcp-server
image: ghcr.io/myorg/platform-mcp-server:latest
command: ["uvicorn", "platform_mcp_server:app", "--host", "0.0.0.0", "--port", "8080"]
env:
- name: AIRFLOW_DB_URI
valueFrom:
secretKeyRef:
name: platform-secrets
key: airflow-db-uri
- name: KAFKA_BOOTSTRAP
value: "production-kafka-bootstrap.kafka.svc.cluster.local:9093"
resources:
requests:
cpu: "0.25"
memory: 256Mi
limits:
cpu: "1"
memory: 512Mi
Anti-Patterns
- No authorization on MCP tools — every tool callable by any agent with access; implement tool-level authorization (read-only tools vs write tools vs admin tools).
- Stateful agents without checkpointing — long-running agentic workflows lose state on failure; checkpoint agent progress to a store and design for resumption.
- Agents that can call each other recursively — two agents triggering each other creates infinite loops; enforce a maximum call depth.
- MCP server with production DB credentials in env — MCP servers are accessible to any connected agent; use read-only replicas for query tools, and separate credentials per tool scope.
- Agentic platform ops without rollback — an agent that applies 50 VPA changes in one cycle is hard to roll back; batch changes in small sets with validation between batches.
References
- MCP Python SDK:
modelcontextprotocol.io/docs/develop/build-server
- FastMCP:
github.com/jlowin/fastmcp
- Claude tool use:
platform.claude.com/docs/en/docs/build-with-claude/tool-use
- Anthropic agentic patterns:
anthropic.com/research/building-effective-agents
- Related skills:
[[platform-engineering-data-platform-api]], [[platform-engineering-internal-developer-platform]], [[aiops-autonomous-incident-response]], [[dataops-self-healing-platform]]