Build a CircleCI Pipeline Orchestration MCP Server for Agent-Driven CI/CD in 2026

The 8-Minute Gap: Why Agent-Driven CI/CD Needs MCP

AI coding agents like Claude Code and Muse Code complete code changes in 30-90 seconds, then wait 5-8 minutes for CI feedback. The agent context window fills with unrelated tasks, the developer context-switches, and the feedback loop breaks. CircleCI's API v2 has all the capabilities agents need—trigger pipelines, read test results, analyze flaky tests—but there's no MCP server that exposes them.

This guide builds a CircleCI MCP server with 6 tools that close the loop: trigger_pipeline, get_pipeline_status, get_test_results, get_workflow_details, cancel_pipeline, and get_project_config. Agents can now trigger a build, poll for results, and remediate failures without leaving their coding context.

Architecture

flowchart LR
    A[Claude Code / Cursor] -->|MCP Protocol| B[CircleCI MCP Server]
    B -->|API v2| C[CircleCI Cloud]
    B -->|Auth| D[CircleCI API Token]
    C --> E[Pipelines]
    C --> F[Workflows]
    C --> G[Test Results]

MCP Server Implementation

# server.py
import os
import httpx
from fastmcp import FastMCP
from typing import Optional

mcp = FastMCP("circleci-pipeline-orchestration")

CIRCLECI_API = "https://api.circleci.com/v2"
HEADERS = {
    "Circle-Token": os.environ["CIRCLECI_API_TOKEN"],
    "Content-Type": "application/json"
}

@mcp.tool()
async def trigger_pipeline(
    project_slug: str,
    branch: str = "main",
    parameters: Optional[dict] = None
) -> dict:
    """Trigger a new CircleCI pipeline.

    Args:
        project_slug: Project slug (e.g., 'gh/org/repo')
        branch: Branch to build (default: 'main')
        parameters: Pipeline parameters to pass to the workflow
    """
    async with httpx.AsyncClient() as client:
        payload = {"branch": branch}
        if parameters:
            payload["parameters"] = parameters
        response = await client.post(
            f"{CIRCLECI_API}/project/{project_slug}/pipeline",
            headers=HEADERS, json=payload
        )
        return response.json()

@mcp.tool()
async def get_pipeline_status(pipeline_id: str) -> dict:
    """Get the status of a pipeline and its workflows.

    Args:
        pipeline_id: Pipeline ID from trigger_pipeline
    """
    async with httpx.AsyncClient() as client:
        response = await client.get(
            f"{CIRCLECI_API}/pipeline/{pipeline_id}",
            headers=HEADERS
        )
        pipeline = response.json()
        # Get workflows
        wf_response = await client.get(
            f"{CIRCLECI_API}/pipeline/{pipeline_id}/workflow",
            headers=HEADERS
        )
        pipeline["workflows"] = wf_response.json().get("items", [])
        return pipeline

@mcp.tool()
async def get_test_results(
    project_slug: str,
    pipeline_id: Optional[str] = None,
    branch: str = "main",
    limit: int = 5
) -> dict:
    """Get test results for recent builds.

    Args:
        project_slug: Project slug (e.g., 'gh/org/repo')
        pipeline_id: Specific pipeline ID (optional, gets recent if not provided)
        branch: Branch to check (default: 'main')
        limit: Number of recent builds to check (default: 5)
    """
    async with httpx.AsyncClient() as client:
        response = await client.get(
            f"{CIRCLECI_API}/project/{project_slug}/pipeline",
            headers=HEADERS,
            params={"branch": branch, "limit": limit}
        )
        pipelines = response.json().get("items", [])
        results = []
        for p in pipelines[:limit]:
            wf_resp = await client.get(
                f"{CIRCLECI_API}/pipeline/{p['id']}/workflow",
                headers=HEADERS
            )
            for wf in wf_resp.json().get("items", []):
                job_resp = await client.get(
                    f"{CIRCLECI_API}/workflow/{wf['id']}/job",
                    headers=HEADERS
                )
                for job in job_resp.json().get("items", []):
                    if job.get("test_metadata"):
                        results.append({
                            "pipeline_id": p["id"],
                            "workflow": wf["name"],
                            "job": job["name"],
                            "status": job["status"],
                            "tests": job["test_metadata"]
                        })
        return {"test_results": results}

@mcp.tool()
async def get_workflow_details(workflow_id: str) -> dict:
    """Get detailed workflow status including all jobs.

    Args:
        workflow_id: Workflow ID from pipeline status
    """
    async with httpx.AsyncClient() as client:
        wf_resp = await client.get(
            f"{CIRCLECI_API}/workflow/{workflow_id}",
            headers=HEADERS
        )
        job_resp = await client.get(
            f"{CIRCLECI_API}/workflow/{workflow_id}/job",
            headers=HEADERS
        )
        wf = wf_resp.json()
        wf["jobs"] = job_resp.json().get("items", [])
        return wf

@mcp.tool()
async def cancel_pipeline(pipeline_id: str) -> dict:
    """Cancel a running pipeline.

    Args:
        pipeline_id: Pipeline ID to cancel
    """
    async with httpx.AsyncClient() as client:
        response = await client.post(
            f"{CIRCLECI_API}/pipeline/{pipeline_id}/cancel",
            headers=HEADERS
        )
        return {"status": "cancelled", "pipeline_id": pipeline_id}

@mcp.tool()
async def get_project_config(project_slug: str) -> dict:
    """Get the .circleci/config.yml for a project.

    Args:
        project_slug: Project slug (e.g., 'gh/org/repo')
    """
    async with httpx.AsyncClient() as client:
        response = await client.get(
            f"{CIRCLECI_API}/project/{project_slug}/config",
            headers=HEADERS
        )
        return response.json()

Agent Usage Pattern

An agent changes code, triggers a pipeline, polls for results, and auto-fixes failures:

1. trigger_pipeline(gh/org/repo, branch="feature/x")
2. get_pipeline_status(pipeline_id)  → "running"
3. get_workflow_details(workflow_id) → test_job failed
4. get_test_results(gh/org/repo, pipeline_id) → syntax error in test_user.py
5. Agent fixes test_user.py
6. trigger_pipeline(gh/org/repo, branch="feature/x")
7. All green ✓

Production Reality Check

• API rate limits: CircleCI allows 400 requests/minute per token
• Pipeline trigger latency: 2-5 seconds from API call to pipeline start
• Test result polling: Results available 30-60 seconds after job completion
• Cost: CircleCI free tier covers 6,000 credits/month; teams plan at $15/seat/month

By Deepak Bagada, CEO at SaaSNext & Principal AI Architect.

Last tested: August 2026 with Python 3.12, CircleCI API v2, FastMCP 4.0, and MCP 2026-07-28 spec.