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