airbyte.mcp.cloud

Airbyte Cloud MCP operations.

cloud module

MCP primitives registered by the cloud module of the airbyte-mcp server: 39 tool(s), 0 prompt(s), 0 resource(s).

Tools (39)

cancel_cloud_sync

Hints: destructive · open-world

Cancel a running sync job on an Airbyte Cloud connection.

When connecting to a hosted MCP server, provide a bearer token via the Authorization header, or client credentials via the transport Client-Id and Client-Secret headers. To discover available organizations and workspaces, call list_cloud_organizations and list_cloud_workspaces before asking the user for an ID. For local or stdio connections, set the AIRBYTE_CLOUD_BEARER_TOKEN environment variable, or both AIRBYTE_CLOUD_CLIENT_ID and AIRBYTE_CLOUD_CLIENT_SECRET. If discovery returns multiple candidates, ask the user to choose one; do not select automatically.

Parameters

Name	Type	Required	Default	Description
`connection_id`	`string`	yes	—	The ID of the Airbyte Cloud connection.
`job_id`	`integer \| null`	no	`null`	Optional job ID to cancel. If not provided, the connection's most recent sync job will be cancelled. Other job types require an explicit job ID.
`workspace_id`	`string \| null`	no	`null`	Workspace ID. Hosted MCP connections pass it via the `X-Airbyte-Workspace-Id` header; local or stdio connections use the `AIRBYTE_CLOUD_WORKSPACE_ID` environment variable.

Show input JSON schema

{
  "additionalProperties": false,
  "properties": {
    "connection_id": {
      "description": "The ID of the Airbyte Cloud connection.",
      "type": "string"
    },
    "job_id": {
      "anyOf": [
        {
          "type": "integer"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "description": "Optional job ID to cancel. If not provided, the connection's most recent sync job will be cancelled. Other job types require an explicit job ID."
    },
    "workspace_id": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "description": "Workspace ID. Hosted MCP connections pass it via the `X-Airbyte-Workspace-Id` header; local or stdio connections use the `AIRBYTE_CLOUD_WORKSPACE_ID` environment variable."
    }
  },
  "required": [
    "connection_id"
  ],
  "type": "object"
}

Show output JSON schema

{
  "description": "Information about a sync job.",
  "properties": {
    "job_id": {
      "type": "integer"
    },
    "status": {
      "type": "string"
    },
    "bytes_synced": {
      "type": "integer"
    },
    "records_synced": {
      "type": "integer"
    },
    "start_time": {
      "type": "string"
    },
    "job_url": {
      "type": "string"
    }
  },
  "required": [
    "job_id",
    "status",
    "bytes_synced",
    "records_synced",
    "start_time",
    "job_url"
  ],
  "type": "object"
}

check_airbyte_cloud_workspace

Hints: read-only · idempotent · open-world

Check if we have a valid Airbyte Cloud connection and return workspace info.

Returns workspace details including workspace ID, name, organization info, and billing status.

Parameters

Name	Type	Required	Default	Description
`workspace_id`	`string \| null`	no	`null`	Workspace ID. Hosted MCP connections pass it via the `X-Airbyte-Workspace-Id` header; local or stdio connections use the `AIRBYTE_CLOUD_WORKSPACE_ID` environment variable.

Show input JSON schema

{
  "additionalProperties": false,
  "properties": {
    "workspace_id": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "description": "Workspace ID. Hosted MCP connections pass it via the `X-Airbyte-Workspace-Id` header; local or stdio connections use the `AIRBYTE_CLOUD_WORKSPACE_ID` environment variable."
    }
  },
  "type": "object"
}

Show output JSON schema

{
  "description": "Information about a workspace in Airbyte Cloud.",
  "properties": {
    "workspace_id": {
      "type": "string"
    },
    "workspace_name": {
      "type": "string"
    },
    "workspace_url": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null
    },
    "organization_id": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ]
    },
    "organization_name": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null
    },
    "payment_status": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null
    },
    "subscription_status": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null
    },
    "is_account_locked": {
      "default": false,
      "type": "boolean"
    }
  },
  "required": [
    "workspace_id",
    "workspace_name",
    "organization_id"
  ],
  "type": "object"
}

check_cloud_destination

Hints: read-only · idempotent · open-world

Check the configuration and credentials of a deployed destination connector.

Parameters

Name	Type	Required	Default	Description
`destination_id`	`string`	yes	—	The ID of the deployed destination connector to check.
`workspace_id`	`string \| null`	no	`null`	Workspace ID. Hosted MCP connections pass it via the `X-Airbyte-Workspace-Id` header; local or stdio connections use the `AIRBYTE_CLOUD_WORKSPACE_ID` environment variable.

Show input JSON schema

{
  "additionalProperties": false,
  "properties": {
    "destination_id": {
      "description": "The ID of the deployed destination connector to check.",
      "type": "string"
    },
    "workspace_id": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "description": "Workspace ID. Hosted MCP connections pass it via the `X-Airbyte-Workspace-Id` header; local or stdio connections use the `AIRBYTE_CLOUD_WORKSPACE_ID` environment variable."
    }
  },
  "required": [
    "destination_id"
  ],
  "type": "object"
}

Show output JSON schema

{
  "description": "Result of a connection check against a deployed Cloud connector.",
  "properties": {
    "connector_id": {
      "type": "string"
    },
    "connector_type": {
      "enum": [
        "source",
        "destination"
      ],
      "type": "string"
    },
    "succeeded": {
      "type": "boolean"
    },
    "message": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ]
    }
  },
  "required": [
    "connector_id",
    "connector_type",
    "succeeded",
    "message"
  ],
  "type": "object"
}

check_cloud_source

Hints: read-only · idempotent · open-world

Check the configuration and credentials of a deployed source connector.

Parameters

Name	Type	Required	Default	Description
`source_id`	`string`	yes	—	The ID of the deployed source connector to check.
`workspace_id`	`string \| null`	no	`null`	Workspace ID. Hosted MCP connections pass it via the `X-Airbyte-Workspace-Id` header; local or stdio connections use the `AIRBYTE_CLOUD_WORKSPACE_ID` environment variable.

Show input JSON schema

{
  "additionalProperties": false,
  "properties": {
    "source_id": {
      "description": "The ID of the deployed source connector to check.",
      "type": "string"
    },
    "workspace_id": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "description": "Workspace ID. Hosted MCP connections pass it via the `X-Airbyte-Workspace-Id` header; local or stdio connections use the `AIRBYTE_CLOUD_WORKSPACE_ID` environment variable."
    }
  },
  "required": [
    "source_id"
  ],
  "type": "object"
}

Show output JSON schema

{
  "description": "Result of a connection check against a deployed Cloud connector.",
  "properties": {
    "connector_id": {
      "type": "string"
    },
    "connector_type": {
      "enum": [
        "source",
        "destination"
      ],
      "type": "string"
    },
    "succeeded": {
      "type": "boolean"
    },
    "message": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ]
    }
  },
  "required": [
    "connector_id",
    "connector_type",
    "succeeded",
    "message"
  ],
  "type": "object"
}

create_connection_on_cloud

Hints: open-world

Create a connection between a deployed source and destination on Airbyte Cloud.

Parameters

Name	Type	Required	Default	Description
`connection_name`	`string`	yes	—	The name of the connection.
`source_id`	`string`	yes	—	The ID of the deployed source.
`destination_id`	`string`	yes	—	The ID of the deployed destination.
`selected_streams`	`string \| array<string>`	yes	—	The selected stream names to sync within the connection. Must be an explicit stream name or list of streams. Cannot be empty or '*'.
`workspace_id`	`string \| null`	no	`null`	Workspace ID. Hosted MCP connections pass it via the `X-Airbyte-Workspace-Id` header; local or stdio connections use the `AIRBYTE_CLOUD_WORKSPACE_ID` environment variable.
`table_prefix`	`string \| null`	no	`null`	Optional table prefix to use when syncing to the destination.

Show input JSON schema

{
  "additionalProperties": false,
  "properties": {
    "connection_name": {
      "description": "The name of the connection.",
      "type": "string"
    },
    "source_id": {
      "description": "The ID of the deployed source.",
      "type": "string"
    },
    "destination_id": {
      "description": "The ID of the deployed destination.",
      "type": "string"
    },
    "selected_streams": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "items": {
            "type": "string"
          },
          "type": "array"
        }
      ],
      "description": "The selected stream names to sync within the connection. Must be an explicit stream name or list of streams. Cannot be empty or '*'."
    },
    "workspace_id": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "description": "Workspace ID. Hosted MCP connections pass it via the `X-Airbyte-Workspace-Id` header; local or stdio connections use the `AIRBYTE_CLOUD_WORKSPACE_ID` environment variable."
    },
    "table_prefix": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "description": "Optional table prefix to use when syncing to the destination."
    }
  },
  "required": [
    "connection_name",
    "source_id",
    "destination_id",
    "selected_streams"
  ],
  "type": "object"
}

Show output JSON schema

{
  "properties": {
    "result": {
      "type": "string"
    }
  },
  "required": [
    "result"
  ],
  "type": "object",
  "x-fastmcp-wrap-result": true
}

deploy_destination_to_cloud

Hints: open-world

