Building a CI/CD Automation Agent with Pydantic AI: Complete Guide

Building a CI/CD Automation Agent with Pydantic AI: Complete Guide

Continuous Integration and Continuous Deployment (CI/CD) pipelines have become the backbone of modern software delivery. However, managing these pipelines—writing YAML configs, diagnosing failures, rolling back deployments, and optimizing build times—remains a tedious, error-prone process. By combining the structured output guarantees of Pydantic with the reasoning capabilities of Large Language Models, Pydantic AI offers a compelling framework for building autonomous agents that can reason about, operate on, and improve your CI/CD workflows.

In this guide, you'll learn what a CI/CD automation agent is, why Pydantic AI is uniquely suited for the task, and how to build a production-ready agent that can inspect pipeline runs, diagnose failures, suggest fixes, and trigger redeployments.

What Is a CI/CD Automation Agent?

A CI/CD automation agent is an LLM-powered system that interacts with your delivery infrastructure as an intelligent operator. Unlike a static script, an agent can:

• Observe pipeline state through tool calls (GitHub Actions API, Jenkins, GitLab CI, ArgoCD, etc.)
• Reason about failures by reading logs, comparing against historical runs, and forming hypotheses
• Act by opening pull requests with fixes, re-running failed jobs, or rolling back deployments
• Report structured findings to engineers via Slack, email, or dashboards

The key distinction from a chatbot is that the agent operates in a loop: it plans, calls tools, observes results, and revises its plan until the goal is met or it determines it needs human input.

Why Pydantic AI?

Pydantic AI is a Python agent framework built by the same team behind Pydantic. It brings several properties that matter enormously for CI/CD automation:

Type-Safe Tool Outputs

When an agent calls a tool like get_failed_jobs, the return value is validated against a Pydantic model. This means the LLM receives well-formed, predictable data instead of free-form strings, dramatically reducing hallucinations and parsing errors.

Structured Final Responses

You can declare a typed result_type for your agent. For a CI/CD agent, this might be a DiagnosisReport with fields like root_cause, confidence, suggested_fix, and action_taken. Downstream systems can consume this output without fragile regex parsing.

Dependency Injection

Pydantic AI supports a typed Deps container, making it trivial to inject API clients, configuration, and mock services for testing.

Model-Agnostic Design

You can swap between OpenAI, Anthropic, Gemini, or local models without rewriting your agent logic—useful when balancing cost, latency, and capability.

Project Setup

Let's build the agent. Start by creating a project directory and installing dependencies:

mkdir cicd-agent && cd cicd-agent
python -m venv .venv
source .venv/bin/activate

pip install pydantic-ai httpx pydantic python-dotenv

Create a .env file with your credentials:

OPENAI_API_KEY=sk-...
GITHUB_TOKEN=ghp_...
GITHUB_REPOSITORY=your-org/your-repo

Defining the Domain Models

Before writing the agent, define the structured types that flow through the system. Create models.py:

from enum import Enum
from pydantic import BaseModel, Field
from typing import Optional


class JobStatus(str, Enum):
    SUCCESS = "success"
    FAILURE = "failure"
    CANCELLED = "cancelled"
    IN_PROGRESS = "in_progress"


class PipelineJob(BaseModel):
    name: str
    status: JobStatus
    conclusion: Optional[JobStatus] = None
    html_url: str
    started_at: str
    completed_at: Optional[str] = None
    steps: list[str] = Field(default_factory=list)


class PipelineRun(BaseModel):
    run_id: int
    branch: str
    commit_sha: str
    commit_message: str
    status: JobStatus
    event: str
    html_url: str
    jobs: list[PipelineJob] = Field(default_factory=list)


class LogSnippet(BaseModel):
    job_name: str
    line_number: int
    text: str


class DiagnosisReport(BaseModel):
    run_id: int
    root_cause: str = Field(description="Concise explanation of why the pipeline failed")
    confidence: float = Field(ge=0.0, le=1.0)
    failing_jobs: list[str]
    suggested_fix: str = Field(description="Actionable fix an engineer can apply")
    action_taken: Optional[str] = Field(
        default=None,
        description="If the agent took action, describe it; otherwise null"
    )

These models serve as the contract between your tools, the LLM, and downstream consumers. The DiagnosisReport is the agent's final structured output.

Building the GitHub Actions Client

