airbyte_ops_mcp.mcp.connector_versions

   1# Copyright (c) 2025 Airbyte, Inc., all rights reserved.
   2"""MCP tools for connector version lifecycle: cloud version overrides, progressive rollouts, and pre-release publishing.
   3
   4## MCP reference
   5
   6.. include:: ../../../docs/mcp-generated/connector_versions.md
   7    :start-line: 2
   8"""
   9
  10# NOTE: We intentionally do NOT use `from __future__ import annotations` here.
  11# FastMCP has issues resolving forward references when PEP 563 deferred annotations
  12# are used. See: https://github.com/jlowin/fastmcp/issues/905
  13# Python 3.12+ supports modern type hint syntax natively, so this is not needed.
  14
  15__all__: list[str] = []
  16
  17import base64
  18import logging
  19from dataclasses import dataclass
  20from enum import StrEnum
  21from typing import Annotated, Literal
  22
  23import requests
  24import yaml
  25from airbyte import constants
  26from airbyte.exceptions import PyAirbyteInputError
  27from fastmcp import Context, FastMCP
  28from fastmcp_extensions import get_mcp_config, mcp_tool, register_mcp_tools
  29from pydantic import BaseModel, Field
  30
  31from airbyte_ops_mcp.airbyte_repo.bump_version import strip_prerelease_suffix
  32from airbyte_ops_mcp.approval_resolution import (
  33    ApprovalStatus,
  34    check_approval_status,
  35)
  36from airbyte_ops_mcp.cloud_admin import api_client
  37from airbyte_ops_mcp.cloud_admin.auth import (
  38    CloudAuthError,
  39    require_internal_admin_flag_only,
  40)
  41from airbyte_ops_mcp.cloud_admin.models import (
  42    ConnectorRolloutFinalizeResult,
  43    ConnectorRolloutProgressResult,
  44    ConnectorRolloutStartResult,
  45    ConnectorVersionInfo,
  46    OrganizationVersionOverrideResult,
  47    VersionOverrideOperationResult,
  48    WorkspaceVersionOverrideResult,
  49)
  50from airbyte_ops_mcp.cloud_admin.version_overrides import (
  51    VersionOverrideTarget,
  52    get_connector_version_info,
  53    set_version_override,
  54)
  55from airbyte_ops_mcp.connector_ops.rollouts._helpers import (
  56    count_eligible_or_pinned_actors,
  57)
  58from airbyte_ops_mcp.constants import ServerConfigKey, WorkspaceAliasEnum
  59from airbyte_ops_mcp.github_actions import trigger_workflow_dispatch
  60from airbyte_ops_mcp.github_api import (
  61    GITHUB_API_BASE,
  62    get_pr_head_ref,
  63    resolve_ci_trigger_github_token,
  64)
  65from airbyte_ops_mcp.mcp.cloud_auth import resolve_cloud_auth
  66from airbyte_ops_mcp.tier_cache import TierFilter, resolve_workspace
  67
  68logger = logging.getLogger(__name__)
  69
  70
  71@mcp_tool(
  72    read_only=True,
  73    idempotent=True,
  74    open_world=True,
  75)
  76def get_cloud_connector_version(
  77    workspace_id: Annotated[
  78        str | WorkspaceAliasEnum,
  79        Field(
  80            description="The Airbyte Cloud workspace ID (UUID) or alias. "
  81            "Accepts '@devin-ai-sandbox' as an alias for the Devin AI sandbox workspace."
  82        ),
  83    ],
  84    actor_id: Annotated[
  85        str, "The ID of the deployed connector (source or destination)"
  86    ],
  87    actor_type: Annotated[
  88        Literal["source", "destination"],
  89        "The type of connector (source or destination)",
  90    ],
  91    config_api_root: Annotated[
  92        str | None,
  93        Field(
  94            description="Optional API root URL override for the Config API. "
  95            "Defaults to Airbyte Cloud (https://cloud.airbyte.com/api/v1). "
  96            "Use this to target local or self-hosted deployments.",
  97            default=None,
  98        ),
  99    ] = None,
 100    *,
 101    ctx: Context,
 102) -> ConnectorVersionInfo:
 103    """Get the current version information for a deployed connector.
 104
 105    Returns version details including the current version string and whether
 106    an override (pin) is applied.
 107
 108    Authentication credentials are resolved in priority order:
 109
 110    1. Bearer token (Authorization header or AIRBYTE_CLOUD_BEARER_TOKEN env var)
 111    2. HTTP headers: X-Airbyte-Cloud-Client-Id, X-Airbyte-Cloud-Client-Secret
 112    3. Environment variables: AIRBYTE_CLOUD_CLIENT_ID, AIRBYTE_CLOUD_CLIENT_SECRET
 113    """
 114    resolved_workspace_id = WorkspaceAliasEnum.resolve(workspace_id)
 115    assert resolved_workspace_id is not None  # workspace_id is required
 116
 117    return get_connector_version_info(
 118        auth=resolve_cloud_auth(ctx),
 119        workspace_id=resolved_workspace_id,
 120        actor_id=actor_id,
 121        actor_type=actor_type,
 122        config_api_root=config_api_root,
 123    )
 124
 125
 126@mcp_tool(
 127    destructive=True,
 128    idempotent=False,
 129    open_world=True,
 130)
 131def set_cloud_connector_version_override(
 132    workspace_id: Annotated[
 133        str | WorkspaceAliasEnum,
 134        Field(
 135            description="The Airbyte Cloud workspace ID (UUID) or alias. "
 136            "Accepts '@devin-ai-sandbox' as an alias for the Devin AI sandbox workspace."
 137        ),
 138    ],
 139    actor_id: Annotated[
 140        str, "The ID of the deployed connector (source or destination)"
 141    ],
 142    actor_type: Annotated[
 143        Literal["source", "destination"],
 144        "The type of connector (source or destination)",
 145    ],
 146    approval_comment_url: Annotated[
 147        str | None,
 148        Field(
 149            description="URL to the Slack approval record. Obtain this by calling the "
 150            "`escalate_to_human` tool with `approval_requested=True`; the backend delivers "
 151            "the approval record URL when a human clicks Approve. "
 152            "Format: https://<workspace>.slack.com/archives/... "
 153            "The admin email is automatically resolved from the approver's identity "
 154            "via the team roster.",
 155            default=None,
 156        ),
 157    ],
 158    version: Annotated[
 159        str | None,
 160        Field(
 161            description="The semver version string to pin to (e.g., '0.1.0'). "
 162            "Must be None if unset is True.",
 163            default=None,
 164        ),
 165    ],
 166    unset: Annotated[
 167        bool,
 168        Field(
 169            description="If True, removes any existing version override. "
 170            "Cannot be True if version is provided.",
 171            default=False,
 172        ),
 173    ],
 174    override_reason: Annotated[
 175        str | None,
 176        Field(
 177            description="Required when setting a version. "
 178            "Explanation for the override (min 10 characters).",
 179            default=None,
 180        ),
 181    ],
 182    override_reason_reference_url: Annotated[
 183        str | None,
 184        Field(
 185            description="Optional URL with more context (e.g., issue link).",
 186            default=None,
 187        ),
 188    ],
 189    issue_url: Annotated[
 190        str | None,
 191        Field(
 192            description="URL to the GitHub issue providing context for this operation. "
 193            "Must be a valid GitHub URL (https://github.com/...). Required for authorization.",
 194            default=None,
 195        ),
 196    ],
 197    ai_agent_session_url: Annotated[
 198        str | None,
 199        Field(
 200            description="URL to the AI agent session driving this operation, if applicable. "
 201            "Provides additional auditability for AI-driven operations.",
 202            default=None,
 203        ),
 204    ] = None,
 205    force: Annotated[
 206        bool,
 207        Field(
 208            description="If `True`, allow overwriting an existing version pin. "
 209            "Existing pins may have been set by rollouts, breaking-change migrations, "
 210            "or other operators. Defaults to `False`. NOTE: `force=True` only "
 211            "bypasses the existing-pin check — major-version crossings are always "
 212            "blocked and cannot be overridden.",
 213            default=False,
 214        ),
 215    ] = False,
 216    config_api_root: Annotated[
 217        str | None,
 218        Field(
 219            description="Optional API root URL override for the Config API. "
 220            "Defaults to Airbyte Cloud (https://cloud.airbyte.com/api/v1). "
 221            "Use this to target local or self-hosted deployments.",
 222            default=None,
 223        ),
 224    ] = None,
 225    customer_tier_filter: Annotated[
 226        TierFilter,
 227        Field(
 228            description=(
 229                "Required tier filter: 'TIER_0', 'TIER_1', 'TIER_2', 'UNKNOWN', or 'ALL'. "
 230                "The operation will be rejected if the actual customer tier does not match. "
 231                "Use 'ALL' to proceed regardless of tier (a warning is shown for sensitive tiers)."
 232            ),
 233        ),
 234    ] = "TIER_2",
 235    *,
 236    ctx: Context,
 237) -> VersionOverrideOperationResult:
 238    """Set or clear a version override for a deployed connector.
 239
 240    **Admin-only operation** - Requires:
 241
 242    - AIRBYTE_INTERNAL_ADMIN_FLAG=airbyte.io environment variable
 243    - issue_url parameter (GitHub issue URL for context)
 244    - approval_comment_url (Slack approval record URL from `escalate_to_human`)
 245
 246    The admin user email is automatically derived from the Slack approval record,
 247    resolving the approver's @airbyte.io email via the team roster.
 248
 249    You must specify EXACTLY ONE of `version` OR `unset=True`, but not both.
 250    When setting a version, `override_reason` is required.
 251
 252    The `customer_tier_filter` parameter gates the operation: the call fails if
 253    the actual tier of the workspace's organization does not match.  Use `ALL`
 254    to bypass the check (a warning is still emitted for sensitive tiers).
 255
 256    Business rules enforced:
 257
 258    - Dev versions (-dev): Only creator can unpin their own dev version override
 259    - Production versions: Require strong justification mentioning customer/support/investigation
 260    - Release candidates (-rc): Any admin can pin/unpin RC versions
 261
 262    Authentication credentials are resolved in priority order:
 263
 264    1. Bearer token (Authorization header or AIRBYTE_CLOUD_BEARER_TOKEN env var)
 265    2. HTTP headers: X-Airbyte-Cloud-Client-Id, X-Airbyte-Cloud-Client-Secret
 266    3. Environment variables: AIRBYTE_CLOUD_CLIENT_ID, AIRBYTE_CLOUD_CLIENT_SECRET
 267    """
 268    resolved_workspace_id = WorkspaceAliasEnum.resolve(workspace_id)
 269    assert resolved_workspace_id is not None  # workspace_id is required
 270
 271    ws_resolution = resolve_workspace(
 272        workspace_id=resolved_workspace_id,
 273        allow_degraded=True,
 274    )
 275    if not ws_resolution.organization_id:
 276        return VersionOverrideOperationResult(
 277            success=False,
 278            message="Could not resolve organization for workspace.",
 279            connector_id=actor_id,
 280            connector_type=actor_type,
 281        )
 282
 283    result = set_version_override(
 284        auth=resolve_cloud_auth(ctx),
 285        target=VersionOverrideTarget(
 286            scope="actor",
 287            organization_id=ws_resolution.organization_id,
 288            workspace_id=resolved_workspace_id,
 289            actor_id=actor_id,
 290            connector_type=actor_type,
 291        ),
 292        approval_comment_url=approval_comment_url,
 293        version=version,
 294        unset=unset,
 295        override_reason=override_reason,
 296        override_reason_reference_url=override_reason_reference_url,
 297        issue_url=issue_url,
 298        ai_agent_session_url=ai_agent_session_url,
 299        customer_tier_filter=customer_tier_filter,
 300        force=force,
 301        config_api_root=config_api_root,
 302    )
 303    assert isinstance(result, VersionOverrideOperationResult)
 304    return result
 305
 306
 307@mcp_tool(
 308    destructive=True,
 309    idempotent=False,
 310    open_world=True,
 311)
 312def set_workspace_connector_version_override(
 313    workspace_id: Annotated[
 314        str | WorkspaceAliasEnum,
 315        Field(
 316            description="The Airbyte Cloud workspace ID (UUID) or alias. "
 317            "Accepts '@devin-ai-sandbox' as an alias for the Devin AI sandbox workspace."
 318        ),
 319    ],
 320    connector_name: Annotated[
 321        str,
 322        Field(
 323            description="The connector name (e.g., 'source-github', 'destination-bigquery')."
 324        ),
 325    ],
 326    connector_type: Annotated[
 327        Literal["source", "destination"],
 328        "The type of connector (source or destination)",
 329    ],
 330    approval_comment_url: Annotated[
 331        str | None,
 332        Field(
 333            description="URL to the Slack approval record. Obtain this by calling the "
 334            "`escalate_to_human` tool with `approval_requested=True`; the backend delivers "
 335            "the approval record URL when a human clicks Approve. "
 336            "Format: https://<workspace>.slack.com/archives/... "
 337            "The admin email is automatically resolved from the approver's identity "
 338            "via the team roster.",
 339            default=None,
 340        ),
 341    ],
 342    version: Annotated[
 343        str | None,
 344        Field(
 345            description="The semver version string to pin to (e.g., '0.1.0'). "
 346            "Must be None if unset is True.",
 347            default=None,
 348        ),
 349    ],
 350    unset: Annotated[
 351        bool,
 352        Field(
 353            description="If True, removes any existing version override. "
 354            "Cannot be True if version is provided.",
 355            default=False,
 356        ),
 357    ],
 358    override_reason: Annotated[
 359        str | None,
 360        Field(
 361            description="Required when setting a version. "
 362            "Explanation for the override (min 10 characters).",
 363            default=None,
 364        ),
 365    ],
 366    override_reason_reference_url: Annotated[
 367        str | None,
 368        Field(
 369            description="Optional URL with more context (e.g., issue link).",
 370            default=None,
 371        ),
 372    ],
 373    issue_url: Annotated[
 374        str | None,
 375        Field(
 376            description="URL to the GitHub issue providing context for this operation. "
 377            "Must be a valid GitHub URL (https://github.com/...). Required for authorization.",
 378            default=None,
 379        ),
 380    ],
 381    ai_agent_session_url: Annotated[
 382        str | None,
 383        Field(
 384            description="URL to the AI agent session driving this operation, if applicable. "
 385            "Provides additional auditability for AI-driven operations.",
 386            default=None,
 387        ),
 388    ] = None,
 389    force: Annotated[
 390        bool,
 391        Field(
 392            description="If `True`, allow overwriting an existing version pin. "
 393            "Existing pins may have been set by rollouts, breaking-change migrations, "
 394            "or other operators. Defaults to `False`. NOTE: `force=True` only "
 395            "bypasses the existing-pin check — major-version crossings are always "
 396            "blocked and cannot be overridden.",
 397            default=False,
 398        ),
 399    ] = False,
 400    config_api_root: Annotated[
 401        str | None,
 402        Field(
 403            description="Optional API root URL override for the Config API. "
 404            "Defaults to Airbyte Cloud (https://cloud.airbyte.com/api/v1). "
 405            "Use this to target local or self-hosted deployments.",
 406            default=None,
 407        ),
 408    ] = None,
 409    customer_tier_filter: Annotated[
 410        TierFilter,
 411        Field(
 412            description=(
 413                "Required tier filter: 'TIER_0', 'TIER_1', 'TIER_2', 'UNKNOWN', or 'ALL'. "
 414                "The operation will be rejected if the actual customer tier does not match. "
 415                "Use 'ALL' to proceed regardless of tier (a warning is shown for sensitive tiers)."
 416            ),
 417        ),
 418    ] = "TIER_2",
 419    *,
 420    ctx: Context,
 421) -> WorkspaceVersionOverrideResult:
 422    """Set or clear a workspace-level version override for a connector type.
 423
 424    This pins ALL instances of a connector type within a workspace to a specific version.
 425    For example, pinning 'source-github' at workspace level means all GitHub sources
 426    in that workspace will use the pinned version.
 427
 428    **Admin-only operation** - Requires:
 429
 430    - AIRBYTE_INTERNAL_ADMIN_FLAG=airbyte.io environment variable
 431    - issue_url parameter (GitHub issue URL for context)
 432    - approval_comment_url (Slack approval record URL from `escalate_to_human`)
 433
 434    You must specify EXACTLY ONE of `version` OR `unset=True`, but not both.
 435    When setting a version, `override_reason` is required.
 436
 437    The `customer_tier_filter` parameter gates the operation: the call fails if
 438    the actual tier of the workspace's organization does not match.  Use `ALL`
 439    to bypass the check (a warning is still emitted for sensitive tiers).
 440    """
 441    resolved_workspace_id = WorkspaceAliasEnum.resolve(workspace_id)
 442    assert resolved_workspace_id is not None  # workspace_id is required
 443
 444    ws_resolution = resolve_workspace(
 445        workspace_id=resolved_workspace_id,
 446        allow_degraded=True,
 447    )
 448    if not ws_resolution.organization_id:
 449        return WorkspaceVersionOverrideResult(
 450            success=False,
 451            message="Could not resolve organization for workspace.",
 452            workspace_id=resolved_workspace_id,
 453            connector_name=connector_name,
 454            connector_type=connector_type,
 455        )
 456
 457    result = set_version_override(
 458        auth=resolve_cloud_auth(ctx),
 459        target=VersionOverrideTarget(
 460            scope="workspace",
 461            organization_id=ws_resolution.organization_id,
 462            workspace_id=resolved_workspace_id,
 463            connector_name=connector_name,
 464            connector_type=connector_type,
 465        ),
 466        approval_comment_url=approval_comment_url,
 467        version=version,
 468        unset=unset,
 469        override_reason=override_reason,
 470        override_reason_reference_url=override_reason_reference_url,
 471        issue_url=issue_url,
 472        ai_agent_session_url=ai_agent_session_url,
 473        customer_tier_filter=customer_tier_filter,
 474        force=force,
 475        config_api_root=config_api_root,
 476    )
 477    assert isinstance(result, WorkspaceVersionOverrideResult)
 478    return result
 479
 480
 481@mcp_tool(
 482    destructive=True,
 483    idempotent=False,
 484    open_world=True,
 485)
 486def set_organization_connector_version_override(
 487    organization_id: Annotated[
 488        str,
 489        Field(description="The Airbyte Cloud organization ID."),
 490    ],
 491    connector_name: Annotated[
 492        str,
 493        Field(
 494            description="The connector name (e.g., 'source-github', 'destination-bigquery')."
 495        ),
 496    ],
 497    connector_type: Annotated[
 498        Literal["source", "destination"],
 499        "The type of connector (source or destination)",
 500    ],
 501    approval_comment_url: Annotated[
 502        str | None,
 503        Field(
 504            description="URL to the Slack approval record. Obtain this by calling the "
 505            "`escalate_to_human` tool with `approval_requested=True`; the backend delivers "
 506            "the approval record URL when a human clicks Approve. "
 507            "Format: https://<workspace>.slack.com/archives/... "
 508            "The admin email is automatically resolved from the approver's identity "
 509            "via the team roster.",
 510            default=None,
 511        ),
 512    ],
 513    version: Annotated[
 514        str | None,
 515        Field(
 516            description="The semver version string to pin to (e.g., '0.1.0'). "
 517            "Must be None if unset is True.",
 518            default=None,
 519        ),
 520    ],
 521    unset: Annotated[
 522        bool,
 523        Field(
 524            description="If True, removes any existing version override. "
 525            "Cannot be True if version is provided.",
 526            default=False,
 527        ),
 528    ],
 529    override_reason: Annotated[
 530        str | None,
 531        Field(
 532            description="Required when setting a version. "
 533            "Explanation for the override (min 10 characters).",
 534            default=None,
 535        ),
 536    ],
 537    override_reason_reference_url: Annotated[
 538        str | None,
 539        Field(
 540            description="Optional URL with more context (e.g., issue link).",
 541            default=None,
 542        ),
 543    ],
 544    issue_url: Annotated[
 545        str | None,
 546        Field(
 547            description="URL to the GitHub issue providing context for this operation. "
 548            "Must be a valid GitHub URL (https://github.com/...). Required for authorization.",
 549            default=None,
 550        ),
 551    ],
 552    ai_agent_session_url: Annotated[
 553        str | None,
 554        Field(
 555            description="URL to the AI agent session driving this operation, if applicable. "
 556            "Provides additional auditability for AI-driven operations.",
 557            default=None,
 558        ),
 559    ] = None,
 560    force: Annotated[
 561        bool,
 562        Field(
 563            description="If `True`, allow overwriting an existing version pin. "
 564            "Existing pins may have been set by rollouts, breaking-change migrations, "
 565            "or other operators. Defaults to `False`. NOTE: `force=True` only "
 566            "bypasses the existing-pin check — major-version crossings are always "
 567            "blocked and cannot be overridden.",
 568            default=False,
 569        ),
 570    ] = False,
 571    config_api_root: Annotated[
 572        str | None,
 573        Field(
 574            description="Optional API root URL override for the Config API. "
 575            "Defaults to Airbyte Cloud (https://cloud.airbyte.com/api/v1). "
 576            "Use this to target local or self-hosted deployments.",
 577            default=None,
 578        ),
 579    ] = None,
 580    customer_tier_filter: Annotated[
 581        TierFilter,
 582        Field(
 583            description=(
 584                "Required tier filter: 'TIER_0', 'TIER_1', 'TIER_2', 'UNKNOWN', or 'ALL'. "
 585                "The operation will be rejected if the actual customer tier does not match. "
 586                "Use 'ALL' to proceed regardless of tier (a warning is shown for sensitive tiers)."
 587            ),
 588        ),
 589    ] = "TIER_2",
 590    *,
 591    ctx: Context,
 592) -> OrganizationVersionOverrideResult:
 593    """Set or clear an organization-level version override for a connector type.
 594
 595    This pins ALL instances of a connector type across an entire organization to a
 596    specific version. For example, pinning 'source-github' at organization level means
 597    all GitHub sources in all workspaces within that organization will use the pinned version.
 598
 599    **Admin-only operation** - Requires:
 600
 601    - AIRBYTE_INTERNAL_ADMIN_FLAG=airbyte.io environment variable
 602    - issue_url parameter (GitHub issue URL for context)
 603    - approval_comment_url (Slack approval record URL from `escalate_to_human`)
 604
 605    You must specify EXACTLY ONE of `version` OR `unset=True`, but not both.
 606    When setting a version, `override_reason` is required.
 607
 608    The `customer_tier_filter` parameter gates the operation: the call fails if
 609    the actual tier of the organization does not match.  Use `ALL` to bypass
 610    the check (a warning is still emitted for sensitive tiers).
 611    """
 612    result = set_version_override(
 613        auth=resolve_cloud_auth(ctx),
 614        target=VersionOverrideTarget(
 615            scope="organization",
 616            organization_id=organization_id,
 617            connector_name=connector_name,
 618            connector_type=connector_type,
 619        ),
 620        approval_comment_url=approval_comment_url,
 621        version=version,
 622        unset=unset,
 623        override_reason=override_reason,
 624        override_reason_reference_url=override_reason_reference_url,
 625        issue_url=issue_url,
 626        ai_agent_session_url=ai_agent_session_url,
 627        customer_tier_filter=customer_tier_filter,
 628        force=force,
 629        config_api_root=config_api_root,
 630    )
 631    assert isinstance(result, OrganizationVersionOverrideResult)
 632    return result
 633
 634
 635@dataclass(frozen=True)
 636class _ResolvedCloudAuth:
 637    """Resolved authentication for Airbyte Cloud API calls.
 638
 639    Either bearer_token OR (client_id AND client_secret) will be set, not both.
 640    """
 641
 642    bearer_token: str | None = None
 643    client_id: str | None = None
 644    client_secret: str | None = None
 645
 646
 647def _resolve_cloud_auth(ctx: Context) -> _ResolvedCloudAuth:
 648    """Resolve authentication credentials for Airbyte Cloud API.
 649
 650    Credentials are resolved in priority order:
 651    1. Bearer token (Authorization header or AIRBYTE_CLOUD_BEARER_TOKEN env var)
 652    2. Client credentials (X-Airbyte-Cloud-Client-Id/Secret headers or env vars)
 653
 654    Args:
 655        ctx: FastMCP Context object from the current tool invocation.
 656
 657    Returns:
 658        _ResolvedCloudAuth with either bearer_token or client credentials set.
 659
 660    Raises:
 661        CloudAuthError: If credentials cannot be resolved from headers or env vars.
 662    """
 663    # Try bearer token first (preferred, but not required)
 664    bearer_token = get_mcp_config(ctx, ServerConfigKey.BEARER_TOKEN)
 665    if bearer_token:
 666        return _ResolvedCloudAuth(bearer_token=bearer_token)
 667
 668    # Fall back to client credentials
 669    try:
 670        client_id = get_mcp_config(ctx, ServerConfigKey.CLIENT_ID)
 671        client_secret = get_mcp_config(ctx, ServerConfigKey.CLIENT_SECRET)
 672        return _ResolvedCloudAuth(
 673            client_id=client_id,
 674            client_secret=client_secret,
 675        )
 676    except ValueError as e:
 677        raise CloudAuthError(
 678            f"Failed to resolve credentials. Ensure credentials are provided "
 679            f"via Authorization header (Bearer token), "
 680            f"HTTP headers (X-Airbyte-Cloud-Client-Id, X-Airbyte-Cloud-Client-Secret), "
 681            f"or environment variables. Error: {e}"
 682        ) from e
 683
 684
 685@mcp_tool(
 686    destructive=True,
 687    idempotent=False,
 688    open_world=True,
 689)
 690def start_connector_rollout(
 691    docker_repository: Annotated[
 692        str,
 693        Field(description="The docker repository (e.g., 'airbyte/source-pokeapi')"),
 694    ],
 695    docker_image_tag: Annotated[
 696        str,
 697        Field(description="The docker image tag (e.g., '0.3.48-rc.1')"),
 698    ],
 699    actor_definition_id: Annotated[
 700        str,
 701        Field(description="The actor definition ID (UUID)"),
 702    ],
 703    approval_comment_url: Annotated[
 704        str | None,
 705        Field(
 706            description="URL to the Slack approval record. Obtain this by calling the "
 707            "`escalate_to_human` tool with `approval_requested=True`; the backend delivers "
 708            "the approval record URL when a human clicks Approve. "
 709            "Format: https://<workspace>.slack.com/archives/... "
 710            "The admin email is automatically resolved from the approver's identity "
 711            "via the team roster.",
 712            default=None,
 713        ),
 714    ],
 715    admin_user_email_override: Annotated[
 716        str | None,
 717        Field(
 718            description="Direct admin email override for webapp-initiated actions. "
 719            "When the Ops Webapp env var is set, this bypasses the approval URL "
 720            "requirement. Ignored in agent/cron environments.",
 721            default=None,
 722        ),
 723    ],
 724    rollout_strategy: Annotated[
 725        Literal["manual", "automated", "overridden"],
 726        Field(
 727            description="The rollout strategy: "
 728            "'manual' for manual control of rollout progression, "
 729            "'automated' for automatic progression based on metrics, "
 730            "'overridden' for special cases where normal rules are bypassed.",
 731            default="manual",
 732        ),
 733    ],
 734    initial_rollout_pct: Annotated[
 735        int | None,
 736        Field(
 737            description="Initial/step percentage for rollout progression (0-100). "
 738            "For automated rollouts, this is the percentage increment per step. "
 739            "For example, 25 means the rollout will advance by 25% each step. "
 740            "Default is 25% if not specified.",
 741            default=None,
 742        ),
 743    ],
 744    final_target_rollout_pct: Annotated[
 745        int | None,
 746        Field(
 747            description="Maximum percentage of actors to pin (0-100). "
 748            "The rollout will not exceed this percentage. "
 749            "For example, 50 means at most 50% of actors will be pinned to the RC. "
 750            "Default is 50% if not specified.",
 751            default=None,
 752        ),
 753    ],
 754    customer_tier: Annotated[
 755        Literal["TIER_0", "TIER_1", "TIER_2", "ALL"] | None,
 756        Field(
 757            description="The customer tier to target for this rollout. "
 758            "Each tier represents a different group of customers: "
 759            "'TIER_0' for the highest-priority customers, "
 760            "'TIER_1' for mid-tier customers, "
 761            "'TIER_2' for the broadest customer group (default if not specified), "
 762            "'ALL' to target all customer tiers. "
 763            "When not specified, the platform defaults to TIER_2 only.",
 764            default=None,
 765        ),
 766    ],
 767    *,
 768    ctx: Context,
 769) -> ConnectorRolloutStartResult:
 770    """Start or configure a connector rollout workflow.
 771
 772    This tool configures and starts a connector rollout workflow. It can be called
 773    multiple times while the rollout is in INITIALIZED state to update the configuration
 774    (strategy, percentages). Once the Temporal workflow starts and the state transitions
 775    to WORKFLOW_STARTED, the configuration is locked and cannot be changed.
 776
 777    **Behavior:**
 778    - If rollout is INITIALIZED: Updates configuration and starts the workflow
 779    - If rollout is already started: Returns an error (configuration is locked)
 780
 781    **Configuration Parameters:**
 782    - rollout_strategy: 'manual' (default), 'automated', or 'overridden'
 783    - initial_rollout_pct: Step size for progression (default: 25%)
 784    - final_target_rollout_pct: Maximum percentage to pin (default: 50%)
 785    - customer_tier: Customer tier to target - 'TIER_0', 'TIER_1', 'TIER_2', or 'ALL' (default: TIER_2)
 786
 787    **Admin-only operation** - Requires:
 788    - AIRBYTE_INTERNAL_ADMIN_FLAG=airbyte.io environment variable
 789    - `approval_comment_url` (Slack approval record URL from `escalate_to_human`),
 790      OR `admin_user_email_override` when running inside the Ops Webapp.
 791    """
 792    # Validate admin access (check env var flag)
 793    try:
 794        require_internal_admin_flag_only()
 795    except CloudAuthError as e:
 796        return ConnectorRolloutStartResult(
 797            success=False,
 798            message=f"Admin authentication failed: {e}",
 799            docker_repository=docker_repository,
 800            docker_image_tag=docker_image_tag,
 801            actor_definition_id=actor_definition_id,
 802        )
 803
 804    # Resolve admin email: webapp bypass or external approval URL
 805    approval = check_approval_status(
 806        approval_comment_url=approval_comment_url,
 807        user_email=admin_user_email_override,
 808    )
 809    if approval.status != ApprovalStatus.APPROVED:
 810        return ConnectorRolloutStartResult(
 811            success=False,
 812            message=approval.reason or "Approval check failed",
 813            docker_repository=docker_repository,
 814            docker_image_tag=docker_image_tag,
 815            actor_definition_id=actor_definition_id,
 816        )
 817    admin_user_email = approval.admin_email
 818
 819    # Resolve auth credentials
 820    try:
 821        auth = _resolve_cloud_auth(ctx)
 822    except CloudAuthError as e:
 823        return ConnectorRolloutStartResult(
 824            success=False,
 825            message=f"Failed to resolve credentials: {e}",
 826            docker_repository=docker_repository,
 827            docker_image_tag=docker_image_tag,
 828            actor_definition_id=actor_definition_id,
 829        )
 830
 831    # Get user ID from admin email
 832    try:
 833        user_id = api_client.get_user_id_by_email(
 834            email=admin_user_email,
 835            config_api_root=constants.CLOUD_CONFIG_API_ROOT,
 836            client_id=auth.client_id,
 837            client_secret=auth.client_secret,
 838            bearer_token=auth.bearer_token,
 839        )
 840    except PyAirbyteInputError as e:
 841        return ConnectorRolloutStartResult(
 842            success=False,
 843            message=f"Failed to get user ID for admin email '{admin_user_email}': {e}",
 844            docker_repository=docker_repository,
 845            docker_image_tag=docker_image_tag,
 846            actor_definition_id=actor_definition_id,
 847        )
 848
 849    # Call the API to start the rollout
 850    try:
 851        api_client.start_connector_rollout(
 852            docker_repository=docker_repository,
 853            docker_image_tag=docker_image_tag,
 854            actor_definition_id=actor_definition_id,
 855            updated_by=user_id,
 856            rollout_strategy=rollout_strategy,
 857            config_api_root=constants.CLOUD_CONFIG_API_ROOT,
 858            initial_rollout_pct=initial_rollout_pct,
 859            final_target_rollout_pct=final_target_rollout_pct,
 860            customer_tier=customer_tier,
 861            client_id=auth.client_id,
 862            client_secret=auth.client_secret,
 863            bearer_token=auth.bearer_token,
 864        )
 865
 866        # Build message with configuration details
 867        config_details = []
 868        if initial_rollout_pct is not None:
 869            config_details.append(f"initial_rollout_pct={initial_rollout_pct}%")
 870        if final_target_rollout_pct is not None:
 871            config_details.append(
 872                f"final_target_rollout_pct={final_target_rollout_pct}%"
 873            )
 874        if customer_tier is not None:
 875            config_details.append(f"customer_tier={customer_tier}")
 876        config_str = (
 877            f" Configuration: {', '.join(config_details)}." if config_details else ""
 878        )
 879
 880        return ConnectorRolloutStartResult(
 881            success=True,
 882            message=f"Successfully started rollout workflow for "
 883            f"{docker_repository}:{docker_image_tag}. "
 884            f"The rollout state has transitioned from INITIALIZED to WORKFLOW_STARTED."
 885            f"{config_str}",
 886            docker_repository=docker_repository,
 887            docker_image_tag=docker_image_tag,
 888            actor_definition_id=actor_definition_id,
 889            rollout_strategy=rollout_strategy,
 890        )
 891
 892    except PyAirbyteInputError as e:
 893        return ConnectorRolloutStartResult(
 894            success=False,
 895            message=str(e),
 896            docker_repository=docker_repository,
 897            docker_image_tag=docker_image_tag,
 898            actor_definition_id=actor_definition_id,
 899        )
 900
 901
 902@mcp_tool(
 903    destructive=True,
 904    idempotent=False,
 905    open_world=True,
 906)
 907def progress_connector_rollout(
 908    docker_repository: Annotated[
 909        str,
 910        Field(description="The docker repository (e.g., 'airbyte/source-pokeapi')"),
 911    ],
 912    docker_image_tag: Annotated[
 913        str,
 914        Field(description="The docker image tag (e.g., '0.3.48-rc.1')"),
 915    ],
 916    actor_definition_id: Annotated[
 917        str,
 918        Field(description="The actor definition ID (UUID)"),
 919    ],
 920    rollout_id: Annotated[
 921        str,
 922        Field(
 923            description="The rollout ID (UUID). Can be found from query_prod_connector_rollouts."
 924        ),
 925    ],
 926    approval_comment_url: Annotated[
 927        str | None,
 928        Field(
 929            description="URL to the Slack approval record. Obtain this by calling the "
 930            "`escalate_to_human` tool with `approval_requested=True`; the backend delivers "
 931            "the approval record URL when a human clicks Approve. "
 932            "Format: https://<workspace>.slack.com/archives/... "
 933            "The admin email is automatically resolved from the approver's identity "
 934            "via the team roster.",
 935            default=None,
 936        ),
 937    ],
 938    admin_user_email_override: Annotated[
 939        str | None,
 940        Field(
 941            description="Direct admin email override for webapp-initiated actions. "
 942            "When the Ops Webapp env var is set, this bypasses the approval URL "
 943            "requirement. Ignored in agent/cron environments.",
 944            default=None,
 945        ),
 946    ],
 947    target_percentage: Annotated[
 948        int | None,
 949        Field(
 950            description="Target percentage of actors to pin to the RC (1-100). "
 951            "Either target_percentage or actor_ids must be provided.",
 952            default=None,
 953        ),
 954    ] = None,
 955    actor_ids: Annotated[
 956        list[str] | None,
 957        Field(
 958            description="Specific actor IDs to pin to the RC. "
 959            "Either target_percentage or actor_ids must be provided.",
 960            default=None,
 961        ),
 962    ] = None,
 963    *,
 964    ctx: Context,
 965) -> ConnectorRolloutProgressResult:
 966    """Progress a connector rollout by pinning actors to the RC version.
 967
 968    This tool progresses a connector rollout by either:
 969    - Setting a target percentage of actors to pin to the RC version
 970    - Specifying specific actor IDs to pin
 971
 972    **Admin-only operation** - Requires:
 973    - AIRBYTE_INTERNAL_ADMIN_FLAG=airbyte.io environment variable
 974    - `approval_comment_url` (Slack approval record URL from `escalate_to_human`),
 975      OR `admin_user_email_override` when running inside the Ops Webapp.
 976    """
 977    # Validate admin access (check env var flag)
 978    try:
 979        require_internal_admin_flag_only()
 980    except CloudAuthError as e:
 981        return ConnectorRolloutProgressResult(
 982            success=False,
 983            message=f"Admin authentication failed: {e}",
 984            rollout_id=rollout_id,
 985            docker_repository=docker_repository,
 986            docker_image_tag=docker_image_tag,
 987        )
 988
 989    # Validate that at least one of target_percentage or actor_ids is provided
 990    if target_percentage is None and actor_ids is None:
 991        return ConnectorRolloutProgressResult(
 992            success=False,
 993            message="Either target_percentage or actor_ids must be provided",
 994            rollout_id=rollout_id,
 995            docker_repository=docker_repository,
 996            docker_image_tag=docker_image_tag,
 997        )
 998
 999    # Resolve admin email: webapp bypass or external approval URL
