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