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