1000    approval = check_approval_status(
1001        approval_comment_url=approval_comment_url,
1002        user_email=admin_user_email_override,
1003    )
1004    if approval.status != ApprovalStatus.APPROVED:
1005        return ConnectorRolloutProgressResult(
1006            success=False,
1007            message=approval.reason or "Approval check failed",
1008            rollout_id=rollout_id,
1009            docker_repository=docker_repository,
1010            docker_image_tag=docker_image_tag,
1011        )
1012    admin_user_email = approval.admin_email
1013
1014    # Resolve auth credentials
1015    try:
1016        auth = _resolve_cloud_auth(ctx)
1017    except CloudAuthError as e:
1018        return ConnectorRolloutProgressResult(
1019            success=False,
1020            message=f"Failed to resolve credentials: {e}",
1021            rollout_id=rollout_id,
1022            docker_repository=docker_repository,
1023            docker_image_tag=docker_image_tag,
1024        )
1025
1026    # Get user ID from admin email
1027    try:
1028        user_id = api_client.get_user_id_by_email(
1029            email=admin_user_email,
1030            config_api_root=constants.CLOUD_CONFIG_API_ROOT,
1031            client_id=auth.client_id,
1032            client_secret=auth.client_secret,
1033            bearer_token=auth.bearer_token,
1034        )
1035    except PyAirbyteInputError as e:
1036        return ConnectorRolloutProgressResult(
1037            success=False,
1038            message=f"Failed to get user ID for admin email '{admin_user_email}': {e}",
1039            rollout_id=rollout_id,
1040            docker_repository=docker_repository,
1041            docker_image_tag=docker_image_tag,
1042        )
1043
1044    # Guard: a percentage-based progression toward a tier with zero eligible
1045    # actors will throw `ConnectorRolloutNotEnoughActorsProblem` server-side
1046    # and silently wedge the rollout at `workflow_started` (the throw happens
1047    # before the `IN_PROGRESS` write and the `@Transactional` rolls back).
1048    # Detect this up front and return an actionable error instead. Skipped when
1049    # pinning specific `actor_ids` (the caller chose the actors explicitly).
1050    if target_percentage is not None and target_percentage > 0 and not actor_ids:
1051        try:
1052            sync_info = api_client.get_actor_sync_info(
1053                rollout_id=rollout_id,
1054                config_api_root=constants.CLOUD_CONFIG_API_ROOT,
1055                client_id=auth.client_id,
1056                client_secret=auth.client_secret,
1057                bearer_token=auth.bearer_token,
1058            )
1059        except (PyAirbyteInputError, requests.exceptions.RequestException):
1060            # If eligibility can't be fetched (bad input or a network-level
1061            # failure), fall through and let the progress call surface the
1062            # underlying error rather than crashing the pre-check.
1063            sync_info = None
1064        if sync_info is not None and count_eligible_or_pinned_actors(sync_info) == 0:
1065            return ConnectorRolloutProgressResult(
1066                success=False,
1067                message=(
1068                    "Zero eligible actors: this rollout's tier has no actors to "
1069                    "pin, so progressing to "
1070                    f"target_percentage={target_percentage}% would throw "
1071                    "ConnectorRolloutNotEnoughActorsProblem server-side and "
1072                    "silently wedge the rollout at 'workflow_started'. "
1073                    "TIER_1/TIER_0 are named strategic accounts; a connector "
1074                    "with no customers in this tier can never pin anyone here. "
1075                    "Do not progress this tier. If the last non-empty tier is "
1076                    "healthy at 100%, cancel this empty rollout "
1077                    "(retain_pins_on_cancellation=true) and finalize the healthy "
1078                    "rollout as 'succeeded' to promote to GA. See the "
1079                    "'Rollout stuck at workflow_started with zero eligible "
1080                    "actors' troubleshooting guide in docs/progressive-rollouts.md."
1081                ),
1082                rollout_id=rollout_id,
1083                docker_repository=docker_repository,
1084                docker_image_tag=docker_image_tag,
1085            )
1086
1087    # Call the API to progress the rollout
1088    try:
1089        api_client.progress_connector_rollout(
1090            docker_repository=docker_repository,
1091            docker_image_tag=docker_image_tag,
1092            actor_definition_id=actor_definition_id,
1093            rollout_id=rollout_id,
1094            updated_by=user_id,
1095            config_api_root=constants.CLOUD_CONFIG_API_ROOT,
1096            target_percentage=target_percentage,
1097            actor_ids=actor_ids,
1098            client_id=auth.client_id,
1099            client_secret=auth.client_secret,
1100            bearer_token=auth.bearer_token,
1101        )
1102
1103        progress_msg = (
1104            f"target_percentage={target_percentage}%"
1105            if target_percentage
1106            else f"{len(actor_ids) if actor_ids else 0} specific actors"
1107        )
1108        return ConnectorRolloutProgressResult(
1109            success=True,
1110            message=f"Successfully progressed rollout for "
1111            f"{docker_repository}:{docker_image_tag} to {progress_msg}.",
1112            rollout_id=rollout_id,
1113            docker_repository=docker_repository,
1114            docker_image_tag=docker_image_tag,
1115            target_percentage=target_percentage,
1116        )
1117
1118    except PyAirbyteInputError as e:
1119        return ConnectorRolloutProgressResult(
1120            success=False,
1121            message=str(e),
1122            rollout_id=rollout_id,
1123            docker_repository=docker_repository,
1124            docker_image_tag=docker_image_tag,
1125        )
1126
1127
1128@mcp_tool(
1129    destructive=True,
1130    idempotent=False,
1131    open_world=True,
1132)
1133def finalize_connector_rollout(
1134    docker_repository: Annotated[
1135        str,
1136        Field(
1137            description="The docker repository (e.g., 'airbyte/source-youtube-analytics')"
1138        ),
1139    ],
1140    docker_image_tag: Annotated[
1141        str,
1142        Field(description="The docker image tag (e.g., '1.2.0-rc.2')"),
1143    ],
1144    actor_definition_id: Annotated[
1145        str,
1146        Field(description="The actor definition ID (UUID)"),
1147    ],
1148    rollout_id: Annotated[
1149        str,
1150        Field(
1151            description="The rollout ID (UUID). Can be found in the 'pin_origin' field "
1152            "of rollout data from query_prod_actors_by_pinned_connector_version."
1153        ),
1154    ],
1155    state: Annotated[
1156        Literal["succeeded", "failed_rolled_back", "canceled"],
1157        Field(
1158            description="The final state for the rollout: "
1159            "'succeeded' promotes the RC to GA (default version for all users), "
1160            "'failed_rolled_back' rolls back the RC, "
1161            "'canceled' cancels the rollout without promotion or rollback."
1162        ),
1163    ],
1164    approval_comment_url: Annotated[
1165        str | None,
1166        Field(
1167            description="URL to the Slack approval record. Obtain this by calling the "
1168            "`escalate_to_human` tool with `approval_requested=True`; the backend delivers "
1169            "the approval record URL when a human clicks Approve. "
1170            "Format: https://<workspace>.slack.com/archives/... "
1171            "The admin email is automatically resolved from the approver's identity "
1172            "via the team roster.",
1173            default=None,
1174        ),
1175    ],
1176    admin_user_email_override: Annotated[
1177        str | None,
1178        Field(
1179            description="Direct admin email override for webapp-initiated actions. "
1180            "When the Ops Webapp env var is set, this bypasses the approval URL "
1181            "requirement. Ignored in agent/cron environments.",
1182            default=None,
1183        ),
1184    ],
1185    error_msg: Annotated[
1186        str | None,
1187        Field(
1188            description="Optional error message for failed/canceled states.",
1189            default=None,
1190        ),
1191    ] = None,
1192    failed_reason: Annotated[
1193        str | None,
1194        Field(
1195            description="Optional failure reason for failed/canceled states.",
1196            default=None,
1197        ),
1198    ] = None,
1199    retain_pins_on_cancellation: Annotated[
1200        bool | None,
1201        Field(
1202            description="If True, retain version pins when canceling. "
1203            "Only applicable when state is 'canceled'.",
1204            default=None,
1205        ),
1206    ] = None,
1207    *,
1208    ctx: Context,
1209) -> ConnectorRolloutFinalizeResult:
1210    """Finalize a connector rollout by promoting, rolling back, or canceling.
1211
1212    This tool allows admins to finalize connector rollouts that are in progress.
1213    Use this after monitoring a rollout and determining it is ready for finalization.
1214
1215    **IMPORTANT: Finalization is asynchronous.** This tool sends a finalization
1216    request to the platform API, which transitions the rollout to `finalizing`
1217    state and triggers a Temporal workflow. The actual promotion (PR creation,
1218    connector publish, registry update) or rollback (GCS cleanup, registry
1219    recompile) happens asynchronously via the `finalize_rollout.yml` GitHub
1220    Actions workflow. A successful response from this tool means the request
1221    was accepted — NOT that the promotion/rollback is complete.
1222
1223    After calling this tool, you MUST verify:
1224    1. The `finalize_rollout.yml` workflow ran successfully in GitHub Actions
1225    2. For promotions: a merged PR exists (e.g., `chore: finalize promote for <connector>`)
1226    3. The rollout state transitioned to its terminal state (`succeeded`,
1227       `failed_rolled_back`, or `canceled`) via `query_prod_connector_rollouts`
1228
1229    **Admin-only operation** - Requires:
1230    - AIRBYTE_INTERNAL_ADMIN_FLAG=airbyte.io environment variable
1231    - `approval_comment_url` (Slack approval record URL from `escalate_to_human`),
1232      OR `admin_user_email_override` when running inside the Ops Webapp.
1233    """
1234    # Validate admin access (check env var flag)
1235    try:
1236        require_internal_admin_flag_only()
1237    except CloudAuthError as e:
1238        return ConnectorRolloutFinalizeResult(
1239            success=False,
1240            message=f"Admin authentication failed: {e}",
1241            rollout_id=rollout_id,
1242            docker_repository=docker_repository,
1243            docker_image_tag=docker_image_tag,
1244        )
1245
1246    # Resolve admin email: webapp bypass or external approval URL
1247    approval = check_approval_status(
1248        approval_comment_url=approval_comment_url,
1249        user_email=admin_user_email_override,
1250    )
1251    if approval.status != ApprovalStatus.APPROVED:
1252        return ConnectorRolloutFinalizeResult(
1253            success=False,
1254            message=approval.reason or "Approval check failed",
1255            rollout_id=rollout_id,
1256            docker_repository=docker_repository,
1257            docker_image_tag=docker_image_tag,
1258        )
1259    admin_user_email = approval.admin_email
1260
1261    # Resolve auth credentials
1262    try:
1263        auth = _resolve_cloud_auth(ctx)
1264    except CloudAuthError as e:
1265        return ConnectorRolloutFinalizeResult(
1266            success=False,
1267            message=f"Failed to resolve credentials: {e}",
1268            rollout_id=rollout_id,
1269            docker_repository=docker_repository,
1270            docker_image_tag=docker_image_tag,
1271        )
1272
1273    # Get user ID from admin email
1274    try:
1275        user_id = api_client.get_user_id_by_email(
1276            email=admin_user_email,
1277            config_api_root=constants.CLOUD_CONFIG_API_ROOT,
1278            client_id=auth.client_id,
1279            client_secret=auth.client_secret,
1280            bearer_token=auth.bearer_token,
1281        )
1282    except PyAirbyteInputError as e:
1283        return ConnectorRolloutFinalizeResult(
1284            success=False,
1285            message=f"Failed to get user ID for admin email '{admin_user_email}': {e}",
1286            rollout_id=rollout_id,
1287            docker_repository=docker_repository,
1288            docker_image_tag=docker_image_tag,
1289        )
1290
1291    # Call the API to finalize the rollout
1292    try:
1293        api_client.finalize_connector_rollout(
1294            docker_repository=docker_repository,
1295            docker_image_tag=docker_image_tag,
1296            actor_definition_id=actor_definition_id,
1297            rollout_id=rollout_id,
1298            updated_by=user_id,
1299            state=state,
1300            config_api_root=constants.CLOUD_CONFIG_API_ROOT,
1301            client_id=auth.client_id,
1302            client_secret=auth.client_secret,
1303            bearer_token=auth.bearer_token,
1304            error_msg=error_msg,
1305            failed_reason=failed_reason,
1306            retain_pins_on_cancellation=retain_pins_on_cancellation,
1307        )
1308
1309        state_descriptions = {
1310            "succeeded": (
1311                "GA promotion has been initiated (state: finalizing). "
1312                "The actual promotion (PR creation, publish, registry update) "
1313                "happens asynchronously via the finalize_rollout.yml GitHub Actions workflow. "
1314                "You MUST verify the workflow completes successfully and the rollout "
1315                "transitions to 'succeeded' state before reporting completion. "
1316                "Check: (1) GitHub Actions for a 'Finalize Progressive Rollout' workflow run, "
1317                "(2) a merged promotion PR, and "
1318                "(3) query_prod_connector_rollouts to confirm state is 'succeeded'."
1319            ),
1320            "failed_rolled_back": (
1321                "rollback has been initiated (state: finalizing). "
1322                "The rollback happens asynchronously via the finalize_rollout.yml workflow. "
1323                "Verify the workflow completes and the rollout transitions to "
1324                "'failed_rolled_back' state."
1325            ),
1326            "canceled": "canceled",
1327        }
1328        state_desc = state_descriptions.get(state, state)
1329
1330        return ConnectorRolloutFinalizeResult(
1331            success=True,
1332            message=f"Finalization request accepted for {docker_repository}:{docker_image_tag}: "
1333            f"{state_desc}",
1334            rollout_id=rollout_id,
1335            docker_repository=docker_repository,
1336            docker_image_tag=docker_image_tag,
1337            state=state,
1338        )
1339
1340    except PyAirbyteInputError as e:
1341        return ConnectorRolloutFinalizeResult(
1342            success=False,
1343            message=str(e),
1344            rollout_id=rollout_id,
1345            docker_repository=docker_repository,
1346            docker_image_tag=docker_image_tag,
1347        )
1348
1349
1350class RolloutActorSelectionInfo(BaseModel):
1351    """Actor selection info for a connector rollout."""
1352
1353    num_actors: int = Field(description="Total actors using this connector")
1354    num_pinned_to_connector_rollout: int = Field(
1355        description="Actors specifically pinned to this rollout"
1356    )
1357    num_actors_eligible_or_already_pinned: int = Field(
1358        description="Actors eligible for pinning or already pinned"
1359    )
1360
1361
1362class RolloutActorSyncStats(BaseModel):
1363    """Per-actor sync stats for a rollout (only syncs using the RC version)."""
1364
1365    actor_id: str = Field(description="Actor UUID")
1366    num_connections: int = Field(description="Number of connections using this actor")
1367    num_succeeded: int = Field(
1368        description="Number of successful syncs using the RC version"
1369    )
1370    num_failed: int = Field(description="Number of failed syncs using the RC version")
1371
1372
1373class RolloutMonitoringResult(BaseModel):
1374    """Complete monitoring result for a rollout from the platform API.
1375
1376    This uses the platform API's /get_actor_sync_info endpoint which filters
1377    sync stats to only include syncs that actually used the RC version
1378    associated with the rollout.
1379    """
1380
1381    rollout_id: str = Field(description="Rollout UUID")
1382    actor_selection_info: RolloutActorSelectionInfo = Field(
1383        description="Actor selection info for the rollout"
1384    )
1385    actor_sync_stats: list[RolloutActorSyncStats] = Field(
1386        description="Per-actor sync stats for actors pinned to the rollout"
1387    )
1388
1389
1390@mcp_tool(
1391    read_only=True,
1392    idempotent=True,
1393)
1394def query_prod_rollout_monitoring_stats(
1395    rollout_id: Annotated[
1396        str,
1397        Field(description="Rollout UUID to get monitoring stats for"),
1398    ],
1399    *,
1400    ctx: Context,
1401) -> RolloutMonitoringResult:
1402    """Get monitoring stats for a connector rollout.
1403
1404    Returns actor selection info and per-actor sync stats for actors
1405    participating in the rollout. This uses the platform API's
1406    /get_actor_sync_info endpoint which filters sync stats to only include
1407    syncs that actually used the RC version associated with the rollout.
1408
1409    This is more accurate than SQL-based approaches which count all syncs
1410    regardless of which connector version was used.
1411    """
1412    auth = _resolve_cloud_auth(ctx)
1413
1414    response = api_client.get_actor_sync_info(
1415        rollout_id=rollout_id,
1416        config_api_root=constants.CLOUD_CONFIG_API_ROOT,
1417        client_id=auth.client_id,
1418        client_secret=auth.client_secret,
1419        bearer_token=auth.bearer_token,
1420    )
1421
1422    data = response.get("data", {})
1423    actor_selection_info_data = data.get("actor_selection_info", {})
1424    syncs_data = data.get("syncs", {})
1425
1426    actor_selection_info = RolloutActorSelectionInfo(
1427        num_actors=actor_selection_info_data.get("num_actors", 0),
1428        num_pinned_to_connector_rollout=actor_selection_info_data.get(
1429            "num_pinned_to_connector_rollout", 0
1430        ),
1431        num_actors_eligible_or_already_pinned=actor_selection_info_data.get(
1432            "num_actors_eligible_or_already_pinned", 0
1433        ),
1434    )
1435
1436    actor_sync_stats = [
1437        RolloutActorSyncStats(
1438            actor_id=actor_id,
1439            num_connections=sync_info.get("num_connections", 0),
1440            num_succeeded=sync_info.get("num_succeeded", 0),
1441            num_failed=sync_info.get("num_failed", 0),
1442        )
1443        for actor_id, sync_info in syncs_data.items()
1444    ]
1445
1446    return RolloutMonitoringResult(
1447        rollout_id=rollout_id,
1448        actor_selection_info=actor_selection_info,
1449        actor_sync_stats=actor_sync_stats,
1450    )
1451
1452
1453class ConnectorRepo(StrEnum):
1454    """Repository where connector code is located."""
1455
1456    AIRBYTE = "airbyte"
1457    AIRBYTE_ENTERPRISE = "airbyte-enterprise"
1458
1459
1460DEFAULT_REPO_OWNER = "airbytehq"
1461
1462DEFAULT_REPO_NAME = ConnectorRepo.AIRBYTE
1463
1464DEFAULT_BRANCH = "master"
1465
1466PRERELEASE_WORKFLOW_FILE = "publish-connectors-prerelease-command.yml"
1467
1468CONNECTOR_PATH_PREFIX = "airbyte-integrations/connectors"
1469
1470ENTERPRISE_REPO_NAME = ConnectorRepo.AIRBYTE_ENTERPRISE
1471
1472ENTERPRISE_DEFAULT_BRANCH = "main"
1473
1474ENTERPRISE_PRERELEASE_WORKFLOW_FILE = "publish_enterprise_connectors.yml"
1475
1476PRERELEASE_TOKEN_ENV_VARS = [
1477    "GITHUB_CONNECTOR_PUBLISHING_PAT",
1478    "GITHUB_CI_WORKFLOW_TRIGGER_PAT",
1479    "GITHUB_TOKEN",
1480]
1481
1482PRERELEASE_TAG_PREFIX = "preview"
1483
1484PRERELEASE_SHA_LENGTH = 7
1485
1486
1487def compute_prerelease_docker_image_tag(base_version: str, sha: str) -> str:
1488    """Compute the pre-release docker image tag.
1489
1490    This is the SINGLE SOURCE OF TRUTH for pre-release version format.
1491    All other code should receive this value as a parameter, not recompute it.
1492
1493    The format is: {base_version}-preview.{short_sha}
1494
1495    Where:
1496        - base_version: The base version from metadata.yaml (e.g., "1.2.3"),
1497          which may already contain a pre-release suffix (e.g., "2.23.16-rc.1").
1498          Any existing pre-release suffix is stripped before applying the preview tag.
1499        - short_sha: The first 7 characters of the git commit SHA
1500
1501    Examples:
1502        >>> compute_prerelease_docker_image_tag("1.2.3", "abcdef1234567890")
1503        '1.2.3-preview.abcdef1'
1504        >>> compute_prerelease_docker_image_tag("0.1.0", "1234567")
1505        '0.1.0-preview.1234567'
1506        >>> compute_prerelease_docker_image_tag("2.23.16-rc.1", "abcdef1234567890")
1507        '2.23.16-preview.abcdef1'
1508
1509    Args:
1510        base_version: The base version from metadata.yaml (e.g., "1.2.3" or "2.23.16-rc.1")
1511        sha: The full git commit SHA (or at least 7 characters)
1512
1513    Returns:
1514        Pre-release version tag (e.g., "1.2.3-preview.abcde12")
1515    """
1516    short_sha = sha[:PRERELEASE_SHA_LENGTH]
1517    clean_version = strip_prerelease_suffix(base_version)
1518    return f"{clean_version}-{PRERELEASE_TAG_PREFIX}.{short_sha}"
1519
1520
1521class PrereleaseWorkflowResult(BaseModel):
1522    """Response model for publish_connector_to_airbyte_registry MCP tool."""
1523
1524    success: bool
1525    message: str
1526    workflow_url: str | None = None
1527    connector_name: str | None = None
1528    pr_number: int | None = None
1529    docker_image: str | None = None
1530    docker_image_tag: str | None = None
1531
1532
1533def _get_connector_metadata(
1534    owner: str,
1535    repo: str,
1536    connector_name: str,
1537    ref: str,
1538    token: str,
1539) -> dict | None:
1540    """Fetch and parse connector metadata.yaml from the repository.
1541
1542    Args:
1543        owner: Repository owner (e.g., "airbytehq")
1544        repo: Repository name (e.g., "airbyte")
1545        connector_name: Connector name (e.g., "source-github")
1546        ref: Git ref to fetch from (branch name or SHA)
1547        token: GitHub API token
1548
1549    Returns:
1550        Parsed metadata dictionary, or None if not found.
1551    """
1552    metadata_path = f"{CONNECTOR_PATH_PREFIX}/{connector_name}/metadata.yaml"
1553    url = f"{GITHUB_API_BASE}/repos/{owner}/{repo}/contents/{metadata_path}"
1554    headers = {
1555        "Authorization": f"Bearer {token}",
1556        "Accept": "application/vnd.github+json",
1557        "X-GitHub-Api-Version": "2022-11-28",
1558    }
1559    params = {"ref": ref}
1560
1561    response = requests.get(url, headers=headers, params=params, timeout=30)
1562
1563    # Guard: Return None if metadata file not found
1564    if response.status_code == 404:
1565        return None
1566
1567    response.raise_for_status()
1568
1569    content_data = response.json()
1570
1571    # Guard: Return None if content is not base64 encoded
1572    if content_data.get("encoding") != "base64":
1573        return None
1574
1575    content = base64.b64decode(content_data["content"]).decode("utf-8")
1576    return yaml.safe_load(content)
1577
1578
1579@mcp_tool(
1580    read_only=False,
1581    destructive=False,
1582    idempotent=False,
1583    open_world=True,
1584)
1585def publish_connector_to_airbyte_registry(
1586    connector_name: Annotated[
1587        str,
1588        Field(
1589            description="The connector name to publish (e.g., 'source-github', 'destination-postgres')"
1590        ),
1591    ],
1592    pr_number: Annotated[
1593        int,
1594        Field(description="The pull request number containing the connector changes"),
1595    ],
1596    repo: Annotated[
1597        ConnectorRepo,
1598        Field(
1599            default=ConnectorRepo.AIRBYTE,
1600            description="Repository where the connector PR is located. "
1601            "Use 'airbyte' for OSS connectors (default) or 'airbyte-enterprise' for enterprise connectors.",
1602        ),
1603    ],
1604    prerelease: Annotated[
1605        Literal[True],
1606        Field(
1607            default=True,
1608            description="Must be True. Only prerelease publishing is supported at this time.",
1609        ),
1610    ],
1611) -> PrereleaseWorkflowResult:
1612    """Publish a connector to the Airbyte registry.
1613
1614    Currently only supports pre-release publishing. This tool triggers the
1615    publish-connectors-prerelease workflow in the airbytehq/airbyte repository
1616    (for OSS connectors) or the publish_enterprise_connectors workflow in
1617    airbytehq/airbyte-enterprise (for enterprise connectors), which publishes
1618    a pre-release version of the specified connector from the PR branch.
1619
1620    Pre-release versions are tagged with the format: {version}-preview.{7-char-git-sha}
1621    These versions are available for version pinning via the scoped_configuration API.
1622
1623    Requires GITHUB_CONNECTOR_PUBLISHING_PAT or GITHUB_TOKEN environment variable
1624    with 'actions:write' permission.
1625    """
1626    # Guard: Only prerelease publishing is supported
1627    if prerelease is not True:
1628        raise NotImplementedError(
1629            "Non-prerelease publishing is not implemented yet. Set prerelease=True."
1630        )
1631
1632    # Guard: Check for required token
1633    token = resolve_ci_trigger_github_token(PRERELEASE_TOKEN_ENV_VARS)
1634
1635    # Determine repo-specific settings
1636    is_enterprise = repo == ConnectorRepo.AIRBYTE_ENTERPRISE
1637    target_repo_name = ENTERPRISE_REPO_NAME if is_enterprise else DEFAULT_REPO_NAME
1638    target_branch = ENTERPRISE_DEFAULT_BRANCH if is_enterprise else DEFAULT_BRANCH
1639    target_workflow = (
1640        ENTERPRISE_PRERELEASE_WORKFLOW_FILE
1641        if is_enterprise
1642        else PRERELEASE_WORKFLOW_FILE
1643    )
1644
1645    # Get the PR's head SHA for computing the docker image tag
1646    # Note: We no longer pass gitref to the workflow - it derives the ref from PR number
1647    head_info = get_pr_head_ref(DEFAULT_REPO_OWNER, target_repo_name, pr_number, token)
1648
1649    # Prepare workflow inputs
1650    workflow_inputs = {
1651        "repo": f"{DEFAULT_REPO_OWNER}/{target_repo_name}",
1652        "pr": str(pr_number),
1653        "connector": connector_name,
1654    }
1655
1656    # Trigger the workflow on the configured default branch.
1657    # The workflow will checkout the PR branch via inputs.gitref
1658    dispatch_result = trigger_workflow_dispatch(
1659        owner=DEFAULT_REPO_OWNER,
1660        repo=target_repo_name,
1661        workflow_file=target_workflow,
1662        ref=target_branch,
1663        inputs=workflow_inputs,
1664        token=token,
1665        find_run=True,
1666    )
1667    # Use the specific run URL if found, otherwise fall back to the workflow URL
1668    workflow_url = dispatch_result.run_url or dispatch_result.workflow_url
1669
1670    # Try to compute docker_image and docker_image_tag from connector metadata
1671    docker_image: str | None = None
1672    docker_image_tag: str | None = None
1673    metadata = _get_connector_metadata(
1674        DEFAULT_REPO_OWNER,
1675        target_repo_name,
1676        connector_name,
1677        head_info.sha,
1678        token,
1679    )
1680    if metadata and "data" in metadata:
1681        data = metadata["data"]
1682        docker_image = data.get("dockerRepository")
1683        base_version = data.get("dockerImageTag")
1684        if base_version:
1685            docker_image_tag = compute_prerelease_docker_image_tag(
1686                base_version, head_info.sha
1687            )
1688
1689    repo_info = f" from {repo}" if is_enterprise else ""
1690    return PrereleaseWorkflowResult(
1691        success=True,
1692        message=f"Successfully triggered pre-release workflow for {connector_name}{repo_info} from PR #{pr_number}",
1693        workflow_url=workflow_url,
1694        connector_name=connector_name,
1695        pr_number=pr_number,
1696        docker_image=docker_image,
1697        docker_image_tag=docker_image_tag,
1698    )
1699
1700
1701def register_connector_version_tools(app: FastMCP) -> None:
1702    """Register connector_versions tools with the FastMCP app."""
1703    register_mcp_tools(app, mcp_module=__name__)