airbyte_ops_mcp.mcp.connector_qa

  1# Copyright (c) 2025 Airbyte, Inc., all rights reserved.
  2"""MCP tools for connector quality assurance: regression tests and connector release blocking.
  3
  4## MCP reference
  5
  6.. include:: ../../../docs/mcp-generated/connector_qa.md
  7    :start-line: 2
  8"""
  9
 10from __future__ import annotations
 11
 12__all__: list[str] = []
 13
 14import uuid
 15from datetime import datetime
 16from enum import Enum
 17from typing import Annotated, Any
 18from urllib.parse import quote
 19
 20import requests
 21import yaml
 22from airbyte.cloud import CloudWorkspace
 23from airbyte.cloud.auth import resolve_cloud_client_id, resolve_cloud_client_secret
 24from airbyte.exceptions import (
 25    AirbyteMissingResourceError,
 26    AirbyteNoCloudCredentialsError,
 27    AirbyteWorkspaceMismatchError,
 28)
 29from fastmcp import FastMCP
 30from fastmcp_extensions import mcp_tool, register_mcp_tools
 31from pydantic import BaseModel, Field
 32
 33from airbyte_ops_mcp.constants import WorkspaceAliasEnum
 34from airbyte_ops_mcp.github_actions import (
 35    resolve_default_workflow_branch,
 36    trigger_workflow_dispatch,
 37)
 38from airbyte_ops_mcp.github_api import (
 39    GITHUB_API_BASE,
 40    get_file_contents_at_ref,
 41    resolve_ci_trigger_github_token,
 42)
 43from airbyte_ops_mcp.mcp.connector_versions import ConnectorRepo
 44
 45REGRESSION_TEST_REPO_OWNER = "airbytehq"
 46
 47REGRESSION_TEST_REPO_NAME = "airbyte-ops-mcp"
 48
 49REGRESSION_TEST_DEFAULT_BRANCH = "main"
 50
 51REGRESSION_TEST_WORKFLOW_FILE = "connector-regression-test.yml"
 52
 53
 54def validate_connection_workspace(
 55    connection_id: str,
 56    workspace_id: str,
 57) -> None:
 58    """Validate that a connection belongs to the expected workspace.
 59
 60    Uses PyAirbyte's CloudConnection.check_is_valid() method to verify that
 61    the connection exists and belongs to the specified workspace.
 62
 63    Raises:
 64        `AirbyteNoCloudCredentialsError`: If Airbyte Cloud credentials are missing.
 65        `AirbyteWorkspaceMismatchError`: If connection belongs to a different workspace.
 66        `AirbyteMissingResourceError`: If connection is not found.
 67    """
 68    client_id = resolve_cloud_client_id()
 69    client_secret = resolve_cloud_client_secret()
 70    if not client_id or not client_secret:
 71        # Only client credentials are resolved; suppress bearer-token guidance.
 72        raise AirbyteNoCloudCredentialsError(_allow_bearer=False)
 73
 74    workspace = CloudWorkspace(
 75        workspace_id=workspace_id,
 76        client_id=client_id,
 77        client_secret=client_secret,
 78    )
 79    connection = workspace.get_connection(connection_id)
 80    connection.check_is_valid()
 81
 82
 83def _get_workflow_run_status(
 84    owner: str,
 85    repo: str,
 86    run_id: int,
 87    token: str,
 88) -> dict[str, Any]:
 89    """Get workflow run details from GitHub API.
 90
 91    Args:
 92        owner: Repository owner (e.g., "airbytehq")
 93        repo: Repository name (e.g., "airbyte-ops-mcp")
 94        run_id: Workflow run ID
 95        token: GitHub API token
 96
 97    Returns:
 98        Workflow run data dictionary.
 99
100    Raises:
101        ValueError: If workflow run not found.
102        requests.HTTPError: If API request fails.
103    """
104    url = f"{GITHUB_API_BASE}/repos/{owner}/{repo}/actions/runs/{run_id}"
105    headers = {
106        "Authorization": f"Bearer {token}",
107        "Accept": "application/vnd.github+json",
108        "X-GitHub-Api-Version": "2022-11-28",
109    }
110
111    response = requests.get(url, headers=headers, timeout=30)
112    if response.status_code == 404:
113        raise ValueError(f"Workflow run {owner}/{repo}/actions/runs/{run_id} not found")
114    response.raise_for_status()
115
116    return response.json()
117
118
119class TestRunStatus(str, Enum):
120    """Status of a test run."""
121
122    QUEUED = "queued"
123    RUNNING = "running"
124    SUCCEEDED = "succeeded"
125    FAILED = "failed"
126
127
128class TestOutcome(str, Enum):
129    """Outcome of a test (execution or comparison)."""
130
131    PENDING = "pending"
132    RUNNING = "running"
133    PASSED = "passed"
134    FAILED = "failed"
135    SKIPPED = "skipped"
136
137
138class ValidationResultModel(BaseModel):
139    """Result of a single validation check."""
140
141    name: str = Field(description="Name of the validation check")
142    passed: bool = Field(description="Whether the validation passed")
143    message: str = Field(description="Human-readable result message")
144    errors: list[str] = Field(
145        default_factory=list,
146        description="List of error messages if validation failed",
147    )
148
149
150class StreamComparisonResultModel(BaseModel):
151    """Result of comparing a single stream between control and target."""
152
153    stream_name: str = Field(description="Name of the stream")
154    passed: bool = Field(description="Whether all comparisons passed")
155    control_record_count: int = Field(description="Number of records in control")
156    target_record_count: int = Field(description="Number of records in target")
157    missing_pks: list[str] = Field(
158        default_factory=list,
159        description="Primary keys present in control but missing in target",
160    )
161    differing_records: int = Field(
162        default=0,
163        description="Number of records that differ between control and target",
164    )
165    message: str = Field(description="Human-readable comparison summary")
166
167
168class RegressionTestExecutionResult(BaseModel):
169    """Results from executing the connector (validations and record counts)."""
170
171    outcome: TestOutcome = Field(description="Outcome of the execution")
172    catalog_validations: list[ValidationResultModel] = Field(
173        default_factory=list,
174        description="Results of catalog validation checks",
175    )
176    record_validations: list[ValidationResultModel] = Field(
177        default_factory=list,
178        description="Results of record validation checks",
179    )
180    record_count: int = Field(
181        default=0,
182        description="Total number of records read",
183    )
184    error_message: str | None = Field(
185        default=None,
186        description="Error message if the execution failed",
187    )
188
189
190class RegressionTestComparisonResult(BaseModel):
191    """Results from comparing target vs control connector versions."""
192
193    outcome: TestOutcome = Field(description="Outcome of the comparison")
194    baseline_version: str | None = Field(
195        default=None,
196        description="Version of the baseline (control) connector",
197    )
198    stream_comparisons: list[StreamComparisonResultModel] = Field(
199        default_factory=list,
200        description="Per-stream comparison results",
201    )
202    error_message: str | None = Field(
203        default=None,
204        description="Error message if the comparison failed",
205    )
206
207
208class RegressionTestResult(BaseModel):
209    """Complete result of a regression test run."""
210
211    run_id: str = Field(description="Unique identifier for this test run")
212    connection_id: str = Field(description="The connection being tested")
213    workspace_id: str = Field(description="The workspace containing the connection")
214    status: TestRunStatus = Field(description="Overall status of the test run")
215    target_version: str | None = Field(
216        default=None,
217        description="Version of the target connector being tested",
218    )
219    baseline_version: str | None = Field(
220        default=None,
221        description="Version of the baseline connector (if comparison mode)",
222    )
223    evaluation_mode: str = Field(
224        default="diagnostic",
225        description="Evaluation mode used (diagnostic or strict)",
226    )
227    compare_versions: bool = Field(
228        default=False,
229        description="Whether comparison mode was used (target vs control)",
230    )
231    execution_result: RegressionTestExecutionResult | None = Field(
232        default=None,
233        description="Results from executing the connector (validations and record counts)",
234    )
235    comparison_result: RegressionTestComparisonResult | None = Field(
236        default=None,
237        description="Results from comparing target vs control connector versions",
238    )
239    artifacts: dict[str, str] = Field(
240        default_factory=dict,
241        description="Paths to generated artifacts (JSONL, DuckDB, HAR files)",
242    )
243    human_summary: str = Field(
244        default="",
245        description="Human-readable summary of the test results",
246    )
247    started_at: datetime | None = Field(
248        default=None,
249        description="When the test run started",
250    )
251    completed_at: datetime | None = Field(
252        default=None,
253        description="When the test run completed",
254    )
255    test_description: str | None = Field(
256        default=None,
257        description="Optional description/context for this test run",
258    )
259
260
261class RunRegressionTestsResponse(BaseModel):
262    """Response from starting a regression test via GitHub Actions workflow."""
263
264    run_id: str = Field(
265        description="Unique identifier for the test run (internal tracking ID)"
266    )
267    status: TestRunStatus = Field(description="Initial status of the test run")
268    message: str = Field(description="Human-readable status message")
269    workflow_url: str | None = Field(
270        default=None,
271        description="URL to view the GitHub Actions workflow file",
272    )
273    github_run_id: int | None = Field(
274        default=None,
275        description="GitHub Actions workflow run ID (use with check_ci_workflow_status)",
276    )
277    github_run_url: str | None = Field(
278        default=None,
279        description="Direct URL to the GitHub Actions workflow run",
280    )
281
282
283@mcp_tool(
284    read_only=False,
285    idempotent=False,
286    open_world=True,
287)
288def run_regression_tests(
289    connector_name: Annotated[
290        str,
291        "Connector name to build from source (e.g., 'source-pokeapi'). Required.",
292    ],
293    pr: Annotated[
294        int,
295        "PR number to checkout and build from (e.g., 70847). Required. "
296        "The PR must be from the repository specified by the 'repo' parameter.",
297    ],
298    repo: Annotated[
299        ConnectorRepo,
300        "Repository where the connector PR is located. "
301        "Use 'airbyte' for OSS connectors (default) or 'airbyte-enterprise' for enterprise connectors.",
302    ],
303    connection_id: Annotated[
304        str | None,
305        "Airbyte Cloud connection ID to fetch config/catalog from. "
306        "If not provided, uses GSM integration test secrets.",
307    ] = None,
308    skip_compare: Annotated[
309        bool,
310        "If True, skip comparison and run single-version tests only. "
311        "If False (default), run comparison tests (target vs control versions).",
312    ] = False,
313    skip_read_action: Annotated[
314        bool,
315        "If True, skip the read action (run only spec, check, discover). "
316        "If False (default), run all verbs including read.",
317    ] = False,
318    override_test_image: Annotated[
319        str | None,
320        "Override test connector image with tag (e.g., 'airbyte/source-github:1.0.0'). "
321        "Ignored if skip_compare=False.",
322    ] = None,
323    override_control_image: Annotated[
324        str | None,
325        "Override control connector image (baseline version) with tag. "
326        "Ignored if skip_compare=True.",
327    ] = None,
328    workspace_id: Annotated[
329        str | WorkspaceAliasEnum | None,
330        "Optional Airbyte Cloud workspace ID (UUID) or alias. If provided with connection_id, "
331        "validates that the connection belongs to this workspace before triggering tests. "
332        "Accepts '@devin-ai-sandbox' as an alias for the Devin AI sandbox workspace.",
333    ] = None,
334    selected_streams: Annotated[
335        list[str] | None,
336        "List of stream names to include in the read. Only these streams will be included "
337        "in the configured catalog. This is useful to limit data volume by testing only "
338        "specific streams. If not provided, all streams are tested.",
339    ] = None,
340    enable_debug_logs: Annotated[
341        bool,
342        "Enable debug-level logging for regression test output. "
343        "Also passed as `LOG_LEVEL=DEBUG` to the connector Docker container.",
344    ] = False,
345    disable_http_replay: Annotated[
346        bool,
347        "Run both connector versions against the live API instead of replaying the "
348        "control run's recorded HTTP responses. Replay is on by default and is what "
349        "makes a record difference attributable to the version under test rather "
350        "than to upstream drift, so only disable it for a connector whose traffic "
351        "cannot be replayed faithfully. Comparison mode only.",
352    ] = False,
353    with_state: Annotated[
354        bool | None,
355        "Fetch and pass the connection's current state to the read command, "
356        "producing a warm read instead of a cold read. Defaults to `True` when "
357        "`connection_id` is provided, `False` otherwise. Has no effect unless "
358        "the command is `read`.",
359    ] = None,
360) -> RunRegressionTestsResponse:
361    """Start a regression test run via GitHub Actions workflow.
362
363    This tool triggers the regression test workflow which builds the connector
364    from the specified PR and runs tests against it.
365
366    Supports both OSS connectors (from airbytehq/airbyte) and enterprise connectors
367    (from airbytehq/airbyte-enterprise). Use the 'repo' parameter to specify which
368    repository contains the connector PR.
369
370    - skip_compare=False (default): Comparison mode - compares the PR version
371      against the baseline (control) version.
372    - skip_compare=True: Single-version mode - runs tests without comparison.
373
374    If connection_id is provided, config/catalog are fetched from Airbyte Cloud.
375    Otherwise, GSM integration test secrets are used.
376
377    Returns immediately with a run_id and workflow URL. Check the workflow URL
378    to monitor progress and view results.
379
380    Requires GITHUB_CI_WORKFLOW_TRIGGER_PAT or GITHUB_TOKEN environment variable
381    with 'actions:write' permission.
382    """
383    # Resolve workspace ID alias
384    resolved_workspace_id = WorkspaceAliasEnum.resolve(workspace_id)
385
386    # Generate a unique run ID for tracking
387    run_id = str(uuid.uuid4())
388
389    # Get GitHub token
390    try:
391        token = resolve_ci_trigger_github_token()
392    except ValueError as e:
393        return RunRegressionTestsResponse(
394            run_id=run_id,
395            status=TestRunStatus.FAILED,
396            message=str(e),
397            workflow_url=None,
398        )
399
400    # Validate workspace membership if workspace_id and connection_id are provided
401    if resolved_workspace_id and connection_id:
402        try:
403            validate_connection_workspace(connection_id, resolved_workspace_id)
404        except (
405            ValueError,
406            AirbyteWorkspaceMismatchError,
407            AirbyteMissingResourceError,
408        ) as e:
409            return RunRegressionTestsResponse(
410                run_id=run_id,
411                status=TestRunStatus.FAILED,
412                message=str(e),
413                workflow_url=None,
414            )
415
416    # Build workflow inputs - connector_name, pr, and repo are required
417    workflow_inputs: dict[str, str] = {
418        "connector_name": connector_name,
419        "pr": str(pr),
420        "repo": repo,
421    }
422
423    # Add optional inputs
424    if connection_id:
425        workflow_inputs["connection_id"] = connection_id
426    if skip_compare:
427        workflow_inputs["skip_compare"] = "true"
428    if skip_read_action:
429        workflow_inputs["skip_read_action"] = "true"
430    if override_test_image:
431        workflow_inputs["override_test_image"] = override_test_image
432    if override_control_image:
433        workflow_inputs["override_control_image"] = override_control_image
434    if selected_streams:
435        workflow_inputs["selected_streams"] = ",".join(selected_streams)
436    if enable_debug_logs:
437        workflow_inputs["enable_debug_logs"] = "true"
438    if disable_http_replay:
439        workflow_inputs["disable_http_replay"] = "true"
440    if with_state is True:
441        workflow_inputs["with_state"] = "true"
442    elif with_state is False:
443        workflow_inputs["with_state"] = "false"
444
445    mode_description = "single-version" if skip_compare else "comparison"
446
447    dispatch_result = trigger_workflow_dispatch(
448        owner=REGRESSION_TEST_REPO_OWNER,
449        repo=REGRESSION_TEST_REPO_NAME,
450        workflow_file=REGRESSION_TEST_WORKFLOW_FILE,
451        ref=resolve_default_workflow_branch(REGRESSION_TEST_DEFAULT_BRANCH),
452        inputs=workflow_inputs,
453        token=token,
454    )
455
456    view_url = dispatch_result.run_url or dispatch_result.workflow_url
457    connection_info = f" for connection {connection_id}" if connection_id else ""
458    repo_info = f" from {repo}" if repo != ConnectorRepo.AIRBYTE else ""
459    return RunRegressionTestsResponse(
460        run_id=run_id,
461        status=TestRunStatus.QUEUED,
462        message=(
463            f"{mode_description.capitalize()} regression test workflow triggered "
464            f"for {connector_name} (PR #{pr}{repo_info}){connection_info}. View progress at: {view_url}"
465        ),
466        workflow_url=dispatch_result.workflow_url,
467        github_run_id=dispatch_result.run_id,
468        github_run_url=dispatch_result.run_url,
469    )
470
471
472AIRBYTE_REPO_OWNER = "airbytehq"
473
474AIRBYTE_REPO_NAME = "airbyte"
475
476BLOCK_RELEASE_WORKFLOW_FILE = "block-release.yml"
477
478DEFAULT_REF = "master"
479
480
481class BlockConnectorReleaseResult(BaseModel):
482    """Result of triggering a release block workflow."""
483
484    success: bool = Field(description="Whether the workflow dispatch succeeded")
485    message: str = Field(description="Human-readable result message")
486    workflow_url: str = Field(description="URL to the workflow")
487    run_id: int | None = Field(
488        default=None, description="Workflow run ID, if discovered"
489    )
490    run_url: str | None = Field(default=None, description="URL to the workflow run")
491
492
493class UnblockConnectorReleaseResult(BaseModel):
494    """Result of triggering a release unblock workflow."""
495
496    success: bool = Field(description="Whether the workflow dispatch succeeded")
497    message: str = Field(description="Human-readable result message")
498    workflow_url: str = Field(description="URL to the workflow")
499    run_id: int | None = Field(
500        default=None, description="Workflow run ID, if discovered"
501    )
502    run_url: str | None = Field(default=None, description="URL to the workflow run")
503
504
505class ListBlockedConnectorsResult(BaseModel):
506    """Result of listing blocked connectors via the GitHub API."""
507
508    blocked_connectors: list[dict] = Field(
509        default_factory=list,
510        description="List of blocked connectors with optional block metadata",
511    )
512    count: int = Field(default=0, description="Number of blocked connectors")
513
514
515@mcp_tool(
516    read_only=False,
517    idempotent=False,
518    open_world=True,
519)
520def block_connector_release(
521    connector_name: Annotated[
522        str,
523        Field(
524            description="Connector technical name (e.g., `source-faker`, `destination-postgres`)"
525        ),
526    ],
527    reason: Annotated[
528        str,
529        Field(description="Human-readable reason for blocking the release"),
530    ],
531    yanked_version: Annotated[
532        str | None,
533        Field(description="Version that was yanked (for reference)"),
534    ] = None,
535    blocked_by: Annotated[
536        str | None,
537        Field(description="Email or identifier of the person requesting the block"),
538    ] = None,
539) -> BlockConnectorReleaseResult:
540    """Block a connector from being released by creating a `block-release.yaml` marker.
541
542    Triggers the `block-release.yml` workflow in the airbyte monorepo, which
543    creates a marker file, opens a PR, and force-merges it to master. While the
544    marker exists, the publish pipeline will skip the connector with a warning.
545
546    Use this after yanking a connector version to prevent CI from accidentally
547    re-publishing the broken code.
548    """
549    token = resolve_ci_trigger_github_token()
550
551    inputs: dict[str, str] = {
552        "connector-name": connector_name,
553        "action": "block",
554        "reason": reason,
555    }
556    if yanked_version:
557        inputs["yanked-version"] = yanked_version
558    if blocked_by:
559        inputs["blocked-by"] = blocked_by
560
561    result = trigger_workflow_dispatch(
562        owner=AIRBYTE_REPO_OWNER,
563        repo=AIRBYTE_REPO_NAME,
564        workflow_file=BLOCK_RELEASE_WORKFLOW_FILE,
565        ref=resolve_default_workflow_branch(DEFAULT_REF),
566        inputs=inputs,
567        token=token,
568        find_run=True,
569    )
570
571    if result.run_id:
572        message = (
573            f"Successfully triggered release block for {connector_name}. "
574            f"Run ID: {result.run_id}"
575        )
576    else:
577        message = (
578            f"Successfully triggered release block for {connector_name}. "
579            "Run ID not yet available."
580        )
581
582    return BlockConnectorReleaseResult(
583        success=True,
584        message=message,
585        workflow_url=result.workflow_url,
586        run_id=result.run_id,
587        run_url=result.run_url,
588    )
589
590
591@mcp_tool(
592    read_only=False,
593    idempotent=False,
594    open_world=True,
595)
596def unblock_connector_release(
597    connector_name: Annotated[
598        str,
599        Field(
600            description="Connector technical name (e.g., `source-faker`, `destination-postgres`)"
601        ),
602    ],
603) -> UnblockConnectorReleaseResult:
604    """Remove a release block for a connector by deleting its `block-release.yaml` marker.
605
606    Triggers the `block-release.yml` workflow with action=unblock, which removes
607    the marker file, opens a PR, and force-merges it to master. After this, the
608    connector can be published normally again.
609    """
610    token = resolve_ci_trigger_github_token()
611
612    result = trigger_workflow_dispatch(
613        owner=AIRBYTE_REPO_OWNER,
614        repo=AIRBYTE_REPO_NAME,
615        workflow_file=BLOCK_RELEASE_WORKFLOW_FILE,
616        ref=resolve_default_workflow_branch(DEFAULT_REF),
617        inputs={
618            "connector-name": connector_name,
619            "action": "unblock",
620        },
621        token=token,
622        find_run=True,
623    )
624
625    if result.run_id:
626        message = (
627            f"Successfully triggered release unblock for {connector_name}. "
628            f"Run ID: {result.run_id}"
629        )
630    else:
631        message = (
632            f"Successfully triggered release unblock for {connector_name}. "
633            "Run ID not yet available."
634        )
635
636    return UnblockConnectorReleaseResult(
637        success=True,
638        message=message,
639        workflow_url=result.workflow_url,
640        run_id=result.run_id,
641        run_url=result.run_url,
642    )
643
644
645@mcp_tool(
646    read_only=True,
647    idempotent=True,
648    open_world=True,
649)
650def list_blocked_connector_releases(
651    connector_name: Annotated[
652        str | None,
653        Field(
654            description="Optional connector name to check. If not provided, scans all connectors."
655        ),
656    ] = None,
657    include_details: Annotated[
658        bool,
659        Field(
660            description="Whether to fetch and parse each marker file for reason and metadata."
661        ),
662    ] = True,
663) -> ListBlockedConnectorsResult:
664    """List connectors that are currently blocked from release.
665
666    Searches the airbyte monorepo for `block-release.yaml` marker files using
667    the GitHub API. Returns a list of blocked connectors, with marker metadata
668    when `include_details` is `True`.
669    """
670
671    token = resolve_ci_trigger_github_token()
672    ref = resolve_default_workflow_branch(DEFAULT_REF)
673    blocked: list[dict] = []
674
675    if connector_name:
676        blocked = _check_single_connector_block(connector_name, token, ref)
677    else:
678        blocked = _search_all_blocked_connectors(token, ref, include_details)
679
680    return ListBlockedConnectorsResult(
681        blocked_connectors=blocked,
682        count=len(blocked),
683    )
684
685
686def _check_single_connector_block(
687    connector_name: str,
688    token: str,
689    ref: str,
690) -> list[dict]:
691    """Check if a single connector has a release block."""
692    path = f"airbyte-integrations/connectors/{connector_name}/block-release.yaml"
693    content = get_file_contents_at_ref(
694        owner=AIRBYTE_REPO_OWNER,
695        repo=AIRBYTE_REPO_NAME,
696        path=path,
697        ref=ref,
698        token=token,
699    )
700    if content is None:
701        return []
702
703    return _parse_block_marker_content(connector_name, content)
704
705
706def _parse_block_marker_content(connector_name: str, content: str) -> list[dict]:
707    """Parse a `block-release.yaml` marker into the MCP response shape."""
708    try:
709        block_file_data = yaml.safe_load(content)
710        if isinstance(block_file_data, dict):
711            return [
712                {
713                    "connector_name": connector_name,
714                    "reason": block_file_data.get("reason", "(no reason provided)"),
715                    "yanked_version": block_file_data.get("yanked_version"),
716                    "blocked_at": block_file_data.get("blocked_at"),
717                    "blocked_by": block_file_data.get("blocked_by"),
718                }
719            ]
720    except yaml.YAMLError:
721        return [
722            {"connector_name": connector_name, "reason": "(unable to parse marker)"}
723        ]
724
725    return [
726        {
727            "connector_name": connector_name,
728            "reason": "(invalid block-release.yaml format)",
729        }
730    ]
731
732
733def _search_all_blocked_connectors(
734    token: str,
735    ref: str,
736    include_details: bool,
737) -> list[dict]:
738    """Search the repo for all `block-release.yaml` files at the requested ref."""
739    tree_ref = quote(ref, safe="")
740    url = (
741        f"{GITHUB_API_BASE}/repos/{AIRBYTE_REPO_OWNER}/{AIRBYTE_REPO_NAME}"
742        f"/git/trees/{tree_ref}"
743    )
744    headers = {
745        "Authorization": f"Bearer {token}",
746        "Accept": "application/vnd.github+json",
747        "X-GitHub-Api-Version": "2022-11-28",
748    }
749    response = requests.get(
750        url,
751        headers=headers,
752        params={"recursive": "1"},
753        timeout=30,
754    )
755    response.raise_for_status()
756
757    tree = response.json().get("tree", [])
758    blocked: list[dict] = []
759
760    for item in tree:
761        file_path = item.get("path", "")
762        if not file_path.endswith("/block-release.yaml"):
763            continue
764
765        parts = file_path.split("/")
766        if (
767            len(parts) == 4
768            and parts[0] == "airbyte-integrations"
769            and parts[1] == "connectors"
770        ):
771            connector_name = parts[2]
772            if not include_details:
773                blocked.append({"connector_name": connector_name})
774                continue
775
776            content = get_file_contents_at_ref(
777                owner=AIRBYTE_REPO_OWNER,
778                repo=AIRBYTE_REPO_NAME,
779                path=file_path,
780                ref=ref,
781                token=token,
782            )
783            if content is not None:
784                blocked.extend(_parse_block_marker_content(connector_name, content))
785
786    return blocked
787
788
789def register_connector_qa_tools(app: FastMCP) -> None:
790    """Register connector_qa tools with the FastMCP app."""
791    register_mcp_tools(app, mcp_module=__name__)