airbyte_ops_mcp.mcp.cloud_connector_versions

  1# Copyright (c) 2025 Airbyte, Inc., all rights reserved.
  2"""MCP tools for cloud connector version management.
  3
  4This module exposes the connector version-override operations as MCP tools.
  5The tools are thin presentation-layer wrappers: they resolve auth from the
  6incoming `fastmcp.Context` and forward to the core
  7`airbyte_ops_mcp.cloud_admin.version_overrides` module which both the MCP
  8and CLI surfaces share.
  9
 10## MCP reference
 11
 12.. include:: ../../../docs/mcp-generated/cloud_connector_versions.md
 13    :start-line: 2
 14"""
 15
 16# NOTE: We intentionally do NOT use `from __future__ import annotations` here.
 17# FastMCP has issues resolving forward references when PEP 563 deferred annotations
 18# are used. See: https://github.com/jlowin/fastmcp/issues/905
 19# Python 3.12+ supports modern type hint syntax natively, so this is not needed.
 20
 21__all__: list[str] = []
 22
 23import logging
 24from typing import Annotated, Literal
 25
 26from fastmcp import Context, FastMCP
 27from fastmcp_extensions import get_mcp_config, mcp_tool, register_mcp_tools
 28from pydantic import Field
 29
 30from airbyte_ops_mcp.cloud_admin.auth import CloudAuthError
 31from airbyte_ops_mcp.cloud_admin.models import (
 32    ConnectorVersionInfo,
 33    OrganizationVersionOverrideResult,
 34    VersionOverrideOperationResult,
 35    WorkspaceVersionOverrideResult,
 36)
 37from airbyte_ops_mcp.cloud_admin.version_overrides import (
 38    ResolvedCloudAuth,
 39    get_connector_version_info,
 40    set_actor_version_override,
 41    set_organization_version_override,
 42    set_workspace_version_override,
 43)
 44from airbyte_ops_mcp.constants import ServerConfigKey, WorkspaceAliasEnum
 45from airbyte_ops_mcp.tier_cache import TierFilter
 46
 47logger = logging.getLogger(__name__)
 48
 49
 50def _resolve_cloud_auth(ctx: Context) -> ResolvedCloudAuth:
 51    """Resolve authentication credentials for Airbyte Cloud API from `ctx`.
 52
 53    Credentials are resolved in priority order:
 54
 55    1. Bearer token (`Authorization` header or `AIRBYTE_CLOUD_BEARER_TOKEN` env var)
 56    2. Client credentials (`X-Airbyte-Cloud-Client-Id`/`Secret` headers or env vars)
 57
 58    Raises `CloudAuthError` if credentials cannot be resolved.
 59    """
 60    bearer_token = get_mcp_config(ctx, ServerConfigKey.BEARER_TOKEN)
 61    if bearer_token:
 62        return ResolvedCloudAuth(bearer_token=bearer_token)
 63
 64    try:
 65        client_id = get_mcp_config(ctx, ServerConfigKey.CLIENT_ID)
 66        client_secret = get_mcp_config(ctx, ServerConfigKey.CLIENT_SECRET)
 67        return ResolvedCloudAuth(
 68            client_id=client_id,
 69            client_secret=client_secret,
 70        )
 71    except ValueError as e:
 72        raise CloudAuthError(
 73            f"Failed to resolve credentials. Ensure credentials are provided "
 74            f"via Authorization header (Bearer token), "
 75            f"HTTP headers (X-Airbyte-Cloud-Client-Id, X-Airbyte-Cloud-Client-Secret), "
 76            f"or environment variables. Error: {e}"
 77        ) from e
 78
 79
 80@mcp_tool(
 81    read_only=True,
 82    idempotent=True,
 83    open_world=True,
 84)
 85def get_cloud_connector_version(
 86    workspace_id: Annotated[
 87        str | WorkspaceAliasEnum,
 88        Field(
 89            description="The Airbyte Cloud workspace ID (UUID) or alias. "
 90            "Accepts '@devin-ai-sandbox' as an alias for the Devin AI sandbox workspace."
 91        ),
 92    ],
 93    actor_id: Annotated[
 94        str, "The ID of the deployed connector (source or destination)"
 95    ],
 96    actor_type: Annotated[
 97        Literal["source", "destination"],
 98        "The type of connector (source or destination)",
 99    ],
100    config_api_root: Annotated[
101        str | None,
102        Field(
103            description="Optional API root URL override for the Config API. "
104            "Defaults to Airbyte Cloud (https://cloud.airbyte.com/api/v1). "
105            "Use this to target local or self-hosted deployments.",
106            default=None,
107        ),
108    ] = None,
109    *,
110    ctx: Context,
111) -> ConnectorVersionInfo:
112    """Get the current version information for a deployed connector.
113
114    Returns version details including the current version string and whether
115    an override (pin) is applied.
116
117    Authentication credentials are resolved in priority order:
118
119    1. Bearer token (Authorization header or AIRBYTE_CLOUD_BEARER_TOKEN env var)
120    2. HTTP headers: X-Airbyte-Cloud-Client-Id, X-Airbyte-Cloud-Client-Secret
121    3. Environment variables: AIRBYTE_CLOUD_CLIENT_ID, AIRBYTE_CLOUD_CLIENT_SECRET
122    """
123    resolved_workspace_id = WorkspaceAliasEnum.resolve(workspace_id)
124    assert resolved_workspace_id is not None  # workspace_id is required
125
126    return get_connector_version_info(
127        auth=_resolve_cloud_auth(ctx),
128        workspace_id=resolved_workspace_id,
129        actor_id=actor_id,
130        actor_type=actor_type,
131        config_api_root=config_api_root,
132    )
133
134
135@mcp_tool(
136    destructive=True,
137    idempotent=False,
138    open_world=True,
139)
140def set_cloud_connector_version_override(
141    workspace_id: Annotated[
142        str | WorkspaceAliasEnum,
143        Field(
144            description="The Airbyte Cloud workspace ID (UUID) or alias. "
145            "Accepts '@devin-ai-sandbox' as an alias for the Devin AI sandbox workspace."
146        ),
147    ],
148    actor_id: Annotated[
149        str, "The ID of the deployed connector (source or destination)"
150    ],
151    actor_type: Annotated[
152        Literal["source", "destination"],
153        "The type of connector (source or destination)",
154    ],
155    approval_comment_url: Annotated[
156        str | None,
157        Field(
158            description="URL to the Slack approval record. Obtain this by calling the "
159            "`escalate_to_human` tool with `approval_requested=True`; the backend delivers "
160            "the approval record URL when a human clicks Approve. "
161            "Format: https://<workspace>.slack.com/archives/... "
162            "The admin email is automatically resolved from the approver's identity "
163            "via the team roster.",
164            default=None,
165        ),
166    ],
167    version: Annotated[
168        str | None,
169        Field(
170            description="The semver version string to pin to (e.g., '0.1.0'). "
171            "Must be None if unset is True.",
172            default=None,
173        ),
174    ],
175    unset: Annotated[
176        bool,
177        Field(
178            description="If True, removes any existing version override. "
179            "Cannot be True if version is provided.",
180            default=False,
181        ),
182    ],
183    override_reason: Annotated[
184        str | None,
185        Field(
186            description="Required when setting a version. "
187            "Explanation for the override (min 10 characters).",
188            default=None,
189        ),
190    ],
191    override_reason_reference_url: Annotated[
192        str | None,
193        Field(
194            description="Optional URL with more context (e.g., issue link).",
195            default=None,
196        ),
197    ],
198    issue_url: Annotated[
199        str | None,
200        Field(
201            description="URL to the GitHub issue providing context for this operation. "
202            "Must be a valid GitHub URL (https://github.com/...). Required for authorization.",
203            default=None,
204        ),
205    ],
206    ai_agent_session_url: Annotated[
207        str | None,
208        Field(
209            description="URL to the AI agent session driving this operation, if applicable. "
210            "Provides additional auditability for AI-driven operations.",
211            default=None,
212        ),
213    ] = None,
214    force: Annotated[
215        bool,
216        Field(
217            description="If `True`, allow overwriting an existing version pin. "
218            "Existing pins may have been set by rollouts, breaking-change migrations, "
219            "or other operators. Defaults to `False`. NOTE: `force=True` only "
220            "bypasses the existing-pin check — major-version crossings are always "
221            "blocked and cannot be overridden.",
222            default=False,
223        ),
224    ] = False,
225    config_api_root: Annotated[
226        str | None,
227        Field(
228            description="Optional API root URL override for the Config API. "
229            "Defaults to Airbyte Cloud (https://cloud.airbyte.com/api/v1). "
230            "Use this to target local or self-hosted deployments.",
231            default=None,
232        ),
233    ] = None,
234    customer_tier_filter: Annotated[
235        TierFilter,
236        Field(
237            description=(
238                "Required tier filter: 'TIER_0', 'TIER_1', 'TIER_2', or 'ALL'. "
239                "The operation will be rejected if the actual customer tier does not match. "
240                "Use 'ALL' to proceed regardless of tier (a warning is shown for sensitive tiers)."
241            ),
242        ),
243    ] = "TIER_2",
244    *,
245    ctx: Context,
246) -> VersionOverrideOperationResult:
247    """Set or clear a version override for a deployed connector.
248
249    **Admin-only operation** - Requires:
250
251    - AIRBYTE_INTERNAL_ADMIN_FLAG=airbyte.io environment variable
252    - issue_url parameter (GitHub issue URL for context)
253    - approval_comment_url (Slack approval record URL from `escalate_to_human`)
254
255    The admin user email is automatically derived from the Slack approval record,
256    resolving the approver's @airbyte.io email via the team roster.
257
258    You must specify EXACTLY ONE of `version` OR `unset=True`, but not both.
259    When setting a version, `override_reason` is required.
260
261    The `customer_tier_filter` parameter gates the operation: the call fails if
262    the actual tier of the workspace's organization does not match.  Use `ALL`
263    to bypass the check (a warning is still emitted for sensitive tiers).
264
265    Business rules enforced:
266
267    - Dev versions (-dev): Only creator can unpin their own dev version override
268    - Production versions: Require strong justification mentioning customer/support/investigation
269    - Release candidates (-rc): Any admin can pin/unpin RC versions
270
271    Authentication credentials are resolved in priority order:
272
273    1. Bearer token (Authorization header or AIRBYTE_CLOUD_BEARER_TOKEN env var)
274    2. HTTP headers: X-Airbyte-Cloud-Client-Id, X-Airbyte-Cloud-Client-Secret
275    3. Environment variables: AIRBYTE_CLOUD_CLIENT_ID, AIRBYTE_CLOUD_CLIENT_SECRET
276    """
277    resolved_workspace_id = WorkspaceAliasEnum.resolve(workspace_id)
278    assert resolved_workspace_id is not None  # workspace_id is required
279
280    return set_actor_version_override(
281        auth=_resolve_cloud_auth(ctx),
282        workspace_id=resolved_workspace_id,
283        actor_id=actor_id,
284        actor_type=actor_type,
285        approval_comment_url=approval_comment_url,
286        version=version,
287        unset=unset,
288        override_reason=override_reason,
289        override_reason_reference_url=override_reason_reference_url,
290        issue_url=issue_url,
291        ai_agent_session_url=ai_agent_session_url,
292        customer_tier_filter=customer_tier_filter,
293        force=force,
294        config_api_root=config_api_root,
295    )
296
297
298@mcp_tool(
299    destructive=True,
300    idempotent=False,
301    open_world=True,
302)
303def set_workspace_connector_version_override(
304    workspace_id: Annotated[
305        str | WorkspaceAliasEnum,
306        Field(
307            description="The Airbyte Cloud workspace ID (UUID) or alias. "
308            "Accepts '@devin-ai-sandbox' as an alias for the Devin AI sandbox workspace."
309        ),
310    ],
311    connector_name: Annotated[
312        str,
313        Field(
314            description="The connector name (e.g., 'source-github', 'destination-bigquery')."
315        ),
316    ],
317    connector_type: Annotated[
318        Literal["source", "destination"],
319        "The type of connector (source or destination)",
320    ],
321    approval_comment_url: Annotated[
322        str | None,
323        Field(
324            description="URL to the Slack approval record. Obtain this by calling the "
325            "`escalate_to_human` tool with `approval_requested=True`; the backend delivers "
326            "the approval record URL when a human clicks Approve. "
327            "Format: https://<workspace>.slack.com/archives/... "
328            "The admin email is automatically resolved from the approver's identity "
329            "via the team roster.",
330            default=None,
331        ),
332    ],
333    version: Annotated[
334        str | None,
335        Field(
336            description="The semver version string to pin to (e.g., '0.1.0'). "
337            "Must be None if unset is True.",
338            default=None,
339        ),
340    ],
341    unset: Annotated[
342        bool,
343        Field(
344            description="If True, removes any existing version override. "
345            "Cannot be True if version is provided.",
346            default=False,
347        ),
348    ],
349    override_reason: Annotated[
350        str | None,
351        Field(
352            description="Required when setting a version. "
353            "Explanation for the override (min 10 characters).",
354            default=None,
355        ),
356    ],
357    override_reason_reference_url: Annotated[
358        str | None,
359        Field(
360            description="Optional URL with more context (e.g., issue link).",
361            default=None,
362        ),
363    ],
364    issue_url: Annotated[
365        str | None,
366        Field(
367            description="URL to the GitHub issue providing context for this operation. "
368            "Must be a valid GitHub URL (https://github.com/...). Required for authorization.",
369            default=None,
370        ),
371    ],
372    ai_agent_session_url: Annotated[
373        str | None,
374        Field(
375            description="URL to the AI agent session driving this operation, if applicable. "
376            "Provides additional auditability for AI-driven operations.",
377            default=None,
378        ),
379    ] = None,
380    force: Annotated[
381        bool,
382        Field(
383            description="If `True`, allow overwriting an existing version pin. "
384            "Existing pins may have been set by rollouts, breaking-change migrations, "
385            "or other operators. Defaults to `False`. NOTE: `force=True` only "
386            "bypasses the existing-pin check — major-version crossings are always "
387            "blocked and cannot be overridden.",
388            default=False,
389        ),
390    ] = False,
391    config_api_root: Annotated[
392        str | None,
393        Field(
394            description="Optional API root URL override for the Config API. "
395            "Defaults to Airbyte Cloud (https://cloud.airbyte.com/api/v1). "
396            "Use this to target local or self-hosted deployments.",
397            default=None,
398        ),
399    ] = None,
400    customer_tier_filter: Annotated[
401        TierFilter,
402        Field(
403            description=(
404                "Required tier filter: 'TIER_0', 'TIER_1', 'TIER_2', or 'ALL'. "
405                "The operation will be rejected if the actual customer tier does not match. "
406                "Use 'ALL' to proceed regardless of tier (a warning is shown for sensitive tiers)."
407            ),
408        ),
409    ] = "TIER_2",
410    *,
411    ctx: Context,
412) -> WorkspaceVersionOverrideResult:
413    """Set or clear a workspace-level version override for a connector type.
414
415    This pins ALL instances of a connector type within a workspace to a specific version.
416    For example, pinning 'source-github' at workspace level means all GitHub sources
417    in that workspace will use the pinned version.
418
419    **Admin-only operation** - Requires:
420
421    - AIRBYTE_INTERNAL_ADMIN_FLAG=airbyte.io environment variable
422    - issue_url parameter (GitHub issue URL for context)
423    - approval_comment_url (Slack approval record URL from `escalate_to_human`)
424
425    You must specify EXACTLY ONE of `version` OR `unset=True`, but not both.
426    When setting a version, `override_reason` is required.
427
428    The `customer_tier_filter` parameter gates the operation: the call fails if
429    the actual tier of the workspace's organization does not match.  Use `ALL`
430    to bypass the check (a warning is still emitted for sensitive tiers).
431    """
432    resolved_workspace_id = WorkspaceAliasEnum.resolve(workspace_id)
433    assert resolved_workspace_id is not None  # workspace_id is required
434
435    return set_workspace_version_override(
436        auth=_resolve_cloud_auth(ctx),
437        workspace_id=resolved_workspace_id,
438        connector_name=connector_name,
439        connector_type=connector_type,
440        approval_comment_url=approval_comment_url,
441        version=version,
442        unset=unset,
443        override_reason=override_reason,
444        override_reason_reference_url=override_reason_reference_url,
445        issue_url=issue_url,
446        ai_agent_session_url=ai_agent_session_url,
447        customer_tier_filter=customer_tier_filter,
448        force=force,
449        config_api_root=config_api_root,
450    )
451
452
453@mcp_tool(
454    destructive=True,
455    idempotent=False,
456    open_world=True,
457)
458def set_organization_connector_version_override(
459    organization_id: Annotated[
460        str,
461        Field(description="The Airbyte Cloud organization ID."),
462    ],
463    connector_name: Annotated[
464        str,
465        Field(
466            description="The connector name (e.g., 'source-github', 'destination-bigquery')."
467        ),
468    ],
469    connector_type: Annotated[
470        Literal["source", "destination"],
471        "The type of connector (source or destination)",
472    ],
473    approval_comment_url: Annotated[
474        str | None,
475        Field(
476            description="URL to the Slack approval record. Obtain this by calling the "
477            "`escalate_to_human` tool with `approval_requested=True`; the backend delivers "
478            "the approval record URL when a human clicks Approve. "
479            "Format: https://<workspace>.slack.com/archives/... "
480            "The admin email is automatically resolved from the approver's identity "
481            "via the team roster.",
482            default=None,
483        ),
484    ],
485    version: Annotated[
486        str | None,
487        Field(
488            description="The semver version string to pin to (e.g., '0.1.0'). "
489            "Must be None if unset is True.",
490            default=None,
491        ),
492    ],
493    unset: Annotated[
494        bool,
495        Field(
496            description="If True, removes any existing version override. "
497            "Cannot be True if version is provided.",
498            default=False,
499        ),
500    ],
501    override_reason: Annotated[
502        str | None,
503        Field(
504            description="Required when setting a version. "
505            "Explanation for the override (min 10 characters).",
506            default=None,
507        ),
508    ],
509    override_reason_reference_url: Annotated[
510        str | None,
511        Field(
512            description="Optional URL with more context (e.g., issue link).",
513            default=None,
514        ),
515    ],
516    issue_url: Annotated[
517        str | None,
518        Field(
519            description="URL to the GitHub issue providing context for this operation. "
520            "Must be a valid GitHub URL (https://github.com/...). Required for authorization.",
521            default=None,
522        ),
523    ],
524    ai_agent_session_url: Annotated[
525        str | None,
526        Field(
527            description="URL to the AI agent session driving this operation, if applicable. "
528            "Provides additional auditability for AI-driven operations.",
529            default=None,
530        ),
531    ] = None,
532    force: Annotated[
533        bool,
534        Field(
535            description="If `True`, allow overwriting an existing version pin. "
536            "Existing pins may have been set by rollouts, breaking-change migrations, "
537            "or other operators. Defaults to `False`. NOTE: `force=True` only "
538            "bypasses the existing-pin check — major-version crossings are always "
539            "blocked and cannot be overridden.",
540            default=False,
541        ),
542    ] = False,
543    config_api_root: Annotated[
544        str | None,
545        Field(
546            description="Optional API root URL override for the Config API. "
547            "Defaults to Airbyte Cloud (https://cloud.airbyte.com/api/v1). "
548            "Use this to target local or self-hosted deployments.",
549            default=None,
550        ),
551    ] = None,
552    customer_tier_filter: Annotated[
553        TierFilter,
554        Field(
555            description=(
556                "Required tier filter: 'TIER_0', 'TIER_1', 'TIER_2', or 'ALL'. "
557                "The operation will be rejected if the actual customer tier does not match. "
558                "Use 'ALL' to proceed regardless of tier (a warning is shown for sensitive tiers)."
559            ),
560        ),
561    ] = "TIER_2",
562    *,
563    ctx: Context,
564) -> OrganizationVersionOverrideResult:
565    """Set or clear an organization-level version override for a connector type.
566
567    This pins ALL instances of a connector type across an entire organization to a
568    specific version. For example, pinning 'source-github' at organization level means
569    all GitHub sources in all workspaces within that organization will use the pinned version.
570
571    **Admin-only operation** - Requires:
572
573    - AIRBYTE_INTERNAL_ADMIN_FLAG=airbyte.io environment variable
574    - issue_url parameter (GitHub issue URL for context)
575    - approval_comment_url (Slack approval record URL from `escalate_to_human`)
576
577    You must specify EXACTLY ONE of `version` OR `unset=True`, but not both.
578    When setting a version, `override_reason` is required.
579
580    The `customer_tier_filter` parameter gates the operation: the call fails if
581    the actual tier of the organization does not match.  Use `ALL` to bypass
582    the check (a warning is still emitted for sensitive tiers).
583    """
584    return set_organization_version_override(
585        auth=_resolve_cloud_auth(ctx),
586        organization_id=organization_id,
587        connector_name=connector_name,
588        connector_type=connector_type,
589        approval_comment_url=approval_comment_url,
590        version=version,
591        unset=unset,
592        override_reason=override_reason,
593        override_reason_reference_url=override_reason_reference_url,
594        issue_url=issue_url,
595        ai_agent_session_url=ai_agent_session_url,
596        customer_tier_filter=customer_tier_filter,
597        force=force,
598        config_api_root=config_api_root,
599    )
600
601
602def register_cloud_connector_version_tools(app: FastMCP) -> None:
603    """Register cloud connector version management tools with the FastMCP app."""
604    register_mcp_tools(app, mcp_module=__name__)