MCP Tool Integrations: Wiring LLM Agents Into GitLab, Airflow, and MLflow

It’s late night and a training pipeline failed. To find out why, you do what everyone does: open GitLab to find the red pipeline, open the failing job’s log, scroll through four thousand lines, find the step that called Airflow, open the Airflow UI to find the DAG run, open the failed task’s log, then open MLflow to see whether the metrics actually regressed — and only then start forming a hypothesis. Four systems, four auth models, four query languages, and the clock is running.

The bottleneck was never the model. It was that the model had no eyes on the systems where the answer already lived.

This post is about fixing that with MCP (Model Context Protocol), connecting an LLM agent to internal data sources and developer tooling so that “which stage broke and why” goes from tens of minutes of context-switching to a few minutes. The integrations that moved the needle were GitLab CI/CD, Airflow DAGs, and MLflow/Metaflow.

The One Diagram

                          ┌──────────────────────────────────────┐
                          │            Agent host                │
                          │  (Claude / IDE / your agent loop)    │
                          │  system prompt + tool-calling model  │
                          └───────────────┬──────────────────────┘
                                          │  MCP (JSON-RPC 2.0)
                     ┌────────────────────┼─────────────────────┐
                     │                    │                     │
             ┌───────▼──────┐    ┌────────▼───────┐    ┌────────▼───────┐
             │ MCP: gitlab  │    │ MCP: airflow   │    │ MCP: mlflow    │
             │ list_failed_ │    │ dag_run_status │    │ compare_runs   │
             │ pipelines    │    │ failing_tasks  │    │ run_metrics    │
             │ job_log_     │    │ task_log_      │    │ (metaflow too) │
             │ excerpt      │    │ excerpt        │    │                │
             └───────┬──────┘    └────────┬───────┘    └────────┬───────┘
                     │ REST               │ REST/DB            │ REST
             ┌───────▼──────┐    ┌────────▼───────┐    ┌────────▼───────┐
             │ GitLab API   │    │ Airflow        │    │ MLflow         │
             │ CI/CD        │    │ metadata + logs│    │ tracking server│
             └──────────────┘    └────────────────┘    └────────────────┘

          Audit + cache: every tool call and result → Postgres / Redis

The agent never talks to GitLab, Airflow, or MLflow directly. It talks to small, typed MCP servers you control, which in turn talk to the real systems with least-privilege credentials. That indirection is where the safety and the reliability come from.

Before and After

The model didn’t get smarter. It got tools and a playbook.

What MCP Actually Is (and Isn’t)

MCP is an open protocol — JSON-RPC 2.0 over a transport — that standardizes how an LLM application talks to external capabilities. A server exposes three primitives:

• Tools — callable functions with typed inputs/outputs (list_failed_pipelines, task_log_excerpt). This is 95% of what you’ll build.
• Resources — readable data addressed by URI, for things the host wants to fetch selectively (a run’s metrics, a DAG’s definition).
• Prompts — reusable instruction templates the host can surface to the user.

Two transports matter:

• stdio — the server runs as a local subprocess; great for developer machines and IDEs.
• streamable HTTP — the server runs as a service; what you want for a shared, air-gapped-by-VPC triage agent.

And what it isn’t: MCP is not an agent framework, not a model, and not a vector database. It’s a port standard — the USB-C between a model and a tool. Write a server once and any MCP-capable client can use it, which is why it’s worth the small amount of ceremony.

Designing Tools, Not Endpoints

The single biggest mistake is exposing a generic passthrough: gitlab_request(method, path). It hands the model an unbounded API surface, invites prompt-injection into your infrastructure, and produces enormous, useless payloads.

Expose task-shaped tools instead. Narrow, typed, and honest about what they return:

• Typed in and out. The schema is the contract; let the SDK generate it from type hints.
• Compact results. Return IDs, statuses, timestamps, and links — not raw blobs. job_log_excerpt returns a bounded excerpt, never the full trace.
• Read-only by default. Anything that mutates (retry a pipeline, clear a task, trigger a DAG) is a separate tool, gated behind human approval.
• Bounded. Hard timeouts, page-size caps, and truncation so one call can’t hang the agent or flood its context.
• Self-describing. Tool descriptions are prompts. The model picks tools almost entirely by their description and schema, so write them like API docs for a very literal reader.

Building the Servers

Three small servers, one per system, each with its own scoped token. The official Python SDK ships a FastMCP helper that turns type-hinted functions into MCP tools.

GitLab CI/CD

