airbyte.mcp.cloud_ops

   1# Copyright (c) 2024 Airbyte, Inc., all rights reserved.
   2"""Airbyte Cloud MCP operations."""
   3
   4from pathlib import Path
   5from typing import Annotated, Any
   6
   7from fastmcp import FastMCP
   8from pydantic import Field
   9
  10from airbyte import cloud, get_destination, get_source
  11from airbyte.cloud.auth import (
  12    resolve_cloud_api_url,
  13    resolve_cloud_client_id,
  14    resolve_cloud_client_secret,
  15    resolve_cloud_workspace_id,
  16)
  17from airbyte.cloud.connections import CloudConnection
  18from airbyte.cloud.connectors import CloudDestination, CloudSource, CustomCloudSourceDefinition
  19from airbyte.cloud.workspaces import CloudWorkspace
  20from airbyte.destinations.util import get_noop_destination
  21from airbyte.mcp._tool_utils import (
  22    check_guid_created_in_session,
  23    mcp_tool,
  24    register_guid_created_in_session,
  25    register_tools,
  26)
  27from airbyte.mcp._util import resolve_config, resolve_list_of_strings
  28
  29
  30def _get_cloud_workspace() -> CloudWorkspace:
  31    """Get an authenticated CloudWorkspace using environment variables."""
  32    return CloudWorkspace(
  33        workspace_id=resolve_cloud_workspace_id(),
  34        client_id=resolve_cloud_client_id(),
  35        client_secret=resolve_cloud_client_secret(),
  36        api_root=resolve_cloud_api_url(),
  37    )
  38
  39
  40@mcp_tool(
  41    domain="cloud",
  42    open_world=True,
  43)
  44def deploy_source_to_cloud(
  45    source_name: Annotated[
  46        str,
  47        Field(description="The name to use when deploying the source."),
  48    ],
  49    source_connector_name: Annotated[
  50        str,
  51        Field(description="The name of the source connector (e.g., 'source-faker')."),
  52    ],
  53    *,
  54    config: Annotated[
  55        dict | str | None,
  56        Field(
  57            description="The configuration for the source connector.",
  58            default=None,
  59        ),
  60    ],
  61    config_secret_name: Annotated[
  62        str | None,
  63        Field(
  64            description="The name of the secret containing the configuration.",
  65            default=None,
  66        ),
  67    ],
  68    unique: Annotated[
  69        bool,
  70        Field(
  71            description="Whether to require a unique name.",
  72            default=True,
  73        ),
  74    ],
  75) -> str:
  76    """Deploy a source connector to Airbyte Cloud.
  77
  78    By default, the `AIRBYTE_CLIENT_ID`, `AIRBYTE_CLIENT_SECRET`, `AIRBYTE_WORKSPACE_ID`,
  79    and `AIRBYTE_API_ROOT` environment variables will be used to authenticate with the
  80    Airbyte Cloud API.
  81    """
  82    try:
  83        source = get_source(
  84            source_connector_name,
  85            install_if_missing=False,
  86        )
  87        config_dict = resolve_config(
  88            config=config,
  89            config_secret_name=config_secret_name,
  90            config_spec_jsonschema=source.config_spec,
  91        )
  92        source.set_config(config_dict)
  93
  94        workspace: CloudWorkspace = _get_cloud_workspace()
  95        deployed_source = workspace.deploy_source(
  96            name=source_name,
  97            source=source,
  98            unique=unique,
  99        )
 100
 101    except Exception as ex:
 102        return f"Failed to deploy source '{source_name}': {ex}"
 103    else:
 104        register_guid_created_in_session(deployed_source.connector_id)
 105        return (
 106            f"Successfully deployed source '{source_name}' with ID '{deployed_source.connector_id}'"
 107            f" and URL: {deployed_source.connector_url}"
 108        )
 109
 110
 111@mcp_tool(
 112    domain="cloud",
 113    open_world=True,
 114)
 115def deploy_destination_to_cloud(
 116    destination_name: Annotated[
 117        str,
 118        Field(description="The name to use when deploying the destination."),
 119    ],
 120    destination_connector_name: Annotated[
 121        str,
 122        Field(description="The name of the destination connector (e.g., 'destination-postgres')."),
 123    ],
 124    *,
 125    config: Annotated[
 126        dict | str | None,
 127        Field(
 128            description="The configuration for the destination connector.",
 129            default=None,
 130        ),
 131    ],
 132    config_secret_name: Annotated[
 133        str | None,
 134        Field(
 135            description="The name of the secret containing the configuration.",
 136            default=None,
 137        ),
 138    ],
 139    unique: Annotated[
 140        bool,
 141        Field(
 142            description="Whether to require a unique name.",
 143            default=True,
 144        ),
 145    ],
 146) -> str:
 147    """Deploy a destination connector to Airbyte Cloud.
 148
 149    By default, the `AIRBYTE_CLIENT_ID`, `AIRBYTE_CLIENT_SECRET`, `AIRBYTE_WORKSPACE_ID`,
 150    and `AIRBYTE_API_ROOT` environment variables will be used to authenticate with the
 151    Airbyte Cloud API.
 152    """
 153    try:
 154        destination = get_destination(
 155            destination_connector_name,
 156            install_if_missing=False,
 157        )
 158        config_dict = resolve_config(
 159            config=config,
 160            config_secret_name=config_secret_name,
 161            config_spec_jsonschema=destination.config_spec,
 162        )
 163        destination.set_config(config_dict)
 164
 165        workspace: CloudWorkspace = _get_cloud_workspace()
 166        deployed_destination = workspace.deploy_destination(
 167            name=destination_name,
 168            destination=destination,
 169            unique=unique,
 170        )
 171
 172    except Exception as ex:
 173        return f"Failed to deploy destination '{destination_name}': {ex}"
 174    else:
 175        register_guid_created_in_session(deployed_destination.connector_id)
 176        return (
 177            f"Successfully deployed destination '{destination_name}' "
 178            f"with ID: {deployed_destination.connector_id}"
 179        )
 180
 181
 182@mcp_tool(
 183    domain="cloud",
 184    open_world=True,
 185)
 186def create_connection_on_cloud(
 187    connection_name: Annotated[
 188        str,
 189        Field(description="The name of the connection."),
 190    ],
 191    source_id: Annotated[
 192        str,
 193        Field(description="The ID of the deployed source."),
 194    ],
 195    destination_id: Annotated[
 196        str,
 197        Field(description="The ID of the deployed destination."),
 198    ],
 199    selected_streams: Annotated[
 200        str | list[str],
 201        Field(
 202            description=(
 203                "The selected stream names to sync within the connection. "
 204                "Must be an explicit stream name or list of streams. "
 205                "Cannot be empty or '*'."
 206            )
 207        ),
 208    ],
 209    table_prefix: Annotated[
 210        str | None,
 211        Field(
 212            description="Optional table prefix to use when syncing to the destination.",
 213            default=None,
 214        ),
 215    ],
 216) -> str:
 217    """Create a connection between a deployed source and destination on Airbyte Cloud.
 218
 219    By default, the `AIRBYTE_CLIENT_ID`, `AIRBYTE_CLIENT_SECRET`, `AIRBYTE_WORKSPACE_ID`,
 220    and `AIRBYTE_API_ROOT` environment variables will be used to authenticate with the
 221    Airbyte Cloud API.
 222    """
 223    resolved_streams_list: list[str] = resolve_list_of_strings(selected_streams)
 224    try:
 225        workspace: CloudWorkspace = _get_cloud_workspace()
 226        deployed_connection = workspace.deploy_connection(
 227            connection_name=connection_name,
 228            source=source_id,
 229            destination=destination_id,
 230            selected_streams=resolved_streams_list,
 231            table_prefix=table_prefix,
 232        )
 233
 234    except Exception as ex:
 235        return f"Failed to create connection '{connection_name}': {ex}"
 236    else:
 237        register_guid_created_in_session(deployed_connection.connection_id)
 238        return (
 239            f"Successfully created connection '{connection_name}' "
 240            f"with ID '{deployed_connection.connection_id}' and "
 241            f"URL: {deployed_connection.connection_url}"
 242        )
 243
 244
 245@mcp_tool(
 246    domain="cloud",
 247    open_world=True,
 248)
 249def run_cloud_sync(
 250    connection_id: Annotated[
 251        str,
 252        Field(description="The ID of the Airbyte Cloud connection."),
 253    ],
 254    *,
 255    wait: Annotated[
 256        bool,
 257        Field(
 258            description="Whether to wait for the sync to complete.",
 259            default=True,
 260        ),
 261    ],
 262    wait_timeout: Annotated[
 263        int,
 264        Field(
 265            description="Maximum time to wait for sync completion (seconds).",
 266            default=300,
 267        ),
 268    ],
 269) -> str:
 270    """Run a sync job on Airbyte Cloud.
 271
 272    By default, the `AIRBYTE_CLIENT_ID`, `AIRBYTE_CLIENT_SECRET`, `AIRBYTE_WORKSPACE_ID`,
 273    and `AIRBYTE_API_ROOT` environment variables will be used to authenticate with the
 274    Airbyte Cloud API.
 275    """
 276    try:
 277        workspace: CloudWorkspace = _get_cloud_workspace()
 278        connection = workspace.get_connection(connection_id=connection_id)
 279        sync_result = connection.run_sync(wait=wait, wait_timeout=wait_timeout)
 280
 281    except Exception as ex:
 282        return f"Failed to run sync for connection '{connection_id}': {ex}"
 283    else:
 284        if wait:
 285            status = sync_result.get_job_status()
 286            return (
 287                f"Sync completed with status: {status}. "  # Sync completed.
 288                f"Job ID is '{sync_result.job_id}' and "
 289                f"job URL is: {sync_result.job_url}"
 290            )
 291        return (
 292            f"Sync started. "  # Sync started.
 293            f"Job ID is '{sync_result.job_id}' and "
 294            f"job URL is: {sync_result.job_url}"
 295        )
 296
 297
 298@mcp_tool(
 299    domain="cloud",
 300    read_only=True,
 301    idempotent=True,
 302    open_world=True,
 303)
 304def check_airbyte_cloud_workspace() -> str:
 305    """Check if we have a valid Airbyte Cloud connection and return workspace info.
 306
 307    By default, the `AIRBYTE_CLIENT_ID`, `AIRBYTE_CLIENT_SECRET`, `AIRBYTE_WORKSPACE_ID`,
 308    and `AIRBYTE_API_ROOT` environment variables will be used to authenticate with the
 309    Airbyte Cloud API.
 310
 311    Returns workspace ID and workspace URL for verification.
 312    """
 313    try:
 314        workspace: CloudWorkspace = _get_cloud_workspace()
 315        workspace.connect()
 316
 317    except Exception as ex:
 318        return f"❌ Failed to connect to Airbyte Cloud workspace: {ex}"
 319    else:
 320        return (
 321            f"✅ Successfully connected to Airbyte Cloud workspace.\n"
 322            f"Workspace ID: {workspace.workspace_id}\n"
 323            f"Workspace URL: {workspace.workspace_url}"
 324        )
 325
 326
 327@mcp_tool(
 328    domain="cloud",
 329    open_world=True,
 330)
 331def deploy_noop_destination_to_cloud(
 332    name: str = "No-op Destination",
 333    *,
 334    unique: bool = True,
 335) -> str:
 336    """Deploy the No-op destination to Airbyte Cloud for testing purposes.
 337
 338    By default, the `AIRBYTE_CLIENT_ID`, `AIRBYTE_CLIENT_SECRET`, `AIRBYTE_WORKSPACE_ID`,
 339    and `AIRBYTE_API_ROOT` environment variables will be used to authenticate with the
 340    Airbyte Cloud API.
 341    """
 342    try:
 343        destination = get_noop_destination()
 344        workspace: CloudWorkspace = _get_cloud_workspace()
 345        deployed_destination = workspace.deploy_destination(
 346            name=name,
 347            destination=destination,
 348            unique=unique,
 349        )
 350    except Exception as ex:
 351        return f"Failed to deploy No-op Destination: {ex}"
 352    else:
 353        register_guid_created_in_session(deployed_destination.connector_id)
 354        return (
 355            f"Successfully deployed No-op Destination "
 356            f"with ID '{deployed_destination.connector_id}' and "
 357            f"URL: {deployed_destination.connector_url}"
 358        )
 359
 360
 361@mcp_tool(
 362    domain="cloud",
 363    read_only=True,
 364    idempotent=True,
 365    open_world=True,
 366)
 367def get_cloud_sync_status(
 368    connection_id: Annotated[
 369        str,
 370        Field(
 371            description="The ID of the Airbyte Cloud connection.",
 372        ),
 373    ],
 374    job_id: Annotated[
 375        int | None,
 376        Field(
 377            description="Optional job ID. If not provided, the latest job will be used.",
 378            default=None,
 379        ),
 380    ],
 381    *,
 382    include_attempts: Annotated[
 383        bool,
 384        Field(
 385            description="Whether to include detailed attempts information.",
 386            default=False,
 387        ),
 388    ],
 389) -> dict[str, Any]:
 390    """Get the status of a sync job from the Airbyte Cloud.
 391
 392    By default, the `AIRBYTE_CLIENT_ID`, `AIRBYTE_CLIENT_SECRET`, `AIRBYTE_WORKSPACE_ID`,
 393    and `AIRBYTE_API_ROOT` environment variables will be used to authenticate with the
 394    Airbyte Cloud API.
 395    """
 396    try:
 397        workspace: CloudWorkspace = _get_cloud_workspace()
 398        connection = workspace.get_connection(connection_id=connection_id)
 399
 400        # If a job ID is provided, get the job by ID.
 401        sync_result: cloud.SyncResult | None = connection.get_sync_result(job_id=job_id)
 402
 403        if not sync_result:
 404            return {"status": None, "job_id": None, "attempts": []}
 405
 406        result = {
 407            "status": sync_result.get_job_status(),
 408            "job_id": sync_result.job_id,
 409            "bytes_synced": sync_result.bytes_synced,
 410            "records_synced": sync_result.records_synced,
 411            "start_time": sync_result.start_time.isoformat(),
 412            "job_url": sync_result.job_url,
 413            "attempts": [],
 414        }
 415
 416        if include_attempts:
 417            attempts = sync_result.get_attempts()
 418            result["attempts"] = [
 419                {
 420                    "attempt_number": attempt.attempt_number,
 421                    "attempt_id": attempt.attempt_id,
 422                    "status": attempt.status,
 423                    "bytes_synced": attempt.bytes_synced,
 424                    "records_synced": attempt.records_synced,
 425                    "created_at": attempt.created_at.isoformat(),
 426                }
 427                for attempt in attempts
 428            ]
 429
 430        return result  # noqa: TRY300
 431
 432    except Exception as ex:
 433        return {
 434            "status": None,
 435            "job_id": job_id,
 436            "error": f"Failed to get sync status for connection '{connection_id}': {ex}",
 437            "attempts": [],
 438        }
 439
 440
 441@mcp_tool(
 442    domain="cloud",
 443    read_only=True,
 444    idempotent=True,
 445    open_world=True,
 446)
 447def list_deployed_cloud_source_connectors() -> list[CloudSource]:
 448    """List all deployed source connectors in the Airbyte Cloud workspace.
 449
 450    By default, the `AIRBYTE_CLIENT_ID`, `AIRBYTE_CLIENT_SECRET`, `AIRBYTE_WORKSPACE_ID`,
 451    and `AIRBYTE_API_ROOT` environment variables will be used to authenticate with the
 452    Airbyte Cloud API.
 453    """
 454    workspace: CloudWorkspace = _get_cloud_workspace()
 455    return workspace.list_sources()
 456
 457
 458@mcp_tool(
 459    domain="cloud",
 460    read_only=True,
 461    idempotent=True,
 462    open_world=True,
 463)
 464def list_deployed_cloud_destination_connectors() -> list[CloudDestination]:
 465    """List all deployed destination connectors in the Airbyte Cloud workspace.
 466
 467    By default, the `AIRBYTE_CLIENT_ID`, `AIRBYTE_CLIENT_SECRET`, `AIRBYTE_WORKSPACE_ID`,
 468    and `AIRBYTE_API_ROOT` environment variables will be used to authenticate with the
 469    Airbyte Cloud API.
 470    """
 471    workspace: CloudWorkspace = _get_cloud_workspace()
 472    return workspace.list_destinations()
 473
 474
 475@mcp_tool(
 476    domain="cloud",
 477    read_only=True,
 478    idempotent=True,
 479    open_world=True,
 480)
 481def get_cloud_sync_logs(
 482    connection_id: Annotated[
 483        str,
 484        Field(description="The ID of the Airbyte Cloud connection."),
 485    ],
 486    job_id: Annotated[
 487        int | None,
 488        Field(description="Optional job ID. If not provided, the latest job will be used."),
 489    ] = None,
 490    attempt_number: Annotated[
 491        int | None,
 492        Field(
 493            description="Optional attempt number. If not provided, the latest attempt will be used."
 494        ),
 495    ] = None,
 496) -> str:
 497    """Get the logs from a sync job attempt on Airbyte Cloud.
 498
 499    By default, the `AIRBYTE_CLIENT_ID`, `AIRBYTE_CLIENT_SECRET`, `AIRBYTE_WORKSPACE_ID`,
 500    and `AIRBYTE_API_ROOT` environment variables will be used to authenticate with the
 501    Airbyte Cloud API.
 502    """
 503    try:
 504        workspace: CloudWorkspace = _get_cloud_workspace()
 505        connection = workspace.get_connection(connection_id=connection_id)
 506
 507        sync_result: cloud.SyncResult | None = connection.get_sync_result(job_id=job_id)
 508
 509        if not sync_result:
 510            return f"No sync job found for connection '{connection_id}'"
 511
 512        attempts = sync_result.get_attempts()
 513
 514        if not attempts:
 515            return f"No attempts found for job '{sync_result.job_id}'"
 516
 517        if attempt_number is not None:
 518            target_attempt = None
 519            for attempt in attempts:
 520                if attempt.attempt_number == attempt_number:
 521                    target_attempt = attempt
 522                    break
 523
 524            if target_attempt is None:
 525                return f"Attempt number {attempt_number} not found for job '{sync_result.job_id}'"
 526        else:
 527            target_attempt = max(attempts, key=lambda a: a.attempt_number)
 528
 529        logs = target_attempt.get_full_log_text()
 530
 531        if not logs:
 532            return (
 533                f"No logs available for job '{sync_result.job_id}', "
 534                f"attempt {target_attempt.attempt_number}"
 535            )
 536
 537        return logs  # noqa: TRY300
 538
 539    except Exception as ex:
 540        return f"Failed to get logs for connection '{connection_id}': {ex}"
 541
 542
 543@mcp_tool(
 544    domain="cloud",
 545    read_only=True,
 546    idempotent=True,
 547    open_world=True,
 548)
 549def list_deployed_cloud_connections() -> list[CloudConnection]:
 550    """List all deployed connections in the Airbyte Cloud workspace.
 551
 552    By default, the `AIRBYTE_CLIENT_ID`, `AIRBYTE_CLIENT_SECRET`, `AIRBYTE_WORKSPACE_ID`,
 553    and `AIRBYTE_API_ROOT` environment variables will be used to authenticate with the
 554    Airbyte Cloud API.
 555    """
 556    workspace: CloudWorkspace = _get_cloud_workspace()
 557    return workspace.list_connections()
 558
 559
 560def _get_custom_source_definition_description(
 561    custom_source: CustomCloudSourceDefinition,
 562) -> str:
 563    return "\n".join(
 564        [
 565            f" - Custom Source Name: {custom_source.name}",
 566            f" - Definition ID: {custom_source.definition_id}",
 567            f" - Definition Version: {custom_source.version}",
 568            f" - Connector Builder Project ID: {custom_source.connector_builder_project_id}",
 569            f" - Connector Builder Project URL: {custom_source.connector_builder_project_url}",
 570        ]
 571    )
 572
 573
 574@mcp_tool(
 575    domain="cloud",
 576    open_world=True,
 577)
 578def publish_custom_source_definition(
 579    name: Annotated[
 580        str,
 581        Field(description="The name for the custom connector definition."),
 582    ],
 583    *,
 584    manifest_yaml: Annotated[
 585        str | Path | None,
 586        Field(
 587            description=(
 588                "The Low-code CDK manifest as a YAML string or file path. "
 589                "Required for YAML connectors."
 590            ),
 591            default=None,
 592        ),
 593    ] = None,
 594    unique: Annotated[
 595        bool,
 596        Field(
 597            description="Whether to require a unique name.",
 598            default=True,
 599        ),
 600    ] = True,
 601    pre_validate: Annotated[
 602        bool,
 603        Field(
 604            description="Whether to validate the manifest client-side before publishing.",
 605            default=True,
 606        ),
 607    ] = True,
 608) -> str:
 609    """Publish a custom YAML source connector definition to Airbyte Cloud.
 610
 611    Note: Only YAML (declarative) connectors are currently supported.
 612    Docker-based custom sources are not yet available.
 613    """
 614    try:
 615        processed_manifest = manifest_yaml
 616        if isinstance(manifest_yaml, str) and "\n" not in manifest_yaml:
 617            processed_manifest = Path(manifest_yaml)
 618
 619        workspace: CloudWorkspace = _get_cloud_workspace()
 620        custom_source = workspace.publish_custom_source_definition(
 621            name=name,
 622            manifest_yaml=processed_manifest,
 623            unique=unique,
 624            pre_validate=pre_validate,
 625        )
 626    except Exception as ex:
 627        return f"Failed to publish custom source definition '{name}': {ex}"
 628    else:
 629        register_guid_created_in_session(custom_source.definition_id)
 630        return (
 631            "Successfully published custom YAML source definition:\n"
 632            + _get_custom_source_definition_description(
 633                custom_source=custom_source,
 634            )
 635            + "\n"
 636        )
 637
 638
 639@mcp_tool(
 640    domain="cloud",
 641    read_only=True,
 642    idempotent=True,
 643    open_world=True,
 644)
 645def list_custom_source_definitions() -> list[dict[str, Any]]:
 646    """List custom YAML source definitions in the Airbyte Cloud workspace.
 647
 648    Note: Only YAML (declarative) connectors are currently supported.
 649    Docker-based custom sources are not yet available.
 650    """
 651    workspace: CloudWorkspace = _get_cloud_workspace()
 652    definitions = workspace.list_custom_source_definitions(
 653        definition_type="yaml",
 654    )
 655
 656    return [
 657        {
 658            "definition_id": d.definition_id,
 659            "name": d.name,
 660            "version": d.version,
 661            "connector_builder_project_url": d.connector_builder_project_url,
 662        }
 663        for d in definitions
 664    ]
 665
 666
 667@mcp_tool(
 668    domain="cloud",
 669    destructive=True,
 670    open_world=True,
 671)
 672def update_custom_source_definition(
 673    definition_id: Annotated[
 674        str,
 675        Field(description="The ID of the definition to update."),
 676    ],
 677    manifest_yaml: Annotated[
 678        str | Path,
 679        Field(
 680            description="New manifest as YAML string or file path.",
 681        ),
 682    ],
 683    *,
 684    pre_validate: Annotated[
 685        bool,
 686        Field(
 687            description="Whether to validate the manifest client-side before updating.",
 688            default=True,
 689        ),
 690    ] = True,
 691) -> str:
 692    """Update a custom YAML source definition in Airbyte Cloud.
 693
 694    Note: Only YAML (declarative) connectors are currently supported.
 695    Docker-based custom sources are not yet available.
 696    """
 697    check_guid_created_in_session(definition_id)
 698    try:
 699        processed_manifest = manifest_yaml
 700        if isinstance(manifest_yaml, str) and "\n" not in manifest_yaml:
 701            processed_manifest = Path(manifest_yaml)
 702
 703        workspace: CloudWorkspace = _get_cloud_workspace()
 704        definition = workspace.get_custom_source_definition(
 705            definition_id=definition_id,
 706            definition_type="yaml",
 707        )
 708        custom_source: CustomCloudSourceDefinition = definition.update_definition(
 709            manifest_yaml=processed_manifest,
 710            pre_validate=pre_validate,
 711        )
 712    except Exception as ex:
 713        return f"Failed to update custom source definition '{definition_id}': {ex}"
 714    else:
 715        return (
 716            "Successfully updated custom YAML source definition:\n"
 717            + _get_custom_source_definition_description(
 718                custom_source=custom_source,
 719            )
 720        )
 721
 722
 723@mcp_tool(
 724    domain="cloud",
 725    destructive=True,
 726    open_world=True,
 727)
 728def permanently_delete_custom_source_definition(
 729    definition_id: Annotated[
 730        str,
 731        Field(description="The ID of the custom source definition to delete."),
 732    ],
 733) -> str:
 734    """Permanently delete a custom YAML source definition from Airbyte Cloud.
 735
 736    IMPORTANT: This operation requires the connector name to either:
 737    1. Start with "delete:" (case insensitive), OR
 738    2. Contain "delete-me" (case insensitive)
 739
 740    If the connector does not meet these requirements, the deletion will be rejected with a
 741    helpful error message. Instruct the user to rename the connector appropriately to authorize
 742    the deletion.
 743
 744    Note: Only YAML (declarative) connectors are currently supported.
 745    Docker-based custom sources are not yet available.
 746    """
 747    check_guid_created_in_session(definition_id)
 748    workspace: CloudWorkspace = _get_cloud_workspace()
 749    definition = workspace.get_custom_source_definition(
 750        definition_id=definition_id,
 751        definition_type="yaml",
 752    )
 753    definition_name: str = definition.name  # Capture name before deletion
 754    definition.permanently_delete(
 755        safe_mode=True,  # Hard-coded safe mode for extra protection when running in LLM agents.
 756    )
 757    return (
 758        f"Successfully deleted custom source definition '{definition_name}' (ID: {definition_id})"
 759    )
 760
 761
 762@mcp_tool(
 763    domain="cloud",
 764    open_world=True,
 765)
 766def rename_cloud_source(
 767    source_id: Annotated[
 768        str,
 769        Field(description="The ID of the deployed source to rename."),
 770    ],
 771    name: Annotated[
 772        str,
 773        Field(description="New name for the source."),
 774    ],
 775) -> str:
 776    """Rename a deployed source connector on Airbyte Cloud.
 777
 778    By default, the `AIRBYTE_CLIENT_ID`, `AIRBYTE_CLIENT_SECRET`, `AIRBYTE_WORKSPACE_ID`,
 779    and `AIRBYTE_API_ROOT` environment variables will be used to authenticate with the
 780    Airbyte Cloud API.
 781    """
 782    workspace: CloudWorkspace = _get_cloud_workspace()
 783    source = workspace.get_source(source_id=source_id)
 784    source.rename(name=name)
 785    return f"Successfully renamed source '{source_id}' to '{name}'. URL: {source.connector_url}"
 786
 787
 788@mcp_tool(
 789    domain="cloud",
 790    destructive=True,
 791    open_world=True,
 792)
 793def update_cloud_source_config(
 794    source_id: Annotated[
 795        str,
 796        Field(description="The ID of the deployed source to update."),
 797    ],
 798    config: Annotated[
 799        dict | str,
 800        Field(
 801            description="New configuration for the source connector.",
 802        ),
 803    ],
 804    config_secret_name: Annotated[
 805        str | None,
 806        Field(
 807            description="The name of the secret containing the configuration.",
 808            default=None,
 809        ),
 810    ] = None,
 811) -> str:
 812    """Update a deployed source connector's configuration on Airbyte Cloud.
 813
 814    This is a destructive operation that can break existing connections if the
 815    configuration is changed incorrectly. Use with caution.
 816
 817    By default, the `AIRBYTE_CLIENT_ID`, `AIRBYTE_CLIENT_SECRET`, `AIRBYTE_WORKSPACE_ID`,
 818    and `AIRBYTE_API_ROOT` environment variables will be used to authenticate with the
 819    Airbyte Cloud API.
 820    """
 821    check_guid_created_in_session(source_id)
 822    workspace: CloudWorkspace = _get_cloud_workspace()
 823    source = workspace.get_source(source_id=source_id)
 824
 825    config_dict = resolve_config(
 826        config=config,
 827        config_secret_name=config_secret_name,
 828        config_spec_jsonschema=None,  # We don't have the spec here
 829    )
 830
 831    source.update_config(config=config_dict)
 832    return f"Successfully updated source '{source_id}'. URL: {source.connector_url}"
 833
 834
 835@mcp_tool(
 836    domain="cloud",
 837    open_world=True,
 838)
 839def rename_cloud_destination(
 840    destination_id: Annotated[
 841        str,
 842        Field(description="The ID of the deployed destination to rename."),
 843    ],
 844    name: Annotated[
 845        str,
 846        Field(description="New name for the destination."),
 847    ],
 848) -> str:
 849    """Rename a deployed destination connector on Airbyte Cloud.
 850
 851    By default, the `AIRBYTE_CLIENT_ID`, `AIRBYTE_CLIENT_SECRET`, `AIRBYTE_WORKSPACE_ID`,
 852    and `AIRBYTE_API_ROOT` environment variables will be used to authenticate with the
 853    Airbyte Cloud API.
 854    """
 855    workspace: CloudWorkspace = _get_cloud_workspace()
 856    destination = workspace.get_destination(destination_id=destination_id)
 857    destination.rename(name=name)
 858    return (
 859        f"Successfully renamed destination '{destination_id}' to '{name}'. "
 860        f"URL: {destination.connector_url}"
 861    )
 862
 863
 864@mcp_tool(
 865    domain="cloud",
 866    destructive=True,
 867    open_world=True,
 868)
 869def update_cloud_destination_config(
 870    destination_id: Annotated[
 871        str,
 872        Field(description="The ID of the deployed destination to update."),
 873    ],
 874    config: Annotated[
 875        dict | str,
 876        Field(
 877            description="New configuration for the destination connector.",
 878        ),
 879    ],
 880    config_secret_name: Annotated[
 881        str | None,
 882        Field(
 883            description="The name of the secret containing the configuration.",
 884            default=None,
 885        ),
 886    ] = None,
 887) -> str:
 888    """Update a deployed destination connector's configuration on Airbyte Cloud.
 889
 890    This is a destructive operation that can break existing connections if the
 891    configuration is changed incorrectly. Use with caution.
 892
 893    By default, the `AIRBYTE_CLIENT_ID`, `AIRBYTE_CLIENT_SECRET`, `AIRBYTE_WORKSPACE_ID`,
 894    and `AIRBYTE_API_ROOT` environment variables will be used to authenticate with the
 895    Airbyte Cloud API.
 896    """
 897    check_guid_created_in_session(destination_id)
 898    workspace: CloudWorkspace = _get_cloud_workspace()
 899    destination = workspace.get_destination(destination_id=destination_id)
 900
 901    config_dict = resolve_config(
 902        config=config,
 903        config_secret_name=config_secret_name,
 904        config_spec_jsonschema=None,  # We don't have the spec here
 905    )
 906
 907    destination.update_config(config=config_dict)
 908    return (
 909        f"Successfully updated destination '{destination_id}'. " f"URL: {destination.connector_url}"
 910    )
 911
 912
 913@mcp_tool(
 914    domain="cloud",
 915    open_world=True,
 916)
 917def rename_cloud_connection(
 918    connection_id: Annotated[
 919        str,
 920        Field(description="The ID of the connection to rename."),
 921    ],
 922    name: Annotated[
 923        str,
 924        Field(description="New name for the connection."),
 925    ],
 926) -> str:
 927    """Rename a connection on Airbyte Cloud.
 928
 929    By default, the `AIRBYTE_CLIENT_ID`, `AIRBYTE_CLIENT_SECRET`, `AIRBYTE_WORKSPACE_ID`,
 930    and `AIRBYTE_API_ROOT` environment variables will be used to authenticate with the
 931    Airbyte Cloud API.
 932    """
 933    workspace: CloudWorkspace = _get_cloud_workspace()
 934    connection = workspace.get_connection(connection_id=connection_id)
 935    connection.rename(name=name)
 936    return (
 937        f"Successfully renamed connection '{connection_id}' to '{name}'. "
 938        f"URL: {connection.connection_url}"
 939    )
 940
 941
 942@mcp_tool(
 943    domain="cloud",
 944    destructive=True,
 945    open_world=True,
 946)
 947def set_cloud_connection_table_prefix(
 948    connection_id: Annotated[
 949        str,
 950        Field(description="The ID of the connection to update."),
 951    ],
 952    prefix: Annotated[
 953        str,
 954        Field(description="New table prefix to use when syncing to the destination."),
 955    ],
 956) -> str:
 957    """Set the table prefix for a connection on Airbyte Cloud.
 958
 959    This is a destructive operation that can break downstream dependencies if the
 960    table prefix is changed incorrectly. Use with caution.
 961
 962    By default, the `AIRBYTE_CLIENT_ID`, `AIRBYTE_CLIENT_SECRET`, `AIRBYTE_WORKSPACE_ID`,
 963    and `AIRBYTE_API_ROOT` environment variables will be used to authenticate with the
 964    Airbyte Cloud API.
 965    """
 966    check_guid_created_in_session(connection_id)
 967    workspace: CloudWorkspace = _get_cloud_workspace()
 968    connection = workspace.get_connection(connection_id=connection_id)
 969    connection.set_table_prefix(prefix=prefix)
 970    return (
 971        f"Successfully set table prefix for connection '{connection_id}' to '{prefix}'. "
 972        f"URL: {connection.connection_url}"
 973    )
 974
 975
 976@mcp_tool(
 977    domain="cloud",
 978    destructive=True,
 979    open_world=True,
 980)
 981def set_cloud_connection_selected_streams(
 982    connection_id: Annotated[
 983        str,
 984        Field(description="The ID of the connection to update."),
 985    ],
 986    stream_names: Annotated[
 987        str | list[str],
 988        Field(
 989            description=(
 990                "The selected stream names to sync within the connection. "
 991                "Must be an explicit stream name or list of streams."
 992            )
 993        ),
 994    ],
 995) -> str:
 996    """Set the selected streams for a connection on Airbyte Cloud.
 997
 998    This is a destructive operation that can break existing connections if the
 999    stream selection is changed incorrectly. Use with caution.
1000
1001    By default, the `AIRBYTE_CLIENT_ID`, `AIRBYTE_CLIENT_SECRET`, `AIRBYTE_WORKSPACE_ID`,
1002    and `AIRBYTE_API_ROOT` environment variables will be used to authenticate with the
1003    Airbyte Cloud API.
1004    """
1005    check_guid_created_in_session(connection_id)
1006    workspace: CloudWorkspace = _get_cloud_workspace()
1007    connection = workspace.get_connection(connection_id=connection_id)
1008
1009    resolved_streams_list: list[str] = resolve_list_of_strings(stream_names)
1010    connection.set_selected_streams(stream_names=resolved_streams_list)
1011
1012    return (
1013        f"Successfully set selected streams for connection '{connection_id}' "
1014        f"to {resolved_streams_list}. URL: {connection.connection_url}"
1015    )
1016
1017
1018def register_cloud_ops_tools(app: FastMCP) -> None:
1019    """@private Register tools with the FastMCP app.
1020
1021    This is an internal function and should not be called directly.
1022
1023    Tools are filtered based on mode settings:
1024    - AIRBYTE_CLOUD_MCP_READONLY_MODE=1: Only read-only tools are registered
1025    - AIRBYTE_CLOUD_MCP_SAFE_MODE=1: All tools are registered, but destructive
1026      operations are protected by runtime session checks
1027    """
1028    register_tools(app, domain="cloud")