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