airbyte.mcp.cloud

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