airbyte_ops_mcp.cli.gh

  1# Copyright (c) 2025 Airbyte, Inc., all rights reserved.
  2"""CLI commands for GitHub operations.
  3
  4Commands:
  5    airbyte-ops gh workflow status - Check GitHub Actions workflow status
  6    airbyte-ops gh workflow trigger - Trigger a GitHub Actions CI workflow
  7
  8## CLI reference
  9
 10The commands below are regenerated by `poe docs-generate` via cyclopts's
 11programmatic docs API; see `docs/generate_cli.py`.
 12
 13.. include:: ../../../docs/generated/cli/gh.md
 14   :start-line: 2
 15"""
 16
 17from __future__ import annotations
 18
 19# Hide Python-level members from the pdoc page for this module; the rendered
 20# docs for this CLI group come entirely from the grafted `.. include::` in
 21# the module docstring above.
 22__all__: list[str] = []
 23
 24import json
 25import time
 26from typing import Annotated
 27
 28from cyclopts import Parameter
 29
 30from airbyte_ops_mcp.cli._base import App, app
 31from airbyte_ops_mcp.cli._shared import exit_with_error, print_json
 32from airbyte_ops_mcp.mcp.github_actions import (
 33    check_ci_workflow_status,
 34    trigger_ci_workflow,
 35)
 36
 37# Create the gh sub-app
 38gh_app = App(name="gh", help="GitHub operations.")
 39app.command(gh_app)
 40
 41# Create the workflow sub-app under gh
 42workflow_app = App(name="workflow", help="GitHub Actions workflow operations.")
 43gh_app.command(workflow_app)
 44
 45
 46@workflow_app.command(name="status")
 47def workflow_status(
 48    url: Annotated[
 49        str | None,
 50        Parameter(
 51            help="Full GitHub Actions workflow run URL "
 52            "(e.g., 'https://github.com/owner/repo/actions/runs/12345')."
 53        ),
 54    ] = None,
 55    owner: Annotated[
 56        str | None,
 57        Parameter(help="Repository owner (e.g., 'airbytehq')."),
 58    ] = None,
 59    repo: Annotated[
 60        str | None,
 61        Parameter(help="Repository name (e.g., 'airbyte')."),
 62    ] = None,
 63    run_id: Annotated[
 64        int | None,
 65        Parameter(help="Workflow run ID."),
 66    ] = None,
 67) -> None:
 68    """Check the status of a GitHub Actions workflow run.
 69
 70    Provide either --url OR all of (--owner, --repo, --run-id).
 71    """
 72    # Validate input parameters
 73    if url:
 74        if owner or repo or run_id:
 75            exit_with_error(
 76                "Cannot specify --url together with --owner/--repo/--run-id. "
 77                "Use either --url OR the component parts."
 78            )
 79    elif not (owner and repo and run_id):
 80        exit_with_error(
 81            "Must provide either --url OR all of (--owner, --repo, --run-id)."
 82        )
 83
 84    result = check_ci_workflow_status(
 85        workflow_url=url,
 86        owner=owner,
 87        repo=repo,
 88        run_id=run_id,
 89    )
 90    print_json(result.model_dump())
 91
 92
 93@workflow_app.command(name="trigger")
 94def workflow_trigger(
 95    owner: Annotated[
 96        str,
 97        Parameter(help="Repository owner (e.g., 'airbytehq')."),
 98    ],
 99    repo: Annotated[
100        str,
101        Parameter(help="Repository name (e.g., 'airbyte')."),
102    ],
103    workflow_file: Annotated[
104        str,
105        Parameter(help="Workflow file name (e.g., 'connector-regression-test.yml')."),
106    ],
107    workflow_definition_ref: Annotated[
108        str | None,
109        Parameter(
110            help="Branch name or PR number for the workflow definition to use. "
111            "If a PR number is provided, it resolves to the PR's head branch name. "
112            "Defaults to 'main' if not specified."
113        ),
114    ] = None,
115    inputs: Annotated[
116        str | None,
117        Parameter(
118            help='Workflow inputs as a JSON string (e.g., \'{"key": "value"}\').'
119        ),
120    ] = None,
121    wait: Annotated[
122        bool,
123        Parameter(help="Wait for the workflow to complete before returning."),
124    ] = False,
125    wait_seconds: Annotated[
126        int,
127        Parameter(
128            help="Maximum seconds to wait for workflow completion (default: 600)."
129        ),
130    ] = 600,
131) -> None:
132    """Trigger a GitHub Actions CI workflow via workflow_dispatch.
133
134    This command triggers a workflow in any GitHub repository that has workflow_dispatch
135    enabled. It resolves PR numbers to branch names automatically.
136    """
137    # Parse inputs JSON if provided
138    parsed_inputs: dict[str, str] | None = None
139    if inputs:
140        try:
141            parsed_inputs = json.loads(inputs)
142        except json.JSONDecodeError as e:
143            exit_with_error(f"Invalid JSON for --inputs: {e}")
144
145    # Trigger the workflow
146    result = trigger_ci_workflow(
147        owner=owner,
148        repo=repo,
149        workflow_file=workflow_file,
150        workflow_definition_ref=workflow_definition_ref,
151        inputs=parsed_inputs,
152    )
153
154    print_json(result.model_dump())
155
156    # If wait is enabled and we have a run_id, poll for completion
157    if wait and result.run_id:
158        print(f"\nWaiting for workflow to complete (timeout: {wait_seconds}s)...")
159        start_time = time.time()
160        poll_interval = 10  # seconds
161
162        while time.time() - start_time < wait_seconds:
163            status_result = check_ci_workflow_status(
164                owner=owner,
165                repo=repo,
166                run_id=result.run_id,
167            )
168
169            if status_result.status == "completed":
170                print(
171                    f"\nWorkflow completed with conclusion: {status_result.conclusion}"
172                )
173                print_json(status_result.model_dump())
174                return
175
176            elapsed = int(time.time() - start_time)
177            print(f"  Status: {status_result.status} (elapsed: {elapsed}s)")
178            time.sleep(poll_interval)
179
180        print(f"\nTimeout reached after {wait_seconds}s. Workflow still running.")
181        # Print final status
182        final_status = check_ci_workflow_status(
183            owner=owner,
184            repo=repo,
185            run_id=result.run_id,
186        )
187        print_json(final_status.model_dump())