Create github_client.py to wrap the GitHub REST API. Keeping this separate from the agent makes it easy to test and swap providers later.

import os
import httpx
from models import PipelineRun, PipelineJob, JobStatus, LogSnippet


class GitHubActionsClient:
    def __init__(self, token: str, repository: str):
        self.token = token
        self.owner, self.repo = repository.split("/")
        self.base_url = "https://api.github.com"
        self.headers = {
            "Authorization": f"Bearer {token}",
            "Accept": "application/vnd.github+json",
            "X-GitHub-Api-Version": "2022-11-28",
        }

    def list_recent_runs(self, limit: int = 10) -> list[PipelineRun]:
        url = f"{self.base_url}/repos/{self.owner}/{self.repo}/actions/runs"
        params = {"per_page": limit}
        with httpx.Client(headers=self.headers, timeout=30) as client:
            resp = client.get(url, params=params)
            resp.raise_for_status()
            data = resp.json()

        runs = []
        for wf in data.get("workflow_runs", []):
            runs.append(
                PipelineRun(
                    run_id=wf["id"],
                    branch=wf["head_branch"],
                    commit_sha=wf["head_sha"],
                    commit_message=wf.get("display_title", ""),
                    status=JobStatus(wf["status"]),
                    event=wf["event"],
                    html_url=wf["html_url"],
                )
            )
        return runs

    def get_jobs(self, run_id: int) -> list[PipelineJob]:
        url = f"{self.base_url}/repos/{self.owner}/{self.repo}/actions/runs/{run_id}/jobs"
        with httpx.Client(headers=self.headers, timeout=30) as client:
            resp = client.get(url)
            resp.raise_for_status()
            data = resp.json()

        jobs = []
        for j in data.get("jobs", []):
            conclusion = JobStatus(j["conclusion"]) if j.get("conclusion") else None
            steps = [s["name"] for s in j.get("steps", [])]
            jobs.append(
                PipelineJob(
                    name=j["name"],
                    status=JobStatus(j["status"]),
                    conclusion=conclusion,
                    html_url=j["html_url"],
                    started_at=j["started_at"],
                    completed_at=j.get("completed_at"),
                    steps=steps,
                )
            )
        return jobs

    def get_logs(self, run_id: int, job_name: str, max_lines: int = 200) -> list[LogSnippet]:
        url = f"{self.base_url}/repos/{self.owner}/{self.repo}/actions/runs/{run_id}/logs"
        with httpx.Client(headers=self.headers, timeout=60) as client:
            resp = client.get(url)
            if resp.status_code != 200:
                return []
            # The logs endpoint returns a zip archive; for simplicity we
            # capture the raw bytes and search for the job name.
            text = resp.content.decode("utf-8", errors="ignore")

        snippets = []
        for i, line in enumerate(text.splitlines()):
            if job_name in line or "error" in line.lower():
                snippets.append(LogSnippet(job_name=job_name, line_number=i, text=line[:500]))
                if len(snippets) >= max_lines:
                    break
        return snippets

    def rerun_failed_jobs(self, run_id: int) -> bool:
        url = f"{self.base_url}/repos/{self.owner}/{self.repo}/actions/runs/{run_id}/rerun-failed-jobs"
        with httpx.Client(headers=self.headers, timeout=30) as client:
            resp = client.post(url)
            return resp.status_code in (200, 201)

This client exposes four operations: listing runs, fetching jobs, retrieving logs, and re-running failed jobs. Each returns validated Pydantic models.

Defining the Agent

Now create agent.py. This is where Pydantic AI ties everything together.

import os
from dataclasses import dataclass
from dotenv import load_dotenv
from pydantic_ai import Agent, RunContext
from github_client import GitHubActionsClient
from models import DiagnosisReport, PipelineRun, JobStatus

load_dotenv()


@dataclass
class Deps:
    gh: GitHubActionsClient
    allow_rerun: bool = True


system_prompt = """\
You are a senior DevOps engineer responsible for diagnosing CI/CD failures.

Your workflow:
1. Call list_recent_runs to find the most recent pipeline runs.
2. Identify any run whose status is FAILURE.
3. For each failing run, call get_jobs to find which jobs failed.
4. For each failing job, call get_logs to retrieve log snippets.
5. Analyze the logs and form a root cause hypothesis.
6. If the failure looks transient (network timeout, flaky test, runner issue)
   and allow_rerun is True, call rerun_failed_jobs.
7. Produce a DiagnosisReport with your findings.

Be precise. Quote log lines when relevant. Do not invent commands or files
that do not appear in the logs.
"""

