airbyte_ops_mcp.mcp.github_actions

  1# Copyright (c) 2025 Airbyte, Inc., all rights reserved.
  2"""MCP tools for GitHub workflow and Docker operations.
  3
  4This module provides MCP tools for checking GitHub Actions workflow status,
  5Docker image availability, and other related operations.
  6
  7## MCP reference
  8
  9.. include:: ../../../docs/mcp-generated/github_actions.md
 10    :start-line: 2
 11"""
 12
 13from __future__ import annotations
 14
 15__all__: list[str] = []
 16
 17import re
 18from typing import Annotated
 19
 20import requests
 21from fastmcp import FastMCP
 22from fastmcp_extensions import mcp_tool, register_mcp_tools
 23from pydantic import BaseModel, Field
 24
 25from airbyte_ops_mcp.github_actions import (
 26    get_workflow_jobs,
 27    resolve_default_workflow_branch,
 28    trigger_workflow_dispatch,
 29)
 30from airbyte_ops_mcp.github_api import (
 31    GITHUB_API_BASE,
 32    get_pr_head_ref,
 33    resolve_ci_trigger_github_token,
 34)
 35
 36DOCKERHUB_API_BASE = "https://hub.docker.com/v2"
 37
 38
 39class JobInfo(BaseModel):
 40    """Information about a single job in a workflow run."""
 41
 42    job_id: int
 43    name: str
 44    status: str
 45    conclusion: str | None = None
 46    started_at: str | None = None
 47    completed_at: str | None = None
 48
 49
 50class WorkflowRunStatus(BaseModel):
 51    """Response model for check_ci_workflow_status MCP tool."""
 52
 53    run_id: int
 54    status: str
 55    conclusion: str | None
 56    workflow_name: str
 57    head_branch: str
 58    head_sha: str
 59    html_url: str
 60    created_at: str
 61    updated_at: str
 62    run_started_at: str | None = None
 63    jobs_url: str
 64    jobs: list[JobInfo] = []
 65
 66
 67def _parse_workflow_url(url: str) -> tuple[str, str, int]:
 68    """Parse a GitHub Actions workflow run URL into components.
 69
 70    Args:
 71        url: GitHub Actions workflow run URL
 72            (e.g., "https://github.com/owner/repo/actions/runs/12345")
 73
 74    Returns:
 75        Tuple of (owner, repo, run_id)
 76
 77    Raises:
 78        ValueError: If URL format is invalid.
 79    """
 80    pattern = r"https://github\.com/([^/]+)/([^/]+)/actions/runs/(\d+)"
 81    match = re.match(pattern, url)
 82    if not match:
 83        raise ValueError(
 84            f"Invalid workflow URL format: {url}. "
 85            "Expected format: https://github.com/owner/repo/actions/runs/12345"
 86        )
 87    return match.group(1), match.group(2), int(match.group(3))
 88
 89
 90def _get_workflow_run(
 91    owner: str,
 92    repo: str,
 93    run_id: int,
 94    token: str,
 95) -> dict:
 96    """Get workflow run details from GitHub API.
 97
 98    Args:
 99        owner: Repository owner (e.g., "airbytehq")
100        repo: Repository name (e.g., "airbyte")
101        run_id: Workflow run ID
102        token: GitHub API token
103
104    Returns:
105        Workflow run data dictionary.
106
107    Raises:
108        ValueError: If workflow run not found.
109        requests.HTTPError: If API request fails.
110    """
111    url = f"{GITHUB_API_BASE}/repos/{owner}/{repo}/actions/runs/{run_id}"
112    headers = {
113        "Authorization": f"Bearer {token}",
114        "Accept": "application/vnd.github+json",
115        "X-GitHub-Api-Version": "2022-11-28",
116    }
117
118    response = requests.get(url, headers=headers, timeout=30)
119    if response.status_code == 404:
120        raise ValueError(f"Workflow run {owner}/{repo}/actions/runs/{run_id} not found")
121    response.raise_for_status()
122
123    return response.json()
124
125
126@mcp_tool(
127    read_only=True,
128    idempotent=True,
129    open_world=True,
130)
131def check_ci_workflow_status(
132    workflow_url: Annotated[
133        str | None,
134        Field(
135            description="Full GitHub Actions workflow run URL (e.g., 'https://github.com/owner/repo/actions/runs/12345')"
136        ),
137    ] = None,
138    owner: Annotated[
139        str | None,
140        Field(description="Repository owner (e.g., 'airbytehq')"),
141    ] = None,
142    repo: Annotated[
143        str | None,
144        Field(description="Repository name (e.g., 'airbyte')"),
145    ] = None,
146    run_id: Annotated[
147        int | None,
148        Field(description="Workflow run ID"),
149    ] = None,
150) -> WorkflowRunStatus:
151    """Check the status of a GitHub Actions workflow run.
152
153    You can provide either:
154    - A full workflow URL (workflow_url parameter), OR
155    - The component parts (owner, repo, run_id parameters)
156
157    Returns the current status, conclusion, and other details about the workflow run.
158
159    Uses the CI trigger token (GITHUB_CI_WORKFLOW_TRIGGER_PAT) so that
160    workflow runs in private repositories are accessible.
161    """
162    # Guard: Validate input parameters
163    if workflow_url:
164        # Parse URL to get components
165        owner, repo, run_id = _parse_workflow_url(workflow_url)
166    elif owner and repo and run_id:
167        # Use provided components
168        pass
169    else:
170        raise ValueError(
171            "Must provide either workflow_url OR all of (owner, repo, run_id)"
172        )
173
174    # Guard: Check for required token
175    # Use the CI trigger token (same as trigger_ci_workflow) so that
176    # private-repo workflow runs are accessible.
177    token = resolve_ci_trigger_github_token()
178
179    # Get workflow run details
180    run_data = _get_workflow_run(owner, repo, run_id, token)
181
182    # Get jobs for the workflow run, passing the same token
183    workflow_jobs = get_workflow_jobs(owner, repo, run_id, token=token)
184
185    # Convert dataclass objects to Pydantic models for the response
186    jobs = [
187        JobInfo(
188            job_id=job.job_id,
189            name=job.name,
190            status=job.status,
191            conclusion=job.conclusion,
192            started_at=job.started_at,
193            completed_at=job.completed_at,
194        )
195        for job in workflow_jobs
196    ]
197
198    return WorkflowRunStatus(
199        run_id=run_data["id"],
200        status=run_data["status"],
201        conclusion=run_data["conclusion"],
202        workflow_name=run_data["name"],
203        head_branch=run_data["head_branch"],
204        head_sha=run_data["head_sha"],
205        html_url=run_data["html_url"],
206        created_at=run_data["created_at"],
207        updated_at=run_data["updated_at"],
208        run_started_at=run_data.get("run_started_at"),
209        jobs_url=run_data["jobs_url"],
210        jobs=jobs,
211    )
212
213
214class TriggerCIWorkflowResult(BaseModel):
215    """Response model for trigger_ci_workflow MCP tool."""
216
217    success: bool
218    message: str
219    workflow_url: str
220    run_id: int | None = None
221    run_url: str | None = None
222
223
224@mcp_tool(
225    read_only=False,
226    idempotent=False,
227    open_world=True,
228)
229def trigger_ci_workflow(
230    owner: Annotated[
231        str,
232        Field(description="Repository owner (e.g., 'airbytehq')"),
233    ],
234    repo: Annotated[
235        str,
236        Field(description="Repository name (e.g., 'airbyte')"),
237    ],
238    workflow_file: Annotated[
239        str,
240        Field(description="Workflow file name (e.g., 'connector-regression-test.yml')"),
241    ],
242    workflow_definition_ref: Annotated[
243        str | None,
244        Field(
245            description="Branch name or PR number for the workflow definition to use. "
246            "If a PR number (integer string) is provided, it resolves to the PR's head branch name. "
247            "If a branch name is provided, it is used directly. "
248            "Defaults to 'main' if not specified, "
249            "or AIRBYTE_OPS_DEFAULT_WORKFLOW_BRANCH_OVERRIDE when set for local testing."
250        ),
251    ] = None,
252    inputs: Annotated[
253        dict[str, str] | None,
254        Field(
255            description="Workflow inputs as a dictionary of string key-value pairs. "
256            "These are passed to the workflow_dispatch event."
257        ),
258    ] = None,
259) -> TriggerCIWorkflowResult:
260    """Trigger a GitHub Actions CI workflow via workflow_dispatch.
261
262    This tool triggers a workflow in any GitHub repository that has workflow_dispatch
263    enabled. It resolves PR numbers to branch names automatically since GitHub's
264    workflow_dispatch API only accepts branch names, not refs/pull/{pr}/head format.
265
266    Requires GITHUB_CI_WORKFLOW_TRIGGER_PAT or GITHUB_TOKEN environment variable
267    with 'actions:write' permission.
268    """
269    # Guard: Check for required token
270    token = resolve_ci_trigger_github_token()
271
272    if workflow_definition_ref:
273        if workflow_definition_ref.isdigit():
274            pr_head_info = get_pr_head_ref(
275                owner,
276                repo,
277                int(workflow_definition_ref),
278                token,
279            )
280            resolved_ref = pr_head_info.ref
281        else:
282            resolved_ref = workflow_definition_ref
283    else:
284        resolved_ref = resolve_default_workflow_branch("main")
285
286    # Trigger the workflow
287    result = trigger_workflow_dispatch(
288        owner=owner,
289        repo=repo,
290        workflow_file=workflow_file,
291        ref=resolved_ref,
292        inputs=inputs or {},
293        token=token,
294        find_run=True,
295    )
296
297    # Build response message
298    if result.run_id:
299        message = f"Successfully triggered workflow {workflow_file} on {owner}/{repo} (ref: {resolved_ref}). Run ID: {result.run_id}"
300    else:
301        message = f"Successfully triggered workflow {workflow_file} on {owner}/{repo} (ref: {resolved_ref}). Run ID not yet available."
302
303    return TriggerCIWorkflowResult(
304        success=True,
305        message=message,
306        workflow_url=result.workflow_url,
307        run_id=result.run_id,
308        run_url=result.run_url,
309    )
310
311
312class DockerImageInfo(BaseModel):
313    """Response model for get_docker_image_info MCP tool."""
314
315    exists: bool
316    image: str
317    tag: str
318    full_name: str
319    digest: str | None = None
320    last_updated: str | None = None
321    size_bytes: int | None = None
322    architecture: str | None = None
323    os: str | None = None
324
325
326def _check_dockerhub_image(
327    image: str,
328    tag: str,
329) -> dict | None:
330    """Check if a Docker image tag exists on DockerHub.
331
332    Args:
333        image: Docker image name (e.g., "airbyte/source-github")
334        tag: Image tag (e.g., "2.1.5-preview.abc1234")
335
336    Returns:
337        Tag data dictionary if found, None if not found.
338    """
339    # DockerHub API endpoint for tag info
340    url = f"{DOCKERHUB_API_BASE}/repositories/{image}/tags/{tag}"
341
342    response = requests.get(url, timeout=30)
343    if response.status_code == 404:
344        return None
345    response.raise_for_status()
346
347    return response.json()
348
349
350@mcp_tool(
351    read_only=True,
352    idempotent=True,
353    open_world=True,
354)
355def get_docker_image_info(
356    image: Annotated[
357        str,
358        Field(description="Docker image name (e.g., 'airbyte/source-github')"),
359    ],
360    tag: Annotated[
361        str,
362        Field(description="Image tag (e.g., '2.1.5-preview.abc1234')"),
363    ],
364) -> DockerImageInfo:
365    """Check if a Docker image exists on DockerHub.
366
367    Returns information about the image if it exists, or indicates if it doesn't exist.
368    This is useful for confirming that a pre-release connector was successfully published.
369    """
370    full_name = f"{image}:{tag}"
371    tag_data = _check_dockerhub_image(image, tag)
372
373    if not tag_data:
374        return DockerImageInfo(
375            exists=False,
376            image=image,
377            tag=tag,
378            full_name=full_name,
379        )
380
381    # Extract image details from the first image in the list (if available)
382    images = tag_data.get("images", [])
383    first_image = images[0] if images else {}
384
385    return DockerImageInfo(
386        exists=True,
387        image=image,
388        tag=tag,
389        full_name=full_name,
390        digest=tag_data.get("digest"),
391        last_updated=tag_data.get("last_updated"),
392        size_bytes=first_image.get("size"),
393        architecture=first_image.get("architecture"),
394        os=first_image.get("os"),
395    )
396
397
398def register_github_actions_tools(app: FastMCP) -> None:
399    """Register GitHub Actions tools with the FastMCP app.
400
401    Args:
402        app: FastMCP application instance
403    """
404    register_mcp_tools(app, mcp_module=__name__)