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, Literal, cast
   6
   7from fastmcp import FastMCP
   8from pydantic import BaseModel, Field
   9
  10from airbyte import cloud, get_destination, get_source
  11from airbyte._util import api_util
  12from airbyte.cloud.connectors import CustomCloudSourceDefinition
  13from airbyte.cloud.constants import FAILED_STATUSES
  14from airbyte.cloud.workspaces import CloudWorkspace
  15from airbyte.destinations.util import get_noop_destination
  16from airbyte.exceptions import AirbyteMissingResourceError, PyAirbyteInputError
  17from airbyte.mcp._tool_utils import (
  18    check_guid_created_in_session,
  19    mcp_tool,
  20    register_guid_created_in_session,
  21    register_tools,
  22)
  23from airbyte.mcp._util import (
  24    resolve_cloud_credentials,
  25    resolve_config,
  26    resolve_list_of_strings,
  27    resolve_workspace_id,
  28)
  29from airbyte.secrets import SecretString
  30
  31
  32CLOUD_AUTH_TIP_TEXT = (
  33    "By default, the `AIRBYTE_CLOUD_CLIENT_ID`, `AIRBYTE_CLOUD_CLIENT_SECRET`, "
  34    "and `AIRBYTE_CLOUD_WORKSPACE_ID` environment variables "
  35    "will be used to authenticate with the Airbyte Cloud API."
  36)
  37WORKSPACE_ID_TIP_TEXT = "Workspace ID. Defaults to `AIRBYTE_CLOUD_WORKSPACE_ID` env var."
  38
  39
  40class CloudSourceResult(BaseModel):
  41    """Information about a deployed source connector in Airbyte Cloud."""
  42
  43    id: str
  44    """The source ID."""
  45    name: str
  46    """Display name of the source."""
  47    url: str
  48    """Web URL for managing this source in Airbyte Cloud."""
  49
  50
  51class CloudDestinationResult(BaseModel):
  52    """Information about a deployed destination connector in Airbyte Cloud."""
  53
  54    id: str
  55    """The destination ID."""
  56    name: str
  57    """Display name of the destination."""
  58    url: str
  59    """Web URL for managing this destination in Airbyte Cloud."""
  60
  61
  62class CloudConnectionResult(BaseModel):
  63    """Information about a deployed connection in Airbyte Cloud."""
  64
  65    id: str
  66    """The connection ID."""
  67    name: str
  68    """Display name of the connection."""
  69    url: str
  70    """Web URL for managing this connection in Airbyte Cloud."""
  71    source_id: str
  72    """ID of the source used by this connection."""
  73    destination_id: str
  74    """ID of the destination used by this connection."""
  75    last_job_status: str | None = None
  76    """Status of the most recent completed sync job (e.g., 'succeeded', 'failed', 'cancelled').
  77    Only populated when with_connection_status=True."""
  78    last_job_id: int | None = None
  79    """Job ID of the most recent completed sync. Only populated when with_connection_status=True."""
  80    last_job_time: str | None = None
  81    """ISO 8601 timestamp of the most recent completed sync.
  82    Only populated when with_connection_status=True."""
  83    currently_running_job_id: int | None = None
  84    """Job ID of a currently running sync, if any.
  85    Only populated when with_connection_status=True."""
  86    currently_running_job_start_time: str | None = None
  87    """ISO 8601 timestamp of when the currently running sync started.
  88    Only populated when with_connection_status=True."""
  89
  90
  91class CloudSourceDetails(BaseModel):
  92    """Detailed information about a deployed source connector in Airbyte Cloud."""
  93
  94    source_id: str
  95    """The source ID."""
  96    source_name: str
  97    """Display name of the source."""
  98    source_url: str
  99    """Web URL for managing this source in Airbyte Cloud."""
 100    connector_definition_id: str
 101    """The connector definition ID (e.g., the ID for 'source-postgres')."""
 102
 103
 104class CloudDestinationDetails(BaseModel):
 105    """Detailed information about a deployed destination connector in Airbyte Cloud."""
 106
 107    destination_id: str
 108    """The destination ID."""
 109    destination_name: str
 110    """Display name of the destination."""
 111    destination_url: str
 112    """Web URL for managing this destination in Airbyte Cloud."""
 113    connector_definition_id: str
 114    """The connector definition ID (e.g., the ID for 'destination-snowflake')."""
 115
 116
 117class CloudConnectionDetails(BaseModel):
 118    """Detailed information about a deployed connection in Airbyte Cloud."""
 119
 120    connection_id: str
 121    """The connection ID."""
 122    connection_name: str
 123    """Display name of the connection."""
 124    connection_url: str
 125    """Web URL for managing this connection in Airbyte Cloud."""
 126    source_id: str
 127    """ID of the source used by this connection."""
 128    source_name: str
 129    """Display name of the source."""
 130    destination_id: str
 131    """ID of the destination used by this connection."""
 132    destination_name: str
 133    """Display name of the destination."""
 134    selected_streams: list[str]
 135    """List of stream names selected for syncing."""
 136    table_prefix: str | None
 137    """Table prefix applied when syncing to the destination."""
 138
 139
 140class CloudOrganizationResult(BaseModel):
 141    """Information about an organization in Airbyte Cloud."""
 142
 143    id: str
 144    """The organization ID."""
 145    name: str
 146    """Display name of the organization."""
 147    email: str
 148    """Email associated with the organization."""
 149
 150
 151class CloudWorkspaceResult(BaseModel):
 152    """Information about a workspace in Airbyte Cloud."""
 153
 154    workspace_id: str
 155    """The workspace ID."""
 156    workspace_name: str
 157    """Display name of the workspace."""
 158    workspace_url: str | None = None
 159    """URL to access the workspace in Airbyte Cloud."""
 160    organization_id: str
 161    """ID of the organization (requires ORGANIZATION_READER permission)."""
 162    organization_name: str | None = None
 163    """Name of the organization (requires ORGANIZATION_READER permission)."""
 164
 165
 166class LogReadResult(BaseModel):
 167    """Result of reading sync logs with pagination support."""
 168
 169    job_id: int
 170    """The job ID the logs belong to."""
 171    attempt_number: int
 172    """The attempt number the logs belong to."""
 173    log_text: str
 174    """The string containing the log text we are returning."""
 175    log_text_start_line: int
 176    """1-based line index of the first line returned."""
 177    log_text_line_count: int
 178    """Count of lines we are returning."""
 179    total_log_lines_available: int
 180    """Total number of log lines available, shows if any lines were missed due to the limit."""
 181
 182
 183class SyncJobResult(BaseModel):
 184    """Information about a sync job."""
 185
 186    job_id: int
 187    """The job ID."""
 188    status: str
 189    """The job status (e.g., 'succeeded', 'failed', 'running', 'pending')."""
 190    bytes_synced: int
 191    """Number of bytes synced in this job."""
 192    records_synced: int
 193    """Number of records synced in this job."""
 194    start_time: str
 195    """ISO 8601 timestamp of when the job started."""
 196    job_url: str
 197    """URL to view the job in Airbyte Cloud."""
 198
 199
 200class SyncJobListResult(BaseModel):
 201    """Result of listing sync jobs with pagination support."""
 202
 203    jobs: list[SyncJobResult]
 204    """List of sync jobs."""
 205    jobs_count: int
 206    """Number of jobs returned in this response."""
 207    jobs_offset: int
 208    """Offset used for this request (0 if not specified)."""
 209    from_tail: bool
 210    """Whether jobs are ordered newest-first (True) or oldest-first (False)."""
 211
 212
 213def _get_cloud_workspace(workspace_id: str | None = None) -> CloudWorkspace:
 214    """Get an authenticated CloudWorkspace.
 215
 216    Resolves credentials from multiple sources in order:
 217    1. HTTP headers (when running as MCP server with HTTP/SSE transport)
 218    2. Environment variables
 219
 220    Args:
 221        workspace_id: Optional workspace ID. If not provided, uses HTTP headers
 222            or the AIRBYTE_CLOUD_WORKSPACE_ID environment variable.
 223    """
 224    credentials = resolve_cloud_credentials()
 225    resolved_workspace_id = resolve_workspace_id(workspace_id)
 226
 227    return CloudWorkspace(
 228        workspace_id=resolved_workspace_id,
 229        client_id=credentials.client_id,
 230        client_secret=credentials.client_secret,
 231        bearer_token=credentials.bearer_token,
 232        api_root=credentials.api_root,
 233    )
 234
 235
 236@mcp_tool(
 237    domain="cloud",
 238    open_world=True,
 239    extra_help_text=CLOUD_AUTH_TIP_TEXT,
 240)
 241def deploy_source_to_cloud(
 242    source_name: Annotated[
 243        str,
 244        Field(description="The name to use when deploying the source."),
 245    ],
 246    source_connector_name: Annotated[
 247        str,
 248        Field(description="The name of the source connector (e.g., 'source-faker')."),
 249    ],
 250    *,
 251    workspace_id: Annotated[
 252        str | None,
 253        Field(
 254            description=WORKSPACE_ID_TIP_TEXT,
 255            default=None,
 256        ),
 257    ],
 258    config: Annotated[
 259        dict | str | None,
 260        Field(
 261            description="The configuration for the source connector.",
 262            default=None,
 263        ),
 264    ],
 265    config_secret_name: Annotated[
 266        str | None,
 267        Field(
 268            description="The name of the secret containing the configuration.",
 269            default=None,
 270        ),
 271    ],
 272    unique: Annotated[
 273        bool,
 274        Field(
 275            description="Whether to require a unique name.",
 276            default=True,
 277        ),
 278    ],
 279) -> str:
 280    """Deploy a source connector to Airbyte Cloud."""
 281    source = get_source(
 282        source_connector_name,
 283        no_executor=True,
 284    )
 285    config_dict = resolve_config(
 286        config=config,
 287        config_secret_name=config_secret_name,
 288        config_spec_jsonschema=source.config_spec,
 289    )
 290    source.set_config(config_dict, validate=True)
 291
 292    workspace: CloudWorkspace = _get_cloud_workspace(workspace_id)
 293    deployed_source = workspace.deploy_source(
 294        name=source_name,
 295        source=source,
 296        unique=unique,
 297    )
 298
 299    register_guid_created_in_session(deployed_source.connector_id)
 300    return (
 301        f"Successfully deployed source '{source_name}' with ID '{deployed_source.connector_id}'"
 302        f" and URL: {deployed_source.connector_url}"
 303    )
 304
 305
 306@mcp_tool(
 307    domain="cloud",
 308    open_world=True,
 309    extra_help_text=CLOUD_AUTH_TIP_TEXT,
 310)
 311def deploy_destination_to_cloud(
 312    destination_name: Annotated[
 313        str,
 314        Field(description="The name to use when deploying the destination."),
 315    ],
 316    destination_connector_name: Annotated[
 317        str,
 318        Field(description="The name of the destination connector (e.g., 'destination-postgres')."),
 319    ],
 320    *,
 321    workspace_id: Annotated[
 322        str | None,
 323        Field(
 324            description=WORKSPACE_ID_TIP_TEXT,
 325            default=None,
 326        ),
 327    ],
 328    config: Annotated[
 329        dict | str | None,
 330        Field(
 331            description="The configuration for the destination connector.",
 332            default=None,
 333        ),
 334    ],
 335    config_secret_name: Annotated[
 336        str | None,
 337        Field(
 338            description="The name of the secret containing the configuration.",
 339            default=None,
 340        ),
 341    ],
 342    unique: Annotated[
 343        bool,
 344        Field(
 345            description="Whether to require a unique name.",
 346            default=True,
 347        ),
 348    ],
 349) -> str:
 350    """Deploy a destination connector to Airbyte Cloud."""
 351    destination = get_destination(
 352        destination_connector_name,
 353        no_executor=True,
 354    )
 355    config_dict = resolve_config(
 356        config=config,
 357        config_secret_name=config_secret_name,
 358        config_spec_jsonschema=destination.config_spec,
 359    )
 360    destination.set_config(config_dict, validate=True)
 361
 362    workspace: CloudWorkspace = _get_cloud_workspace(workspace_id)
 363    deployed_destination = workspace.deploy_destination(
 364        name=destination_name,
 365        destination=destination,
 366        unique=unique,
 367    )
 368
 369    register_guid_created_in_session(deployed_destination.connector_id)
 370    return (
 371        f"Successfully deployed destination '{destination_name}' "
 372        f"with ID: {deployed_destination.connector_id}"
 373    )
 374
 375
 376@mcp_tool(
 377    domain="cloud",
 378    open_world=True,
 379    extra_help_text=CLOUD_AUTH_TIP_TEXT,
 380)
 381def create_connection_on_cloud(
 382    connection_name: Annotated[
 383        str,
 384        Field(description="The name of the connection."),
 385    ],
 386    source_id: Annotated[
 387        str,
 388        Field(description="The ID of the deployed source."),
 389    ],
 390    destination_id: Annotated[
 391        str,
 392        Field(description="The ID of the deployed destination."),
 393    ],
 394    selected_streams: Annotated[
 395        str | list[str],
 396        Field(
 397            description=(
 398                "The selected stream names to sync within the connection. "
 399                "Must be an explicit stream name or list of streams. "
 400                "Cannot be empty or '*'."
 401            )
 402        ),
 403    ],
 404    *,
 405    workspace_id: Annotated[
 406        str | None,
 407        Field(
 408            description=WORKSPACE_ID_TIP_TEXT,
 409            default=None,
 410        ),
 411    ],
 412    table_prefix: Annotated[
 413        str | None,
 414        Field(
 415            description="Optional table prefix to use when syncing to the destination.",
 416            default=None,
 417        ),
 418    ],
 419) -> str:
 420    """Create a connection between a deployed source and destination on Airbyte Cloud."""
 421    resolved_streams_list: list[str] = resolve_list_of_strings(selected_streams)
 422    workspace: CloudWorkspace = _get_cloud_workspace(workspace_id)
 423    deployed_connection = workspace.deploy_connection(
 424        connection_name=connection_name,
 425        source=source_id,
 426        destination=destination_id,
 427        selected_streams=resolved_streams_list,
 428        table_prefix=table_prefix,
 429    )
 430
 431    register_guid_created_in_session(deployed_connection.connection_id)
 432    return (
 433        f"Successfully created connection '{connection_name}' "
 434        f"with ID '{deployed_connection.connection_id}' and "
 435        f"URL: {deployed_connection.connection_url}"
 436    )
 437
 438
 439@mcp_tool(
 440    domain="cloud",
 441    open_world=True,
 442    extra_help_text=CLOUD_AUTH_TIP_TEXT,
 443)
 444def run_cloud_sync(
 445    connection_id: Annotated[
 446        str,
 447        Field(description="The ID of the Airbyte Cloud connection."),
 448    ],
 449    *,
 450    workspace_id: Annotated[
 451        str | None,
 452        Field(
 453            description=WORKSPACE_ID_TIP_TEXT,
 454            default=None,
 455        ),
 456    ],
 457    wait: Annotated[
 458        bool,
 459        Field(
 460            description=(
 461                "Whether to wait for the sync to complete. Since a sync can take between several "
 462                "minutes and several hours, this option is not recommended for most "
 463                "scenarios."
 464            ),
 465            default=False,
 466        ),
 467    ],
 468    wait_timeout: Annotated[
 469        int,
 470        Field(
 471            description="Maximum time to wait for sync completion (seconds).",
 472            default=300,
 473        ),
 474    ],
 475) -> str:
 476    """Run a sync job on Airbyte Cloud."""
 477    workspace: CloudWorkspace = _get_cloud_workspace(workspace_id)
 478    connection = workspace.get_connection(connection_id=connection_id)
 479    sync_result = connection.run_sync(wait=wait, wait_timeout=wait_timeout)
 480
 481    if wait:
 482        status = sync_result.get_job_status()
 483        return (
 484            f"Sync completed with status: {status}. "
 485            f"Job ID is '{sync_result.job_id}' and "
 486            f"job URL is: {sync_result.job_url}"
 487        )
 488    return f"Sync started. Job ID is '{sync_result.job_id}' and job URL is: {sync_result.job_url}"
 489
 490
 491@mcp_tool(
 492    domain="cloud",
 493    read_only=True,
 494    idempotent=True,
 495    open_world=True,
 496    extra_help_text=CLOUD_AUTH_TIP_TEXT,
 497)
 498def check_airbyte_cloud_workspace(
 499    *,
 500    workspace_id: Annotated[
 501        str | None,
 502        Field(
 503            description=WORKSPACE_ID_TIP_TEXT,
 504            default=None,
 505        ),
 506    ],
 507) -> CloudWorkspaceResult:
 508    """Check if we have a valid Airbyte Cloud connection and return workspace info.
 509
 510    Returns workspace details including workspace ID, name, and organization info.
 511    """
 512    workspace: CloudWorkspace = _get_cloud_workspace(workspace_id)
 513
 514    # Get workspace details from the public API using workspace's credentials
 515    workspace_response = api_util.get_workspace(
 516        workspace_id=workspace.workspace_id,
 517        api_root=workspace.api_root,
 518        client_id=workspace.client_id,
 519        client_secret=workspace.client_secret,
 520        bearer_token=workspace.bearer_token,
 521    )
 522
 523    # Try to get organization info, but fail gracefully if we don't have permissions.
 524    # Fetching organization info requires ORGANIZATION_READER permissions on the organization,
 525    # which may not be available with workspace-scoped credentials.
 526    organization = workspace.get_organization(raise_on_error=False)
 527
 528    return CloudWorkspaceResult(
 529        workspace_id=workspace_response.workspace_id,
 530        workspace_name=workspace_response.name,
 531        workspace_url=workspace.workspace_url,
 532        organization_id=(
 533            organization.organization_id
 534            if organization
 535            else "[unavailable - requires ORGANIZATION_READER permission]"
 536        ),
 537        organization_name=organization.organization_name if organization else None,
 538    )
 539
 540
 541@mcp_tool(
 542    domain="cloud",
 543    open_world=True,
 544    extra_help_text=CLOUD_AUTH_TIP_TEXT,
 545)
 546def deploy_noop_destination_to_cloud(
 547    name: str = "No-op Destination",
 548    *,
 549    workspace_id: Annotated[
 550        str | None,
 551        Field(
 552            description=WORKSPACE_ID_TIP_TEXT,
 553            default=None,
 554        ),
 555    ],
 556    unique: bool = True,
 557) -> str:
 558    """Deploy the No-op destination to Airbyte Cloud for testing purposes."""
 559    destination = get_noop_destination()
 560    workspace: CloudWorkspace = _get_cloud_workspace(workspace_id)
 561    deployed_destination = workspace.deploy_destination(
 562        name=name,
 563        destination=destination,
 564        unique=unique,
 565    )
 566    register_guid_created_in_session(deployed_destination.connector_id)
 567    return (
 568        f"Successfully deployed No-op Destination "
 569        f"with ID '{deployed_destination.connector_id}' and "
 570        f"URL: {deployed_destination.connector_url}"
 571    )
 572
 573
 574@mcp_tool(
 575    domain="cloud",
 576    read_only=True,
 577    idempotent=True,
 578    open_world=True,
 579    extra_help_text=CLOUD_AUTH_TIP_TEXT,
 580)
 581def get_cloud_sync_status(
 582    connection_id: Annotated[
 583        str,
 584        Field(
 585            description="The ID of the Airbyte Cloud connection.",
 586        ),
 587    ],
 588    job_id: Annotated[
 589        int | None,
 590        Field(
 591            description="Optional job ID. If not provided, the latest job will be used.",
 592            default=None,
 593        ),
 594    ],
 595    *,
 596    workspace_id: Annotated[
 597        str | None,
 598        Field(
 599            description=WORKSPACE_ID_TIP_TEXT,
 600            default=None,
 601        ),
 602    ],
 603    include_attempts: Annotated[
 604        bool,
 605        Field(
 606            description="Whether to include detailed attempts information.",
 607            default=False,
 608        ),
 609    ],
 610) -> dict[str, Any]:
 611    """Get the status of a sync job from the Airbyte Cloud."""
 612    workspace: CloudWorkspace = _get_cloud_workspace(workspace_id)
 613    connection = workspace.get_connection(connection_id=connection_id)
 614
 615    # If a job ID is provided, get the job by ID.
 616    sync_result: cloud.SyncResult | None = connection.get_sync_result(job_id=job_id)
 617
 618    if not sync_result:
 619        return {"status": None, "job_id": None, "attempts": []}
 620
 621    result = {
 622        "status": sync_result.get_job_status(),
 623        "job_id": sync_result.job_id,
 624        "bytes_synced": sync_result.bytes_synced,
 625        "records_synced": sync_result.records_synced,
 626        "start_time": sync_result.start_time.isoformat(),
 627        "job_url": sync_result.job_url,
 628        "attempts": [],
 629    }
 630
 631    if include_attempts:
 632        attempts = sync_result.get_attempts()
 633        result["attempts"] = [
 634            {
 635                "attempt_number": attempt.attempt_number,
 636                "attempt_id": attempt.attempt_id,
 637                "status": attempt.status,
 638                "bytes_synced": attempt.bytes_synced,
 639                "records_synced": attempt.records_synced,
 640                "created_at": attempt.created_at.isoformat(),
 641            }
 642            for attempt in attempts
 643        ]
 644
 645    return result
 646
 647
 648@mcp_tool(
 649    domain="cloud",
 650    read_only=True,
 651    idempotent=True,
 652    open_world=True,
 653    extra_help_text=CLOUD_AUTH_TIP_TEXT,
 654)
 655def list_cloud_sync_jobs(
 656    connection_id: Annotated[
 657        str,
 658        Field(description="The ID of the Airbyte Cloud connection."),
 659    ],
 660    *,
 661    workspace_id: Annotated[
 662        str | None,
 663        Field(
 664            description=WORKSPACE_ID_TIP_TEXT,
 665            default=None,
 666        ),
 667    ],
 668    max_jobs: Annotated[
 669        int,
 670        Field(
 671            description=(
 672                "Maximum number of jobs to return. "
 673                "Defaults to 20 if not specified. "
 674                "Maximum allowed value is 500."
 675            ),
 676            default=20,
 677        ),
 678    ],
 679    from_tail: Annotated[
 680        bool | None,
 681        Field(
 682            description=(
 683                "When True, jobs are ordered newest-first (createdAt DESC). "
 684                "When False, jobs are ordered oldest-first (createdAt ASC). "
 685                "Defaults to True if `jobs_offset` is not specified. "
 686                "Cannot combine `from_tail=True` with `jobs_offset`."
 687            ),
 688            default=None,
 689        ),
 690    ],
 691    jobs_offset: Annotated[
 692        int | None,
 693        Field(
 694            description=(
 695                "Number of jobs to skip from the beginning. "
 696                "Cannot be combined with `from_tail=True`."
 697            ),
 698            default=None,
 699        ),
 700    ],
 701) -> SyncJobListResult:
 702    """List sync jobs for a connection with pagination support.
 703
 704    This tool allows you to retrieve a list of sync jobs for a connection,
 705    with control over ordering and pagination. By default, jobs are returned
 706    newest-first (from_tail=True).
 707    """
 708    # Validate that jobs_offset and from_tail are not both set
 709    if jobs_offset is not None and from_tail is True:
 710        raise PyAirbyteInputError(
 711            message="Cannot specify both 'jobs_offset' and 'from_tail=True' parameters.",
 712            context={"jobs_offset": jobs_offset, "from_tail": from_tail},
 713        )
 714
 715    # Default to from_tail=True if neither is specified
 716    if from_tail is None and jobs_offset is None:
 717        from_tail = True
 718    elif from_tail is None:
 719        from_tail = False
 720
 721    workspace: CloudWorkspace = _get_cloud_workspace(workspace_id)
 722    connection = workspace.get_connection(connection_id=connection_id)
 723
 724    # Cap at 500 to avoid overloading agent context
 725    effective_limit = min(max_jobs, 500) if max_jobs > 0 else 20
 726
 727    sync_results = connection.get_previous_sync_logs(
 728        limit=effective_limit,
 729        offset=jobs_offset,
 730        from_tail=from_tail,
 731    )
 732
 733    jobs = [
 734        SyncJobResult(
 735            job_id=sync_result.job_id,
 736            status=str(sync_result.get_job_status()),
 737            bytes_synced=sync_result.bytes_synced,
 738            records_synced=sync_result.records_synced,
 739            start_time=sync_result.start_time.isoformat(),
 740            job_url=sync_result.job_url,
 741        )
 742        for sync_result in sync_results
 743    ]
 744
 745    return SyncJobListResult(
 746        jobs=jobs,
 747        jobs_count=len(jobs),
 748        jobs_offset=jobs_offset or 0,
 749        from_tail=from_tail,
 750    )
 751
 752
 753@mcp_tool(
 754    domain="cloud",
 755    read_only=True,
 756    idempotent=True,
 757    open_world=True,
 758    extra_help_text=CLOUD_AUTH_TIP_TEXT,
 759)
 760def list_deployed_cloud_source_connectors(
 761    *,
 762    workspace_id: Annotated[
 763        str | None,
 764        Field(
 765            description=WORKSPACE_ID_TIP_TEXT,
 766            default=None,
 767        ),
 768    ],
 769    name_contains: Annotated[
 770        str | None,
 771        Field(
 772            description="Optional case-insensitive substring to filter sources by name",
 773            default=None,
 774        ),
 775    ],
 776    max_items_limit: Annotated[
 777        int | None,
 778        Field(
 779            description="Optional maximum number of items to return (default: no limit)",
 780            default=None,
 781        ),
 782    ],
 783) -> list[CloudSourceResult]:
 784    """List all deployed source connectors in the Airbyte Cloud workspace."""
 785    workspace: CloudWorkspace = _get_cloud_workspace(workspace_id)
 786    sources = workspace.list_sources()
 787
 788    # Filter by name if requested
 789    if name_contains:
 790        needle = name_contains.lower()
 791        sources = [s for s in sources if s.name is not None and needle in s.name.lower()]
 792
 793    # Apply limit if requested
 794    if max_items_limit is not None:
 795        sources = sources[:max_items_limit]
 796
 797    # Note: name and url are guaranteed non-null from list API responses
 798    return [
 799        CloudSourceResult(
 800            id=source.source_id,
 801            name=cast(str, source.name),
 802            url=cast(str, source.connector_url),
 803        )
 804        for source in sources
 805    ]
 806
 807
 808@mcp_tool(
 809    domain="cloud",
 810    read_only=True,
 811    idempotent=True,
 812    open_world=True,
 813    extra_help_text=CLOUD_AUTH_TIP_TEXT,
 814)
 815def list_deployed_cloud_destination_connectors(
 816    *,
 817    workspace_id: Annotated[
 818        str | None,
 819        Field(
 820            description=WORKSPACE_ID_TIP_TEXT,
 821            default=None,
 822        ),
 823    ],
 824    name_contains: Annotated[
 825        str | None,
 826        Field(
 827            description="Optional case-insensitive substring to filter destinations by name",
 828            default=None,
 829        ),
 830    ],
 831    max_items_limit: Annotated[
 832        int | None,
 833        Field(
 834            description="Optional maximum number of items to return (default: no limit)",
 835            default=None,
 836        ),
 837    ],
 838) -> list[CloudDestinationResult]:
 839    """List all deployed destination connectors in the Airbyte Cloud workspace."""
 840    workspace: CloudWorkspace = _get_cloud_workspace(workspace_id)
 841    destinations = workspace.list_destinations()
 842
 843    # Filter by name if requested
 844    if name_contains:
 845        needle = name_contains.lower()
 846        destinations = [d for d in destinations if d.name is not None and needle in d.name.lower()]
 847
 848    # Apply limit if requested
 849    if max_items_limit is not None:
 850        destinations = destinations[:max_items_limit]
 851
 852    # Note: name and url are guaranteed non-null from list API responses
 853    return [
 854        CloudDestinationResult(
 855            id=destination.destination_id,
 856            name=cast(str, destination.name),
 857            url=cast(str, destination.connector_url),
 858        )
 859        for destination in destinations
 860    ]
 861
 862
 863@mcp_tool(
 864    domain="cloud",
 865    read_only=True,
 866    idempotent=True,
 867    open_world=True,
 868    extra_help_text=CLOUD_AUTH_TIP_TEXT,
 869)
 870def describe_cloud_source(
 871    source_id: Annotated[
 872        str,
 873        Field(description="The ID of the source to describe."),
 874    ],
 875    *,
 876    workspace_id: Annotated[
 877        str | None,
 878        Field(
 879            description=WORKSPACE_ID_TIP_TEXT,
 880            default=None,
 881        ),
 882    ],
 883) -> CloudSourceDetails:
 884    """Get detailed information about a specific deployed source connector."""
 885    workspace: CloudWorkspace = _get_cloud_workspace(workspace_id)
 886    source = workspace.get_source(source_id=source_id)
 887
 888    # Access name property to ensure _connector_info is populated
 889    source_name = cast(str, source.name)
 890
 891    return CloudSourceDetails(
 892        source_id=source.source_id,
 893        source_name=source_name,
 894        source_url=source.connector_url,
 895        connector_definition_id=source._connector_info.definition_id,  # noqa: SLF001  # type: ignore[union-attr]
 896    )
 897
 898
 899@mcp_tool(
 900    domain="cloud",
 901    read_only=True,
 902    idempotent=True,
 903    open_world=True,
 904    extra_help_text=CLOUD_AUTH_TIP_TEXT,
 905)
 906def describe_cloud_destination(
 907    destination_id: Annotated[
 908        str,
 909        Field(description="The ID of the destination to describe."),
 910    ],
 911    *,
 912    workspace_id: Annotated[
 913        str | None,
 914        Field(
 915            description=WORKSPACE_ID_TIP_TEXT,
 916            default=None,
 917        ),
 918    ],
 919) -> CloudDestinationDetails:
 920    """Get detailed information about a specific deployed destination connector."""
 921    workspace: CloudWorkspace = _get_cloud_workspace(workspace_id)
 922    destination = workspace.get_destination(destination_id=destination_id)
 923
 924    # Access name property to ensure _connector_info is populated
 925    destination_name = cast(str, destination.name)
 926
 927    return CloudDestinationDetails(
 928        destination_id=destination.destination_id,
 929        destination_name=destination_name,
 930        destination_url=destination.connector_url,
 931        connector_definition_id=destination._connector_info.definition_id,  # noqa: SLF001  # type: ignore[union-attr]
 932    )
 933
 934
 935@mcp_tool(
 936    domain="cloud",
 937    read_only=True,
 938    idempotent=True,
 939    open_world=True,
 940    extra_help_text=CLOUD_AUTH_TIP_TEXT,
 941)
 942def describe_cloud_connection(
 943    connection_id: Annotated[
 944        str,
 945        Field(description="The ID of the connection to describe."),
 946    ],
 947    *,
 948    workspace_id: Annotated[
 949        str | None,
 950        Field(
 951            description=WORKSPACE_ID_TIP_TEXT,
 952            default=None,
 953        ),
 954    ],
 955) -> CloudConnectionDetails:
 956    """Get detailed information about a specific deployed connection."""
 957    workspace: CloudWorkspace = _get_cloud_workspace(workspace_id)
 958    connection = workspace.get_connection(connection_id=connection_id)
 959
 960    return CloudConnectionDetails(
 961        connection_id=connection.connection_id,
 962        connection_name=cast(str, connection.name),
 963        connection_url=cast(str, connection.connection_url),
 964        source_id=connection.source_id,
 965        source_name=cast(str, connection.source.name),
 966        destination_id=connection.destination_id,
 967        destination_name=cast(str, connection.destination.name),
 968        selected_streams=connection.stream_names,
 969        table_prefix=connection.table_prefix,
 970    )
 971
 972
 973@mcp_tool(
 974    domain="cloud",
 975    read_only=True,
 976    idempotent=True,
 977    open_world=True,
 978    extra_help_text=CLOUD_AUTH_TIP_TEXT,
 979)
 980def get_cloud_sync_logs(
 981    connection_id: Annotated[
 982        str,
 983        Field(description="The ID of the Airbyte Cloud connection."),
 984    ],
 985    job_id: Annotated[
 986        int | None,
 987        Field(description="Optional job ID. If not provided, the latest job will be used."),
 988    ] = None,
 989    attempt_number: Annotated[
 990        int | None,
 991        Field(
 992            description="Optional attempt number. If not provided, the latest attempt will be used."
 993        ),
 994    ] = None,
 995    *,
 996    workspace_id: Annotated[
 997        str | None,
 998        Field(
 999            description=WORKSPACE_ID_TIP_TEXT,
1000            default=None,
1001        ),
1002    ],
1003    max_lines: Annotated[
1004        int,
1005        Field(
1006            description=(
1007                "Maximum number of lines to return. "
1008                "Defaults to 4000 if not specified. "
1009                "If '0' is provided, no limit is applied."
1010            ),
1011            default=4000,
1012        ),
1013    ],
1014    from_tail: Annotated[
1015        bool | None,
1016        Field(
1017            description=(
1018                "Pull from the end of the log text if total lines is greater than 'max_lines'. "
1019                "Defaults to True if `line_offset` is not specified. "
1020                "Cannot combine `from_tail=True` with `line_offset`."
1021            ),
1022            default=None,
1023        ),
1024    ],
1025    line_offset: Annotated[
1026        int | None,
1027        Field(
1028            description=(
1029                "Number of lines to skip from the beginning of the logs. "
1030                "Cannot be combined with `from_tail=True`."
1031            ),
1032            default=None,
1033        ),
1034    ],
1035) -> LogReadResult:
1036    """Get the logs from a sync job attempt on Airbyte Cloud."""
1037    # Validate that line_offset and from_tail are not both set
1038    if line_offset is not None and from_tail:
1039        raise PyAirbyteInputError(
1040            message="Cannot specify both 'line_offset' and 'from_tail' parameters.",
1041            context={"line_offset": line_offset, "from_tail": from_tail},
1042        )
1043
1044    if from_tail is None and line_offset is None:
1045        from_tail = True
1046    workspace: CloudWorkspace = _get_cloud_workspace(workspace_id)
1047    connection = workspace.get_connection(connection_id=connection_id)
1048
1049    sync_result: cloud.SyncResult | None = connection.get_sync_result(job_id=job_id)
1050
1051    if not sync_result:
1052        raise AirbyteMissingResourceError(
1053            resource_type="sync job",
1054            resource_name_or_id=connection_id,
1055        )
1056
1057    attempts = sync_result.get_attempts()
1058
1059    if not attempts:
1060        raise AirbyteMissingResourceError(
1061            resource_type="sync attempt",
1062            resource_name_or_id=str(sync_result.job_id),
1063        )
1064
1065    if attempt_number is not None:
1066        target_attempt = None
1067        for attempt in attempts:
1068            if attempt.attempt_number == attempt_number:
1069                target_attempt = attempt
1070                break
1071
1072        if target_attempt is None:
1073            raise AirbyteMissingResourceError(
1074                resource_type="sync attempt",
1075                resource_name_or_id=f"job {sync_result.job_id}, attempt {attempt_number}",
1076            )
1077    else:
1078        target_attempt = max(attempts, key=lambda a: a.attempt_number)
1079
1080    logs = target_attempt.get_full_log_text()
1081
1082    if not logs:
1083        # Return empty result with zero lines
1084        return LogReadResult(
1085            log_text=(
1086                f"[No logs available for job '{sync_result.job_id}', "
1087                f"attempt {target_attempt.attempt_number}.]"
1088            ),
1089            log_text_start_line=1,
1090            log_text_line_count=0,
1091            total_log_lines_available=0,
1092            job_id=sync_result.job_id,
1093            attempt_number=target_attempt.attempt_number,
1094        )
1095
1096    # Apply line limiting
1097    log_lines = logs.splitlines()
1098    total_lines = len(log_lines)
1099
1100    # Determine effective max_lines (0 means no limit)
1101    effective_max = total_lines if max_lines == 0 else max_lines
1102
1103    # Calculate start_index and slice based on from_tail or line_offset
1104    if from_tail:
1105        start_index = max(0, total_lines - effective_max)
1106        selected_lines = log_lines[start_index:][:effective_max]
1107    else:
1108        start_index = line_offset or 0
1109        selected_lines = log_lines[start_index : start_index + effective_max]
1110
1111    return LogReadResult(
1112        log_text="\n".join(selected_lines),
1113        log_text_start_line=start_index + 1,  # Convert to 1-based index
1114        log_text_line_count=len(selected_lines),
1115        total_log_lines_available=total_lines,
1116        job_id=sync_result.job_id,
1117        attempt_number=target_attempt.attempt_number,
1118    )
1119
1120
1121@mcp_tool(
1122    domain="cloud",
1123    read_only=True,
1124    idempotent=True,
1125    open_world=True,
1126    extra_help_text=CLOUD_AUTH_TIP_TEXT,
1127)
1128def list_deployed_cloud_connections(
1129    *,
1130    workspace_id: Annotated[
1131        str | None,
1132        Field(
1133            description=WORKSPACE_ID_TIP_TEXT,
1134            default=None,
1135        ),
1136    ],
1137    name_contains: Annotated[
1138        str | None,
1139        Field(
1140            description="Optional case-insensitive substring to filter connections by name",
1141            default=None,
1142        ),
1143    ],
1144    max_items_limit: Annotated[
1145        int | None,
1146        Field(
1147            description="Optional maximum number of items to return (default: no limit)",
1148            default=None,
1149        ),
1150    ],
1151    with_connection_status: Annotated[
1152        bool | None,
1153        Field(
1154            description="If True, include status info for each connection's most recent sync job",
1155            default=False,
1156        ),
1157    ],
1158    failing_connections_only: Annotated[
1159        bool | None,
1160        Field(
1161            description="If True, only return connections with failed/cancelled last sync",
1162            default=False,
1163        ),
1164    ],
1165) -> list[CloudConnectionResult]:
1166    """List all deployed connections in the Airbyte Cloud workspace.
1167
1168    When with_connection_status is True, each connection result will include
1169    information about the most recent sync job status, skipping over any
1170    currently in-progress syncs to find the last completed job.
1171
1172    When failing_connections_only is True, only connections where the most
1173    recent completed sync job failed or was cancelled will be returned.
1174    This implicitly enables with_connection_status.
1175    """
1176    workspace: CloudWorkspace = _get_cloud_workspace(workspace_id)
1177    connections = workspace.list_connections()
1178
1179    # Filter by name if requested
1180    if name_contains:
1181        needle = name_contains.lower()
1182        connections = [c for c in connections if c.name is not None and needle in c.name.lower()]
1183
1184    # If failing_connections_only is True, implicitly enable with_connection_status
1185    if failing_connections_only:
1186        with_connection_status = True
1187
1188    results: list[CloudConnectionResult] = []
1189
1190    for connection in connections:
1191        last_job_status: str | None = None
1192        last_job_id: int | None = None
1193        last_job_time: str | None = None
1194        currently_running_job_id: int | None = None
1195        currently_running_job_start_time: str | None = None
1196
1197        if with_connection_status:
1198            sync_logs = connection.get_previous_sync_logs(limit=5)
1199            last_completed_job_status = None  # Keep enum for comparison
1200
1201            for sync_result in sync_logs:
1202                job_status = sync_result.get_job_status()
1203
1204                if not sync_result.is_job_complete():
1205                    currently_running_job_id = sync_result.job_id
1206                    currently_running_job_start_time = sync_result.start_time.isoformat()
1207                    continue
1208
1209                last_completed_job_status = job_status
1210                last_job_status = str(job_status.value) if job_status else None
1211                last_job_id = sync_result.job_id
1212                last_job_time = sync_result.start_time.isoformat()
1213                break
1214
1215            if failing_connections_only and (
1216                last_completed_job_status is None
1217                or last_completed_job_status not in FAILED_STATUSES
1218            ):
1219                continue
1220
1221        results.append(
1222            CloudConnectionResult(
1223                id=connection.connection_id,
1224                name=cast(str, connection.name),
1225                url=cast(str, connection.connection_url),
1226                source_id=connection.source_id,
1227                destination_id=connection.destination_id,
1228                last_job_status=last_job_status,
1229                last_job_id=last_job_id,
1230                last_job_time=last_job_time,
1231                currently_running_job_id=currently_running_job_id,
1232                currently_running_job_start_time=currently_running_job_start_time,
1233            )
1234        )
1235
1236        if max_items_limit is not None and len(results) >= max_items_limit:
1237            break
1238
1239    return results
1240
1241
1242def _resolve_organization(
1243    organization_id: str | None,
1244    organization_name: str | None,
1245    *,
1246    api_root: str,
1247    client_id: SecretString | None,
1248    client_secret: SecretString | None,
1249    bearer_token: SecretString | None = None,
1250) -> api_util.models.OrganizationResponse:
1251    """Resolve organization from either ID or exact name match.
1252
1253    Args:
1254        organization_id: The organization ID (if provided directly)
1255        organization_name: The organization name (exact match required)
1256        api_root: The API root URL
1257        client_id: OAuth client ID (optional if bearer_token is provided)
1258        client_secret: OAuth client secret (optional if bearer_token is provided)
1259        bearer_token: Bearer token for authentication (optional if client credentials provided)
1260
1261    Returns:
1262        The resolved OrganizationResponse object
1263
1264    Raises:
1265        PyAirbyteInputError: If neither or both parameters are provided,
1266            or if no organization matches the exact name
1267        AirbyteMissingResourceError: If the organization is not found
1268    """
1269    if organization_id and organization_name:
1270        raise PyAirbyteInputError(
1271            message="Provide either 'organization_id' or 'organization_name', not both."
1272        )
1273    if not organization_id and not organization_name:
1274        raise PyAirbyteInputError(
1275            message="Either 'organization_id' or 'organization_name' must be provided."
1276        )
1277
1278    # Get all organizations for the user
1279    orgs = api_util.list_organizations_for_user(
1280        api_root=api_root,
1281        client_id=client_id,
1282        client_secret=client_secret,
1283        bearer_token=bearer_token,
1284    )
1285
1286    if organization_id:
1287        # Find by ID
1288        matching_orgs = [org for org in orgs if org.organization_id == organization_id]
1289        if not matching_orgs:
1290            raise AirbyteMissingResourceError(
1291                resource_type="organization",
1292                context={
1293                    "organization_id": organization_id,
1294                    "message": f"No organization found with ID '{organization_id}' "
1295                    "for the current user.",
1296                },
1297            )
1298        return matching_orgs[0]
1299
1300    # Find by exact name match (case-sensitive)
1301    matching_orgs = [org for org in orgs if org.organization_name == organization_name]
1302
1303    if not matching_orgs:
1304        raise AirbyteMissingResourceError(
1305            resource_type="organization",
1306            context={
1307                "organization_name": organization_name,
1308                "message": f"No organization found with exact name '{organization_name}' "
1309                "for the current user.",
1310            },
1311        )
1312
1313    if len(matching_orgs) > 1:
1314        raise PyAirbyteInputError(
1315            message=f"Multiple organizations found with name '{organization_name}'. "
1316            "Please use 'organization_id' instead to specify the exact organization."
1317        )
1318
1319    return matching_orgs[0]
1320
1321
1322def _resolve_organization_id(
1323    organization_id: str | None,
1324    organization_name: str | None,
1325    *,
1326    api_root: str,
1327    client_id: SecretString | None,
1328    client_secret: SecretString | None,
1329    bearer_token: SecretString | None = None,
1330) -> str:
1331    """Resolve organization ID from either ID or exact name match.
1332
1333    This is a convenience wrapper around _resolve_organization that returns just the ID.
1334    """
1335    org = _resolve_organization(
1336        organization_id=organization_id,
1337        organization_name=organization_name,
1338        api_root=api_root,
1339        client_id=client_id,
1340        client_secret=client_secret,
1341        bearer_token=bearer_token,
1342    )
1343    return org.organization_id
1344
1345
1346@mcp_tool(
1347    domain="cloud",
1348    read_only=True,
1349    idempotent=True,
1350    open_world=True,
1351    extra_help_text=CLOUD_AUTH_TIP_TEXT,
1352)
1353def list_cloud_workspaces(
1354    *,
1355    organization_id: Annotated[
1356        str | None,
1357        Field(
1358            description="Organization ID. Required if organization_name is not provided.",
1359            default=None,
1360        ),
1361    ],
1362    organization_name: Annotated[
1363        str | None,
1364        Field(
1365            description=(
1366                "Organization name (exact match). " "Required if organization_id is not provided."
1367            ),
1368            default=None,
1369        ),
1370    ],
1371    name_contains: Annotated[
1372        str | None,
1373        Field(
1374            description="Optional substring to filter workspaces by name (server-side filtering)",
1375            default=None,
1376        ),
1377    ],
1378    max_items_limit: Annotated[
1379        int | None,
1380        Field(
1381            description="Optional maximum number of items to return (default: no limit)",
1382            default=None,
1383        ),
1384    ],
1385) -> list[CloudWorkspaceResult]:
1386    """List all workspaces in a specific organization.
1387
1388    Requires either organization_id OR organization_name (exact match) to be provided.
1389    This tool will NOT list workspaces across all organizations - you must specify
1390    which organization to list workspaces from.
1391    """
1392    credentials = resolve_cloud_credentials()
1393
1394    resolved_org_id = _resolve_organization_id(
1395        organization_id=organization_id,
1396        organization_name=organization_name,
1397        api_root=credentials.api_root,
1398        client_id=credentials.client_id,
1399        client_secret=credentials.client_secret,
1400        bearer_token=credentials.bearer_token,
1401    )
1402
1403    workspaces = api_util.list_workspaces_in_organization(
1404        organization_id=resolved_org_id,
1405        api_root=credentials.api_root,
1406        client_id=credentials.client_id,
1407        client_secret=credentials.client_secret,
1408        bearer_token=credentials.bearer_token,
1409        name_contains=name_contains,
1410        max_items_limit=max_items_limit,
1411    )
1412
1413    return [
1414        CloudWorkspaceResult(
1415            workspace_id=ws.get("workspaceId", ""),
1416            workspace_name=ws.get("name", ""),
1417            organization_id=ws.get("organizationId", ""),
1418        )
1419        for ws in workspaces
1420    ]
1421
1422
1423@mcp_tool(
1424    domain="cloud",
1425    read_only=True,
1426    idempotent=True,
1427    open_world=True,
1428    extra_help_text=CLOUD_AUTH_TIP_TEXT,
1429)
1430def describe_cloud_organization(
1431    *,
1432    organization_id: Annotated[
1433        str | None,
1434        Field(
1435            description="Organization ID. Required if organization_name is not provided.",
1436            default=None,
1437        ),
1438    ],
1439    organization_name: Annotated[
1440        str | None,
1441        Field(
1442            description=(
1443                "Organization name (exact match). " "Required if organization_id is not provided."
1444            ),
1445            default=None,
1446        ),
1447    ],
1448) -> CloudOrganizationResult:
1449    """Get details about a specific organization.
1450
1451    Requires either organization_id OR organization_name (exact match) to be provided.
1452    This tool is useful for looking up an organization's ID from its name, or vice versa.
1453    """
1454    credentials = resolve_cloud_credentials()
1455
1456    org = _resolve_organization(
1457        organization_id=organization_id,
1458        organization_name=organization_name,
1459        api_root=credentials.api_root,
1460        client_id=credentials.client_id,
1461        client_secret=credentials.client_secret,
1462        bearer_token=credentials.bearer_token,
1463    )
1464
1465    return CloudOrganizationResult(
1466        id=org.organization_id,
1467        name=org.organization_name,
1468        email=org.email,
1469    )
1470
1471
1472def _get_custom_source_definition_description(
1473    custom_source: CustomCloudSourceDefinition,
1474) -> str:
1475    return "\n".join(
1476        [
1477            f" - Custom Source Name: {custom_source.name}",
1478            f" - Definition ID: {custom_source.definition_id}",
1479            f" - Definition Version: {custom_source.version}",
1480            f" - Connector Builder Project ID: {custom_source.connector_builder_project_id}",
1481            f" - Connector Builder Project URL: {custom_source.connector_builder_project_url}",
1482        ]
1483    )
1484
1485
1486@mcp_tool(
1487    domain="cloud",
1488    open_world=True,
1489    extra_help_text=CLOUD_AUTH_TIP_TEXT,
1490)
1491def publish_custom_source_definition(
1492    name: Annotated[
1493        str,
1494        Field(description="The name for the custom connector definition."),
1495    ],
1496    *,
1497    workspace_id: Annotated[
1498        str | None,
1499        Field(
1500            description=WORKSPACE_ID_TIP_TEXT,
1501            default=None,
1502        ),
1503    ],
1504    manifest_yaml: Annotated[
1505        str | Path | None,
1506        Field(
1507            description=(
1508                "The Low-code CDK manifest as a YAML string or file path. "
1509                "Required for YAML connectors."
1510            ),
1511            default=None,
1512        ),
1513    ] = None,
1514    unique: Annotated[
1515        bool,
1516        Field(
1517            description="Whether to require a unique name.",
1518            default=True,
1519        ),
1520    ] = True,
1521    pre_validate: Annotated[
1522        bool,
1523        Field(
1524            description="Whether to validate the manifest client-side before publishing.",
1525            default=True,
1526        ),
1527    ] = True,
1528    testing_values: Annotated[
1529        dict | str | None,
1530        Field(
1531            description=(
1532                "Optional testing configuration values for the Builder UI. "
1533                "Can be provided as a JSON object or JSON string. "
1534                "Supports inline secret refs via 'secret_reference::ENV_VAR_NAME' syntax. "
1535                "If provided, these values replace any existing testing values "
1536                "for the connector builder project, allowing immediate test read operations."
1537            ),
1538            default=None,
1539        ),
1540    ],
1541    testing_values_secret_name: Annotated[
1542        str | None,
1543        Field(
1544            description=(
1545                "Optional name of a secret containing testing configuration values "
1546                "in JSON or YAML format. The secret will be resolved by the MCP "
1547                "server and merged into testing_values, with secret values taking "
1548                "precedence. This lets the agent reference secrets without sending "
1549                "raw values as tool arguments."
1550            ),
1551            default=None,
1552        ),
1553    ],
1554) -> str:
1555    """Publish a custom YAML source connector definition to Airbyte Cloud.
1556
1557    Note: Only YAML (declarative) connectors are currently supported.
1558    Docker-based custom sources are not yet available.
1559    """
1560    processed_manifest = manifest_yaml
1561    if isinstance(manifest_yaml, str) and "\n" not in manifest_yaml:
1562        processed_manifest = Path(manifest_yaml)
1563
1564    # Resolve testing values from inline config and/or secret
1565    testing_values_dict: dict[str, Any] | None = None
1566    if testing_values is not None or testing_values_secret_name is not None:
1567        testing_values_dict = (
1568            resolve_config(
1569                config=testing_values,
1570                config_secret_name=testing_values_secret_name,
1571            )
1572            or None
1573        )
1574
1575    workspace: CloudWorkspace = _get_cloud_workspace(workspace_id)
1576    custom_source = workspace.publish_custom_source_definition(
1577        name=name,
1578        manifest_yaml=processed_manifest,
1579        unique=unique,
1580        pre_validate=pre_validate,
1581        testing_values=testing_values_dict,
1582    )
1583    register_guid_created_in_session(custom_source.definition_id)
1584    return (
1585        "Successfully published custom YAML source definition:\n"
1586        + _get_custom_source_definition_description(
1587            custom_source=custom_source,
1588        )
1589        + "\n"
1590    )
1591
1592
1593@mcp_tool(
1594    domain="cloud",
1595    read_only=True,
1596    idempotent=True,
1597    open_world=True,
1598)
1599def list_custom_source_definitions(
1600    *,
1601    workspace_id: Annotated[
1602        str | None,
1603        Field(
1604            description=WORKSPACE_ID_TIP_TEXT,
1605            default=None,
1606        ),
1607    ],
1608) -> list[dict[str, Any]]:
1609    """List custom YAML source definitions in the Airbyte Cloud workspace.
1610
1611    Note: Only YAML (declarative) connectors are currently supported.
1612    Docker-based custom sources are not yet available.
1613    """
1614    workspace: CloudWorkspace = _get_cloud_workspace(workspace_id)
1615    definitions = workspace.list_custom_source_definitions(
1616        definition_type="yaml",
1617    )
1618
1619    return [
1620        {
1621            "definition_id": d.definition_id,
1622            "name": d.name,
1623            "version": d.version,
1624            "connector_builder_project_url": d.connector_builder_project_url,
1625        }
1626        for d in definitions
1627    ]
1628
1629
1630@mcp_tool(
1631    domain="cloud",
1632    read_only=True,
1633    idempotent=True,
1634    open_world=True,
1635)
1636def get_custom_source_definition(
1637    definition_id: Annotated[
1638        str,
1639        Field(description="The ID of the custom source definition to retrieve."),
1640    ],
1641    *,
1642    workspace_id: Annotated[
1643        str | None,
1644        Field(
1645            description=WORKSPACE_ID_TIP_TEXT,
1646            default=None,
1647        ),
1648    ],
1649) -> dict[str, Any]:
1650    """Get a custom YAML source definition from Airbyte Cloud, including its manifest.
1651
1652    Returns the full definition details including the manifest YAML content,
1653    which can be used to inspect or store the connector configuration locally.
1654
1655    Note: Only YAML (declarative) connectors are currently supported.
1656    Docker-based custom sources are not yet available.
1657    """
1658    workspace: CloudWorkspace = _get_cloud_workspace(workspace_id)
1659    definition = workspace.get_custom_source_definition(
1660        definition_id=definition_id,
1661        definition_type="yaml",
1662    )
1663
1664    return {
1665        "definition_id": definition.definition_id,
1666        "name": definition.name,
1667        "version": definition.version,
1668        "connector_builder_project_id": definition.connector_builder_project_id,
1669        "connector_builder_project_url": definition.connector_builder_project_url,
1670        "manifest": definition.manifest,
1671    }
1672
1673
1674@mcp_tool(
1675    domain="cloud",
1676    destructive=True,
1677    open_world=True,
1678)
1679def update_custom_source_definition(
1680    definition_id: Annotated[
1681        str,
1682        Field(description="The ID of the definition to update."),
1683    ],
1684    manifest_yaml: Annotated[
1685        str | Path | None,
1686        Field(
1687            description=(
1688                "New manifest as YAML string or file path. "
1689                "Optional; omit to update only testing values."
1690            ),
1691            default=None,
1692        ),
1693    ] = None,
1694    *,
1695    workspace_id: Annotated[
1696        str | None,
1697        Field(
1698            description=WORKSPACE_ID_TIP_TEXT,
1699            default=None,
1700        ),
1701    ],
1702    pre_validate: Annotated[
1703        bool,
1704        Field(
1705            description="Whether to validate the manifest client-side before updating.",
1706            default=True,
1707        ),
1708    ] = True,
1709    testing_values: Annotated[
1710        dict | str | None,
1711        Field(
1712            description=(
1713                "Optional testing configuration values for the Builder UI. "
1714                "Can be provided as a JSON object or JSON string. "
1715                "Supports inline secret refs via 'secret_reference::ENV_VAR_NAME' syntax. "
1716                "If provided, these values replace any existing testing values "
1717                "for the connector builder project. The entire testing values object "
1718                "is overwritten, so pass the full set of values you want to persist."
1719            ),
1720            default=None,
1721        ),
1722    ],
1723    testing_values_secret_name: Annotated[
1724        str | None,
1725        Field(
1726            description=(
1727                "Optional name of a secret containing testing configuration values "
1728                "in JSON or YAML format. The secret will be resolved by the MCP "
1729                "server and merged into testing_values, with secret values taking "
1730                "precedence. This lets the agent reference secrets without sending "
1731                "raw values as tool arguments."
1732            ),
1733            default=None,
1734        ),
1735    ],
1736) -> str:
1737    """Update a custom YAML source definition in Airbyte Cloud.
1738
1739    Updates the manifest and/or testing values for an existing custom source definition.
1740    At least one of manifest_yaml, testing_values, or testing_values_secret_name must be provided.
1741    """
1742    check_guid_created_in_session(definition_id)
1743
1744    workspace: CloudWorkspace = _get_cloud_workspace(workspace_id)
1745
1746    if manifest_yaml is None and testing_values is None and testing_values_secret_name is None:
1747        raise PyAirbyteInputError(
1748            message=(
1749                "At least one of manifest_yaml, testing_values, or testing_values_secret_name "
1750                "must be provided to update a custom source definition."
1751            ),
1752            context={
1753                "definition_id": definition_id,
1754                "workspace_id": workspace.workspace_id,
1755            },
1756        )
1757
1758    processed_manifest: str | Path | None = manifest_yaml
1759    if isinstance(manifest_yaml, str) and "\n" not in manifest_yaml:
1760        processed_manifest = Path(manifest_yaml)
1761
1762    # Resolve testing values from inline config and/or secret
1763    testing_values_dict: dict[str, Any] | None = None
1764    if testing_values is not None or testing_values_secret_name is not None:
1765        testing_values_dict = (
1766            resolve_config(
1767                config=testing_values,
1768                config_secret_name=testing_values_secret_name,
1769            )
1770            or None
1771        )
1772
1773    definition = workspace.get_custom_source_definition(
1774        definition_id=definition_id,
1775        definition_type="yaml",
1776    )
1777    custom_source: CustomCloudSourceDefinition = definition
1778
1779    if processed_manifest is not None:
1780        custom_source = definition.update_definition(
1781            manifest_yaml=processed_manifest,
1782            pre_validate=pre_validate,
1783        )
1784
1785    if testing_values_dict is not None:
1786        custom_source.set_testing_values(testing_values_dict)
1787
1788    return (
1789        "Successfully updated custom YAML source definition:\n"
1790        + _get_custom_source_definition_description(
1791            custom_source=custom_source,
1792        )
1793    )
1794
1795
1796@mcp_tool(
1797    domain="cloud",
1798    destructive=True,
1799    open_world=True,
1800)
1801def permanently_delete_custom_source_definition(
1802    definition_id: Annotated[
1803        str,
1804        Field(description="The ID of the custom source definition to delete."),
1805    ],
1806    name: Annotated[
1807        str,
1808        Field(description="The expected name of the custom source definition (for verification)."),
1809    ],
1810    *,
1811    workspace_id: Annotated[
1812        str | None,
1813        Field(
1814            description=WORKSPACE_ID_TIP_TEXT,
1815            default=None,
1816        ),
1817    ],
1818) -> str:
1819    """Permanently delete a custom YAML source definition from Airbyte Cloud.
1820
1821    IMPORTANT: This operation requires the connector name to contain "delete-me" or "deleteme"
1822    (case insensitive).
1823
1824    If the connector does not meet this requirement, the deletion will be rejected with a
1825    helpful error message. Instruct the user to rename the connector appropriately to authorize
1826    the deletion.
1827
1828    The provided name must match the actual name of the definition for the operation to proceed.
1829    This is a safety measure to ensure you are deleting the correct resource.
1830
1831    Note: Only YAML (declarative) connectors are currently supported.
1832    Docker-based custom sources are not yet available.
1833    """
1834    check_guid_created_in_session(definition_id)
1835    workspace: CloudWorkspace = _get_cloud_workspace(workspace_id)
1836    definition = workspace.get_custom_source_definition(
1837        definition_id=definition_id,
1838        definition_type="yaml",
1839    )
1840    actual_name: str = definition.name
1841
1842    # Verify the name matches
1843    if actual_name != name:
1844        raise PyAirbyteInputError(
1845            message=(
1846                f"Name mismatch: expected '{name}' but found '{actual_name}'. "
1847                "The provided name must exactly match the definition's actual name. "
1848                "This is a safety measure to prevent accidental deletion."
1849            ),
1850            context={
1851                "definition_id": definition_id,
1852                "expected_name": name,
1853                "actual_name": actual_name,
1854            },
1855        )
1856
1857    definition.permanently_delete(
1858        safe_mode=True,  # Hard-coded safe mode for extra protection when running in LLM agents.
1859    )
1860    return f"Successfully deleted custom source definition '{actual_name}' (ID: {definition_id})"
1861
1862
1863@mcp_tool(
1864    domain="cloud",
1865    destructive=True,
1866    open_world=True,
1867    extra_help_text=CLOUD_AUTH_TIP_TEXT,
1868)
1869def permanently_delete_cloud_source(
1870    source_id: Annotated[
1871        str,
1872        Field(description="The ID of the deployed source to delete."),
1873    ],
1874    name: Annotated[
1875        str,
1876        Field(description="The expected name of the source (for verification)."),
1877    ],
1878) -> str:
1879    """Permanently delete a deployed source connector from Airbyte Cloud.
1880
1881    IMPORTANT: This operation requires the source name to contain "delete-me" or "deleteme"
1882    (case insensitive).
1883
1884    If the source does not meet this requirement, the deletion will be rejected with a
1885    helpful error message. Instruct the user to rename the source appropriately to authorize
1886    the deletion.
1887
1888    The provided name must match the actual name of the source for the operation to proceed.
1889    This is a safety measure to ensure you are deleting the correct resource.
1890    """
1891    check_guid_created_in_session(source_id)
1892    workspace: CloudWorkspace = _get_cloud_workspace()
1893    source = workspace.get_source(source_id=source_id)
1894    actual_name: str = cast(str, source.name)
1895
1896    # Verify the name matches
1897    if actual_name != name:
1898        raise PyAirbyteInputError(
1899            message=(
1900                f"Name mismatch: expected '{name}' but found '{actual_name}'. "
1901                "The provided name must exactly match the source's actual name. "
1902                "This is a safety measure to prevent accidental deletion."
1903            ),
1904            context={
1905                "source_id": source_id,
1906                "expected_name": name,
1907                "actual_name": actual_name,
1908            },
1909        )
1910
1911    # Safe mode is hard-coded to True for extra protection when running in LLM agents
1912    workspace.permanently_delete_source(
1913        source=source_id,
1914        safe_mode=True,  # Requires name to contain "delete-me" or "deleteme" (case insensitive)
1915    )
1916    return f"Successfully deleted source '{actual_name}' (ID: {source_id})"
1917
1918
1919@mcp_tool(
1920    domain="cloud",
1921    destructive=True,
1922    open_world=True,
1923    extra_help_text=CLOUD_AUTH_TIP_TEXT,
1924)
1925def permanently_delete_cloud_destination(
1926    destination_id: Annotated[
1927        str,
1928        Field(description="The ID of the deployed destination to delete."),
1929    ],
1930    name: Annotated[
1931        str,
1932        Field(description="The expected name of the destination (for verification)."),
1933    ],
1934) -> str:
1935    """Permanently delete a deployed destination connector from Airbyte Cloud.
1936
1937    IMPORTANT: This operation requires the destination name to contain "delete-me" or "deleteme"
1938    (case insensitive).
1939
1940    If the destination does not meet this requirement, the deletion will be rejected with a
1941    helpful error message. Instruct the user to rename the destination appropriately to authorize
1942    the deletion.
1943
1944    The provided name must match the actual name of the destination for the operation to proceed.
1945    This is a safety measure to ensure you are deleting the correct resource.
1946    """
1947    check_guid_created_in_session(destination_id)
1948    workspace: CloudWorkspace = _get_cloud_workspace()
1949    destination = workspace.get_destination(destination_id=destination_id)
1950    actual_name: str = cast(str, destination.name)
1951
1952    # Verify the name matches
1953    if actual_name != name:
1954        raise PyAirbyteInputError(
1955            message=(
1956                f"Name mismatch: expected '{name}' but found '{actual_name}'. "
1957                "The provided name must exactly match the destination's actual name. "
1958                "This is a safety measure to prevent accidental deletion."
1959            ),
1960            context={
1961                "destination_id": destination_id,
1962                "expected_name": name,
1963                "actual_name": actual_name,
1964            },
1965        )
1966
1967    # Safe mode is hard-coded to True for extra protection when running in LLM agents
1968    workspace.permanently_delete_destination(
1969        destination=destination_id,
1970        safe_mode=True,  # Requires name-based delete disposition ("delete-me" or "deleteme")
1971    )
1972    return f"Successfully deleted destination '{actual_name}' (ID: {destination_id})"
1973
1974
1975@mcp_tool(
1976    domain="cloud",
1977    destructive=True,
1978    open_world=True,
1979    extra_help_text=CLOUD_AUTH_TIP_TEXT,
1980)
1981def permanently_delete_cloud_connection(
1982    connection_id: Annotated[
1983        str,
1984        Field(description="The ID of the connection to delete."),
1985    ],
1986    name: Annotated[
1987        str,
1988        Field(description="The expected name of the connection (for verification)."),
1989    ],
1990    *,
1991    cascade_delete_source: Annotated[
1992        bool,
1993        Field(
1994            description=(
1995                "Whether to also delete the source connector associated with this connection."
1996            ),
1997            default=False,
1998        ),
1999    ] = False,
2000    cascade_delete_destination: Annotated[
2001        bool,
2002        Field(
2003            description=(
2004                "Whether to also delete the destination connector associated with this connection."
2005            ),
2006            default=False,
2007        ),
2008    ] = False,
2009) -> str:
2010    """Permanently delete a connection from Airbyte Cloud.
2011
2012    IMPORTANT: This operation requires the connection name to contain "delete-me" or "deleteme"
2013    (case insensitive).
2014
2015    If the connection does not meet this requirement, the deletion will be rejected with a
2016    helpful error message. Instruct the user to rename the connection appropriately to authorize
2017    the deletion.
2018
2019    The provided name must match the actual name of the connection for the operation to proceed.
2020    This is a safety measure to ensure you are deleting the correct resource.
2021    """
2022    check_guid_created_in_session(connection_id)
2023    workspace: CloudWorkspace = _get_cloud_workspace()
2024    connection = workspace.get_connection(connection_id=connection_id)
2025    actual_name: str = cast(str, connection.name)
2026
2027    # Verify the name matches
2028    if actual_name != name:
2029        raise PyAirbyteInputError(
2030            message=(
2031                f"Name mismatch: expected '{name}' but found '{actual_name}'. "
2032                "The provided name must exactly match the connection's actual name. "
2033                "This is a safety measure to prevent accidental deletion."
2034            ),
2035            context={
2036                "connection_id": connection_id,
2037                "expected_name": name,
2038                "actual_name": actual_name,
2039            },
2040        )
2041
2042    # Safe mode is hard-coded to True for extra protection when running in LLM agents
2043    workspace.permanently_delete_connection(
2044        safe_mode=True,  # Requires name-based delete disposition ("delete-me" or "deleteme")
2045        connection=connection_id,
2046        cascade_delete_source=cascade_delete_source,
2047        cascade_delete_destination=cascade_delete_destination,
2048    )
2049    return f"Successfully deleted connection '{actual_name}' (ID: {connection_id})"
2050
2051
2052@mcp_tool(
2053    domain="cloud",
2054    open_world=True,
2055    extra_help_text=CLOUD_AUTH_TIP_TEXT,
2056)
2057def rename_cloud_source(
2058    source_id: Annotated[
2059        str,
2060        Field(description="The ID of the deployed source to rename."),
2061    ],
2062    name: Annotated[
2063        str,
2064        Field(description="New name for the source."),
2065    ],
2066    *,
2067    workspace_id: Annotated[
2068        str | None,
2069        Field(
2070            description=WORKSPACE_ID_TIP_TEXT,
2071            default=None,
2072        ),
2073    ],
2074) -> str:
2075    """Rename a deployed source connector on Airbyte Cloud."""
2076    workspace: CloudWorkspace = _get_cloud_workspace(workspace_id)
2077    source = workspace.get_source(source_id=source_id)
2078    source.rename(name=name)
2079    return f"Successfully renamed source '{source_id}' to '{name}'. URL: {source.connector_url}"
2080
2081
2082@mcp_tool(
2083    domain="cloud",
2084    destructive=True,
2085    open_world=True,
2086    extra_help_text=CLOUD_AUTH_TIP_TEXT,
2087)
2088def update_cloud_source_config(
2089    source_id: Annotated[
2090        str,
2091        Field(description="The ID of the deployed source to update."),
2092    ],
2093    config: Annotated[
2094        dict | str,
2095        Field(
2096            description="New configuration for the source connector.",
2097        ),
2098    ],
2099    config_secret_name: Annotated[
2100        str | None,
2101        Field(
2102            description="The name of the secret containing the configuration.",
2103            default=None,
2104        ),
2105    ] = None,
2106    *,
2107    workspace_id: Annotated[
2108        str | None,
2109        Field(
2110            description=WORKSPACE_ID_TIP_TEXT,
2111            default=None,
2112        ),
2113    ],
2114) -> str:
2115    """Update a deployed source connector's configuration on Airbyte Cloud.
2116
2117    This is a destructive operation that can break existing connections if the
2118    configuration is changed incorrectly. Use with caution.
2119    """
2120    check_guid_created_in_session(source_id)
2121    workspace: CloudWorkspace = _get_cloud_workspace(workspace_id)
2122    source = workspace.get_source(source_id=source_id)
2123
2124    config_dict = resolve_config(
2125        config=config,
2126        config_secret_name=config_secret_name,
2127        config_spec_jsonschema=None,  # We don't have the spec here
2128    )
2129
2130    source.update_config(config=config_dict)
2131    return f"Successfully updated source '{source_id}'. URL: {source.connector_url}"
2132
2133
2134@mcp_tool(
2135    domain="cloud",
2136    open_world=True,
2137    extra_help_text=CLOUD_AUTH_TIP_TEXT,
2138)
2139def rename_cloud_destination(
2140    destination_id: Annotated[
2141        str,
2142        Field(description="The ID of the deployed destination to rename."),
2143    ],
2144    name: Annotated[
2145        str,
2146        Field(description="New name for the destination."),
2147    ],
2148    *,
2149    workspace_id: Annotated[
2150        str | None,
2151        Field(
2152            description=WORKSPACE_ID_TIP_TEXT,
2153            default=None,
2154        ),
2155    ],
2156) -> str:
2157    """Rename a deployed destination connector on Airbyte Cloud."""
2158    workspace: CloudWorkspace = _get_cloud_workspace(workspace_id)
2159    destination = workspace.get_destination(destination_id=destination_id)
2160    destination.rename(name=name)
2161    return (
2162        f"Successfully renamed destination '{destination_id}' to '{name}'. "
2163        f"URL: {destination.connector_url}"
2164    )
2165
2166
2167@mcp_tool(
2168    domain="cloud",
2169    destructive=True,
2170    open_world=True,
2171    extra_help_text=CLOUD_AUTH_TIP_TEXT,
2172)
2173def update_cloud_destination_config(
2174    destination_id: Annotated[
2175        str,
2176        Field(description="The ID of the deployed destination to update."),
2177    ],
2178    config: Annotated[
2179        dict | str,
2180        Field(
2181            description="New configuration for the destination connector.",
2182        ),
2183    ],
2184    config_secret_name: Annotated[
2185        str | None,
2186        Field(
2187            description="The name of the secret containing the configuration.",
2188            default=None,
2189        ),
2190    ],
2191    *,
2192    workspace_id: Annotated[
2193        str | None,
2194        Field(
2195            description=WORKSPACE_ID_TIP_TEXT,
2196            default=None,
2197        ),
2198    ],
2199) -> str:
2200    """Update a deployed destination connector's configuration on Airbyte Cloud.
2201
2202    This is a destructive operation that can break existing connections if the
2203    configuration is changed incorrectly. Use with caution.
2204    """
2205    check_guid_created_in_session(destination_id)
2206    workspace: CloudWorkspace = _get_cloud_workspace(workspace_id)
2207    destination = workspace.get_destination(destination_id=destination_id)
2208
2209    config_dict = resolve_config(
2210        config=config,
2211        config_secret_name=config_secret_name,
2212        config_spec_jsonschema=None,  # We don't have the spec here
2213    )
2214
2215    destination.update_config(config=config_dict)
2216    return (
2217        f"Successfully updated destination '{destination_id}'. " f"URL: {destination.connector_url}"
2218    )
2219
2220
2221@mcp_tool(
2222    domain="cloud",
2223    open_world=True,
2224    extra_help_text=CLOUD_AUTH_TIP_TEXT,
2225)
2226def rename_cloud_connection(
2227    connection_id: Annotated[
2228        str,
2229        Field(description="The ID of the connection to rename."),
2230    ],
2231    name: Annotated[
2232        str,
2233        Field(description="New name for the connection."),
2234    ],
2235    *,
2236    workspace_id: Annotated[
2237        str | None,
2238        Field(
2239            description=WORKSPACE_ID_TIP_TEXT,
2240            default=None,
2241        ),
2242    ],
2243) -> str:
2244    """Rename a connection on Airbyte Cloud."""
2245    workspace: CloudWorkspace = _get_cloud_workspace(workspace_id)
2246    connection = workspace.get_connection(connection_id=connection_id)
2247    connection.rename(name=name)
2248    return (
2249        f"Successfully renamed connection '{connection_id}' to '{name}'. "
2250        f"URL: {connection.connection_url}"
2251    )
2252
2253
2254@mcp_tool(
2255    domain="cloud",
2256    destructive=True,
2257    open_world=True,
2258    extra_help_text=CLOUD_AUTH_TIP_TEXT,
2259)
2260def set_cloud_connection_table_prefix(
2261    connection_id: Annotated[
2262        str,
2263        Field(description="The ID of the connection to update."),
2264    ],
2265    prefix: Annotated[
2266        str,
2267        Field(description="New table prefix to use when syncing to the destination."),
2268    ],
2269    *,
2270    workspace_id: Annotated[
2271        str | None,
2272        Field(
2273            description=WORKSPACE_ID_TIP_TEXT,
2274            default=None,
2275        ),
2276    ],
2277) -> str:
2278    """Set the table prefix for a connection on Airbyte Cloud.
2279
2280    This is a destructive operation that can break downstream dependencies if the
2281    table prefix is changed incorrectly. Use with caution.
2282    """
2283    check_guid_created_in_session(connection_id)
2284    workspace: CloudWorkspace = _get_cloud_workspace(workspace_id)
2285    connection = workspace.get_connection(connection_id=connection_id)
2286    connection.set_table_prefix(prefix=prefix)
2287    return (
2288        f"Successfully set table prefix for connection '{connection_id}' to '{prefix}'. "
2289        f"URL: {connection.connection_url}"
2290    )
2291
2292
2293@mcp_tool(
2294    domain="cloud",
2295    destructive=True,
2296    open_world=True,
2297    extra_help_text=CLOUD_AUTH_TIP_TEXT,
2298)
2299def set_cloud_connection_selected_streams(
2300    connection_id: Annotated[
2301        str,
2302        Field(description="The ID of the connection to update."),
2303    ],
2304    stream_names: Annotated[
2305        str | list[str],
2306        Field(
2307            description=(
2308                "The selected stream names to sync within the connection. "
2309                "Must be an explicit stream name or list of streams."
2310            )
2311        ),
2312    ],
2313    *,
2314    workspace_id: Annotated[
2315        str | None,
2316        Field(
2317            description=WORKSPACE_ID_TIP_TEXT,
2318            default=None,
2319        ),
2320    ],
2321) -> str:
2322    """Set the selected streams for a connection on Airbyte Cloud.
2323
2324    This is a destructive operation that can break existing connections if the
2325    stream selection is changed incorrectly. Use with caution.
2326    """
2327    check_guid_created_in_session(connection_id)
2328    workspace: CloudWorkspace = _get_cloud_workspace(workspace_id)
2329    connection = workspace.get_connection(connection_id=connection_id)
2330
2331    resolved_streams_list: list[str] = resolve_list_of_strings(stream_names)
2332    connection.set_selected_streams(stream_names=resolved_streams_list)
2333
2334    return (
2335        f"Successfully set selected streams for connection '{connection_id}' "
2336        f"to {resolved_streams_list}. URL: {connection.connection_url}"
2337    )
2338
2339
2340@mcp_tool(
2341    domain="cloud",
2342    open_world=True,
2343    destructive=True,
2344    extra_help_text=CLOUD_AUTH_TIP_TEXT,
2345)
2346def update_cloud_connection(
2347    connection_id: Annotated[
2348        str,
2349        Field(description="The ID of the connection to update."),
2350    ],
2351    *,
2352    enabled: Annotated[
2353        bool | None,
2354        Field(
2355            description=(
2356                "Set the connection's enabled status. "
2357                "True enables the connection (status='active'), "
2358                "False disables it (status='inactive'). "
2359                "Leave unset to keep the current status."
2360            ),
2361            default=None,
2362        ),
2363    ],
2364    cron_expression: Annotated[
2365        str | None,
2366        Field(
2367            description=(
2368                "A cron expression defining when syncs should run. "
2369                "Examples: '0 0 * * *' (daily at midnight UTC), "
2370                "'0 */6 * * *' (every 6 hours), "
2371                "'0 0 * * 0' (weekly on Sunday at midnight UTC). "
2372                "Leave unset to keep the current schedule. "
2373                "Cannot be used together with 'manual_schedule'."
2374            ),
2375            default=None,
2376        ),
2377    ],
2378    manual_schedule: Annotated[
2379        bool | None,
2380        Field(
2381            description=(
2382                "Set to True to disable automatic syncs (manual scheduling only). "
2383                "Syncs will only run when manually triggered. "
2384                "Cannot be used together with 'cron_expression'."
2385            ),
2386            default=None,
2387        ),
2388    ],
2389    workspace_id: Annotated[
2390        str | None,
2391        Field(
2392            description=WORKSPACE_ID_TIP_TEXT,
2393            default=None,
2394        ),
2395    ],
2396) -> str:
2397    """Update a connection's settings on Airbyte Cloud.
2398
2399    This tool allows updating multiple connection settings in a single call:
2400    - Enable or disable the connection
2401    - Set a cron schedule for automatic syncs
2402    - Switch to manual scheduling (no automatic syncs)
2403
2404    At least one setting must be provided. The 'cron_expression' and 'manual_schedule'
2405    parameters are mutually exclusive.
2406    """
2407    check_guid_created_in_session(connection_id)
2408
2409    # Validate that at least one setting is provided
2410    if enabled is None and cron_expression is None and manual_schedule is None:
2411        raise ValueError(
2412            "At least one setting must be provided: 'enabled', 'cron_expression', "
2413            "or 'manual_schedule'."
2414        )
2415
2416    # Validate mutually exclusive schedule options
2417    if cron_expression is not None and manual_schedule is True:
2418        raise ValueError(
2419            "Cannot specify both 'cron_expression' and 'manual_schedule=True'. "
2420            "Use 'cron_expression' for scheduled syncs or 'manual_schedule=True' "
2421            "for manual-only syncs."
2422        )
2423
2424    workspace: CloudWorkspace = _get_cloud_workspace(workspace_id)
2425    connection = workspace.get_connection(connection_id=connection_id)
2426
2427    changes_made: list[str] = []
2428
2429    # Apply enabled status change
2430    if enabled is not None:
2431        connection.set_enabled(enabled=enabled)
2432        status_str = "enabled" if enabled else "disabled"
2433        changes_made.append(f"status set to '{status_str}'")
2434
2435    # Apply schedule change
2436    if cron_expression is not None:
2437        connection.set_schedule(cron_expression=cron_expression)
2438        changes_made.append(f"schedule set to '{cron_expression}'")
2439    elif manual_schedule is True:
2440        connection.set_manual_schedule()
2441        changes_made.append("schedule set to 'manual'")
2442
2443    changes_summary = ", ".join(changes_made)
2444    return (
2445        f"Successfully updated connection '{connection_id}': {changes_summary}. "
2446        f"URL: {connection.connection_url}"
2447    )
2448
2449
2450@mcp_tool(
2451    domain="cloud",
2452    read_only=True,
2453    idempotent=True,
2454    open_world=True,
2455    extra_help_text=CLOUD_AUTH_TIP_TEXT,
2456)
2457def get_connection_artifact(
2458    connection_id: Annotated[
2459        str,
2460        Field(description="The ID of the Airbyte Cloud connection."),
2461    ],
2462    artifact_type: Annotated[
2463        Literal["state", "catalog"],
2464        Field(description="The type of artifact to retrieve: 'state' or 'catalog'."),
2465    ],
2466    *,
2467    workspace_id: Annotated[
2468        str | None,
2469        Field(
2470            description=WORKSPACE_ID_TIP_TEXT,
2471            default=None,
2472        ),
2473    ],
2474) -> dict[str, Any] | list[dict[str, Any]]:
2475    """Get a connection artifact (state or catalog) from Airbyte Cloud.
2476
2477    Retrieves the specified artifact for a connection:
2478    - 'state': Returns the persisted state for incremental syncs as a list of
2479      stream state objects, or {"ERROR": "..."} if no state is set.
2480    - 'catalog': Returns the configured catalog (syncCatalog) as a dict,
2481      or {"ERROR": "..."} if not found.
2482    """
2483    workspace: CloudWorkspace = _get_cloud_workspace(workspace_id)
2484    connection = workspace.get_connection(connection_id=connection_id)
2485
2486    if artifact_type == "state":
2487        result = connection.get_state_artifacts()
2488        if result is None:
2489            return {"ERROR": "No state is set for this connection (stateType: not_set)"}
2490        return result
2491
2492    # artifact_type == "catalog"
2493    result = connection.get_catalog_artifact()
2494    if result is None:
2495        return {"ERROR": "No catalog found for this connection"}
2496    return result
2497
2498
2499def register_cloud_ops_tools(app: FastMCP) -> None:
2500    """@private Register tools with the FastMCP app.
2501
2502    This is an internal function and should not be called directly.
2503
2504    Tools are filtered based on mode settings:
2505    - AIRBYTE_CLOUD_MCP_READONLY_MODE=1: Only read-only tools are registered
2506    - AIRBYTE_CLOUD_MCP_SAFE_MODE=1: All tools are registered, but destructive
2507      operations are protected by runtime session checks
2508    """
2509    register_tools(app, domain="cloud")