# mcp_gitlab/server.py
import os

import httpx
from mcp.server.fastmcp import FastMCP

mcp = FastMCP("gitlab-ci")

GITLAB = os.environ["GITLAB_URL"]
# A read-only project/group token — never a personal admin token.
client = httpx.Client(
    base_url=f"{GITLAB}/api/v4",
    headers={"PRIVATE-TOKEN": os.environ["GITLAB_READ_TOKEN"]},
    timeout=20.0,
)


def _project_id(project: str) -> str:
    # Accept "group/name"; URL-encode for the API path.
    return httpx.URL("").copy_with(path=project).path.lstrip("/").replace("/", "%2F")


@mcp.tool()
def list_failed_pipelines(project: str, since_iso: str, limit: int = 10) -> list[dict]:
    """List failed pipelines in a project updated since an ISO-8601 timestamp,
    newest first. Returns id, ref, short sha, created_at, and web_url."""
    r = client.get(
        f"/projects/{_project_id(project)}/pipelines",
        params={"status": "failed", "updated_after": since_iso, "per_page": limit},
    )
    r.raise_for_status()
    return [
        {"id": p["id"], "ref": p["ref"], "sha": p["sha"][:8],
         "created_at": p["created_at"], "web_url": p["web_url"]}
        for p in r.json()
    ]


@mcp.tool()
def failing_jobs(project: str, pipeline_id: int) -> list[dict]:
    """List the jobs in a pipeline that did not succeed, with stage and name."""
    r = client.get(f"/projects/{_project_id(project)}/pipelines/{pipeline_id}/jobs",
                   params={"per_page": 100})
    r.raise_for_status()
    bad = [j for j in r.json() if j["status"] not in ("success", "skipped", "manual")]
    return [{"id": j["id"], "name": j["name"], "stage": j["stage"],
             "status": j["status"], "web_url": j["web_url"]} for j in bad]


@mcp.tool()
def job_log_excerpt(project: str, job_id: int, grep: str | None = None,
                    max_lines: int = 200) -> str:
    """Return an excerpt of a job's log: lines matching `grep` if given, else the
    tail. Bounded to max_lines so it never floods the context window."""
    trace = client.get(f"/projects/{_project_id(project)}/jobs/{job_id}/trace").text
    lines = trace.splitlines()
    if grep:
        lines = [ln for ln in lines if grep.lower() in ln.lower()]
    return "\n".join(lines[-max_lines:])

Airflow DAGs

# mcp_airflow/server.py
import httpx
from mcp.server.fastmcp import FastMCP

mcp = FastMCP("airflow")
api = httpx.Client(
    base_url=f'{os.environ["AIRFLOW_URL"]}/api/v1',
    headers={"Authorization": f'Bearer {os.environ["AIRFLOW_TOKEN"]}'},
    timeout=20.0,
)


@mcp.tool()
def dag_run_status(dag_id: str, run_id: str) -> dict:
    """Get one DAG run's state, start/end times, and execution date."""
    r = api.get(f"/dags/{dag_id}/dagRuns/{run_id}")
    r.raise_for_status()
    d = r.json()
    return {"state": d["state"], "start_date": d.get("start_date"),
            "end_date": d.get("end_date"), "logical_date": d.get("logical_date")}


@mcp.tool()
def failed_tasks(dag_id: str, run_id: str) -> list[dict]:
    """List task instances in a DAG run that failed, with try_number and duration."""
    r = api.get(f"/dags/{dag_id}/dagRuns/{run_id}/taskInstances",
                params={"state": "failed"})
    r.raise_for_status()
    return [{"task_id": t["task_id"], "try_number": t["try_number"],
             "duration": t.get("duration")} for t in r.json()["task_instances"]]


@mcp.tool()
def task_instance_history(dag_id: str, task_id: str, limit: int = 5) -> list[dict]:
    """Recent runs of a task. Distinguish a one-off flake from a new regression."""
    r = api.get(f"/dags/{dag_id}/dagRuns", params={"order_by": "-execution_date",
                "limit": limit})
    r.raise_for_status()
    out = []
    for run in r.json()["dag_runs"]:
        ti = api.get(f"/dags/{dag_id}/dagRuns/{run['dag_run_id']}"
                     f"/taskInstances/{task_id}").json()
        out.append({"run_id": run["dag_run_id"], "state": ti.get("state"),
                    "try_number": ti.get("try_number")})
    return out