Deploy a destination connector to Airbyte Cloud.

Parameters

Name	Type	Required	Default	Description
`destination_name`	`string`	yes	—	The name to use when deploying the destination.
`destination_connector_name`	`string`	yes	—	The name of the destination connector (e.g., 'destination-postgres').
`workspace_id`	`string \| null`	no	`null`	Workspace ID. Hosted MCP connections pass it via the `X-Airbyte-Workspace-Id` header; local or stdio connections use the `AIRBYTE_CLOUD_WORKSPACE_ID` environment variable.
`config`	`object \| string \| null`	no	`null`	The configuration for the destination connector.
`config_secret_name`	`string \| null`	no	`null`	The name of the secret containing the configuration.
`unique`	`boolean`	no	`true`	Whether to require a unique name.

Show input JSON schema

{
  "additionalProperties": false,
  "properties": {
    "destination_name": {
      "description": "The name to use when deploying the destination.",
      "type": "string"
    },
    "destination_connector_name": {
      "description": "The name of the destination connector (e.g., 'destination-postgres').",
      "type": "string"
    },
    "workspace_id": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "description": "Workspace ID. Hosted MCP connections pass it via the `X-Airbyte-Workspace-Id` header; local or stdio connections use the `AIRBYTE_CLOUD_WORKSPACE_ID` environment variable."
    },
    "config": {
      "anyOf": [
        {
          "additionalProperties": true,
          "type": "object"
        },
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "description": "The configuration for the destination connector."
    },
    "config_secret_name": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "description": "The name of the secret containing the configuration."
    },
    "unique": {
      "default": true,
      "description": "Whether to require a unique name.",
      "type": "boolean"
    }
  },
  "required": [
    "destination_name",
    "destination_connector_name"
  ],
  "type": "object"
}

Show output JSON schema

{
  "properties": {
    "result": {
      "type": "string"
    }
  },
  "required": [
    "result"
  ],
  "type": "object",
  "x-fastmcp-wrap-result": true
}

deploy_noop_destination_to_cloud

Hints: open-world

Deploy the No-op destination to Airbyte Cloud for testing purposes.

Parameters

Name	Type	Required	Default	Description
`name`	`string`	no	`"No-op Destination"`
`workspace_id`	`string \| null`	no	`null`	Workspace ID. Hosted MCP connections pass it via the `X-Airbyte-Workspace-Id` header; local or stdio connections use the `AIRBYTE_CLOUD_WORKSPACE_ID` environment variable.
`unique`	`boolean`	no	`true`

Show input JSON schema

{
  "additionalProperties": false,
  "properties": {
    "name": {
      "default": "No-op Destination",
      "type": "string"
    },
    "workspace_id": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "description": "Workspace ID. Hosted MCP connections pass it via the `X-Airbyte-Workspace-Id` header; local or stdio connections use the `AIRBYTE_CLOUD_WORKSPACE_ID` environment variable."
    },
    "unique": {
      "default": true,
      "type": "boolean"
    }
  },
  "type": "object"
}

Show output JSON schema

{
  "properties": {
    "result": {
      "type": "string"
    }
  },
  "required": [
    "result"
  ],
  "type": "object",
  "x-fastmcp-wrap-result": true
}

deploy_source_to_cloud

Hints: open-world

Deploy a source connector to Airbyte Cloud.

Parameters

Name	Type	Required	Default	Description
`source_name`	`string`	yes	—	The name to use when deploying the source.
`source_connector_name`	`string`	yes	—	The name of the source connector (e.g., 'source-faker').
`workspace_id`	`string \| null`	no	`null`	Workspace ID. Hosted MCP connections pass it via the `X-Airbyte-Workspace-Id` header; local or stdio connections use the `AIRBYTE_CLOUD_WORKSPACE_ID` environment variable.
`config`	`object \| string \| null`	no	`null`	The configuration for the source connector.
`config_secret_name`	`string \| null`	no	`null`	The name of the secret containing the configuration.
`unique`	`boolean`	no	`true`	Whether to require a unique name.

Show input JSON schema

{
  "additionalProperties": false,
  "properties": {
    "source_name": {
      "description": "The name to use when deploying the source.",
      "type": "string"
    },
    "source_connector_name": {
      "description": "The name of the source connector (e.g., 'source-faker').",
      "type": "string"
    },
    "workspace_id": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "description": "Workspace ID. Hosted MCP connections pass it via the `X-Airbyte-Workspace-Id` header; local or stdio connections use the `AIRBYTE_CLOUD_WORKSPACE_ID` environment variable."
    },
    "config": {
      "anyOf": [
        {
          "additionalProperties": true,
          "type": "object"
        },
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "description": "The configuration for the source connector."
    },
    "config_secret_name": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "description": "The name of the secret containing the configuration."
    },
    "unique": {
      "default": true,
      "description": "Whether to require a unique name.",
      "type": "boolean"
    }
  },
  "required": [
    "source_name",
    "source_connector_name"
  ],
  "type": "object"
}

Show output JSON schema

{
  "properties": {
    "result": {
      "type": "string"
    }
  },
  "required": [
    "result"
  ],
  "type": "object",
  "x-fastmcp-wrap-result": true
}

describe_cloud_connection

Hints: read-only · idempotent · open-world

Get detailed information about a specific deployed connection.

Parameters

Name	Type	Required	Default	Description
`connection_id`	`string`	yes	—	The ID of the connection to describe.
`workspace_id`	`string \| null`	no	`null`	Workspace ID. Hosted MCP connections pass it via the `X-Airbyte-Workspace-Id` header; local or stdio connections use the `AIRBYTE_CLOUD_WORKSPACE_ID` environment variable.

Show input JSON schema

{
  "additionalProperties": false,
  "properties": {
    "connection_id": {
      "description": "The ID of the connection to describe.",
      "type": "string"
    },
    "workspace_id": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "description": "Workspace ID. Hosted MCP connections pass it via the `X-Airbyte-Workspace-Id` header; local or stdio connections use the `AIRBYTE_CLOUD_WORKSPACE_ID` environment variable."
    }
  },
  "required": [
    "connection_id"
  ],
  "type": "object"
}

Show output JSON schema

{
  "description": "Detailed information about a deployed connection in Airbyte Cloud.",
  "properties": {
    "connection_id": {
      "type": "string"
    },
    "connection_name": {
      "type": "string"
    },
    "connection_url": {
      "type": "string"
    },
    "source_id": {
      "type": "string"
    },
    "source_name": {
      "type": "string"
    },
    "destination_id": {
      "type": "string"
    },
    "destination_name": {
      "type": "string"
    },
    "selected_streams": {
      "items": {
        "type": "string"
      },
      "type": "array"
    },
    "table_prefix": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ]
    }
  },
  "required": [
    "connection_id",
    "connection_name",
    "connection_url",
    "source_id",
    "source_name",
    "destination_id",
    "destination_name",
    "selected_streams",
    "table_prefix"
  ],
  "type": "object"
}

describe_cloud_destination

Hints: read-only · idempotent · open-world

Get detailed information about a specific deployed destination connector.

Parameters

Name	Type	Required	Default	Description
`destination_id`	`string`	yes	—	The ID of the destination to describe.
`workspace_id`	`string \| null`	no	`null`	Workspace ID. Hosted MCP connections pass it via the `X-Airbyte-Workspace-Id` header; local or stdio connections use the `AIRBYTE_CLOUD_WORKSPACE_ID` environment variable.

Show input JSON schema

{
  "additionalProperties": false,
  "properties": {
    "destination_id": {
      "description": "The ID of the destination to describe.",
      "type": "string"
    },
    "workspace_id": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "description": "Workspace ID. Hosted MCP connections pass it via the `X-Airbyte-Workspace-Id` header; local or stdio connections use the `AIRBYTE_CLOUD_WORKSPACE_ID` environment variable."
    }
  },
  "required": [
    "destination_id"
  ],
  "type": "object"
}

Show output JSON schema

{
  "description": "Detailed information about a deployed destination connector in Airbyte Cloud.",
  "properties": {
    "destination_id": {
      "type": "string"
    },
    "destination_name": {
      "type": "string"
    },
    "destination_url": {
      "type": "string"
    },
    "connector_definition_id": {
      "type": "string"
    }
  },
  "required": [
    "destination_id",
    "destination_name",
    "destination_url",
    "connector_definition_id"
  ],
  "type": "object"
}

describe_cloud_organization

Hints: read-only · idempotent · open-world

Get details about a specific organization including billing status.

Requires either organization_id OR organization_name (exact match) to be provided.
This tool is useful for looking up an organization's ID from its name, or vice versa.

Parameters

Name	Type	Required	Default	Description
`organization_id`	`string \| null`	no	`null`	Organization ID. Required if organization_name is not provided.
`organization_name`	`string \| null`	no	`null`	Organization name (exact match). Required if organization_id is not provided.

Show input JSON schema

{
  "additionalProperties": false,
  "properties": {
    "organization_id": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "description": "Organization ID. Required if organization_name is not provided."
    },
    "organization_name": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "description": "Organization name (exact match). Required if organization_id is not provided."
    }
  },
  "type": "object"
}

Show output JSON schema