agent = Agent(
    model="openai:gpt-4o",
    deps_type=Deps,
    output_type=DiagnosisReport,
    system_prompt=system_prompt,
)


@agent.tool
def list_recent_runs(ctx: RunContext[Deps], limit: int = 10) -> list[PipelineRun]:
    """List the most recent workflow runs in the repository."""
    return ctx.deps.gh.list_recent_runs(limit=limit)


@agent.tool
def get_jobs(ctx: RunContext[Deps], run_id: int) -> list:
    """Get all jobs for a given workflow run, including their statuses."""
    from models import PipelineJob
    return ctx.deps.gh.get_jobs(run_id)


@agent.tool
def get_logs(ctx: RunContext[Deps], run_id: int, job_name: str) -> list:
    """Retrieve log snippets for a specific job in a workflow run."""
    from models import LogSnippet
    return ctx.deps.gh.get_logs(run_id, job_name)


@agent.tool
def rerun_failed_jobs(ctx: RunContext[Deps], run_id: int) -> str:
    """Re-run only the failed jobs of a workflow run. Use only for transient failures."""
    if not ctx.deps.allow_rerun:
        return "Reruns are disabled in this configuration."
    success = ctx.deps.gh.rerun_failed_jobs(run_id)
    return "Rerun triggered successfully." if success else "Rerun request failed."


async def diagnose_latest_failure() -> DiagnosisReport:
    gh = GitHubActionsClient(
        token=os.environ["GITHUB_TOKEN"],
        repository=os.environ["GITHUB_REPOSITORY"],
    )
    deps = Deps(gh=gh, allow_rerun=True)
    result = await agent.run(
        "Diagnose the most recent failing CI/CD run in this repository.",
        deps=deps,
    )
    return result.output

Notice how each tool is decorated with @agent.tool and receives a RunContext[Deps]. The output_type=DiagnosisReport declaration tells Pydantic AI to enforce that the final answer conforms to our schema.

Running the Agent

Create main.py to invoke the agent and print the structured report:

import asyncio
import json
from agent import diagnose_latest_failure


async def main():
    report = await diagnose_latest_failure()
    print(json.dumps(report.model_dump(), indent=2))


if __name__ == "__main__":
    asyncio.run(main())

Run it:

python main.py

A typical output looks like:

{
  "run_id": 88421309,
  "root_cause": "The 'test' job failed because pytest could not connect to the Postgres service container. Log line 412 shows 'connection refused' on port 5432, indicating the service had not finished booting when tests started.",
  "confidence": 0.85,
  "failing_jobs": ["test"],
  "suggested_fix": "Add a health check step before running pytest, e.g. 'wait-for-it postgres:5432 -- pytest'.",
  "action_taken": "Rerun triggered successfully."
}

Integrating with Slack Notifications

A diagnosis is only useful if it reaches the right people. Let's extend the agent to post results to Slack. Add a small notifier:

import os
import httpx
from models import DiagnosisReport


def post_to_slack(report: DiagnosisReport, webhook_url: str) -> None:
    blocks = [
        {"type": "header", "text": {"type": "plain_text", "text": "CI/CD Diagnosis Report"}},
        {"type": "section", "text": {"type": "mrkdwn", "text": f"*Run:* <https://github.com/runs/{report.run_id}|#{report.run_id}>\n*Confidence:* {report.confidence}"}},
        {"type": "section", "text": {"type": "mrkdwn", "text": f"*Root Cause:*\n{report.root_cause}"}},
        {"type": "section", "text": {"type": "mrkdwn", "text": f"*Suggested Fix:*\n{report.suggested_fix}"}},
    ]
    if report.action_taken:
        blocks.append({"type": "section", "text": {"type": "mrkdwn", "text": f"*Action Taken:* {report.action_taken}"}})

    with httpx.Client(timeout=10) as client:
        client.post(webhook_url, json={"blocks": blocks})