Task logs in Airflow aren’t a clean REST resource everywhere; expose them via the same worker you’d use to read them, or a small endpoint, and reuse the exact job_log_excerpt pattern (grep + tail + bound).

MLflow / Metaflow

# mcp_mlflow/server.py
from mlflow.tracking import MlflowClient
from mcp.server.fastmcp import FastMCP

mcp = FastMCP("mlflow")
client = MlflowClient(tracking_uri=os.environ["MLFLOW_TRACKING_URI"])


@mcp.tool()
def list_recent_runs(experiment: str, limit: int = 10) -> list[dict]:
    """Recent runs in an experiment: id, name, status, start_time."""
    exp = client.get_experiment_by_name(experiment)
    runs = client.search_runs([exp.experiment_id],
                              order_by=["attributes.start_time DESC"],
                              max_results=limit)
    return [{"run_id": r.info.run_id, "status": r.info.status,
             "start_time": r.info.start_time} for r in runs]


@mcp.tool()
def compare_runs(run_ids: list[str], metric: str) -> list[dict]:
    """Compare one metric across runs. Answers 'did this regress vs last good run?'"""
    out = []
    for rid in run_ids:
        m = client.get_run(rid).data.metrics
        out.append({"run_id": rid, metric: m.get(metric)})
    return out

Metaflow follows the same shape: list_flows, latest_runs(flow), step_log_excerpt(run_id, step). Same pattern, different backend.

One server per system, or one monolith? Prefer one server per system. Each gets its own least-privilege credential, its own deploy, and its own blast radius; a prompt-injection bug in the GitLab server can’t touch Airflow. Compose them in the agent host.

The Triage Agent (the Loop)

The servers are the plumbing. The playbook is the product. A “pipeline failure triage” agent is a loop: discover tools, then follow a fixed investigation order, accumulating evidence until it can name a root cause.

# agent/triage.py
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client

SERVERS = {
    "gitlab":  StdioServerParameters(command="python", args=["-m", "mcp_gitlab.server"]),
    "airflow": StdioServerParameters(command="python", args=["-m", "mcp_airflow.server"]),
    "mlflow":  StdioServerParameters(command="python", args=["-m", "mcp_mlflow.server"]),
}

async def open_sessions():
    """Open one MCP session per server and collect their tool schemas."""
    sessions, tools = {}, []
    for name, params in SERVERS.items():
        read, write = await stdio_client(params).__aenter__()
        session = await ClientSession(read, write).__aenter__()
        await session.initialize()
        sessions[name] = session
        tools += [t.model_dump() for t in (await session.list_tools()).tools]
    return sessions, tools

The model then gets tools and a system prompt that encodes the playbook:

You are a pipeline-failure triage agent. Investigate in this order and cite
evidence for every claim using the ids/links returned by tools:

1. list_failed_pipelines(project, since)         -> pick the failing pipeline
2. failing_jobs(project, pipeline_id)            -> find the failed stage
3. job_log_excerpt(..., grep='Error'|'Exception')-> the failure signature
4. If the job triggered Airflow:
     dag_run_status -> failed_tasks -> task_log_excerpt
5. If an ML run is involved:
     list_recent_runs -> compare_runs([bad, last_good], metric)
6. task_instance_history(...) to separate a flake from a regression.

Stop as soon as you can name a root cause. Never state a cause you cannot tie
to a tool result. Cap yourself at 8 tool calls.

A real answer from this loop looks like:

Root cause: Airflow task `train_model` failed in dag `fraud_training`,
run `scheduled__2026-10-14T02:00:00Z` (state: failed, try 1/3).

Evidence:
 - GitLab pipeline 8842 (sha a1b2c3d) failed stage `train`; job `train-and-log`.
 - Task log: ValueError: feature column 'chargeback_7d' not found.
 - MLflow run 3f9c1e: n_features 41 vs 42 on the last good run 55aa02.
 - task_instance_history: last 5 runs succeeded -> not a flake.

Likely cause: upstream feature rename in commit a1b2c3d dropped a column.
Suggested fix: update the feature list or pin the dataset version, then rerun.

The two things that make this trustworthy: every claim cites a tool result, and the agent is capped (tool calls, tokens) so it can’t spiral.

Context Is the Real Constraint

The reason this works at all — and the reason naive attempts fail — is that tool outputs, not model capability, dominate the context window. A single raw CI trace can blow it.

Design every tool around that reality:

• grep before tail before return. job_log_excerpt(grep="Error") first; only fall back to the tail.
• Return structure, not text. Counts, IDs, statuses, and short values beat paragraphs.
• Let the agent ask for more. Bounded results force a second, targeted call instead of one giant dump.
• Use resources for big read-only data. Expose a run’s full metrics as an MCP resource the host fetches selectively, rather than stuffing it into a tool result.

Security and Guardrails

Giving a model access to your infrastructure is a serious move. Do it deliberately:

• Least privilege, per system. Separate, read-scoped tokens for GitLab (project/group), Airflow, MLflow. Never a shared admin token.
• Read-only by default. Writes — retry pipeline, clear task, trigger DAG — are separate tools behind explicit human approval, never in the default toolset.
• No arbitrary passthrough. Enumerate tools; never let the model supply a URL, path, or shell command to the backing API.
• Scrub secrets. Regex out tokens, PRIVATE-TOKEN, and env dumps from any log excerpt before returning it.
• Protect the backends. Rate limits, timeouts, page caps, and query-parameter allowlists so the agent can’t accidentally DoS GitLab or Airflow.
• Audit everything. Log every tool call and result to Postgres: actor, tool, args, latency, result size, and the final diagnosis. This becomes both your security trail and your incident memory.
• Keep egress in mind. Run the MCP servers inside your VPC. With a hosted model, only compact tool outputs leave; or self-host the model (vLLM) so nothing leaves at all.

Observability and Evaluation

You can’t improve a diagnosis loop you don’t measure:

• Instrument tool calls: latency per backend, error rates, token spend per incident, number of tool calls to resolution.
• Track MTTR before and after, split by failure class (CI, scheduler, model/data).
• Build an eval set from past incidents: (pipeline_id, true_root_cause). Score whether the agent finds the cause, in how many tool calls and tokens.
• Require citations. Reject conclusions that don’t reference a tool-returned id or link — the cheapest defense against confident hallucinations.

What Actually Changed

• Mean time to diagnosis for “which stage broke and why” went from tens of minutes to a few. The on-call stopped being a human API gateway between four UIs.
• Consistency. The playbook is in the system prompt and the audit log, not in one senior engineer’s head, so a new on-call gets the same investigation order.
• Institutional memory. The audit log plus the eval set turn each incident into something the next one benefits from.
• Honest caveats. A vague tool description or a stale cache will send the agent down a rabbit hole, and the agent still can’t fix a flaky host — it just points at it faster. It’s a force multiplier on diagnosis, not a replacement for judgment.

The Gotchas Nobody Mentions First

Tool descriptions are prompts. Terse or ambiguous descriptions cause wrong tool selection. Write them like documentation for a literal reader.

Too many tools beat correct tools. Past a few dozen, selection quality drops. Keep the set small and composable; prefer five sharp tools over thirty thin ones.

Returning raw logs is the #1 self-inflicted wound. It burns budget, degrades reasoning, and buries the signal. Bound everything.

Stale data lies. Cache with short TTLs, and make sure returned IDs and tags are current — an agent arguing with a five-minute-old cache looks broken.

Tokens rotate. Read tokens expire and get revoked. Surface auth failures as explicit, actionable tool errors rather than empty results the agent misreads as “no failures.”

Pick the transport on purpose. stdio for local/IDE use; streamable HTTP, with real authn, for the shared agent. Don’t expose a stdio server over the network by accident.

Version your tools. Renaming a tool or changing its arguments silently breaks every agent that used it. Treat the tool surface as a public API.

TL;DR

• The model wasn’t the bottleneck; the lack of tools was. MCP gave the agent typed eyes on GitLab, Airflow, and MLflow/Metaflow.
• Build small task-shaped tools, not API passthroughs: typed, bounded, read-only by default, with descriptions written as prompts.
• Run one MCP server per system, each with least-privilege credentials, so blast radius and deploys stay independent.
• The value is the playbook: a fixed investigation order (pipeline → failing job → log excerpt → DAG run → task log → MLflow compare → flake-vs-regression) that every claim must cite from a tool result.
• Context is the constraint. grep, tail, structure, and bound every output; use resources for big read-only data.
• Guardrails are non-negotiable: no passthrough, no default writes, scrub secrets, rate-limit backends, audit every call.
• Measure it: tool-call latency, token spend, and MTTR on an eval set of past incidents. Require citations to catch hallucinations.
• Result: mean pipeline-failure diagnosis dropped from tens of minutes to a few minutes, and the investigation order stopped living in one person’s head.

The lesson generalizes: before you reach for a bigger model, give the one you have better tools and a narrower job.