{
  "description": "Information about an organization in Airbyte Cloud.",
  "properties": {
    "id": {
      "type": "string"
    },
    "name": {
      "type": "string"
    },
    "email": {
      "type": "string"
    },
    "payment_status": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null
    },
    "subscription_status": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null
    },
    "is_account_locked": {
      "default": false,
      "type": "boolean"
    }
  },
  "required": [
    "id",
    "name",
    "email"
  ],
  "type": "object"
}

describe_cloud_source

Hints: read-only · idempotent · open-world

Get detailed information about a specific deployed source connector.

Parameters

Name	Type	Required	Default	Description
`source_id`	`string`	yes	—	The ID of the source to describe.
`workspace_id`	`string \| null`	no	`null`	Workspace ID. Hosted MCP connections pass it via the `X-Airbyte-Workspace-Id` header; local or stdio connections use the `AIRBYTE_CLOUD_WORKSPACE_ID` environment variable.

Show input JSON schema

{
  "additionalProperties": false,
  "properties": {
    "source_id": {
      "description": "The ID of the source to describe.",
      "type": "string"
    },
    "workspace_id": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "description": "Workspace ID. Hosted MCP connections pass it via the `X-Airbyte-Workspace-Id` header; local or stdio connections use the `AIRBYTE_CLOUD_WORKSPACE_ID` environment variable."
    }
  },
  "required": [
    "source_id"
  ],
  "type": "object"
}

Show output JSON schema

{
  "description": "Detailed information about a deployed source connector in Airbyte Cloud.",
  "properties": {
    "source_id": {
      "type": "string"
    },
    "source_name": {
      "type": "string"
    },
    "source_url": {
      "type": "string"
    },
    "connector_definition_id": {
      "type": "string"
    }
  },
  "required": [
    "source_id",
    "source_name",
    "source_url",
    "connector_definition_id"
  ],
  "type": "object"
}

get_cloud_sync_logs

Hints: read-only · idempotent · open-world

Get the logs from a sync job attempt on Airbyte Cloud.

Parameters

Name	Type	Required	Default	Description
`connection_id`	`string`	yes	—	The ID of the Airbyte Cloud connection.
`job_id`	`integer \| null \| null`	no	`null`
`attempt_number`	`integer \| null \| null`	no	`null`
`workspace_id`	`string \| null`	no	`null`	Workspace ID. Hosted MCP connections pass it via the `X-Airbyte-Workspace-Id` header; local or stdio connections use the `AIRBYTE_CLOUD_WORKSPACE_ID` environment variable.
`max_lines`	`integer`	no	`4000`	Maximum number of lines to return. Defaults to 4000 if not specified. If '0' is provided, no limit is applied.
`from_tail`	`boolean \| null`	no	`null`	Pull from the end of the log text if total lines is greater than 'max_lines'. Defaults to True if `line_offset` is not specified. Cannot combine `from_tail=True` with `line_offset`.
`line_offset`	`integer \| null`	no	`null`	Number of lines to skip from the beginning of the logs. Cannot be combined with `from_tail=True`.

Show input JSON schema

{
  "additionalProperties": false,
  "properties": {
    "connection_id": {
      "description": "The ID of the Airbyte Cloud connection.",
      "type": "string"
    },
    "job_id": {
      "anyOf": [
        {
          "anyOf": [
            {
              "type": "integer"
            },
            {
              "type": "null"
            }
          ],
          "description": "Optional job ID. If not provided, the latest job will be used."
        },
        {
          "type": "null"
        }
      ],
      "default": null
    },
    "attempt_number": {
      "anyOf": [
        {
          "anyOf": [
            {
              "type": "integer"
            },
            {
              "type": "null"
            }
          ],
          "description": "Optional attempt number. If not provided, the latest attempt will be used."
        },
        {
          "type": "null"
        }
      ],
      "default": null
    },
    "workspace_id": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "description": "Workspace ID. Hosted MCP connections pass it via the `X-Airbyte-Workspace-Id` header; local or stdio connections use the `AIRBYTE_CLOUD_WORKSPACE_ID` environment variable."
    },
    "max_lines": {
      "default": 4000,
      "description": "Maximum number of lines to return. Defaults to 4000 if not specified. If '0' is provided, no limit is applied.",
      "type": "integer"
    },
    "from_tail": {
      "anyOf": [
        {
          "type": "boolean"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "description": "Pull from the end of the log text if total lines is greater than 'max_lines'. Defaults to True if `line_offset` is not specified. Cannot combine `from_tail=True` with `line_offset`."
    },
    "line_offset": {
      "anyOf": [
        {
          "type": "integer"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "description": "Number of lines to skip from the beginning of the logs. Cannot be combined with `from_tail=True`."
    }
  },
  "required": [
    "connection_id"
  ],
  "type": "object"
}

Show output JSON schema

{
  "description": "Result of reading sync logs with pagination support.",
  "properties": {
    "job_id": {
      "type": "integer"
    },
    "attempt_number": {
      "type": "integer"
    },
    "log_text": {
      "type": "string"
    },
    "log_text_start_line": {
      "type": "integer"
    },
    "log_text_line_count": {
      "type": "integer"
    },
    "total_log_lines_available": {
      "type": "integer"
    }
  },
  "required": [
    "job_id",
    "attempt_number",
    "log_text",
    "log_text_start_line",
    "log_text_line_count",
    "total_log_lines_available"
  ],
  "type": "object"
}

get_cloud_sync_status

Hints: read-only · idempotent · open-world

Get the status of a sync job from the Airbyte Cloud.

Parameters

Name	Type	Required	Default	Description
`connection_id`	`string`	yes	—	The ID of the Airbyte Cloud connection.
`job_id`	`integer \| null`	no	`null`	Optional job ID. If not provided, the latest job will be used.
`workspace_id`	`string \| null`	no	`null`	Workspace ID. Hosted MCP connections pass it via the `X-Airbyte-Workspace-Id` header; local or stdio connections use the `AIRBYTE_CLOUD_WORKSPACE_ID` environment variable.
`include_attempts`	`boolean`	no	`false`	Whether to include detailed attempts information.

Show input JSON schema

{
  "additionalProperties": false,
  "properties": {
    "connection_id": {
      "description": "The ID of the Airbyte Cloud connection.",
      "type": "string"
    },
    "job_id": {
      "anyOf": [
        {
          "type": "integer"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "description": "Optional job ID. If not provided, the latest job will be used."
    },
    "workspace_id": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "description": "Workspace ID. Hosted MCP connections pass it via the `X-Airbyte-Workspace-Id` header; local or stdio connections use the `AIRBYTE_CLOUD_WORKSPACE_ID` environment variable."
    },
    "include_attempts": {
      "default": false,
      "description": "Whether to include detailed attempts information.",
      "type": "boolean"
    }
  },
  "required": [
    "connection_id"
  ],
  "type": "object"
}

Show output JSON schema

{
  "additionalProperties": true,
  "type": "object"
}

get_connection_artifact

Hints: read-only · idempotent · open-world

Get a connection artifact (state or catalog) from Airbyte Cloud.

By default, returns artifacts in Airbyte protocol format (snake_case,
suitable for passing to connector CLI flags like `--state` or `--catalog`).

Retrieves the specified artifact for a connection:
- `state`: Returns a list of protocol-format `AirbyteStateMessage` dicts,
  or `{"ERROR": "..."}` if no state is set.
- `catalog`: Returns the protocol-format `ConfiguredAirbyteCatalog` dict,
  or `{"ERROR": "..."}` if not found.

Parameters

Name	Type	Required	Default	Description
`connection_id`	`string`	yes	—	The ID of the Airbyte Cloud connection.
`artifact_type`	`enum("state", "catalog")`	yes	—	The type of artifact to retrieve: 'state' or 'catalog'.
`workspace_id`	`string \| null`	no	`null`	Workspace ID. Hosted MCP connections pass it via the `X-Airbyte-Workspace-Id` header; local or stdio connections use the `AIRBYTE_CLOUD_WORKSPACE_ID` environment variable.

Show input JSON schema

{
  "additionalProperties": false,
  "properties": {
    "connection_id": {
      "description": "The ID of the Airbyte Cloud connection.",
      "type": "string"
    },
    "artifact_type": {
      "description": "The type of artifact to retrieve: 'state' or 'catalog'.",
      "enum": [
        "state",
        "catalog"
      ],
      "type": "string"
    },
    "workspace_id": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "description": "Workspace ID. Hosted MCP connections pass it via the `X-Airbyte-Workspace-Id` header; local or stdio connections use the `AIRBYTE_CLOUD_WORKSPACE_ID` environment variable."
    }
  },
  "required": [
    "connection_id",
    "artifact_type"
  ],
  "type": "object"
}

Show output JSON schema

{
  "properties": {
    "result": {
      "anyOf": [
        {
          "additionalProperties": true,
          "type": "object"
        },
        {
          "items": {
            "additionalProperties": true,
            "type": "object"
          },
          "type": "array"
        }
      ]
    }
  },
  "required": [
    "result"
  ],
  "type": "object",
  "x-fastmcp-wrap-result": true
}

get_connector_builder_draft_manifest

Hints: read-only · idempotent · open-world

Get the Connector Builder draft manifest for a custom source definition.

Returns the working draft manifest that has been saved in the Connector Builder UI but not yet published. This is useful for inspecting what a user is currently working on before they publish their changes.

If no draft exists, 'has_draft' will be False and 'draft_manifest' will be None. The published manifest is always included for comparison.

Parameters

Name	Type	Required	Default	Description
`definition_id`	`string`	yes	—	The ID of the custom source definition to retrieve the draft for.
`workspace_id`	`string \| null`	no	`null`	Workspace ID. Hosted MCP connections pass it via the `X-Airbyte-Workspace-Id` header; local or stdio connections use the `AIRBYTE_CLOUD_WORKSPACE_ID` environment variable.

Show input JSON schema

{
  "additionalProperties": false,
  "properties": {
    "definition_id": {
      "description": "The ID of the custom source definition to retrieve the draft for.",
      "type": "string"
    },
    "workspace_id": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "description": "Workspace ID. Hosted MCP connections pass it via the `X-Airbyte-Workspace-Id` header; local or stdio connections use the `AIRBYTE_CLOUD_WORKSPACE_ID` environment variable."
    }
  },
  "required": [
    "definition_id"
  ],
  "type": "object"
}

Show output JSON schema

{
  "additionalProperties": true,
  "type": "object"
}

get_custom_source_definition

Hints: read-only · idempotent · open-world

Get a custom YAML source definition from Airbyte Cloud, including its manifest.

Returns the full definition details including the published manifest YAML content. Optionally includes the Connector Builder draft manifest (unpublished changes) when include_draft=True.

Note: Only YAML (declarative) connectors are currently supported. Docker-based custom sources are not yet available.

Parameters

Name	Type	Required	Default	Description
`definition_id`	`string`	yes	—	The ID of the custom source definition to retrieve.
`workspace_id`	`string \| null`	no	`null`	Workspace ID. Hosted MCP connections pass it via the `X-Airbyte-Workspace-Id` header; local or stdio connections use the `AIRBYTE_CLOUD_WORKSPACE_ID` environment variable.
`include_draft`	`boolean`	no	`false`	Whether to include the Connector Builder draft manifest in the response. If True and a draft exists, the response will include 'has_draft' and 'draft_manifest' fields. Defaults to False.

Show input JSON schema

{
  "additionalProperties": false,
  "properties": {
    "definition_id": {
      "description": "The ID of the custom source definition to retrieve.",
      "type": "string"
    },
    "workspace_id": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "description": "Workspace ID. Hosted MCP connections pass it via the `X-Airbyte-Workspace-Id` header; local or stdio connections use the `AIRBYTE_CLOUD_WORKSPACE_ID` environment variable."
    },
    "include_draft": {
      "default": false,
      "description": "Whether to include the Connector Builder draft manifest in the response. If True and a draft exists, the response will include 'has_draft' and 'draft_manifest' fields. Defaults to False.",
      "type": "boolean"
    }
  },
  "required": [
    "definition_id"
  ],
  "type": "object"
}

Show output JSON schema

{
  "additionalProperties": true,
  "type": "object"
}

list_cloud_organizations

Hints: read-only · idempotent · open-world

List organizations visible to the authenticated Airbyte Cloud credentials.

Parameters

_No parameters._

Show input JSON schema

{
  "additionalProperties": false,
  "properties": {},
  "type": "object"
}

Show output JSON schema

{
  "description": "Result of discovering organizations in Airbyte Cloud.",
  "properties": {
    "organizations": {
      "items": {
        "description": "Information about an organization in Airbyte Cloud.",
        "properties": {
          "id": {
            "type": "string"
          },
          "name": {
            "type": "string"
          },
          "email": {
            "type": "string"
          },
          "payment_status": {
            "anyOf": [
              {
                "type": "string"
              },
              {
                "type": "null"
              }
            ],
            "default": null
          },
          "subscription_status": {
            "anyOf": [
              {
                "type": "string"
              },
              {
                "type": "null"
              }
            ],
            "default": null
          },
          "is_account_locked": {
            "default": false,
            "type": "boolean"
          }
        },
        "required": [
          "id",
          "name",
          "email"
        ],
        "type": "object"
      },
      "type": "array"
    },
    "message": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null
    }
  },
  "required": [
    "organizations"
  ],
  "type": "object"
}

