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    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 'main' if not specified, "
248            "or AIRBYTE_OPS_DEFAULT_WORKFLOW_BRANCH_OVERRIDE when set for local testing."
249        ),
250    ] = None,
251    inputs: Annotated[
252        dict[str, str] | None,
253        Field(
254            description="Workflow inputs as a dictionary of string key-value pairs. "
255            "These are passed to the workflow_dispatch event."
256        ),
257    ] = None,
258) -> TriggerCIWorkflowResult:
259    """Trigger a GitHub Actions CI workflow via workflow_dispatch.
260
261    This tool triggers a workflow in any GitHub repository that has workflow_dispatch
262    enabled. It resolves PR numbers to branch names automatically since GitHub's
263    workflow_dispatch API only accepts branch names, not refs/pull/{pr}/head format.
264
265    Requires GITHUB_CI_WORKFLOW_TRIGGER_PAT or GITHUB_TOKEN environment variable
266    with 'actions:write' permission.
267    """
268    # Guard: Check for required token
269    token = resolve_ci_trigger_github_token()
270
271    if workflow_definition_ref:
272        if workflow_definition_ref.isdigit():
273            pr_head_info = get_pr_head_ref(
274                owner,
275                repo,
276                int(workflow_definition_ref),
277                token,
278            )
279            resolved_ref = pr_head_info.ref
280        else:
281            resolved_ref = workflow_definition_ref
282    else:
283        resolved_ref = resolve_default_workflow_branch("main")
284
285    # Trigger the workflow
286    result = trigger_workflow_dispatch(
287        owner=owner,
288        repo=repo,
289        workflow_file=workflow_file,
290        ref=resolved_ref,
291        inputs=inputs or {},
292        token=token,
293        find_run=True,
294    )
295
296    # Build response message
297    if result.run_id:
298        message = f"Successfully triggered workflow {workflow_file} on {owner}/{repo} (ref: {resolved_ref}). Run ID: {result.run_id}"
299    else:
300        message = f"Successfully triggered workflow {workflow_file} on {owner}/{repo} (ref: {resolved_ref}). Run ID not yet available."
301
302    return TriggerCIWorkflowResult(
303        success=True,
304        message=message,
305        workflow_url=result.workflow_url,
306        run_id=result.run_id,
307        run_url=result.run_url,
308    )
309
310
311class DockerImageInfo(BaseModel):
312    """Response model for get_docker_image_info MCP tool."""
313
314    exists: bool
315    image: str
316    tag: str
317    full_name: str
318    digest: str | None = None
319    last_updated: str | None = None
320    size_bytes: int | None = None
321    architecture: str | None = None
322    os: str | None = None
323
324
325def _check_dockerhub_image(
326    image: str,
327    tag: str,
328) -> dict | None:
329    """Check if a Docker image tag exists on DockerHub.
330
331    Args:
332        image: Docker image name (e.g., "airbyte/source-github")
333        tag: Image tag (e.g., "2.1.5-preview.abc1234")
334
335    Returns:
336        Tag data dictionary if found, None if not found.
337    """
338    # DockerHub API endpoint for tag info
339    url = f"{DOCKERHUB_API_BASE}/repositories/{image}/tags/{tag}"
340
341    response = requests.get(url, timeout=30)
342    if response.status_code == 404:
343        return None
344    response.raise_for_status()
345
346    return response.json()
347
348
349@mcp_tool(
350    read_only=True,
351    idempotent=True,
352    open_world=True,
353)
354def get_docker_image_info(
355    image: Annotated[
356        str,
357        Field(description="Docker image name (e.g., 'airbyte/source-github')"),
358    ],
359    tag: Annotated[
360        str,
361        Field(description="Image tag (e.g., '2.1.5-preview.abc1234')"),
362    ],
363) -> DockerImageInfo:
364    """Check if a Docker image exists on DockerHub.
365
366    Returns information about the image if it exists, or indicates if it doesn't exist.
367    This is useful for confirming that a pre-release connector was successfully published.
368    """
369    full_name = f"{image}:{tag}"
370    tag_data = _check_dockerhub_image(image, tag)
371
372    if not tag_data:
373        return DockerImageInfo(
374            exists=False,
375            image=image,
376            tag=tag,
377            full_name=full_name,
378        )
379
380    # Extract image details from the first image in the list (if available)
381    images = tag_data.get("images", [])
382    first_image = images[0] if images else {}
383
384    return DockerImageInfo(
385        exists=True,
386        image=image,
387        tag=tag,
388        full_name=full_name,
389        digest=tag_data.get("digest"),
390        last_updated=tag_data.get("last_updated"),
391        size_bytes=first_image.get("size"),
392        architecture=first_image.get("architecture"),
393        os=first_image.get("os"),
394    )
395
396
397logger = logging.getLogger(__name__)
398
399SUBSCRIPTION_API_URL_ENV = "SUBSCRIPTION_API_URL"
400
401SUBSCRIPTION_API_TOKEN_ENV = "SUBSCRIPTION_API_BEARER_TOKEN"
402
403
404def _get_api_url() -> str:
405    """Get the subscription API base URL."""
406    url = os.environ.get(SUBSCRIPTION_API_URL_ENV)
407    if not url:
408        raise ValueError(
409            f"{SUBSCRIPTION_API_URL_ENV} environment variable is not set. "
410            "Cannot reach the GitHub subscriptions backend."
411        )
412    return url.rstrip("/")
413
414
415def _get_api_token() -> str:
416    """Get the subscription API bearer token."""
417    token = os.environ.get(SUBSCRIPTION_API_TOKEN_ENV)
418    if not token:
419        raise ValueError(
420            f"{SUBSCRIPTION_API_TOKEN_ENV} environment variable is not set. "
421            "Cannot authenticate to the GitHub subscriptions backend."
422        )
423    return token
424
425
426def _api_headers() -> dict[str, str]:
427    """Build headers for API requests."""
428    return {
429        "Authorization": f"Bearer {_get_api_token()}",
430        "Content-Type": "application/json",
431    }
432
433
434class SubscribeResponse(BaseModel):
435    """Response from the subscribe_to_github_issue tool."""
436
437    success: bool = Field(
438        description="Whether the subscription was created successfully"
439    )
440    message: str = Field(description="Human-readable status message")
441    subscription_id: str | None = Field(
442        default=None,
443        description="ID of the created or updated subscription",
444    )
445    github_url: str | None = Field(
446        default=None,
447        description="GitHub URL being watched",
448    )
449    expires_at: str | None = Field(
450        default=None,
451        description="When the subscription expires (ISO 8601)",
452    )
453
454
455class UnsubscribeResponse(BaseModel):
456    """Response from the unsubscribe_from_github_issue tool."""
457
458    success: bool = Field(description="Whether the unsubscribe was successful")
459    message: str = Field(description="Human-readable status message")
460    deleted_count: int = Field(
461        default=0,
462        description="Number of subscriptions removed",
463    )
464
465
466class ListSubscriptionsResponse(BaseModel):
467    """Response from the list_github_subscriptions tool."""
468
469    success: bool = Field(description="Whether the listing was successful")
470    message: str = Field(description="Human-readable status message")
471    subscriptions: list[dict[str, str]] = Field(
472        default_factory=list,
473        description="List of active subscriptions with id, github_url, expires_at",
474    )
475
476
477@mcp_tool(
478    read_only=False,
479    idempotent=True,
480    open_world=True,
481)
482def subscribe_to_github_issue(
483    github_url: Annotated[
484        str,
485        "The GitHub issue or PR URL to subscribe to. "
486        "Examples: https://github.com/airbytehq/airbyte/issues/123 "
487        "or https://github.com/airbytehq/airbyte/pull/456",
488    ],
489    agent_session_url: Annotated[
490        str,
491        "Your Devin session URL so notifications can be delivered back to "
492        "your session. Use the session URL from your system prompt.",
493    ],
494    watch_events: Annotated[
495        list[str] | None,
496        "Optional list of event types to watch. Valid values: "
497        "'comment', 'close', 'merge', 'reopen', 'label', 'synchronize', "
498        "'ready_for_review', 'assigned'. Defaults to all events if not specified.",
499    ] = None,
500    ttl_hours: Annotated[
501        int,
502        "Number of hours until the subscription expires. Default is 240 (10 days).",
503    ] = 240,
504    slack_users_cc: Annotated[
505        str | None,
506        "Optional comma-delimited list of Slack user tags to CC on "
507        "notifications. Example: '<@U12345>, <@U67890>'.",
508    ] = None,
509) -> SubscribeResponse:
510    """Subscribe to notifications on a GitHub issue or pull request.
511
512    Creates a subscription that will deliver real-time notifications back
513    to your Devin session when activity occurs on the specified GitHub
514    issue or PR. Notifications are triggered by GitHub webhooks and
515    delivered within seconds.
516
517    If you are already subscribed to the same issue/PR, the subscription
518    is updated (TTL extended, watch events merged).
519
520    Use this tool when you need to monitor a GitHub issue or PR for
521    changes, new comments, merges, closures, or other activity.
522    """
523    try:
524        api_url = _get_api_url()
525        body: dict[str, str | list[str] | int | None] = {
526            "github_url": github_url,
527            "session_url": agent_session_url,
528            "ttl_hours": ttl_hours,
529        }
530        if watch_events:
531            body["watch_events"] = watch_events
532        if slack_users_cc:
533            body["slack_users_cc"] = slack_users_cc
534
535        response = requests.post(
536            f"{api_url}/subscriptions",
537            json=body,
538            headers=_api_headers(),
539            timeout=10,
540        )
541        response.raise_for_status()
542        data = response.json()
543
544        return SubscribeResponse(
545            success=True,
546            message=(
547                f"Subscribed to {github_url}. "
548                f"You will receive notifications in this session until "
549                f"{data.get('expires_at', 'expiry unknown')}."
550            ),
551            subscription_id=data.get("id"),
552            github_url=github_url,
553            expires_at=data.get("expires_at"),
554        )
555
556    except ValueError as e:
557        return SubscribeResponse(
558            success=False,
559            message=f"Configuration error: {e}",
560        )
561    except requests.RequestException as e:
562        logger.exception("Failed to create subscription")
563        return SubscribeResponse(
564            success=False,
565            message=f"Failed to create subscription: {e}",
566        )
567
568
569@mcp_tool(
570    read_only=False,
571    idempotent=True,
572    open_world=True,
573)
574def unsubscribe_from_github_issue(
575    agent_session_url: Annotated[
576        str,
577        "Your Devin session URL. Use the session URL from your system prompt.",
578    ],
579    github_url: Annotated[
580        str | None,
581        "The GitHub issue or PR URL to unsubscribe from. "
582        "If not provided, all subscriptions for this session are removed.",
583    ] = None,
584    subscription_id: Annotated[
585        str | None,
586        "Optional specific subscription ID to remove. "
587        "Use this if you know the exact subscription to cancel.",
588    ] = None,
589) -> UnsubscribeResponse:
590    """Unsubscribe from notifications on a GitHub issue or pull request.
591
592    Removes an active subscription so you will no longer receive
593    notifications for the specified issue/PR.
594
595    You can unsubscribe by:
596    - Providing a specific subscription_id
597    - Providing a github_url + session_url to unsubscribe from that specific issue/PR
598    - Providing only session_url to unsubscribe from all issues/PRs
599    """
600    try:
601        api_url = _get_api_url()
602
603        if subscription_id:
604            # Delete by ID
605            response = requests.delete(
606                f"{api_url}/subscriptions/{subscription_id}",
607                headers=_api_headers(),
608                timeout=10,
609            )
610        else:
611            # Delete by match
612            params: dict[str, str] = {"session_url": agent_session_url}
613            if github_url:
614                params["github_url"] = github_url
615            response = requests.delete(
616                f"{api_url}/subscriptions",
617                params=params,
618                headers=_api_headers(),
619                timeout=10,
620            )
621
622        response.raise_for_status()
623        data = response.json()
624        count = data.get("deleted_count", 0)
625
626        return UnsubscribeResponse(
627            success=True,
628            message=f"Removed {count} subscription(s).",
629            deleted_count=count,
630        )
631
632    except ValueError as e:
633        return UnsubscribeResponse(
634            success=False,
635            message=f"Configuration error: {e}",
636        )
637    except requests.RequestException as e:
638        logger.exception("Failed to unsubscribe")
639        return UnsubscribeResponse(
640            success=False,
641            message=f"Failed to unsubscribe: {e}",
642        )
643
644
645@mcp_tool(
646    read_only=True,
647    idempotent=True,
648    open_world=True,
649)
650def list_github_subscriptions(
651    agent_session_url: Annotated[
652        str,
653        "Your Devin session URL. Use the session URL from your system prompt.",
654    ],
655) -> ListSubscriptionsResponse:
656    """List all active GitHub issue/PR subscriptions for this session.
657
658    Returns the list of GitHub issues and PRs that this session is
659    currently subscribed to, along with their expiry times.
660    """
661    try:
662        api_url = _get_api_url()
663
664        response = requests.get(
665            f"{api_url}/subscriptions",
666            params={"session_url": agent_session_url},
667            headers=_api_headers(),
668            timeout=10,
669        )
670        response.raise_for_status()
671        data = response.json()
672
673        subs = [
674            {
675                "id": s["id"],
676                "github_url": s["github_url"],
677                "watch_events": ", ".join(s.get("watch_events", [])),
678                "expires_at": s.get("expires_at", "unknown"),
679            }
680            for s in data
681        ]
682
683        if not subs:
684            return ListSubscriptionsResponse(
685                success=True,
686                message="No active subscriptions for this session.",
687                subscriptions=[],
688            )
689
690        return ListSubscriptionsResponse(
691            success=True,
692            message=f"Found {len(subs)} active subscription(s).",
693            subscriptions=subs,
694        )
695
696    except ValueError as e:
697        return ListSubscriptionsResponse(
698            success=False,
699            message=f"Configuration error: {e}",
700        )
701    except requests.RequestException as e:
702        logger.exception("Failed to list subscriptions")
703        return ListSubscriptionsResponse(
704            success=False,
705            message=f"Failed to list subscriptions: {e}",
706        )
707
708
709def register_github_ops_tools(app: FastMCP) -> None:
710    """Register github_ops tools with the FastMCP app."""
711    register_mcp_tools(app, mcp_module=__name__)