airbyte.mcp.cloud

Airbyte Cloud MCP operations.

cloud module

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

Tools (35)

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.

When connecting to a hosted MCP server, provide a bearer token via the Authorization header, or client credentials via the X-Airbyte-Cloud-Client-Id and X-Airbyte-Cloud-Client-Secret headers, plus the workspace ID via the X-Airbyte-Workspace-Id header. For local or stdio connections, set the AIRBYTE_CLOUD_BEARER_TOKEN environment variable, or both AIRBYTE_CLOUD_CLIENT_ID and AIRBYTE_CLOUD_CLIENT_SECRET, plus AIRBYTE_CLOUD_WORKSPACE_ID for the 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.

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": {
      "type": "string"
    },
    "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"
}

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_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 in a specific organization.

Requires either organization_id OR organization_name (exact match) to be provided.
This tool will NOT list workspaces across all organizations - you must specify
which organization to list workspaces from.

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.
`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": "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."
    },
    "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

{
  "properties": {
    "result": {
      "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": {
            "type": "string"
          },
          "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"
    }
  },
  "required": [
    "result"
  ],
  "type": "object",
  "x-fastmcp-wrap-result": true
}

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