airbyte_ops_mcp.mcp.github_ops

  1# Copyright (c) 2025 Airbyte, Inc., all rights reserved.
  2"""MCP tools for GitHub operations: CI workflow triggering/status, Docker image info, and issue/PR subscriptions.
  3
  4## MCP reference
  5
  6.. include:: ../../../docs/mcp-generated/github_ops.md
  7    :start-line: 2
  8"""
  9
 10from __future__ import annotations
 11
 12__all__: list[str] = []
 13
 14import logging
 15import os
 16import re
 17from typing import Annotated
 18
 19import requests
 20from fastmcp import FastMCP
 21from fastmcp_extensions import mcp_tool, register_mcp_tools
 22from pydantic import BaseModel, Field
 23
 24from airbyte_ops_mcp.github_actions import (
 25    get_workflow_jobs,
 26    resolve_default_workflow_branch,
 27    trigger_workflow_dispatch,
 28)
 29from airbyte_ops_mcp.github_api import (
 30    GITHUB_API_BASE,
 31    AgentEnum,
 32    get_pr_head_ref,
 33    resolve_ci_trigger_github_token,
 34    resolve_copilot_review_github_token,
 35)
 36from airbyte_ops_mcp.github_api import (
 37    request_pr_ai_review as request_pr_ai_review_api,
 38)
 39
 40DOCKERHUB_API_BASE = "https://hub.docker.com/v2"
 41
 42
 43class JobInfo(BaseModel):
 44    """Information about a single job in a workflow run."""
 45
 46    job_id: int
 47    name: str
 48    status: str
 49    conclusion: str | None = None
 50    started_at: str | None = None
 51    completed_at: str | None = None
 52
 53
 54class WorkflowRunStatus(BaseModel):
 55    """Response model for check_ci_workflow_status MCP tool."""
 56
 57    run_id: int
 58    status: str
 59    conclusion: str | None
 60    workflow_name: str
 61    head_branch: str
 62    head_sha: str
 63    html_url: str
 64    created_at: str
 65    updated_at: str
 66    run_started_at: str | None = None
 67    jobs_url: str
 68    jobs: list[JobInfo] = []
 69
 70
 71def _parse_workflow_url(url: str) -> tuple[str, str, int]:
 72    """Parse a GitHub Actions workflow run URL into components.
 73
 74    Args:
 75        url: GitHub Actions workflow run URL
 76            (e.g., "https://github.com/owner/repo/actions/runs/12345")
 77
 78    Returns:
 79        Tuple of (owner, repo, run_id)
 80
 81    Raises:
 82        ValueError: If URL format is invalid.
 83    """
 84    pattern = r"https://github\.com/([^/]+)/([^/]+)/actions/runs/(\d+)"
 85    match = re.match(pattern, url)
 86    if not match:
 87        raise ValueError(
 88            f"Invalid workflow URL format: {url}. "
 89            "Expected format: https://github.com/owner/repo/actions/runs/12345"
 90        )
 91    return match.group(1), match.group(2), int(match.group(3))
 92
 93
 94def _get_workflow_run(
 95    owner: str,
 96    repo: str,
 97    run_id: int,
 98    token: str,
 99) -> dict:
100    """Get workflow run details from GitHub API.
101
102    Args:
103        owner: Repository owner (e.g., "airbytehq")
104        repo: Repository name (e.g., "airbyte")
105        run_id: Workflow run ID
106        token: GitHub API token
107
108    Returns:
109        Workflow run data dictionary.
110
111    Raises:
112        ValueError: If workflow run not found.
113        requests.HTTPError: If API request fails.
114    """
115    url = f"{GITHUB_API_BASE}/repos/{owner}/{repo}/actions/runs/{run_id}"
116    headers = {
117        "Authorization": f"Bearer {token}",
118        "Accept": "application/vnd.github+json",
119        "X-GitHub-Api-Version": "2022-11-28",
120    }
121
122    response = requests.get(url, headers=headers, timeout=30)
123    if response.status_code == 404:
124        raise ValueError(f"Workflow run {owner}/{repo}/actions/runs/{run_id} not found")
125    response.raise_for_status()
126
127    return response.json()
128
129
130@mcp_tool(
131    read_only=True,
132    idempotent=True,
133    open_world=True,
134)
135def check_ci_workflow_status(
136    workflow_url: Annotated[
137        str | None,
138        Field(
139            description="Full GitHub Actions workflow run URL (e.g., 'https://github.com/owner/repo/actions/runs/12345')"
140        ),
141    ] = None,
142    owner: Annotated[
143        str | None,
144        Field(description="Repository owner (e.g., 'airbytehq')"),
145    ] = None,
146    repo: Annotated[
147        str | None,
148        Field(description="Repository name (e.g., 'airbyte')"),
149    ] = None,
150    run_id: Annotated[
151        int | None,
152        Field(description="Workflow run ID"),
153    ] = None,
154) -> WorkflowRunStatus:
155    """Check the status of a GitHub Actions workflow run.
156
157    You can provide either:
158    - A full workflow URL (workflow_url parameter), OR
159    - The component parts (owner, repo, run_id parameters)
160
161    Returns the current status, conclusion, and other details about the workflow run.
162
163    Uses the CI trigger token (GITHUB_CI_WORKFLOW_TRIGGER_PAT) so that
164    workflow runs in private repositories are accessible.
165    """
166    # Guard: Validate input parameters
167    if workflow_url:
168        # Parse URL to get components
169        owner, repo, run_id = _parse_workflow_url(workflow_url)
170    elif owner and repo and run_id:
171        # Use provided components
172        pass
173    else:
174        raise ValueError(
175            "Must provide either workflow_url OR all of (owner, repo, run_id)"
176        )
177
178    # Guard: Check for required token
179    # Use the CI trigger token (same as trigger_ci_workflow) so that
180    # private-repo workflow runs are accessible.
181    token = resolve_ci_trigger_github_token()
182
183    # Get workflow run details
184    run_data = _get_workflow_run(owner, repo, run_id, token)
185
186    # Get jobs for the workflow run, passing the same token
187    workflow_jobs = get_workflow_jobs(owner, repo, run_id, token=token)
188
189    # Convert dataclass objects to Pydantic models for the response
190    jobs = [
191        JobInfo(
192            job_id=job.job_id,
193            name=job.name,
194            status=job.status,
195            conclusion=job.conclusion,
196            started_at=job.started_at,
197            completed_at=job.completed_at,
198        )
199        for job in workflow_jobs
200    ]
201
202    return WorkflowRunStatus(
203        run_id=run_data["id"],
204        status=run_data["status"],
205        conclusion=run_data["conclusion"],
206        workflow_name=run_data["name"],
207        head_branch=run_data["head_branch"],
208        head_sha=run_data["head_sha"],
209        html_url=run_data["html_url"],
210        created_at=run_data["created_at"],
211        updated_at=run_data["updated_at"],
212        run_started_at=run_data.get("run_started_at"),
213        jobs_url=run_data["jobs_url"],
214        jobs=jobs,
215    )
216
217
218class TriggerCIWorkflowResult(BaseModel):
219    """Response model for trigger_ci_workflow MCP tool."""
220
221    success: bool
222    message: str
223    workflow_url: str
224    run_id: int | None = None
225    run_url: str | None = None
226
227
228class AIReviewResult(BaseModel):
229    """Response model for `request_pr_ai_review`."""
230
231    requested: bool
232    reviewers: list[str]
233    message: str
234
235
236def _normalize_airbyte_repo(repo: str) -> str:
237    """Normalize an Airbyte repository name and reject other owners."""
238    if not repo or repo.count("/") > 1:
239        raise ValueError(
240            f"Invalid repository '{repo}': expected '<repo>' or 'airbytehq/<repo>'."
241        )
242    if "/" not in repo:
243        return repo
244    owner, repository = repo.split("/")
245    if owner != "airbytehq":
246        raise ValueError(
247            f"Repository owner must be 'airbytehq', but received '{owner}'."
248        )
249    if not repository:
250        raise ValueError(
251            f"Invalid repository '{repo}': repository name cannot be empty."
252        )
253    return repository
254
255
256@mcp_tool(
257    read_only=False,
258    idempotent=False,
259    open_world=True,
260)
261def request_pr_ai_review(
262    repo: Annotated[
263        str,
264        Field(
265            description="Airbyte repository name, optionally prefixed with 'airbytehq/'"
266        ),
267    ],
268    pr_number: Annotated[
269        int,
270        Field(description="Pull request number"),
271    ],
272    request_to: Annotated[
273        AgentEnum | list[AgentEnum],
274        Field(
275            description="AI reviewer to request; defaults to Copilot and accepts "
276            "a single reviewer or a list of reviewers"
277        ),
278    ] = AgentEnum.DEFAULT,
279) -> AIReviewResult:
280    """Request an AI code review on a pull request.
281
282    Requires `GITHUB_CI_WORKFLOW_TRIGGER_PAT`, a PAT for a GitHub user with a
283    Copilot seat. The request is verified through GraphQL `reviewRequests`;
284    a successful mutation response alone is not sufficient.
285    """
286    normalized_repo = _normalize_airbyte_repo(repo)
287    token = resolve_copilot_review_github_token()
288    result = request_pr_ai_review_api(
289        "airbytehq", normalized_repo, pr_number, token, request_to
290    )
291    return AIReviewResult(
292        requested=result.requested,
293        reviewers=result.reviewers,
294        message=result.message,
295    )
296
297
298@mcp_tool(
299    read_only=False,
300    idempotent=False,
301    open_world=True,
302)
303def trigger_ci_workflow(
304    owner: Annotated[
305        str,
306        Field(description="Repository owner (e.g., 'airbytehq')"),
307    ],
308    repo: Annotated[
309        str,
310        Field(description="Repository name (e.g., 'airbyte')"),
311    ],
312    workflow_file: Annotated[
313        str,
314        Field(description="Workflow file name (e.g., 'connector-regression-test.yml')"),
315    ],
316    workflow_definition_ref: Annotated[
317        str | None,
318        Field(
319            description="Branch name or PR number for the workflow definition to use. "
320            "If a PR number (integer string) is provided, it resolves to the PR's head branch name. "
321            "If a branch name is provided, it is used directly. "
322            "Defaults to 'main' if not specified, "
323            "or AIRBYTE_OPS_DEFAULT_WORKFLOW_BRANCH_OVERRIDE when set for local testing."
324        ),
325    ] = None,
326    inputs: Annotated[
327        dict[str, str] | None,
328        Field(
329            description="Workflow inputs as a dictionary of string key-value pairs. "
330            "These are passed to the workflow_dispatch event."
331        ),
332    ] = None,
333) -> TriggerCIWorkflowResult:
334    """Trigger a GitHub Actions CI workflow via workflow_dispatch.
335
336    This tool triggers a workflow in any GitHub repository that has workflow_dispatch
337    enabled. It resolves PR numbers to branch names automatically since GitHub's
338    workflow_dispatch API only accepts branch names, not refs/pull/{pr}/head format.
339
340    Requires GITHUB_CI_WORKFLOW_TRIGGER_PAT or GITHUB_TOKEN environment variable
341    with 'actions:write' permission.
342    """
343    # Guard: Check for required token
344    token = resolve_ci_trigger_github_token()
345
346    if workflow_definition_ref:
347        if workflow_definition_ref.isdigit():
348            pr_head_info = get_pr_head_ref(
349                owner,
350                repo,
351                int(workflow_definition_ref),
352                token,
353            )
354            resolved_ref = pr_head_info.ref
355        else:
356            resolved_ref = workflow_definition_ref
357    else:
358        resolved_ref = resolve_default_workflow_branch("main")
359
360    # Trigger the workflow
361    result = trigger_workflow_dispatch(
362        owner=owner,
363        repo=repo,
364        workflow_file=workflow_file,
365        ref=resolved_ref,
366        inputs=inputs or {},
367        token=token,
368        find_run=True,
369    )
370
371    # Build response message
372    if result.run_id:
373        message = f"Successfully triggered workflow {workflow_file} on {owner}/{repo} (ref: {resolved_ref}). Run ID: {result.run_id}"
374    else:
375        message = f"Successfully triggered workflow {workflow_file} on {owner}/{repo} (ref: {resolved_ref}). Run ID not yet available."
376
377    return TriggerCIWorkflowResult(
378        success=True,
379        message=message,
380        workflow_url=result.workflow_url,
381        run_id=result.run_id,
382        run_url=result.run_url,
383    )
384
385
386class DockerImageInfo(BaseModel):
387    """Response model for get_docker_image_info MCP tool."""
388
389    exists: bool
390    image: str
391    tag: str
392    full_name: str
393    digest: str | None = None
394    last_updated: str | None = None
395    size_bytes: int | None = None
396    architecture: str | None = None
397    os: str | None = None
398
399
400def _check_dockerhub_image(
401    image: str,
402    tag: str,
403) -> dict | None:
404    """Check if a Docker image tag exists on DockerHub.
405
406    Args:
407        image: Docker image name (e.g., "airbyte/source-github")
408        tag: Image tag (e.g., "2.1.5-preview.abc1234")
409
410    Returns:
411        Tag data dictionary if found, None if not found.
412    """
413    # DockerHub API endpoint for tag info
414    url = f"{DOCKERHUB_API_BASE}/repositories/{image}/tags/{tag}"
415
416    response = requests.get(url, timeout=30)
417    if response.status_code == 404:
418        return None
419    response.raise_for_status()
420
421    return response.json()
422
423
424@mcp_tool(
425    read_only=True,
426    idempotent=True,
427    open_world=True,
428)
429def get_docker_image_info(
430    image: Annotated[
431        str,
432        Field(description="Docker image name (e.g., 'airbyte/source-github')"),
433    ],
434    tag: Annotated[
435        str,
436        Field(description="Image tag (e.g., '2.1.5-preview.abc1234')"),
437    ],
438) -> DockerImageInfo:
439    """Check if a Docker image exists on DockerHub.
440
441    Returns information about the image if it exists, or indicates if it doesn't exist.
442    This is useful for confirming that a pre-release connector was successfully published.
443    """
444    full_name = f"{image}:{tag}"
445    tag_data = _check_dockerhub_image(image, tag)
446
447    if not tag_data:
448        return DockerImageInfo(
449            exists=False,
450            image=image,
451            tag=tag,
452            full_name=full_name,
453        )
454
455    # Extract image details from the first image in the list (if available)
456    images = tag_data.get("images", [])
457    first_image = images[0] if images else {}
458
459    return DockerImageInfo(
460        exists=True,
461        image=image,
462        tag=tag,
463        full_name=full_name,
464        digest=tag_data.get("digest"),
465        last_updated=tag_data.get("last_updated"),
466        size_bytes=first_image.get("size"),
467        architecture=first_image.get("architecture"),
468        os=first_image.get("os"),
469    )
470
471
472logger = logging.getLogger(__name__)
473
474SUBSCRIPTION_API_URL_ENV = "SUBSCRIPTION_API_URL"
475
476SUBSCRIPTION_API_TOKEN_ENV = "SUBSCRIPTION_API_BEARER_TOKEN"
477
478
479def _get_api_url() -> str:
480    """Get the subscription API base URL."""
481    url = os.environ.get(SUBSCRIPTION_API_URL_ENV)
482    if not url:
483        raise ValueError(
484            f"{SUBSCRIPTION_API_URL_ENV} environment variable is not set. "
485            "Cannot reach the GitHub subscriptions backend."
486        )
487    return url.rstrip("/")
488
489
490def _get_api_token() -> str:
491    """Get the subscription API bearer token."""
492    token = os.environ.get(SUBSCRIPTION_API_TOKEN_ENV)
493    if not token:
494        raise ValueError(
495            f"{SUBSCRIPTION_API_TOKEN_ENV} environment variable is not set. "
496            "Cannot authenticate to the GitHub subscriptions backend."
497        )
498    return token
499
500
501def _api_headers() -> dict[str, str]:
502    """Build headers for API requests."""
503    return {
504        "Authorization": f"Bearer {_get_api_token()}",
505        "Content-Type": "application/json",
506    }
507
508
509class SubscribeResponse(BaseModel):
510    """Response from the subscribe_to_github_issue tool."""
511
512    success: bool = Field(
513        description="Whether the subscription was created successfully"
514    )
515    message: str = Field(description="Human-readable status message")
516    subscription_id: str | None = Field(
517        default=None,
518        description="ID of the created or updated subscription",
519    )
520    github_url: str | None = Field(
521        default=None,
522        description="GitHub URL being watched",
523    )
524    expires_at: str | None = Field(
525        default=None,
526        description="When the subscription expires (ISO 8601)",
527    )
528
529
530class UnsubscribeResponse(BaseModel):
531    """Response from the unsubscribe_from_github_issue tool."""
532
533    success: bool = Field(description="Whether the unsubscribe was successful")
534    message: str = Field(description="Human-readable status message")
535    deleted_count: int = Field(
536        default=0,
537        description="Number of subscriptions removed",
538    )
539
540
541class ListSubscriptionsResponse(BaseModel):
542    """Response from the list_github_subscriptions tool."""
543
544    success: bool = Field(description="Whether the listing was successful")
545    message: str = Field(description="Human-readable status message")
546    subscriptions: list[dict[str, str]] = Field(
547        default_factory=list,
548        description="List of active subscriptions with id, github_url, expires_at",
549    )
550
551
552@mcp_tool(
553    read_only=False,
554    idempotent=True,
555    open_world=True,
556)
557def subscribe_to_github_issue(
558    github_url: Annotated[
559        str,
560        "The GitHub issue or PR URL to subscribe to. "
561        "Examples: https://github.com/airbytehq/airbyte/issues/123 "
562        "or https://github.com/airbytehq/airbyte/pull/456",
563    ],
564    agent_session_url: Annotated[
565        str,
566        "Your Devin session URL so notifications can be delivered back to "
567        "your session. Use the session URL from your system prompt.",
568    ],
569    watch_events: Annotated[
570        list[str] | None,
571        "Optional list of event types to watch. Valid values: "
572        "'comment', 'close', 'merge', 'reopen', 'label', 'synchronize', "
573        "'ready_for_review', 'assigned'. Defaults to all events if not specified.",
574    ] = None,
575    ttl_hours: Annotated[
576        int,
577        "Number of hours until the subscription expires. Default is 240 (10 days).",
578    ] = 240,
579    slack_users_cc: Annotated[
580        str | None,
581        "Optional comma-delimited list of Slack user tags to CC on "
582        "notifications. Example: '<@U12345>, <@U67890>'.",
583    ] = None,
584) -> SubscribeResponse:
585    """Subscribe to notifications on a GitHub issue or pull request.
586
587    Creates a subscription that will deliver real-time notifications back
588    to your Devin session when activity occurs on the specified GitHub
589    issue or PR. Notifications are triggered by GitHub webhooks and
590    delivered within seconds.
591
592    If you are already subscribed to the same issue/PR, the subscription
593    is updated (TTL extended, watch events merged).
594
595    Use this tool when you need to monitor a GitHub issue or PR for
596    changes, new comments, merges, closures, or other activity.
597    """
598    try:
599        api_url = _get_api_url()
600        body: dict[str, str | list[str] | int | None] = {
601            "github_url": github_url,
602            "session_url": agent_session_url,
603            "ttl_hours": ttl_hours,
604        }
605        if watch_events:
606            body["watch_events"] = watch_events
607        if slack_users_cc:
608            body["slack_users_cc"] = slack_users_cc
609
610        response = requests.post(
611            f"{api_url}/subscriptions",
612            json=body,
613            headers=_api_headers(),
614            timeout=10,
615        )
616        response.raise_for_status()
617        data = response.json()
618
619        return SubscribeResponse(
620            success=True,
621            message=(
622                f"Subscribed to {github_url}. "
623                f"You will receive notifications in this session until "
624                f"{data.get('expires_at', 'expiry unknown')}."
625            ),
626            subscription_id=data.get("id"),
627            github_url=github_url,
628            expires_at=data.get("expires_at"),
629        )
630
631    except ValueError as e:
632        return SubscribeResponse(
633            success=False,
634            message=f"Configuration error: {e}",
635        )
636    except requests.RequestException as e:
637        logger.exception("Failed to create subscription")
638        return SubscribeResponse(
639            success=False,
640            message=f"Failed to create subscription: {e}",
641        )
642
643
644@mcp_tool(
645    read_only=False,
646    idempotent=True,
647    open_world=True,
648)
649def unsubscribe_from_github_issue(
650    agent_session_url: Annotated[
651        str,
652        "Your Devin session URL. Use the session URL from your system prompt.",
653    ],
654    github_url: Annotated[
655        str | None,
656        "The GitHub issue or PR URL to unsubscribe from. "
657        "If not provided, all subscriptions for this session are removed.",
658    ] = None,
659    subscription_id: Annotated[
660        str | None,
661        "Optional specific subscription ID to remove. "
662        "Use this if you know the exact subscription to cancel.",
663    ] = None,
664) -> UnsubscribeResponse:
665    """Unsubscribe from notifications on a GitHub issue or pull request.
666
667    Removes an active subscription so you will no longer receive
668    notifications for the specified issue/PR.
669
670    You can unsubscribe by:
671    - Providing a specific subscription_id
672    - Providing a github_url + session_url to unsubscribe from that specific issue/PR
673    - Providing only session_url to unsubscribe from all issues/PRs
674    """
675    try:
676        api_url = _get_api_url()
677
678        if subscription_id:
679            # Delete by ID
680            response = requests.delete(
681                f"{api_url}/subscriptions/{subscription_id}",
682                headers=_api_headers(),
683                timeout=10,
684            )
685        else:
686            # Delete by match
687            params: dict[str, str] = {"session_url": agent_session_url}
688            if github_url:
689                params["github_url"] = github_url
690            response = requests.delete(
691                f"{api_url}/subscriptions",
692                params=params,
693                headers=_api_headers(),
694                timeout=10,
695            )
696
697        response.raise_for_status()
698        data = response.json()
699        count = data.get("deleted_count", 0)
700
701        return UnsubscribeResponse(
702            success=True,
703            message=f"Removed {count} subscription(s).",
704            deleted_count=count,
705        )
706
707    except ValueError as e:
708        return UnsubscribeResponse(
709            success=False,
710            message=f"Configuration error: {e}",
711        )
712    except requests.RequestException as e:
713        logger.exception("Failed to unsubscribe")
714        return UnsubscribeResponse(
715            success=False,
716            message=f"Failed to unsubscribe: {e}",
717        )
718
719
720@mcp_tool(
721    read_only=True,
722    idempotent=True,
723    open_world=True,
724)
725def list_github_subscriptions(
726    agent_session_url: Annotated[
727        str,
728        "Your Devin session URL. Use the session URL from your system prompt.",
729    ],
730) -> ListSubscriptionsResponse:
731    """List all active GitHub issue/PR subscriptions for this session.
732
733    Returns the list of GitHub issues and PRs that this session is
734    currently subscribed to, along with their expiry times.
735    """
736    try:
737        api_url = _get_api_url()
738
739        response = requests.get(
740            f"{api_url}/subscriptions",
741            params={"session_url": agent_session_url},
742            headers=_api_headers(),
743            timeout=10,
744        )
745        response.raise_for_status()
746        data = response.json()
747
748        subs = [
749            {
750                "id": s["id"],
751                "github_url": s["github_url"],
752                "watch_events": ", ".join(s.get("watch_events", [])),
753                "expires_at": s.get("expires_at", "unknown"),
754            }
755            for s in data
756        ]
757
758        if not subs:
759            return ListSubscriptionsResponse(
760                success=True,
761                message="No active subscriptions for this session.",
762                subscriptions=[],
763            )
764
765        return ListSubscriptionsResponse(
766            success=True,
767            message=f"Found {len(subs)} active subscription(s).",
768            subscriptions=subs,
769        )
770
771    except ValueError as e:
772        return ListSubscriptionsResponse(
773            success=False,
774            message=f"Configuration error: {e}",
775        )
776    except requests.RequestException as e:
777        logger.exception("Failed to list subscriptions")
778        return ListSubscriptionsResponse(
779            success=False,
780            message=f"Failed to list subscriptions: {e}",
781        )
782
783
784def register_github_ops_tools(app: FastMCP) -> None:
785    """Register github_ops tools with the FastMCP app."""
786    register_mcp_tools(app, mcp_module=__name__)