if __name__ == "__main__":
    # Example usage after a diagnosis run
    import asyncio
    from agent import diagnose_latest_failure

    async def go():
        report = await diagnose_latest_failure()
        post_to_slack(report, os.environ["SLACK_WEBHOOK_URL"])
        print("Posted to Slack.")

    asyncio.run(go())

Best Practices

1. Keep Tools Small and Composable

Each tool should do one thing and return a typed model. Avoid "god tools" that fetch everything at once—the LLM reasons better when it can choose which tool to call next.

2. Validate Tool Outputs Aggressively

Use Pydantic validators on your domain models. If the GitHub API returns unexpected data, fail fast at the tool boundary rather than letting malformed data confuse the model.

3. Constrain Autonomous Actions

Actions like re-running jobs, merging PRs, or rolling back deployments should be gated behind flags (like allow_rerun) or require explicit human approval. An agent that can act should also be able to ask for permission.

4. Log Every Tool Call

For auditability, wrap your tools to record inputs, outputs, and timestamps. This is invaluable when reviewing why an agent took a particular action.

import logging
import functools

logger = logging.getLogger("cicd_agent")


def logged_tool(fn):
    @functools.wraps(fn)
    async def wrapper(ctx, *args, **kwargs):
        logger.info("tool_call name=%s args=%s", fn.__name__, args)
        result = await fn(ctx, *args, **kwargs)
        logger.info("tool_result name=%s result_len=%s", fn.__name__, len(str(result)))
        return result
    return wrapper

5. Use Streaming for Long-Running Diagnoses

Pydantic AI supports streaming. For interactive use cases, stream the agent's intermediate messages so engineers can watch the diagnosis unfold in real time.

6. Test with Mock Dependencies

Because dependencies are injected via the Deps container, you can substitute a fake GitHubActionsClient in tests. This lets you verify agent behavior against known failure scenarios without hitting real APIs.

import pytest
from agent import agent, Deps
from models import PipelineRun, PipelineJob, JobStatus


class FakeClient:
    def list_recent_runs(self, limit=10):
        return [PipelineRun(
            run_id=1, branch="main", commit_sha="abc",
            commit_message="bump deps", status=JobStatus.FAILURE,
            event="push", html_url="https://example.com/1",
        )]

    def get_jobs(self, run_id):
        return [PipelineJob(
            name="test", status=JobStatus.FAILURE, conclusion=JobStatus.FAILURE,
            html_url="https://example.com/jobs/1", started_at="2024-01-01T00:00:00Z",
            steps=["pytest"],
        )]

    def get_logs(self, run_id, job_name):
        from models import LogSnippet
        return [LogSnippet(job_name="test", line_number=10, text="ERROR connection refused")]

    def rerun_failed_jobs(self, run_id):
        return True


@pytest.mark.asyncio
async def test_agent_diagnoses_failure():
    deps = Deps(gh=FakeClient(), allow_rerun=False)
    result = await agent.run("Diagnose the latest failure.", deps=deps)
    report = result.output
    assert report.run_id == 1
    assert "connection refused" in report.root_cause.lower()
    assert report.action_taken is None

7. Choose the Right Model for the Task

Complex log analysis benefits from a strong reasoning model like GPT-4o or Claude 3.5 Sonnet. Simple status checks can use a cheaper, faster model. Pydantic AI's model-agnostic design lets you route different tasks to different models.

Extending the Agent

Once the core diagnosis loop works, you can extend the agent with additional tools:

• open_pr_with_fix — generate a patch and open a pull request
• rollback_deployment — trigger an ArgoCD rollback to a previous revision
• compare_with_last_green_run — diff failing run against the last successful one to isolate regressions
• query_metrics — pull build duration trends from Prometheus to detect flaky tests
• notify_on_call — page the on-call engineer via PagerDuty for high-severity failures

Each new tool follows the same pattern: define a typed return model, implement the function, decorate it with @agent.tool, and update the system prompt to explain when to use it.

Conclusion

Building a CI/CD automation agent with Pydantic AI gives you a system that combines the flexibility of LLM reasoning with the safety of typed, validated data. By wrapping your delivery infrastructure in well-defined tools, injecting dependencies cleanly, and enforcing a structured output schema, you create an agent that can diagnose failures, take measured actions, and report findings in a format your existing systems can consume directly. Start small with a single diagnosis loop, add tools incrementally, gate autonomous actions behind explicit permissions, and you'll have a reliable teammate that scales with your pipeline complexity.