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    destructive=True,
1633    open_world=True,
1634)
1635def update_custom_source_definition(
1636    definition_id: Annotated[
1637        str,
1638        Field(description="The ID of the definition to update."),
1639    ],
1640    manifest_yaml: Annotated[
1641        str | Path | None,
1642        Field(
1643            description=(
1644                "New manifest as YAML string or file path. "
1645                "Optional; omit to update only testing values."
1646            ),
1647            default=None,
1648        ),
1649    ] = None,
1650    *,
1651    workspace_id: Annotated[
1652        str | None,
1653        Field(
1654            description=WORKSPACE_ID_TIP_TEXT,
1655            default=None,
1656        ),
1657    ],
1658    pre_validate: Annotated[
1659        bool,
1660        Field(
1661            description="Whether to validate the manifest client-side before updating.",
1662            default=True,
1663        ),
1664    ] = True,
1665    testing_values: Annotated[
1666        dict | str | None,
1667        Field(
1668            description=(
1669                "Optional testing configuration values for the Builder UI. "
1670                "Can be provided as a JSON object or JSON string. "
1671                "Supports inline secret refs via 'secret_reference::ENV_VAR_NAME' syntax. "
1672                "If provided, these values replace any existing testing values "
1673                "for the connector builder project. The entire testing values object "
1674                "is overwritten, so pass the full set of values you want to persist."
1675            ),
1676            default=None,
1677        ),
1678    ],
1679    testing_values_secret_name: Annotated[
1680        str | None,
1681        Field(
1682            description=(
1683                "Optional name of a secret containing testing configuration values "
1684                "in JSON or YAML format. The secret will be resolved by the MCP "
1685                "server and merged into testing_values, with secret values taking "
1686                "precedence. This lets the agent reference secrets without sending "
1687                "raw values as tool arguments."
1688            ),
1689            default=None,
1690        ),
1691    ],
1692) -> str:
1693    """Update a custom YAML source definition in Airbyte Cloud.
1694
1695    Updates the manifest and/or testing values for an existing custom source definition.
1696    At least one of manifest_yaml, testing_values, or testing_values_secret_name must be provided.
1697    """
1698    check_guid_created_in_session(definition_id)
1699
1700    workspace: CloudWorkspace = _get_cloud_workspace(workspace_id)
1701
1702    if manifest_yaml is None and testing_values is None and testing_values_secret_name is None:
1703        raise PyAirbyteInputError(
1704            message=(
1705                "At least one of manifest_yaml, testing_values, or testing_values_secret_name "
1706                "must be provided to update a custom source definition."
1707            ),
1708            context={
1709                "definition_id": definition_id,
1710                "workspace_id": workspace.workspace_id,
1711            },
1712        )
1713
1714    processed_manifest: str | Path | None = manifest_yaml
1715    if isinstance(manifest_yaml, str) and "\n" not in manifest_yaml:
1716        processed_manifest = Path(manifest_yaml)
1717
1718    # Resolve testing values from inline config and/or secret
1719    testing_values_dict: dict[str, Any] | None = None
1720    if testing_values is not None or testing_values_secret_name is not None:
1721        testing_values_dict = (
1722            resolve_config(
1723                config=testing_values,
1724                config_secret_name=testing_values_secret_name,
1725            )
1726            or None
1727        )
1728
1729    definition = workspace.get_custom_source_definition(
1730        definition_id=definition_id,
1731        definition_type="yaml",
1732    )
1733    custom_source: CustomCloudSourceDefinition = definition
1734
1735    if processed_manifest is not None:
1736        custom_source = definition.update_definition(
1737            manifest_yaml=processed_manifest,
1738            pre_validate=pre_validate,
1739        )
1740
1741    if testing_values_dict is not None:
1742        custom_source.set_testing_values(testing_values_dict)
1743
1744    return (
1745        "Successfully updated custom YAML source definition:\n"
1746        + _get_custom_source_definition_description(
1747            custom_source=custom_source,
1748        )
1749    )
1750
1751
1752@mcp_tool(
1753    domain="cloud",
1754    destructive=True,
1755    open_world=True,
1756)
1757def permanently_delete_custom_source_definition(
1758    definition_id: Annotated[
1759        str,
1760        Field(description="The ID of the custom source definition to delete."),
1761    ],
1762    name: Annotated[
1763        str,
1764        Field(description="The expected name of the custom source definition (for verification)."),
1765    ],
1766    *,
1767    workspace_id: Annotated[
1768        str | None,
1769        Field(
1770            description=WORKSPACE_ID_TIP_TEXT,
1771            default=None,
1772        ),
1773    ],
1774) -> str:
1775    """Permanently delete a custom YAML source definition from Airbyte Cloud.
1776
1777    IMPORTANT: This operation requires the connector name to contain "delete-me" or "deleteme"
1778    (case insensitive).
1779
1780    If the connector does not meet this requirement, the deletion will be rejected with a
1781    helpful error message. Instruct the user to rename the connector appropriately to authorize
1782    the deletion.
1783
1784    The provided name must match the actual name of the definition for the operation to proceed.
1785    This is a safety measure to ensure you are deleting the correct resource.
1786
1787    Note: Only YAML (declarative) connectors are currently supported.
1788    Docker-based custom sources are not yet available.
1789    """
1790    check_guid_created_in_session(definition_id)
1791    workspace: CloudWorkspace = _get_cloud_workspace(workspace_id)
1792    definition = workspace.get_custom_source_definition(
1793        definition_id=definition_id,
1794        definition_type="yaml",
1795    )
1796    actual_name: str = definition.name
1797
1798    # Verify the name matches
1799    if actual_name != name:
1800        raise PyAirbyteInputError(
1801            message=(
1802                f"Name mismatch: expected '{name}' but found '{actual_name}'. "
1803                "The provided name must exactly match the definition's actual name. "
1804                "This is a safety measure to prevent accidental deletion."
1805            ),
1806            context={
1807                "definition_id": definition_id,
1808                "expected_name": name,
1809                "actual_name": actual_name,
1810            },
1811        )
1812
1813    definition.permanently_delete(
1814        safe_mode=True,  # Hard-coded safe mode for extra protection when running in LLM agents.
1815    )
1816    return f"Successfully deleted custom source definition '{actual_name}' (ID: {definition_id})"
1817
1818
1819@mcp_tool(
1820    domain="cloud",
1821    destructive=True,
1822    open_world=True,
1823    extra_help_text=CLOUD_AUTH_TIP_TEXT,
1824)
1825def permanently_delete_cloud_source(
1826    source_id: Annotated[
1827        str,
1828        Field(description="The ID of the deployed source to delete."),
1829    ],
1830    name: Annotated[
1831        str,
1832        Field(description="The expected name of the source (for verification)."),
1833    ],
1834) -> str:
1835    """Permanently delete a deployed source connector from Airbyte Cloud.
1836
1837    IMPORTANT: This operation requires the source name to contain "delete-me" or "deleteme"
1838    (case insensitive).
1839
1840    If the source does not meet this requirement, the deletion will be rejected with a
1841    helpful error message. Instruct the user to rename the source appropriately to authorize
1842    the deletion.
1843
1844    The provided name must match the actual name of the source for the operation to proceed.
1845    This is a safety measure to ensure you are deleting the correct resource.
1846    """
1847    check_guid_created_in_session(source_id)
1848    workspace: CloudWorkspace = _get_cloud_workspace()
1849    source = workspace.get_source(source_id=source_id)
1850    actual_name: str = cast(str, source.name)
1851
1852    # Verify the name matches
1853    if actual_name != name:
1854        raise PyAirbyteInputError(
1855            message=(
1856                f"Name mismatch: expected '{name}' but found '{actual_name}'. "
1857                "The provided name must exactly match the source's actual name. "
1858                "This is a safety measure to prevent accidental deletion."
1859            ),
1860            context={
1861                "source_id": source_id,
1862                "expected_name": name,
1863                "actual_name": actual_name,
1864            },
1865        )
1866
1867    # Safe mode is hard-coded to True for extra protection when running in LLM agents
1868    workspace.permanently_delete_source(
1869        source=source_id,
1870        safe_mode=True,  # Requires name to contain "delete-me" or "deleteme" (case insensitive)
1871    )
1872    return f"Successfully deleted source '{actual_name}' (ID: {source_id})"
1873
1874
1875@mcp_tool(
1876    domain="cloud",
1877    destructive=True,
1878    open_world=True,
1879    extra_help_text=CLOUD_AUTH_TIP_TEXT,
1880)
1881def permanently_delete_cloud_destination(
1882    destination_id: Annotated[
1883        str,
1884        Field(description="The ID of the deployed destination to delete."),
1885    ],
1886    name: Annotated[
1887        str,
1888        Field(description="The expected name of the destination (for verification)."),
1889    ],
1890) -> str:
1891    """Permanently delete a deployed destination connector from Airbyte Cloud.
1892
1893    IMPORTANT: This operation requires the destination name to contain "delete-me" or "deleteme"
1894    (case insensitive).
1895
1896    If the destination does not meet this requirement, the deletion will be rejected with a
1897    helpful error message. Instruct the user to rename the destination appropriately to authorize
1898    the deletion.
1899
1900    The provided name must match the actual name of the destination for the operation to proceed.
1901    This is a safety measure to ensure you are deleting the correct resource.
1902    """
1903    check_guid_created_in_session(destination_id)
1904    workspace: CloudWorkspace = _get_cloud_workspace()
1905    destination = workspace.get_destination(destination_id=destination_id)
1906    actual_name: str = cast(str, destination.name)
1907
1908    # Verify the name matches
1909    if actual_name != name:
1910        raise PyAirbyteInputError(
1911            message=(
1912                f"Name mismatch: expected '{name}' but found '{actual_name}'. "
1913                "The provided name must exactly match the destination's actual name. "
1914                "This is a safety measure to prevent accidental deletion."
1915            ),
1916            context={
1917                "destination_id": destination_id,
1918                "expected_name": name,
1919                "actual_name": actual_name,
1920            },
1921        )
1922
1923    # Safe mode is hard-coded to True for extra protection when running in LLM agents
1924    workspace.permanently_delete_destination(
1925        destination=destination_id,
1926        safe_mode=True,  # Requires name-based delete disposition ("delete-me" or "deleteme")
1927    )
1928    return f"Successfully deleted destination '{actual_name}' (ID: {destination_id})"
1929
1930
1931@mcp_tool(
1932    domain="cloud",
1933    destructive=True,
1934    open_world=True,
1935    extra_help_text=CLOUD_AUTH_TIP_TEXT,
1936)
1937def permanently_delete_cloud_connection(
1938    connection_id: Annotated[
1939        str,
1940        Field(description="The ID of the connection to delete."),
1941    ],
1942    name: Annotated[
1943        str,
1944        Field(description="The expected name of the connection (for verification)."),
1945    ],
1946    *,
1947    cascade_delete_source: Annotated[
1948        bool,
1949        Field(
1950            description=(
1951                "Whether to also delete the source connector associated with this connection."
1952            ),
1953            default=False,
1954        ),
1955    ] = False,
1956    cascade_delete_destination: Annotated[
1957        bool,
1958        Field(
1959            description=(
1960                "Whether to also delete the destination connector associated with this connection."
1961            ),
1962            default=False,
1963        ),
1964    ] = False,
1965) -> str:
1966    """Permanently delete a connection from Airbyte Cloud.
1967
1968    IMPORTANT: This operation requires the connection name to contain "delete-me" or "deleteme"
1969    (case insensitive).
1970
1971    If the connection does not meet this requirement, the deletion will be rejected with a
1972    helpful error message. Instruct the user to rename the connection appropriately to authorize
1973    the deletion.
1974
1975    The provided name must match the actual name of the connection for the operation to proceed.
1976    This is a safety measure to ensure you are deleting the correct resource.
1977    """
1978    check_guid_created_in_session(connection_id)
1979    workspace: CloudWorkspace = _get_cloud_workspace()
1980    connection = workspace.get_connection(connection_id=connection_id)
1981    actual_name: str = cast(str, connection.name)
1982
1983    # Verify the name matches
1984    if actual_name != name:
1985        raise PyAirbyteInputError(
1986            message=(
1987                f"Name mismatch: expected '{name}' but found '{actual_name}'. "
1988                "The provided name must exactly match the connection's actual name. "
1989                "This is a safety measure to prevent accidental deletion."
1990            ),
1991            context={
1992                "connection_id": connection_id,
1993                "expected_name": name,
1994                "actual_name": actual_name,
1995            },
1996        )
1997
1998    # Safe mode is hard-coded to True for extra protection when running in LLM agents
1999    workspace.permanently_delete_connection(
2000        safe_mode=True,  # Requires name-based delete disposition ("delete-me" or "deleteme")
2001        connection=connection_id,
2002        cascade_delete_source=cascade_delete_source,
2003        cascade_delete_destination=cascade_delete_destination,
2004    )
2005    return f"Successfully deleted connection '{actual_name}' (ID: {connection_id})"
2006
2007
2008@mcp_tool(
2009    domain="cloud",
2010    open_world=True,
2011    extra_help_text=CLOUD_AUTH_TIP_TEXT,
2012)
2013def rename_cloud_source(
2014    source_id: Annotated[
2015        str,
2016        Field(description="The ID of the deployed source to rename."),
2017    ],
2018    name: Annotated[
2019        str,
2020        Field(description="New name for the source."),
2021    ],
2022    *,
2023    workspace_id: Annotated[
2024        str | None,
2025        Field(
2026            description=WORKSPACE_ID_TIP_TEXT,
2027            default=None,
2028        ),
2029    ],
2030) -> str:
2031    """Rename a deployed source connector on Airbyte Cloud."""
2032    workspace: CloudWorkspace = _get_cloud_workspace(workspace_id)
2033    source = workspace.get_source(source_id=source_id)
2034    source.rename(name=name)
2035    return f"Successfully renamed source '{source_id}' to '{name}'. URL: {source.connector_url}"
2036
2037
2038@mcp_tool(
2039    domain="cloud",
2040    destructive=True,
2041    open_world=True,
2042    extra_help_text=CLOUD_AUTH_TIP_TEXT,
2043)
2044def update_cloud_source_config(
2045    source_id: Annotated[
2046        str,
2047        Field(description="The ID of the deployed source to update."),
2048    ],
2049    config: Annotated[
2050        dict | str,
2051        Field(
2052            description="New configuration for the source connector.",
2053        ),
2054    ],
2055    config_secret_name: Annotated[
2056        str | None,
2057        Field(
2058            description="The name of the secret containing the configuration.",
2059            default=None,
2060        ),
2061    ] = None,
2062    *,
2063    workspace_id: Annotated[
2064        str | None,
2065        Field(
2066            description=WORKSPACE_ID_TIP_TEXT,
2067            default=None,
2068        ),
2069    ],
2070) -> str:
2071    """Update a deployed source connector's configuration on Airbyte Cloud.
2072
2073    This is a destructive operation that can break existing connections if the
2074    configuration is changed incorrectly. Use with caution.
2075    """
2076    check_guid_created_in_session(source_id)
2077    workspace: CloudWorkspace = _get_cloud_workspace(workspace_id)
2078    source = workspace.get_source(source_id=source_id)
2079
2080    config_dict = resolve_config(
2081        config=config,
2082        config_secret_name=config_secret_name,
2083        config_spec_jsonschema=None,  # We don't have the spec here
2084    )
2085
2086    source.update_config(config=config_dict)
2087    return f"Successfully updated source '{source_id}'. URL: {source.connector_url}"
2088
2089
2090@mcp_tool(
2091    domain="cloud",
2092    open_world=True,
2093    extra_help_text=CLOUD_AUTH_TIP_TEXT,
2094)
2095def rename_cloud_destination(
2096    destination_id: Annotated[
2097        str,
2098        Field(description="The ID of the deployed destination to rename."),
2099    ],
2100    name: Annotated[
2101        str,
2102        Field(description="New name for the destination."),
2103    ],
2104    *,
2105    workspace_id: Annotated[
2106        str | None,
2107        Field(
2108            description=WORKSPACE_ID_TIP_TEXT,
2109            default=None,
2110        ),
2111    ],
2112) -> str:
2113    """Rename a deployed destination connector on Airbyte Cloud."""
2114    workspace: CloudWorkspace = _get_cloud_workspace(workspace_id)
2115    destination = workspace.get_destination(destination_id=destination_id)
2116    destination.rename(name=name)
2117    return (
2118        f"Successfully renamed destination '{destination_id}' to '{name}'. "
2119        f"URL: {destination.connector_url}"
2120    )
2121
2122
2123@mcp_tool(
2124    domain="cloud",
2125    destructive=True,
2126    open_world=True,
2127    extra_help_text=CLOUD_AUTH_TIP_TEXT,
2128)
2129def update_cloud_destination_config(
2130    destination_id: Annotated[
2131        str,
2132        Field(description="The ID of the deployed destination to update."),
2133    ],
2134    config: Annotated[
2135        dict | str,
2136        Field(
2137            description="New configuration for the destination connector.",
2138        ),
2139    ],
2140    config_secret_name: Annotated[
2141        str | None,
2142        Field(
2143            description="The name of the secret containing the configuration.",
2144            default=None,
2145        ),
2146    ],
2147    *,
2148    workspace_id: Annotated[
2149        str | None,
2150        Field(
2151            description=WORKSPACE_ID_TIP_TEXT,
2152            default=None,
2153        ),
2154    ],
2155) -> str:
2156    """Update a deployed destination connector's configuration on Airbyte Cloud.
2157
2158    This is a destructive operation that can break existing connections if the
2159    configuration is changed incorrectly. Use with caution.
2160    """
2161    check_guid_created_in_session(destination_id)
2162    workspace: CloudWorkspace = _get_cloud_workspace(workspace_id)
2163    destination = workspace.get_destination(destination_id=destination_id)
2164
2165    config_dict = resolve_config(
2166        config=config,
2167        config_secret_name=config_secret_name,
2168        config_spec_jsonschema=None,  # We don't have the spec here
2169    )
2170
2171    destination.update_config(config=config_dict)
2172    return (
2173        f"Successfully updated destination '{destination_id}'. " f"URL: {destination.connector_url}"
2174    )
2175
2176
2177@mcp_tool(
2178    domain="cloud",
2179    open_world=True,
2180    extra_help_text=CLOUD_AUTH_TIP_TEXT,
2181)
2182def rename_cloud_connection(
2183    connection_id: Annotated[
2184        str,
2185        Field(description="The ID of the connection to rename."),
2186    ],
2187    name: Annotated[
2188        str,
2189        Field(description="New name for the connection."),
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    """Rename a connection on Airbyte Cloud."""
2201    workspace: CloudWorkspace = _get_cloud_workspace(workspace_id)
2202    connection = workspace.get_connection(connection_id=connection_id)
2203    connection.rename(name=name)
2204    return (
2205        f"Successfully renamed connection '{connection_id}' to '{name}'. "
2206        f"URL: {connection.connection_url}"
2207    )
2208
2209
2210@mcp_tool(
2211    domain="cloud",
2212    destructive=True,
2213    open_world=True,
2214    extra_help_text=CLOUD_AUTH_TIP_TEXT,
2215)
2216def set_cloud_connection_table_prefix(
2217    connection_id: Annotated[
2218        str,
2219        Field(description="The ID of the connection to update."),
2220    ],
2221    prefix: Annotated[
2222        str,
2223        Field(description="New table prefix to use when syncing to the destination."),
2224    ],
2225    *,
2226    workspace_id: Annotated[
2227        str | None,
2228        Field(
2229            description=WORKSPACE_ID_TIP_TEXT,
2230            default=None,
2231        ),
2232    ],
2233) -> str:
2234    """Set the table prefix for a connection on Airbyte Cloud.
2235
2236    This is a destructive operation that can break downstream dependencies if the
2237    table prefix is changed incorrectly. Use with caution.
2238    """
2239    check_guid_created_in_session(connection_id)
2240    workspace: CloudWorkspace = _get_cloud_workspace(workspace_id)
2241    connection = workspace.get_connection(connection_id=connection_id)
2242    connection.set_table_prefix(prefix=prefix)
2243    return (
2244        f"Successfully set table prefix for connection '{connection_id}' to '{prefix}'. "
2245        f"URL: {connection.connection_url}"
2246    )
2247
2248
2249@mcp_tool(
2250    domain="cloud",
2251    destructive=True,
2252    open_world=True,
2253    extra_help_text=CLOUD_AUTH_TIP_TEXT,
2254)
2255def set_cloud_connection_selected_streams(
2256    connection_id: Annotated[
2257        str,
2258        Field(description="The ID of the connection to update."),
2259    ],
2260    stream_names: Annotated[
2261        str | list[str],
2262        Field(
2263            description=(
2264                "The selected stream names to sync within the connection. "
2265                "Must be an explicit stream name or list of streams."
2266            )
2267        ),
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 selected streams for a connection on Airbyte Cloud.
2279
2280    This is a destructive operation that can break existing connections if the
2281    stream selection 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
2287    resolved_streams_list: list[str] = resolve_list_of_strings(stream_names)
2288    connection.set_selected_streams(stream_names=resolved_streams_list)
2289
2290    return (
2291        f"Successfully set selected streams for connection '{connection_id}' "
2292        f"to {resolved_streams_list}. URL: {connection.connection_url}"
2293    )
2294
2295
2296@mcp_tool(
2297    domain="cloud",
2298    read_only=True,
2299    idempotent=True,
2300    open_world=True,
2301    extra_help_text=CLOUD_AUTH_TIP_TEXT,
2302)
2303def get_connection_artifact(
2304    connection_id: Annotated[
2305        str,
2306        Field(description="The ID of the Airbyte Cloud connection."),
2307    ],
2308    artifact_type: Annotated[
2309        Literal["state", "catalog"],
2310        Field(description="The type of artifact to retrieve: 'state' or 'catalog'."),
2311    ],
2312    *,
2313    workspace_id: Annotated[
2314        str | None,
2315        Field(
2316            description=WORKSPACE_ID_TIP_TEXT,
2317            default=None,
2318        ),
2319    ],
2320) -> dict[str, Any] | list[dict[str, Any]]:
2321    """Get a connection artifact (state or catalog) from Airbyte Cloud.
2322
2323    Retrieves the specified artifact for a connection:
2324    - 'state': Returns the persisted state for incremental syncs as a list of
2325      stream state objects, or {"ERROR": "..."} if no state is set.
2326    - 'catalog': Returns the configured catalog (syncCatalog) as a dict,
2327      or {"ERROR": "..."} if not found.
2328    """
2329    workspace: CloudWorkspace = _get_cloud_workspace(workspace_id)
2330    connection = workspace.get_connection(connection_id=connection_id)
2331
2332    if artifact_type == "state":
2333        result = connection.get_state_artifacts()
2334        if result is None:
2335            return {"ERROR": "No state is set for this connection (stateType: not_set)"}
2336        return result
2337
2338    # artifact_type == "catalog"
2339    result = connection.get_catalog_artifact()
2340    if result is None:
2341        return {"ERROR": "No catalog found for this connection"}
2342    return result
2343
2344
2345def register_cloud_ops_tools(app: FastMCP) -> None:
2346    """@private Register tools with the FastMCP app.
2347
2348    This is an internal function and should not be called directly.
2349
2350    Tools are filtered based on mode settings:
2351    - AIRBYTE_CLOUD_MCP_READONLY_MODE=1: Only read-only tools are registered
2352    - AIRBYTE_CLOUD_MCP_SAFE_MODE=1: All tools are registered, but destructive
2353      operations are protected by runtime session checks
2354    """
2355    register_tools(app, domain="cloud")