list_cloud_sync_jobs

Hints: read-only · idempotent · open-world

List sync jobs for a connection with limit support.

This tool allows you to retrieve a list of sync jobs for a connection,
with control over ordering and result limit. By default, jobs are returned
newest-first (`from_tail=True`).

Parameters

Name	Type	Required	Default	Description
`connection_id`	`string`	yes	—	The ID of the Airbyte Cloud connection.
`workspace_id`	`string \| null`	no	`null`	Workspace ID. Hosted MCP connections pass it via the `X-Airbyte-Workspace-Id` header; local or stdio connections use the `AIRBYTE_CLOUD_WORKSPACE_ID` environment variable.
`max_jobs`	`integer`	no	`20`	Maximum number of jobs to return. Defaults to 20 if not specified. Maximum allowed value is 500.
`from_tail`	`boolean \| null`	no	`null`	When True, jobs are ordered newest-first (createdAt DESC). When False, jobs are ordered oldest-first (createdAt ASC). Defaults to True.
`job_type`	`enum("sync", "reset", "refresh", "clear") \| null`	no	`null`	Filter by job type. Options: 'sync', 'reset', 'refresh', 'clear'. If not specified, defaults to sync and reset jobs only (API default). Use 'refresh' to find refresh jobs or 'clear' to find clear jobs.

Show input JSON schema

{
  "additionalProperties": false,
  "properties": {
    "connection_id": {
      "description": "The ID of the Airbyte Cloud connection.",
      "type": "string"
    },
    "workspace_id": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "description": "Workspace ID. Hosted MCP connections pass it via the `X-Airbyte-Workspace-Id` header; local or stdio connections use the `AIRBYTE_CLOUD_WORKSPACE_ID` environment variable."
    },
    "max_jobs": {
      "default": 20,
      "description": "Maximum number of jobs to return. Defaults to 20 if not specified. Maximum allowed value is 500.",
      "type": "integer"
    },
    "from_tail": {
      "anyOf": [
        {
          "type": "boolean"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "description": "When True, jobs are ordered newest-first (createdAt DESC). When False, jobs are ordered oldest-first (createdAt ASC). Defaults to True."
    },
    "job_type": {
      "anyOf": [
        {
          "description": "Job type values for Airbyte Cloud jobs.",
          "enum": [
            "sync",
            "reset",
            "refresh",
            "clear"
          ],
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "description": "Filter by job type. Options: 'sync', 'reset', 'refresh', 'clear'. If not specified, defaults to sync and reset jobs only (API default). Use 'refresh' to find refresh jobs or 'clear' to find clear jobs."
    }
  },
  "required": [
    "connection_id"
  ],
  "type": "object"
}

Show output JSON schema

{
  "description": "Result of listing sync jobs with limit support.",
  "properties": {
    "jobs": {
      "items": {
        "description": "Information about a sync job.",
        "properties": {
          "job_id": {
            "type": "integer"
          },
          "status": {
            "type": "string"
          },
          "bytes_synced": {
            "type": "integer"
          },
          "records_synced": {
            "type": "integer"
          },
          "start_time": {
            "type": "string"
          },
          "job_url": {
            "type": "string"
          }
        },
        "required": [
          "job_id",
          "status",
          "bytes_synced",
          "records_synced",
          "start_time",
          "job_url"
        ],
        "type": "object"
      },
      "type": "array"
    },
    "jobs_count": {
      "type": "integer"
    },
    "from_tail": {
      "type": "boolean"
    }
  },
  "required": [
    "jobs",
    "jobs_count",
    "from_tail"
  ],
  "type": "object"
}

list_cloud_workspaces

Hints: read-only · idempotent · open-world

List all workspaces visible to the authenticated credentials.

When an organization ID or exact organization name is provided, the Config API
lists workspaces in that organization. When neither is provided and the client
has no default organization, the public API lists workspaces across organizations
visible to the current credentials. Otherwise, results are scoped to the client's
default organization.

Parameters

Name	Type	Required	Default	Description
`organization_id`	`string \| null`	no	`null`	Optional organization ID to list workspaces within.
`organization_name`	`string \| null`	no	`null`	Optional organization name (exact match) to list workspaces within.
`name_contains`	`string \| null`	no	`null`	Optional substring to filter workspaces by name (server-side filtering)
`limit`	`integer \| null`	no	`null`	Optional maximum number of items to return (default: no limit)

Show input JSON schema

{
  "additionalProperties": false,
  "properties": {
    "organization_id": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "description": "Optional organization ID to list workspaces within."
    },
    "organization_name": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "description": "Optional organization name (exact match) to list workspaces within."
    },
    "name_contains": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "description": "Optional substring to filter workspaces by name (server-side filtering)"
    },
    "limit": {
      "anyOf": [
        {
          "type": "integer"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "description": "Optional maximum number of items to return (default: no limit)"
    }
  },
  "type": "object"
}

Show output JSON schema

{
  "description": "Result of discovering workspaces in Airbyte Cloud.",
  "properties": {
    "workspaces": {
      "items": {
        "description": "Information about a workspace in Airbyte Cloud.",
        "properties": {
          "workspace_id": {
            "type": "string"
          },
          "workspace_name": {
            "type": "string"
          },
          "workspace_url": {
            "anyOf": [
              {
                "type": "string"
              },
              {
                "type": "null"
              }
            ],
            "default": null
          },
          "organization_id": {
            "anyOf": [
              {
                "type": "string"
              },
              {
                "type": "null"
              }
            ]
          },
          "organization_name": {
            "anyOf": [
              {
                "type": "string"
              },
              {
                "type": "null"
              }
            ],
            "default": null
          },
          "payment_status": {
            "anyOf": [
              {
                "type": "string"
              },
              {
                "type": "null"
              }
            ],
            "default": null
          },
          "subscription_status": {
            "anyOf": [
              {
                "type": "string"
              },
              {
                "type": "null"
              }
            ],
            "default": null
          },
          "is_account_locked": {
            "default": false,
            "type": "boolean"
          }
        },
        "required": [
          "workspace_id",
          "workspace_name",
          "organization_id"
        ],
        "type": "object"
      },
      "type": "array"
    },
    "message": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null
    }
  },
  "required": [
    "workspaces"
  ],
  "type": "object"
}

list_custom_source_definitions

Hints: read-only · idempotent · open-world

List custom YAML source definitions in the Airbyte Cloud workspace.

Note: Only YAML (declarative) connectors are currently supported. Docker-based custom sources are not yet available.

Parameters

Name	Type	Required	Default	Description
`workspace_id`	`string \| null`	no	`null`	Workspace ID. Hosted MCP connections pass it via the `X-Airbyte-Workspace-Id` header; local or stdio connections use the `AIRBYTE_CLOUD_WORKSPACE_ID` environment variable.

Show input JSON schema

{
  "additionalProperties": false,
  "properties": {
    "workspace_id": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "description": "Workspace ID. Hosted MCP connections pass it via the `X-Airbyte-Workspace-Id` header; local or stdio connections use the `AIRBYTE_CLOUD_WORKSPACE_ID` environment variable."
    }
  },
  "type": "object"
}

