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    airbyte-ops gh connector info - Get connector metadata from GitHub
  8    airbyte-ops gh connector get-version - Get connector version from GitHub
  9    airbyte-ops gh connector list - List connectors via GitHub API
 10
 11## CLI reference
 12
 13The commands below are regenerated by `poe docs-generate` via cyclopts's
 14programmatic docs API; see `docs/generate_cli.py`.
 15
 16.. include:: ../../../docs/generated/cli/gh.md
 17   :start-line: 2
 18"""
 19
 20from __future__ import annotations
 21
 22# Hide Python-level members from the pdoc page for this module; the rendered
 23# docs for this CLI group come entirely from the grafted `.. include::` in
 24# the module docstring above.
 25__all__: list[str] = []
 26
 27import json
 28import sys
 29from typing import Annotated, Literal
 30
 31from cyclopts import Parameter
 32from fastmcp_extensions.cli import exit_with_error, print_json
 33
 34from airbyte_ops_mcp.airbyte_repo.list_connectors import (
 35    format_github_actions_connector_matrix,
 36    get_modified_connectors_from_github,
 37)
 38from airbyte_ops_mcp.airbyte_repo.utils import parse_pr_info
 39from airbyte_ops_mcp.cli._base import App, app
 40from airbyte_ops_mcp.connector_metadata import (
 41    ConnectorMetadataDpathError,
 42    ConnectorMetadataDpathNotFoundError,
 43    format_metadata_dpath_value,
 44    get_connector_version_from_metadata,
 45    load_raw_connector_metadata_from_github,
 46)
 47from airbyte_ops_mcp.github_api import GitHubAPIError, resolve_default_github_token
 48from airbyte_ops_mcp.mcp.github_ops import (
 49    check_ci_workflow_status,
 50    trigger_ci_workflow,
 51)
 52
 53AIRBYTE_REPO_OWNER = "airbytehq"
 54AIRBYTE_REPO_NAME = "airbyte"
 55
 56# Create the gh sub-app
 57gh_app = App(name="gh", help="GitHub operations.")
 58app.command(gh_app)
 59
 60# Create the workflow sub-app under gh
 61workflow_app = App(name="workflow", help="GitHub Actions workflow operations.")
 62gh_app.command(workflow_app)
 63
 64# Create the connector sub-app under gh
 65connector_app = App(name="connector", help="Connector operations via GitHub API.")
 66gh_app.command(connector_app)
 67
 68
 69ConnectorListOutputFormat = Literal["lines", "csv", "json", "json-gh-matrix"]
 70
 71
 72def _parse_pr_details(pr: str) -> tuple[int, str | None, str | None]:
 73    """Parse pull request details from a number or GitHub PR URL."""
 74    pr_number, pr_owner, pr_repo = parse_pr_info(pr)
 75    if pr_number is None:
 76        exit_with_error(
 77            "PR must be a pull request number or GitHub URL like "
 78            "https://github.com/airbytehq/airbyte/pull/123."
 79        )
 80    return pr_number, pr_owner, pr_repo
 81
 82
 83def _print_connector_list(
 84    connectors: list[str],
 85    output_format: ConnectorListOutputFormat,
 86) -> None:
 87    """Print connector names in the requested format."""
 88    if output_format == "lines":
 89        for connector in connectors:
 90            sys.stdout.write(connector + "\n")
 91    elif output_format == "csv":
 92        sys.stdout.write(",".join(connectors) + "\n")
 93    elif output_format == "json":
 94        sys.stdout.write(json.dumps(connectors, separators=(",", ":")) + "\n")
 95    elif output_format == "json-gh-matrix":
 96        sys.stdout.write(
 97            json.dumps(
 98                format_github_actions_connector_matrix(connectors),
 99                separators=(",", ":"),
100            )
101            + "\n"
102        )
103
104
105@connector_app.command(name="info")
106def connector_info(
107    name: Annotated[
108        str,
109        Parameter(help="Connector technical name (e.g., source-github)."),
110    ],
111    ref: Annotated[
112        str,
113        Parameter(help="Git ref to read metadata.yaml from."),
114    ],
115    owner: Annotated[
116        str,
117        Parameter(help="Repository owner (e.g., airbytehq)."),
118    ] = AIRBYTE_REPO_OWNER,
119    repo: Annotated[
120        str,
121        Parameter(help="Repository name (e.g., airbyte or airbyte-enterprise)."),
122    ] = AIRBYTE_REPO_NAME,
123    gh_token: Annotated[
124        str | None,
125        Parameter(
126            help=(
127                "GitHub API token. If omitted, uses `resolve_default_github_token`; "
128                "public repos can fall back to unauthenticated requests."
129            )
130        ),
131    ] = None,
132    dpath_expression: Annotated[
133        str | None,
134        Parameter(
135            name="--dpath",
136            help=(
137                "Evaluate this dpath expression against the parsed metadata.yaml "
138                "object and print only that value (e.g., data/dockerImageTag)."
139            ),
140        ),
141    ] = None,
142) -> None:
143    """Get connector metadata from GitHub."""
144    try:
145        value = load_raw_connector_metadata_from_github(
146            name,
147            owner=owner,
148            repo=repo,
149            ref=ref,
150            gh_token=gh_token or resolve_default_github_token(allow_none=True),
151            dpath_expression=dpath_expression,
152        )
153    except (
154        ConnectorMetadataDpathError,
155        ConnectorMetadataDpathNotFoundError,
156        FileNotFoundError,
157        ValueError,
158    ) as e:
159        exit_with_error(str(e))
160
161    if dpath_expression is None:
162        print_json(value)
163        return
164
165    sys.stdout.write(format_metadata_dpath_value(value) + "\n")
166
167
168@connector_app.command(name="get-version")
169def connector_get_version(
170    name: Annotated[
171        str,
172        Parameter(help="Connector technical name (e.g., source-github)."),
173    ],
174    ref: Annotated[
175        str,
176        Parameter(help="Git ref to read metadata.yaml from."),
177    ],
178    owner: Annotated[
179        str,
180        Parameter(help="Repository owner (e.g., airbytehq)."),
181    ] = AIRBYTE_REPO_OWNER,
182    repo: Annotated[
183        str,
184        Parameter(help="Repository name (e.g., airbyte or airbyte-enterprise)."),
185    ] = AIRBYTE_REPO_NAME,
186    gh_token: Annotated[
187        str | None,
188        Parameter(
189            help=(
190                "GitHub API token. If omitted, uses `resolve_default_github_token`; "
191                "public repos can fall back to unauthenticated requests."
192            )
193        ),
194    ] = None,
195) -> None:
196    """Get connector version from GitHub.
197
198    This is analogous to `gh connector info --dpath data/dockerImageTag` and uses
199    the same dpath evaluation internally.
200    """
201    try:
202        metadata = load_raw_connector_metadata_from_github(
203            name,
204            owner=owner,
205            repo=repo,
206            ref=ref,
207            gh_token=gh_token or resolve_default_github_token(allow_none=True),
208        )
209        version = get_connector_version_from_metadata(metadata)
210    except (
211        ConnectorMetadataDpathError,
212        ConnectorMetadataDpathNotFoundError,
213        FileNotFoundError,
214        ValueError,
215    ) as e:
216        exit_with_error(str(e))
217    sys.stdout.write(version + "\n")
218
219
220@connector_app.command(name="list")
221def connector_list(
222    modified_only: Annotated[
223        bool,
224        Parameter(help="Include only modified connectors from the provided PR."),
225    ] = False,
226    pr: Annotated[
227        str | None,
228        Parameter(
229            help="Pull request number or GitHub URL to inspect for changed connector files."
230        ),
231    ] = None,
232    owner: Annotated[
233        str,
234        Parameter(help="Repository owner (e.g., airbytehq)."),
235    ] = AIRBYTE_REPO_OWNER,
236    repo: Annotated[
237        str,
238        Parameter(help="Repository name (e.g., airbyte or airbyte-enterprise)."),
239    ] = AIRBYTE_REPO_NAME,
240    gh_token: Annotated[
241        str | None,
242        Parameter(
243            help=("GitHub API token. If omitted, uses `resolve_default_github_token`.")
244        ),
245    ] = None,
246    output_format: Annotated[
247        ConnectorListOutputFormat,
248        Parameter(
249            help=(
250                'Output format: "lines", "csv", "json", or "json-gh-matrix". '
251                'The GitHub Actions matrix format is {"connector":["source-x"]} '
252                'and returns {"connector":[""]} when no connectors changed.'
253            )
254        ),
255    ] = "lines",
256) -> None:
257    """List connector names from GitHub without a local checkout."""
258    if not modified_only:
259        exit_with_error("`--modified-only` is required for GitHub connector listing.")
260    if pr is None:
261        exit_with_error("`--pr` is required when `--modified-only` is set.")
262
263    pr_number, pr_owner, pr_repo = _parse_pr_details(pr)
264    try:
265        connectors = get_modified_connectors_from_github(
266            pr_number=pr_number,
267            pr_owner=pr_owner or owner,
268            pr_repo=pr_repo or repo,
269            gh_token=gh_token or resolve_default_github_token(),
270        )
271    except (GitHubAPIError, ValueError) as e:
272        exit_with_error(str(e))
273    _print_connector_list(connectors, output_format)
274
275
276@workflow_app.command(name="status")
277def workflow_status(
278    url: Annotated[
279        str | None,
280        Parameter(
281            help="Full GitHub Actions workflow run URL "
282            "(e.g., 'https://github.com/owner/repo/actions/runs/12345')."
283        ),
284    ] = None,
285    owner: Annotated[
286        str | None,
287        Parameter(help="Repository owner (e.g., 'airbytehq')."),
288    ] = None,
289    repo: Annotated[
290        str | None,
291        Parameter(help="Repository name (e.g., 'airbyte')."),
292    ] = None,
293    run_id: Annotated[
294        int | None,
295        Parameter(help="Workflow run ID."),
296    ] = None,
297) -> None:
298    """Check the status of a GitHub Actions workflow run.
299
300    Provide either --url OR all of (--owner, --repo, --run-id).
301    """
302    # Validate input parameters
303    if url:
304        if owner or repo or run_id:
305            exit_with_error(
306                "Cannot specify --url together with --owner/--repo/--run-id. "
307                "Use either --url OR the component parts."
308            )
309    elif not (owner and repo and run_id):
310        exit_with_error(
311            "Must provide either --url OR all of (--owner, --repo, --run-id)."
312        )
313
314    result = check_ci_workflow_status(
315        workflow_url=url,
316        owner=owner,
317        repo=repo,
318        run_id=run_id,
319    )
320    print_json(result.model_dump())
321
322
323@workflow_app.command(name="trigger")
324def workflow_trigger(
325    owner: Annotated[
326        str,
327        Parameter(help="Repository owner (e.g., 'airbytehq')."),
328    ],
329    repo: Annotated[
330        str,
331        Parameter(help="Repository name (e.g., 'airbyte')."),
332    ],
333    workflow_file: Annotated[
334        str,
335        Parameter(help="Workflow file name (e.g., 'connector-regression-test.yml')."),
336    ],
337    workflow_definition_ref: Annotated[
338        str | None,
339        Parameter(
340            help="Branch name or PR number for the workflow definition to use. "
341            "If a PR number is provided, it resolves to the PR's head branch name. "
342            "Defaults to 'main' if not specified."
343        ),
344    ] = None,
345    inputs: Annotated[
346        str | None,
347        Parameter(
348            help='Workflow inputs as a JSON string (e.g., \'{"key": "value"}\').'
349        ),
350    ] = None,
351    wait: Annotated[
352        bool,
353        Parameter(help="Wait for the workflow to complete before returning."),
354    ] = False,
355    wait_seconds: Annotated[
356        int,
357        Parameter(
358            help="Maximum seconds to wait for workflow completion (default: 600)."
359        ),
360    ] = 600,
361) -> None:
362    """Trigger a GitHub Actions CI workflow via workflow_dispatch.
363
364    This command triggers a workflow in any GitHub repository that has workflow_dispatch
365    enabled. It resolves PR numbers to branch names automatically.
366    """
367    # Parse inputs JSON if provided
368    parsed_inputs: dict[str, str] | None = None
369    if inputs:
370        try:
371            parsed_inputs = json.loads(inputs)
372        except json.JSONDecodeError as e:
373            exit_with_error(f"Invalid JSON for --inputs: {e}")
374
375    if wait:
376        print(
377            f"Waiting for workflow to complete (timeout: {wait_seconds}s)...",
378            file=sys.stderr,
379        )
380
381    result = trigger_ci_workflow(
382        owner=owner,
383        repo=repo,
384        workflow_file=workflow_file,
385        workflow_definition_ref=workflow_definition_ref,
386        inputs=parsed_inputs,
387        wait_for_completion=wait,
388        max_wait_seconds=wait_seconds,
389    )
390
391    print_json(result.model_dump())
392    if not result.success:
393        sys.exit(1)