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