Show output JSON schema

{
  "properties": {
    "result": {
      "items": {
        "additionalProperties": true,
        "type": "object"
      },
      "type": "array"
    }
  },
  "required": [
    "result"
  ],
  "type": "object",
  "x-fastmcp-wrap-result": true
}

list_deployed_cloud_connections

Hints: read-only · idempotent · open-world

List all deployed connections in the Airbyte Cloud workspace.

When with_connection_status is True, each connection result will include
information about the most recent sync job status, skipping over any
currently in-progress syncs to find the last completed job.

When failing_connections_only is True, only connections where the most
recent completed sync job failed or was cancelled will be returned.
This implicitly enables with_connection_status.

Parameters

Name	Type	Required	Default	Description
`workspace_id`	`string \| null`	no	`null`	Workspace ID. Hosted MCP connections pass it via the `X-Airbyte-Workspace-Id` header; local or stdio connections use the `AIRBYTE_CLOUD_WORKSPACE_ID` environment variable.
`name_contains`	`string \| null`	no	`null`	Optional case-insensitive substring to filter connections by name
`limit`	`integer \| null`	no	`null`	Optional maximum number of items to return (default: no limit)
`with_connection_status`	`boolean \| null`	no	`false`	If True, include status info for each connection's most recent sync job
`failing_connections_only`	`boolean \| null`	no	`false`	If True, only return connections with failed/cancelled last sync

Show input JSON schema

{
  "additionalProperties": false,
  "properties": {
    "workspace_id": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "description": "Workspace ID. Hosted MCP connections pass it via the `X-Airbyte-Workspace-Id` header; local or stdio connections use the `AIRBYTE_CLOUD_WORKSPACE_ID` environment variable."
    },
    "name_contains": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "description": "Optional case-insensitive substring to filter connections by name"
    },
    "limit": {
      "anyOf": [
        {
          "type": "integer"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "description": "Optional maximum number of items to return (default: no limit)"
    },
    "with_connection_status": {
      "anyOf": [
        {
          "type": "boolean"
        },
        {
          "type": "null"
        }
      ],
      "default": false,
      "description": "If True, include status info for each connection's most recent sync job"
    },
    "failing_connections_only": {
      "anyOf": [
        {
          "type": "boolean"
        },
        {
          "type": "null"
        }
      ],
      "default": false,
      "description": "If True, only return connections with failed/cancelled last sync"
    }
  },
  "type": "object"
}

Show output JSON schema

{
  "properties": {
    "result": {
      "items": {
        "description": "Information about a deployed connection in Airbyte Cloud.",
        "properties": {
          "id": {
            "type": "string"
          },
          "name": {
            "type": "string"
          },
          "url": {
            "type": "string"
          },
          "source_id": {
            "type": "string"
          },
          "destination_id": {
            "type": "string"
          },
          "last_job_status": {
            "anyOf": [
              {
                "type": "string"
              },
              {
                "type": "null"
              }
            ],
            "default": null
          },
          "last_job_id": {
            "anyOf": [
              {
                "type": "integer"
              },
              {
                "type": "null"
              }
            ],
            "default": null
          },
          "last_job_time": {
            "anyOf": [
              {
                "type": "string"
              },
              {
                "type": "null"
              }
            ],
            "default": null
          },
          "currently_running_job_id": {
            "anyOf": [
              {
                "type": "integer"
              },
              {
                "type": "null"
              }
            ],
            "default": null
          },
          "currently_running_job_start_time": {
            "anyOf": [
              {
                "type": "string"
              },
              {
                "type": "null"
              }
            ],
            "default": null
          }
        },
        "required": [
          "id",
          "name",
          "url",
          "source_id",
          "destination_id"
        ],
        "type": "object"
      },
      "type": "array"
    }
  },
  "required": [
    "result"
  ],
  "type": "object",
  "x-fastmcp-wrap-result": true
}

list_deployed_cloud_destination_connectors

Hints: read-only · idempotent · open-world

List all deployed destination connectors in the Airbyte Cloud workspace.

Parameters

Name	Type	Required	Default	Description
`workspace_id`	`string \| null`	no	`null`	Workspace ID. Hosted MCP connections pass it via the `X-Airbyte-Workspace-Id` header; local or stdio connections use the `AIRBYTE_CLOUD_WORKSPACE_ID` environment variable.
`name_contains`	`string \| null`	no	`null`	Optional case-insensitive substring to filter destinations by name
`limit`	`integer \| null`	no	`null`	Optional maximum number of items to return (default: no limit)

Show input JSON schema

{
  "additionalProperties": false,
  "properties": {
    "workspace_id": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "description": "Workspace ID. Hosted MCP connections pass it via the `X-Airbyte-Workspace-Id` header; local or stdio connections use the `AIRBYTE_CLOUD_WORKSPACE_ID` environment variable."
    },
    "name_contains": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "description": "Optional case-insensitive substring to filter destinations by name"
    },
    "limit": {
      "anyOf": [
        {
          "type": "integer"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "description": "Optional maximum number of items to return (default: no limit)"
    }
  },
  "type": "object"
}

Show output JSON schema

{
  "properties": {
    "result": {
      "items": {
        "description": "Information about a deployed destination connector in Airbyte Cloud.",
        "properties": {
          "id": {
            "type": "string"
          },
          "name": {
            "type": "string"
          },
          "url": {
            "type": "string"
          }
        },
        "required": [
          "id",
          "name",
          "url"
        ],
        "type": "object"
      },
      "type": "array"
    }
  },
  "required": [
    "result"
  ],
  "type": "object",
  "x-fastmcp-wrap-result": true
}

list_deployed_cloud_source_connectors

Hints: read-only · idempotent · open-world

List all deployed source connectors in the Airbyte Cloud workspace.

Parameters

Name	Type	Required	Default	Description
`workspace_id`	`string \| null`	no	`null`	Workspace ID. Hosted MCP connections pass it via the `X-Airbyte-Workspace-Id` header; local or stdio connections use the `AIRBYTE_CLOUD_WORKSPACE_ID` environment variable.
`name_contains`	`string \| null`	no	`null`	Optional case-insensitive substring to filter sources by name
`limit`	`integer \| null`	no	`null`	Optional maximum number of items to return (default: no limit)

Show input JSON schema

{
  "additionalProperties": false,
  "properties": {
    "workspace_id": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "description": "Workspace ID. Hosted MCP connections pass it via the `X-Airbyte-Workspace-Id` header; local or stdio connections use the `AIRBYTE_CLOUD_WORKSPACE_ID` environment variable."
    },
    "name_contains": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "description": "Optional case-insensitive substring to filter sources by name"
    },
    "limit": {
      "anyOf": [
        {
          "type": "integer"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "description": "Optional maximum number of items to return (default: no limit)"
    }
  },
  "type": "object"
}

Show output JSON schema

{
  "properties": {
    "result": {
      "items": {
        "description": "Information about a deployed source connector in Airbyte Cloud.",
        "properties": {
          "id": {
            "type": "string"
          },
          "name": {
            "type": "string"
          },
          "url": {
            "type": "string"
          }
        },
        "required": [
          "id",
          "name",
          "url"
        ],
        "type": "object"
      },
      "type": "array"
    }
  },
  "required": [
    "result"
  ],
  "type": "object",
  "x-fastmcp-wrap-result": true
}

permanently_delete_cloud_connection

Hints: destructive · open-world

Permanently delete a connection from Airbyte Cloud.

IMPORTANT: This operation requires the connection name to contain "delete-me" or "deleteme"
(case insensitive).

If the connection does not meet this requirement, the deletion will be rejected with a
helpful error message. Instruct the user to rename the connection appropriately to authorize
the deletion.

The provided name must match the actual name of the connection for the operation to proceed.
This is a safety measure to ensure you are deleting the correct resource.

Parameters

Name	Type	Required	Default	Description
`connection_id`	`string`	yes	—	The ID of the connection to delete.
`name`	`string`	yes	—	The expected name of the connection (for verification).
`cascade_delete_source`	`boolean`	no	`false`	Whether to also delete the source connector associated with this connection.
`cascade_delete_destination`	`boolean`	no	`false`	Whether to also delete the destination connector associated with this connection.

Show input JSON schema

{
  "additionalProperties": false,
  "properties": {
    "connection_id": {
      "description": "The ID of the connection to delete.",
      "type": "string"
    },
    "name": {
      "description": "The expected name of the connection (for verification).",
      "type": "string"
    },
    "cascade_delete_source": {
      "default": false,
      "description": "Whether to also delete the source connector associated with this connection.",
      "type": "boolean"
    },
    "cascade_delete_destination": {
      "default": false,
      "description": "Whether to also delete the destination connector associated with this connection.",
      "type": "boolean"
    }
  },
  "required": [
    "connection_id",
    "name"
  ],
  "type": "object"
}

Show output JSON schema

{
  "properties": {
    "result": {
      "type": "string"
    }
  },
  "required": [
    "result"
  ],
  "type": "object",
  "x-fastmcp-wrap-result": true
}

permanently_delete_cloud_destination

Hints: destructive · open-world

Permanently delete a deployed destination connector from Airbyte Cloud.

IMPORTANT: This operation requires the destination name to contain "delete-me" or "deleteme"
(case insensitive).

If the destination does not meet this requirement, the deletion will be rejected with a
helpful error message. Instruct the user to rename the destination appropriately to authorize
the deletion.

The provided name must match the actual name of the destination for the operation to proceed.
This is a safety measure to ensure you are deleting the correct resource.

Parameters

Name	Type	Required	Default	Description
`destination_id`	`string`	yes	—	The ID of the deployed destination to delete.
`name`	`string`	yes	—	The expected name of the destination (for verification).

Show input JSON schema

{
  "additionalProperties": false,
  "properties": {
    "destination_id": {
      "description": "The ID of the deployed destination to delete.",
      "type": "string"
    },
    "name": {
      "description": "The expected name of the destination (for verification).",
      "type": "string"
    }
  },
  "required": [
    "destination_id",
    "name"
  ],
  "type": "object"
}

Show output JSON schema

{
  "properties": {
    "result": {
      "type": "string"
    }
  },
  "required": [
    "result"
  ],
  "type": "object",
  "x-fastmcp-wrap-result": true
}

permanently_delete_cloud_source

Hints: destructive · open-world

Permanently delete a deployed source connector from Airbyte Cloud.

IMPORTANT: This operation requires the source name to contain "delete-me" or "deleteme"
(case insensitive).

If the source does not meet this requirement, the deletion will be rejected with a
helpful error message. Instruct the user to rename the source appropriately to authorize
the deletion.

The provided name must match the actual name of the source for the operation to proceed.
This is a safety measure to ensure you are deleting the correct resource.

Parameters

Name	Type	Required	Default	Description
`source_id`	`string`	yes	—	The ID of the deployed source to delete.
`name`	`string`	yes	—	The expected name of the source (for verification).

Show input JSON schema

{
  "additionalProperties": false,
  "properties": {
    "source_id": {
      "description": "The ID of the deployed source to delete.",
      "type": "string"
    },
    "name": {
      "description": "The expected name of the source (for verification).",
      "type": "string"
    }
  },
  "required": [
    "source_id",
    "name"
  ],
  "type": "object"
}

Show output JSON schema

{
  "properties": {
    "result": {
      "type": "string"
    }
  },
  "required": [
    "result"
  ],
  "type": "object",
  "x-fastmcp-wrap-result": true
}

permanently_delete_custom_source_definition

Hints: destructive · open-world

Permanently delete a custom YAML source definition from Airbyte Cloud.

IMPORTANT: This operation requires the connector name to contain "delete-me" or "deleteme" (case insensitive).

If the connector does not meet this requirement, the deletion will be rejected with a helpful error message. Instruct the user to rename the connector appropriately to authorize the deletion.

The provided name must match the actual name of the definition for the operation to proceed. This is a safety measure to ensure you are deleting the correct resource.

Note: Only YAML (declarative) connectors are currently supported. Docker-based custom sources are not yet available.

Parameters

Name	Type	Required	Default	Description
`definition_id`	`string`	yes	—	The ID of the custom source definition to delete.
`name`	`string`	yes	—	The expected name of the custom source definition (for verification).
`workspace_id`	`string \| null`	no	`null`	Workspace ID. Hosted MCP connections pass it via the `X-Airbyte-Workspace-Id` header; local or stdio connections use the `AIRBYTE_CLOUD_WORKSPACE_ID` environment variable.

Show input JSON schema

{
  "additionalProperties": false,
  "properties": {
    "definition_id": {
      "description": "The ID of the custom source definition to delete.",
      "type": "string"
    },
    "name": {
      "description": "The expected name of the custom source definition (for verification).",
      "type": "string"
    },
    "workspace_id": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "description": "Workspace ID. Hosted MCP connections pass it via the `X-Airbyte-Workspace-Id` header; local or stdio connections use the `AIRBYTE_CLOUD_WORKSPACE_ID` environment variable."
    }
  },
  "required": [
    "definition_id",
    "name"
  ],
  "type": "object"
}

Show output JSON schema

{
  "properties": {
    "result": {
      "type": "string"
    }
  },
  "required": [
    "result"
  ],
  "type": "object",
  "x-fastmcp-wrap-result": true
}

publish_custom_source_definition

Hints: open-world

Publish a custom YAML source connector definition to Airbyte Cloud.

Note: Only YAML (declarative) connectors are currently supported.
Docker-based custom sources are not yet available.

Parameters

Name	Type	Required	Default	Description
`name`	`string`	yes	—	The name for the custom connector definition.
`workspace_id`	`string \| null`	no	`null`	Workspace ID. Hosted MCP connections pass it via the `X-Airbyte-Workspace-Id` header; local or stdio connections use the `AIRBYTE_CLOUD_WORKSPACE_ID` environment variable.
`manifest_yaml`	`string \| string \| null \| null`	no	`null`
`unique`	`boolean`	no	`true`	Whether to require a unique name.
`pre_validate`	`boolean`	no	`true`	Whether to validate the manifest client-side before publishing.
`testing_values`	`object \| string \| null`	no	`null`	Optional testing configuration values for the Builder UI. Can be provided as a JSON object or JSON string. Supports inline secret refs via 'secret_reference::ENV_VAR_NAME' syntax. If provided, these values replace any existing testing values for the connector builder project, allowing immediate test read operations.
`testing_values_secret_name`	`string \| null`	no	`null`	Optional name of a secret containing testing configuration values in JSON or YAML format. The secret will be resolved by the MCP server and merged into testing_values, with secret values taking precedence. This lets the agent reference secrets without sending raw values as tool arguments.

Show input JSON schema

{
  "additionalProperties": false,
  "properties": {
    "name": {
      "description": "The name for the custom connector definition.",
      "type": "string"
    },
    "workspace_id": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "description": "Workspace ID. Hosted MCP connections pass it via the `X-Airbyte-Workspace-Id` header; local or stdio connections use the `AIRBYTE_CLOUD_WORKSPACE_ID` environment variable."
    },
    "manifest_yaml": {
      "anyOf": [
        {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "format": "path",
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "description": "The Low-code CDK manifest as a YAML string or file path. Required for YAML connectors."
        },
        {
          "type": "null"
        }
      ],
      "default": null
    },
    "unique": {
      "default": true,
      "description": "Whether to require a unique name.",
      "type": "boolean"
    },
    "pre_validate": {
      "default": true,
      "description": "Whether to validate the manifest client-side before publishing.",
      "type": "boolean"
    },
    "testing_values": {
      "anyOf": [
        {
          "additionalProperties": true,
          "type": "object"
        },
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "description": "Optional testing configuration values for the Builder UI. Can be provided as a JSON object or JSON string. Supports inline secret refs via 'secret_reference::ENV_VAR_NAME' syntax. If provided, these values replace any existing testing values for the connector builder project, allowing immediate test read operations."
    },
    "testing_values_secret_name": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "description": "Optional name of a secret containing testing configuration values in JSON or YAML format. The secret will be resolved by the MCP server and merged into testing_values, with secret values taking precedence. This lets the agent reference secrets without sending raw values as tool arguments."
    }
  },
  "required": [
    "name"
  ],
  "type": "object"
}

Show output JSON schema

{
  "properties": {
    "result": {
      "type": "string"
    }
  },
  "required": [
    "result"
  ],
  "type": "object",
  "x-fastmcp-wrap-result": true
}

rename_cloud_connection

Hints: open-world

Rename a connection on Airbyte Cloud.

Parameters

Name	Type	Required	Default	Description
`connection_id`	`string`	yes	—	The ID of the connection to rename.
`name`	`string`	yes	—	New name for the connection.
`workspace_id`	`string \| null`	no	`null`	Workspace ID. Hosted MCP connections pass it via the `X-Airbyte-Workspace-Id` header; local or stdio connections use the `AIRBYTE_CLOUD_WORKSPACE_ID` environment variable.

Show input JSON schema

{
  "additionalProperties": false,
  "properties": {
    "connection_id": {
      "description": "The ID of the connection to rename.",
      "type": "string"
    },
    "name": {
      "description": "New name for the connection.",
      "type": "string"
    },
    "workspace_id": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "description": "Workspace ID. Hosted MCP connections pass it via the `X-Airbyte-Workspace-Id` header; local or stdio connections use the `AIRBYTE_CLOUD_WORKSPACE_ID` environment variable."
    }
  },
  "required": [
    "connection_id",
    "name"
  ],
  "type": "object"
}

Show output JSON schema

{
  "properties": {
    "result": {
      "type": "string"
    }
  },
  "required": [
    "result"
  ],
  "type": "object",
  "x-fastmcp-wrap-result": true
}

rename_cloud_destination

Hints: open-world

Rename a deployed destination connector on Airbyte Cloud.

Parameters

Name	Type	Required	Default	Description
`destination_id`	`string`	yes	—	The ID of the deployed destination to rename.
`name`	`string`	yes	—	New name for the destination.
`workspace_id`	`string \| null`	no	`null`	Workspace ID. Hosted MCP connections pass it via the `X-Airbyte-Workspace-Id` header; local or stdio connections use the `AIRBYTE_CLOUD_WORKSPACE_ID` environment variable.

Show input JSON schema

{
  "additionalProperties": false,
  "properties": {
    "destination_id": {
      "description": "The ID of the deployed destination to rename.",
      "type": "string"
    },
    "name": {
      "description": "New name for the destination.",
      "type": "string"
    },
    "workspace_id": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "description": "Workspace ID. Hosted MCP connections pass it via the `X-Airbyte-Workspace-Id` header; local or stdio connections use the `AIRBYTE_CLOUD_WORKSPACE_ID` environment variable."
    }
  },
  "required": [
    "destination_id",
    "name"
  ],
  "type": "object"
}

Show output JSON schema

{
  "properties": {
    "result": {
      "type": "string"
    }
  },
  "required": [
    "result"
  ],
  "type": "object",
  "x-fastmcp-wrap-result": true
}

rename_cloud_source

Hints: open-world

Rename a deployed source connector on Airbyte Cloud.

Parameters

Name	Type	Required	Default	Description
`source_id`	`string`	yes	—	The ID of the deployed source to rename.
`name`	`string`	yes	—	New name for the source.
`workspace_id`	`string \| null`	no	`null`	Workspace ID. Hosted MCP connections pass it via the `X-Airbyte-Workspace-Id` header; local or stdio connections use the `AIRBYTE_CLOUD_WORKSPACE_ID` environment variable.

Show input JSON schema

{
  "additionalProperties": false,
  "properties": {
    "source_id": {
      "description": "The ID of the deployed source to rename.",
      "type": "string"
    },
    "name": {
      "description": "New name for the source.",
      "type": "string"
    },
    "workspace_id": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "description": "Workspace ID. Hosted MCP connections pass it via the `X-Airbyte-Workspace-Id` header; local or stdio connections use the `AIRBYTE_CLOUD_WORKSPACE_ID` environment variable."
    }
  },
  "required": [
    "source_id",
    "name"
  ],
  "type": "object"
}

Show output JSON schema

{
  "properties": {
    "result": {
      "type": "string"
    }
  },
  "required": [
    "result"
  ],
  "type": "object",
  "x-fastmcp-wrap-result": true
}

run_cloud_sync

Hints: open-world

Run a sync job on Airbyte Cloud.

Parameters

Name	Type	Required	Default	Description
`connection_id`	`string`	yes	—	The ID of the Airbyte Cloud connection.
`workspace_id`	`string \| null`	no	`null`	Workspace ID. Hosted MCP connections pass it via the `X-Airbyte-Workspace-Id` header; local or stdio connections use the `AIRBYTE_CLOUD_WORKSPACE_ID` environment variable.
`wait`	`boolean`	no	`false`	Whether to wait for the sync to complete. Since a sync can take between several minutes and several hours, this option is not recommended for most scenarios.
`wait_timeout`	`integer`	no	`300`	Maximum time to wait for sync completion (seconds).

Show input JSON schema

{
  "additionalProperties": false,
  "properties": {
    "connection_id": {
      "description": "The ID of the Airbyte Cloud connection.",
      "type": "string"
    },
    "workspace_id": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "description": "Workspace ID. Hosted MCP connections pass it via the `X-Airbyte-Workspace-Id` header; local or stdio connections use the `AIRBYTE_CLOUD_WORKSPACE_ID` environment variable."
    },
    "wait": {
      "default": false,
      "description": "Whether to wait for the sync to complete. Since a sync can take between several minutes and several hours, this option is not recommended for most scenarios.",
      "type": "boolean"
    },
    "wait_timeout": {
      "default": 300,
      "description": "Maximum time to wait for sync completion (seconds).",
      "type": "integer"
    }
  },
  "required": [
    "connection_id"
  ],
  "type": "object"
}

Show output JSON schema

{
  "properties": {
    "result": {
      "type": "string"
    }
  },
  "required": [
    "result"
  ],
  "type": "object",
  "x-fastmcp-wrap-result": true
}

set_cloud_connection_selected_streams

Hints: destructive · open-world

Set the selected streams for a connection on Airbyte Cloud.

This is a destructive operation that can break existing connections if the
stream selection is changed incorrectly. Use with caution.

Parameters

Name	Type	Required	Default	Description
`connection_id`	`string`	yes	—	The ID of the connection to update.
`stream_names`	`string \| array<string>`	yes	—	The selected stream names to sync within the connection. Must be an explicit stream name or list of streams.
`workspace_id`	`string \| null`	no	`null`	Workspace ID. Hosted MCP connections pass it via the `X-Airbyte-Workspace-Id` header; local or stdio connections use the `AIRBYTE_CLOUD_WORKSPACE_ID` environment variable.

Show input JSON schema

{
  "additionalProperties": false,
  "properties": {
    "connection_id": {
      "description": "The ID of the connection to update.",
      "type": "string"
    },
    "stream_names": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "items": {
            "type": "string"
          },
          "type": "array"
        }
      ],
      "description": "The selected stream names to sync within the connection. Must be an explicit stream name or list of streams."
    },
    "workspace_id": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "description": "Workspace ID. Hosted MCP connections pass it via the `X-Airbyte-Workspace-Id` header; local or stdio connections use the `AIRBYTE_CLOUD_WORKSPACE_ID` environment variable."
    }
  },
  "required": [
    "connection_id",
    "stream_names"
  ],
  "type": "object"
}

Show output JSON schema

{
  "properties": {
    "result": {
      "type": "string"
    }
  },
  "required": [
    "result"
  ],
  "type": "object",
  "x-fastmcp-wrap-result": true
}

set_cloud_connection_table_prefix

Hints: destructive · open-world

Set the table prefix for a connection on Airbyte Cloud.

This is a destructive operation that can break downstream dependencies if the
table prefix is changed incorrectly. Use with caution.

Parameters

Name	Type	Required	Default	Description
`connection_id`	`string`	yes	—	The ID of the connection to update.
`prefix`	`string`	yes	—	New table prefix to use when syncing to the destination.
`workspace_id`	`string \| null`	no	`null`	Workspace ID. Hosted MCP connections pass it via the `X-Airbyte-Workspace-Id` header; local or stdio connections use the `AIRBYTE_CLOUD_WORKSPACE_ID` environment variable.

Show input JSON schema

{
  "additionalProperties": false,
  "properties": {
    "connection_id": {
      "description": "The ID of the connection to update.",
      "type": "string"
    },
    "prefix": {
      "description": "New table prefix to use when syncing to the destination.",
      "type": "string"
    },
    "workspace_id": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "description": "Workspace ID. Hosted MCP connections pass it via the `X-Airbyte-Workspace-Id` header; local or stdio connections use the `AIRBYTE_CLOUD_WORKSPACE_ID` environment variable."
    }
  },
  "required": [
    "connection_id",
    "prefix"
  ],
  "type": "object"
}

Show output JSON schema

{
  "properties": {
    "result": {
      "type": "string"
    }
  },
  "required": [
    "result"
  ],
  "type": "object",
  "x-fastmcp-wrap-result": true
}

update_cloud_connection

Hints: destructive · open-world

Update a connection's settings on Airbyte Cloud.

This tool allows updating multiple connection settings in a single call:
- Enable or disable the connection
- Set a cron schedule for automatic syncs
- Switch to manual scheduling (no automatic syncs)

At least one setting must be provided. The 'cron_expression' and 'manual_schedule'
parameters are mutually exclusive.

Parameters

Name	Type	Required	Default	Description
`connection_id`	`string`	yes	—	The ID of the connection to update.
`enabled`	`boolean \| null`	no	`null`	Set the connection's enabled status. True enables the connection (status='active'), False disables it (status='inactive'). Leave unset to keep the current status.
`cron_expression`	`string \| null`	no	`null`	A cron expression defining when syncs should run. Examples: '0 0 * * ' (daily at midnight UTC), '0 /6 * * ' (every 6 hours), '0 0 * 0' (weekly on Sunday at midnight UTC). Leave unset to keep the current schedule. Cannot be used together with 'manual_schedule'.
`manual_schedule`	`boolean \| null`	no	`null`	Set to True to disable automatic syncs (manual scheduling only). Syncs will only run when manually triggered. Cannot be used together with 'cron_expression'.
`workspace_id`	`string \| null`	no	`null`	Workspace ID. Hosted MCP connections pass it via the `X-Airbyte-Workspace-Id` header; local or stdio connections use the `AIRBYTE_CLOUD_WORKSPACE_ID` environment variable.

Show input JSON schema

{
  "additionalProperties": false,
  "properties": {
    "connection_id": {
      "description": "The ID of the connection to update.",
      "type": "string"
    },
    "enabled": {
      "anyOf": [
        {
          "type": "boolean"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "description": "Set the connection's enabled status. True enables the connection (status='active'), False disables it (status='inactive'). Leave unset to keep the current status."
    },
    "cron_expression": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "description": "A cron expression defining when syncs should run. Examples: '0 0 * * *' (daily at midnight UTC), '0 */6 * * *' (every 6 hours), '0 0 * * 0' (weekly on Sunday at midnight UTC). Leave unset to keep the current schedule. Cannot be used together with 'manual_schedule'."
    },
    "manual_schedule": {
      "anyOf": [
        {
          "type": "boolean"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "description": "Set to True to disable automatic syncs (manual scheduling only). Syncs will only run when manually triggered. Cannot be used together with 'cron_expression'."
    },
    "workspace_id": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "description": "Workspace ID. Hosted MCP connections pass it via the `X-Airbyte-Workspace-Id` header; local or stdio connections use the `AIRBYTE_CLOUD_WORKSPACE_ID` environment variable."
    }
  },
  "required": [
    "connection_id"
  ],
  "type": "object"
}

Show output JSON schema

{
  "properties": {
    "result": {
      "type": "string"
    }
  },
  "required": [
    "result"
  ],
  "type": "object",
  "x-fastmcp-wrap-result": true
}

update_cloud_destination_config

Hints: destructive · open-world

Update a deployed destination connector's configuration on Airbyte Cloud.

This is a destructive operation that can break existing connections if the
configuration is changed incorrectly. Use with caution.

Parameters

Name	Type	Required	Default	Description
`destination_id`	`string`	yes	—	The ID of the deployed destination to update.
`config`	`object \| string`	yes	—	New configuration for the destination connector.
`config_secret_name`	`string \| null`	no	`null`	The name of the secret containing the configuration.
`workspace_id`	`string \| null`	no	`null`	Workspace ID. Hosted MCP connections pass it via the `X-Airbyte-Workspace-Id` header; local or stdio connections use the `AIRBYTE_CLOUD_WORKSPACE_ID` environment variable.

Show input JSON schema

{
  "additionalProperties": false,
  "properties": {
    "destination_id": {
      "description": "The ID of the deployed destination to update.",
      "type": "string"
    },
    "config": {
      "anyOf": [
        {
          "additionalProperties": true,
          "type": "object"
        },
        {
          "type": "string"
        }
      ],
      "description": "New configuration for the destination connector."
    },
    "config_secret_name": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "description": "The name of the secret containing the configuration."
    },
    "workspace_id": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "description": "Workspace ID. Hosted MCP connections pass it via the `X-Airbyte-Workspace-Id` header; local or stdio connections use the `AIRBYTE_CLOUD_WORKSPACE_ID` environment variable."
    }
  },
  "required": [
    "destination_id",
    "config"
  ],
  "type": "object"
}

Show output JSON schema

{
  "properties": {
    "result": {
      "type": "string"
    }
  },
  "required": [
    "result"
  ],
  "type": "object",
  "x-fastmcp-wrap-result": true
}

update_cloud_source_config

Hints: destructive · open-world

Update a deployed source connector's configuration on Airbyte Cloud.

This is a destructive operation that can break existing connections if the
configuration is changed incorrectly. Use with caution.

Parameters

Name	Type	Required	Default	Description
`source_id`	`string`	yes	—	The ID of the deployed source to update.
`config`	`object \| string`	yes	—	New configuration for the source connector.
`config_secret_name`	`string \| null \| null`	no	`null`
`workspace_id`	`string \| null`	no	`null`	Workspace ID. Hosted MCP connections pass it via the `X-Airbyte-Workspace-Id` header; local or stdio connections use the `AIRBYTE_CLOUD_WORKSPACE_ID` environment variable.

Show input JSON schema

{
  "additionalProperties": false,
  "properties": {
    "source_id": {
      "description": "The ID of the deployed source to update.",
      "type": "string"
    },
    "config": {
      "anyOf": [
        {
          "additionalProperties": true,
          "type": "object"
        },
        {
          "type": "string"
        }
      ],
      "description": "New configuration for the source connector."
    },
    "config_secret_name": {
      "anyOf": [
        {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "description": "The name of the secret containing the configuration."
        },
        {
          "type": "null"
        }
      ],
      "default": null
    },
    "workspace_id": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "description": "Workspace ID. Hosted MCP connections pass it via the `X-Airbyte-Workspace-Id` header; local or stdio connections use the `AIRBYTE_CLOUD_WORKSPACE_ID` environment variable."
    }
  },
  "required": [
    "source_id",
    "config"
  ],
  "type": "object"
}

Show output JSON schema

{
  "properties": {
    "result": {
      "type": "string"
    }
  },
  "required": [
    "result"
  ],
  "type": "object",
  "x-fastmcp-wrap-result": true
}

update_custom_source_definition

Hints: destructive · open-world

Update a custom YAML source definition in Airbyte Cloud.

Updates the manifest and/or testing values for an existing custom source definition. At least one of manifest_yaml, testing_values, or testing_values_secret_name must be provided.

Parameters

Name	Type	Required	Default	Description
`definition_id`	`string`	yes	—	The ID of the definition to update.
`manifest_yaml`	`string \| string \| null \| null`	no	`null`
`workspace_id`	`string \| null`	no	`null`	Workspace ID. Hosted MCP connections pass it via the `X-Airbyte-Workspace-Id` header; local or stdio connections use the `AIRBYTE_CLOUD_WORKSPACE_ID` environment variable.
`pre_validate`	`boolean`	no	`true`	Whether to validate the manifest client-side before updating.
`testing_values`	`object \| string \| null`	no	`null`	Optional testing configuration values for the Builder UI. Can be provided as a JSON object or JSON string. Supports inline secret refs via 'secret_reference::ENV_VAR_NAME' syntax. If provided, these values replace any existing testing values for the connector builder project. The entire testing values object is overwritten, so pass the full set of values you want to persist.
`testing_values_secret_name`	`string \| null`	no	`null`	Optional name of a secret containing testing configuration values in JSON or YAML format. The secret will be resolved by the MCP server and merged into testing_values, with secret values taking precedence. This lets the agent reference secrets without sending raw values as tool arguments.

Show input JSON schema

{
  "additionalProperties": false,
  "properties": {
    "definition_id": {
      "description": "The ID of the definition to update.",
      "type": "string"
    },
    "manifest_yaml": {
      "anyOf": [
        {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "format": "path",
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "description": "New manifest as YAML string or file path. Optional; omit to update only testing values."
        },
        {
          "type": "null"
        }
      ],
      "default": null
    },
    "workspace_id": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "description": "Workspace ID. Hosted MCP connections pass it via the `X-Airbyte-Workspace-Id` header; local or stdio connections use the `AIRBYTE_CLOUD_WORKSPACE_ID` environment variable."
    },
    "pre_validate": {
      "default": true,
      "description": "Whether to validate the manifest client-side before updating.",
      "type": "boolean"
    },
    "testing_values": {
      "anyOf": [
        {
          "additionalProperties": true,
          "type": "object"
        },
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "description": "Optional testing configuration values for the Builder UI. Can be provided as a JSON object or JSON string. Supports inline secret refs via 'secret_reference::ENV_VAR_NAME' syntax. If provided, these values replace any existing testing values for the connector builder project. The entire testing values object is overwritten, so pass the full set of values you want to persist."
    },
    "testing_values_secret_name": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "description": "Optional name of a secret containing testing configuration values in JSON or YAML format. The secret will be resolved by the MCP server and merged into testing_values, with secret values taking precedence. This lets the agent reference secrets without sending raw values as tool arguments."
    }
  },
  "required": [
    "definition_id"
  ],
  "type": "object"
}

Show output JSON schema

{
  "properties": {
    "result": {
      "type": "string"
    }
  },
  "required": [
    "result"
  ],
  "type": "object",
  "x-fastmcp-wrap-result": true
}

View Source

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