airbyte.mcp.cloud

Airbyte Cloud MCP operations.

cloud module

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

Tools (43)

cancel_cloud_sync

Hints: destructive · open-world

Cancel a running sync job on an Airbyte Cloud connection.

When connecting to a hosted MCP server, provide a bearer token via the Authorization header, or client credentials via the transport Client-Id and Client-Secret headers. When no workspace ID is provided, the authenticated user's default workspace (and its organization) is used automatically. Call get_default_cloud_context to inspect the resolved context. To discover other workspaces, call list_cloud_workspaces with an organization ID or broader privilege scope. Only call list_cloud_organizations when you need to search organizations by name, passing name_contains. For local or stdio connections, set the AIRBYTE_CLOUD_BEARER_TOKEN environment variable, or both AIRBYTE_CLOUD_CLIENT_ID and AIRBYTE_CLOUD_CLIENT_SECRET. If discovery returns multiple candidates, ask the user to choose one; do not select automatically.

Parameters

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

Show input JSON schema

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

Show output JSON schema

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

check_cloud_connector_setup

Hints: open-world

Check whether a connector deployed with deferred credentials is ready to use.

Call this once after the person confirms they authenticated, tested and saved the connector
in Airbyte Cloud. It runs a single connection check and reports only whether it passed;
failure details stay in Cloud, where the person corrects the configuration.

Parameters

Name	Type	Required	Default	Description
`connector_type`	`enum("source", "destination")`	yes	—	Whether the connector is a `source` or a `destination`.
`connector_id`	`string`	yes	—	The ID returned by a deployment with `defer_credentials=True`.
`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": {
    "connector_type": {
      "description": "Whether the connector is a `source` or a `destination`.",
      "enum": [
        "source",
        "destination"
      ],
      "type": "string"
    },
    "connector_id": {
      "description": "The ID returned by a deployment with `defer_credentials=True`.",
      "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": [
    "connector_type",
    "connector_id"
  ],
  "type": "object"
}

Show output JSON schema

{
  "description": "Result of checking a connector deployed with deferred credentials.",
  "properties": {
    "connector_id": {
      "type": "string"
    },
    "connector_type": {
      "enum": [
        "source",
        "destination"
      ],
      "type": "string"
    },
    "setup_complete": {
      "type": "boolean"
    },
    "settings_url": {
      "type": "string"
    },
    "guidance": {
      "type": "string"
    }
  },
  "required": [
    "connector_id",
    "connector_type",
    "setup_complete",
    "settings_url",
    "guidance"
  ],
  "type": "object"
}

check_cloud_destination

Hints: read-only · idempotent · open-world

Check the configuration and credentials of a deployed destination connector.

Parameters

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

Show input JSON schema

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

Show output JSON schema

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

check_cloud_source

Hints: read-only · idempotent · open-world

Check the configuration and credentials of a deployed source connector.

Parameters

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

Show input JSON schema

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

Show output JSON schema

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

create_connection_on_cloud

Hints: open-world

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

Parameters

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

Show input JSON schema

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

Show output JSON schema

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

deploy_destination_to_cloud

Hints: open-world

Deploy a destination connector to Airbyte Cloud.

With `defer_credentials=True`, returns a JSON `DeferredDeployResult` whose `settings_url`
is where a person completes the credentials.

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.
`defer_credentials`	`boolean`	no	`false`	Create a draft connector so a person can complete OAuth or enter secrets in Airbyte Cloud. Pass only non-secret configuration in `config` (no credentials, secret references or `config_secret_name`). When the connector offers several authentication methods, include the method's selector field in `config`. The result includes a settings link for the person to complete any missing fields and test the draft. After they report a successful test and save, call `check_cloud_connector_setup`.

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"
    },
    "defer_credentials": {
      "default": false,
      "description": "Create a draft connector so a person can complete OAuth or enter secrets in Airbyte Cloud. Pass only non-secret configuration in `config` (no credentials, secret references or `config_secret_name`). When the connector offers several authentication methods, include the method's selector field in `config`. The result includes a settings link for the person to complete any missing fields and test the draft. After they report a successful test and save, call `check_cloud_connector_setup`.",
      "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
`destination_name`	`string`	no	`"No-op Destination"`	The name to use when deploying the 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`	Whether to require a unique name.

Show input JSON schema

{
  "additionalProperties": false,
  "properties": {
    "destination_name": {
      "default": "No-op Destination",
      "description": "The name to use when deploying the 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,
      "description": "Whether to require a unique name.",
      "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.

With `defer_credentials=True`, returns a JSON `DeferredDeployResult` whose `settings_url`
is where a person completes the credentials.

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.
`defer_credentials`	`boolean`	no	`false`	Create a draft connector so a person can complete OAuth or enter secrets in Airbyte Cloud. Pass only non-secret configuration in `config` (no credentials, secret references or `config_secret_name`). When the connector offers several authentication methods, include the method's selector field in `config`. The result includes a settings link for the person to complete any missing fields and test the draft. After they report a successful test and save, call `check_cloud_connector_setup`.

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"
    },
    "defer_credentials": {
      "default": false,
      "description": "Create a draft connector so a person can complete OAuth or enter secrets in Airbyte Cloud. Pass only non-secret configuration in `config` (no credentials, secret references or `config_secret_name`). When the connector offers several authentication methods, include the method's selector field in `config`. The result includes a settings link for the person to complete any missing fields and test the draft. After they report a successful test and save, call `check_cloud_connector_setup`.",
      "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 basic details about an organization (ID, name, email).

Billing/account status is available via `get_cloud_organization_billing_status`.

With no arguments, resolves the organization from the configured default or the
authenticated user's sole membership. With multiple memberships, the error lists
candidate organization IDs. Use organization_id or organization_name (exact match)
to look up a specific organization.

Parameters

Name	Type	Required	Default	Description
`organization_id`	`string \| null`	no	`null`	Organization ID. With no arguments, resolves from the configured default or the authenticated user's sole membership.
`organization_name`	`string \| null`	no	`null`	Organization name (exact match). With no arguments, resolves from the configured default or the authenticated user's sole membership. With multiple memberships, the error lists candidate organization IDs.

Show input JSON schema

{
  "additionalProperties": false,
  "properties": {
    "organization_id": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "description": "Organization ID. With no arguments, resolves from the configured default or the authenticated user's sole membership."
    },
    "organization_name": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "description": "Organization name (exact match). With no arguments, resolves from the configured default or the authenticated user's sole membership. With multiple memberships, the error lists candidate organization IDs."
    }
  },
  "type": "object"
}

Show output JSON schema

{
  "description": "Information about an organization in Airbyte Cloud.",
  "properties": {
    "id": {
      "type": "string"
    },
    "name": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null
    },
    "email": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null
    }
  },
  "required": [
    "id"
  ],
  "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"
}

describe_cloud_workspace

Hints: read-only · idempotent · open-world

Get basic details about a workspace (ID, name, URL, parent organization).

Does not include billing/account status; use `get_cloud_organization_billing_status` for that.

Parameters

Name	Type	Required	Default	Description
`workspace_id`	`string \| null`	no	`null`	Workspace ID. With no argument, resolves the configured default or the authenticated user's default workspace.

Show input JSON schema

{
  "additionalProperties": false,
  "properties": {
    "workspace_id": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "description": "Workspace ID. With no argument, resolves the configured default or the authenticated user's default workspace."
    }
  },
  "type": "object"
}

Show output JSON schema

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

get_cloud_organization_billing_status

Hints: read-only · idempotent · open-world

Get billing and account status for an organization.

This generally requires elevated `ORGANIZATION_READER` or administrator permissions.

Parameters

Name	Type	Required	Default	Description
`organization_id`	`string \| null`	no	`null`	Organization ID, when known.
`organization_name`	`string \| null`	no	`null`	Organization name for an exact match, when ID is not provided.

Show input JSON schema

{
  "additionalProperties": false,
  "properties": {
    "organization_id": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "description": "Organization ID, when known."
    },
    "organization_name": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "description": "Organization name for an exact match, when ID is not provided."
    }
  },
  "type": "object"
}

Show output JSON schema

{
  "description": "Billing and account status for an Airbyte organization.",
  "properties": {
    "organization_id": {
      "type": "string"
    },
    "organization_name": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null
    },
    "billing_info_available": {
      "type": "boolean"
    },
    "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"
    },
    "message": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null
    }
  },
  "required": [
    "organization_id",
    "billing_info_available"
  ],
  "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"
}

get_default_cloud_context

Hints: read-only · idempotent · open-world

Return the authenticated user's default Cloud context.

This is the one-call orientation entry point: it resolves the default
workspace and its parent organization in a single call, along with the
user's explicit workspace and organization memberships.

Parameters

No parameters.

Show input JSON schema

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

Show output JSON schema

{
  "description": "Explicit authenticated Cloud affinities and discovery guidance.",
  "properties": {
    "user_id": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ]
    },
    "user_name": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ]
    },
    "user_email": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ]
    },
    "default_workspace_id": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ]
    },
    "default_workspace_name": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ]
    },
    "default_workspace_verified": {
      "type": "boolean"
    },
    "unvalidated_workspace_count": {
      "default": 0,
      "type": "integer"
    },
    "default_organization_id": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ]
    },
    "default_organization_name": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ]
    },
    "configured_workspace_id": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ]
    },
    "configured_organization_id": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ]
    },
    "member_organizations": {
      "items": {
        "description": "Information about an Airbyte organization.",
        "properties": {
          "organizationId": {
            "type": "string"
          },
          "organizationName": {
            "anyOf": [
              {
                "type": "string"
              },
              {
                "type": "null"
              }
            ],
            "default": null
          }
        },
        "required": [
          "organizationId"
        ],
        "type": "object"
      },
      "type": "array"
    },
    "member_workspaces": {
      "items": {
        "description": "Information about a workspace in Airbyte Cloud.",
        "properties": {
          "workspace_id": {
            "type": "string"
          },
          "workspace_name": {
            "type": "string"
          },
          "workspace_url": {
            "anyOf": [
              {
                "type": "string"
              },
              {
                "type": "null"
              }
            ],
            "default": null
          },
          "organization_id": {
            "anyOf": [
              {
                "type": "string"
              },
              {
                "type": "null"
              }
            ]
          },
          "organization_name": {
            "anyOf": [
              {
                "type": "string"
              },
              {
                "type": "null"
              }
            ],
            "default": null
          }
        },
        "required": [
          "workspace_id",
          "workspace_name",
          "organization_id"
        ],
        "type": "object"
      },
      "type": "array"
    },
    "member_organizations_truncated": {
      "type": "boolean"
    },
    "member_workspaces_truncated": {
      "type": "boolean"
    },
    "discovery_hints": {
      "items": {
        "type": "string"
      },
      "type": "array"
    },
    "message": {
      "type": "string"
    }
  },
  "required": [
    "user_id",
    "user_name",
    "user_email",
    "default_workspace_id",
    "default_workspace_name",
    "default_workspace_verified",
    "default_organization_id",
    "default_organization_name",
    "configured_workspace_id",
    "configured_organization_id",
    "member_organizations",
    "member_workspaces",
    "member_organizations_truncated",
    "member_workspaces_truncated",
    "discovery_hints",
    "message"
  ],
  "type": "object"
}

list_cloud_organizations

Hints: read-only · idempotent · open-world

List organizations visible to the authenticated Airbyte Cloud credentials.

Parameters

Name	Type	Required	Default	Description
`name_contains`	`string \| null \| null`	no	`null`
`limit`	`integer \| null \| null`	no	`null`

Show input JSON schema

{
  "additionalProperties": false,
  "properties": {
    "name_contains": {
      "anyOf": [
        {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "description": "Optional case-insensitive substring to filter organization names."
        },
        {
          "type": "null"
        }
      ],
      "default": null
    },
    "limit": {
      "anyOf": [
        {
          "anyOf": [
            {
              "type": "integer"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "description": "Optional maximum number of organizations to return (default: 100)."
        },
        {
          "type": "null"
        }
      ],
      "default": null
    }
  },
  "type": "object"
}

Show output JSON schema

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

list_cloud_sync_jobs

Hints: read-only · idempotent · open-world

List sync jobs for a connection with limit support.

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

Parameters

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

Show input JSON schema

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

Show output JSON schema

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

list_cloud_workspaces

Hints: read-only · idempotent · open-world

List all workspaces visible to the authenticated credentials.

The default returns direct workspace memberships. Use `organization_id` or a broader
`privilege_scope` to discover more workspaces.

Parameters

Name	Type	Required	Default	Description
`organization_id`	`string \| null`	no	`null`	Optional organization ID to list workspaces within.
`organization_name`	`string \| null`	no	`null`	Optional organization name (exact match) to list workspaces within.
`name_contains`	`string \| null`	no	`null`	Optional substring to filter workspaces by name (server-side filtering)
`limit`	`integer \| null`	no	`null`	Optional maximum number of items to return (default: no limit)
`privilege_scope`	`enum("member_of", "organization_admin", "instance_admin", "any")`	no	`"member_of"`	How broadly to search: direct memberships by default, organization memberships, instance-wide admin access, or any available scope.

Show input JSON schema

{
  "additionalProperties": false,
  "properties": {
    "organization_id": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "description": "Optional organization ID to list workspaces within."
    },
    "organization_name": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "description": "Optional organization name (exact match) to list workspaces within."
    },
    "name_contains": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "description": "Optional substring to filter workspaces by name (server-side filtering)"
    },
    "limit": {
      "anyOf": [
        {
          "type": "integer"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "description": "Optional maximum number of items to return (default: no limit)"
    },
    "privilege_scope": {
      "description": "How broadly to search: direct memberships by default, organization memberships, instance-wide admin access, or any available scope.",
      "enum": [
        "member_of",
        "organization_admin",
        "instance_admin",
        "any"
      ],
      "type": "string",
      "default": "member_of"
    }
  },
  "type": "object"
}

Show output JSON schema

{
  "description": "Result of discovering workspaces in Airbyte Cloud.",
  "properties": {
    "workspaces": {
      "items": {
        "description": "Information about a workspace in Airbyte Cloud.",
        "properties": {
          "workspace_id": {
            "type": "string"
          },
          "workspace_name": {
            "type": "string"
          },
          "workspace_url": {
            "anyOf": [
              {
                "type": "string"
              },
              {
                "type": "null"
              }
            ],
            "default": null
          },
          "organization_id": {
            "anyOf": [
              {
                "type": "string"
              },
              {
                "type": "null"
              }
            ]
          },
          "organization_name": {
            "anyOf": [
              {
                "type": "string"
              },
              {
                "type": "null"
              }
            ],
            "default": null
          }
        },
        "required": [
          "workspace_id",
          "workspace_name",
          "organization_id"
        ],
        "type": "object"
      },
      "type": "array"
    },
    "message": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null
    },
    "available_organizations": {
      "anyOf": [
        {
          "items": {
            "description": "Information about an organization in Airbyte Cloud.",
            "properties": {
              "id": {
                "type": "string"
              },
              "name": {
                "anyOf": [
                  {
                    "type": "string"
                  },
                  {
                    "type": "null"
                  }
                ],
                "default": null
              },
              "email": {
                "anyOf": [
                  {
                    "type": "string"
                  },
                  {
                    "type": "null"
                  }
                ],
                "default": null
              }
            },
            "required": [
              "id"
            ],
            "type": "object"
          },
          "type": "array"
        },
        {
          "type": "null"
        }
      ],
      "default": null
    }
  },
  "required": [
    "workspaces"
  ],
  "type": "object"
}

list_custom_source_definitions

Hints: read-only · idempotent · open-world

List custom YAML source definitions in the Airbyte Cloud workspace.

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

Parameters

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

Show input JSON schema

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

Show output JSON schema

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

list_deployed_cloud_connections

Hints: read-only · idempotent · open-world

List all deployed connections in the Airbyte Cloud workspace.

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

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

Parameters

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

Show input JSON schema

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

Show output JSON schema

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

list_deployed_cloud_destination_connectors

Hints: read-only · idempotent · open-world

List all deployed destination connectors in the Airbyte Cloud workspace.

Parameters

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

Show input JSON schema

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

Show output JSON schema

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

list_deployed_cloud_source_connectors

Hints: read-only · idempotent · open-world

List all deployed source connectors in the Airbyte Cloud workspace.

Parameters

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

Show input JSON schema

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

Show output JSON schema

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

permanently_delete_cloud_connection

Hints: destructive · open-world

Permanently delete a connection from Airbyte Cloud.

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

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

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

Parameters

Name	Type	Required	Default	Description
`connection_id`	`string`	yes	—	The ID of the connection to delete.
`name`	`string`	yes	—	The expected name of the connection (for verification).
`cascade_delete_source`	`boolean`	no	`false`	Whether to also delete the source connector associated with this connection.
`cascade_delete_destination`	`boolean`	no	`false`	Whether to also delete the destination connector associated with this connection.
`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 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"
    },
    "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
}

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).
`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 delete.",
      "type": "string"
    },
    "name": {
      "description": "The expected name of the destination (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": [
    "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).
`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 delete.",
      "type": "string"
    },
    "name": {
      "description": "The expected name of the source (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": [
    "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 inline YAML or a file path. File paths are honored only when trusted execution is enabled; otherwise, the value is treated as inline YAML. 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
}

set_default_cloud_workspace

Hints: destructive · idempotent · open-world

Durably set the authenticated user's default Airbyte Cloud workspace.

WARNING: This is a persistent, account-level change. It updates the user's
stored default workspace in Airbyte Cloud, which affects both future MCP
sessions (default_workspace_id in get_default_cloud_context and every tool
that falls back to the default workspace) AND the Airbyte Cloud web app,
where this workspace becomes the user's default landing workspace.
Call get_default_cloud_context first to confirm the current user and to
discover member workspaces.

Parameters

Name	Type	Required	Default	Description
`user_email`	`string`	yes	—	Email of the authenticated Airbyte Cloud user this change applies to. Must match the current credentials' user (compared case-insensitively, ignoring surrounding whitespace; see get_default_cloud_context); mismatches fail with a validation error. Required as a safety confirmation.
`workspace_id`	`string`	yes	—	ID of the workspace to make the durable default. The user must be an explicit member of the workspace or its organization; tombstoned workspaces are rejected.

Show input JSON schema

{
  "additionalProperties": false,
  "properties": {
    "user_email": {
      "description": "Email of the authenticated Airbyte Cloud user this change applies to. Must match the current credentials' user (compared case-insensitively, ignoring surrounding whitespace; see get_default_cloud_context); mismatches fail with a validation error. Required as a safety confirmation.",
      "type": "string"
    },
    "workspace_id": {
      "description": "ID of the workspace to make the durable default. The user must be an explicit member of the workspace or its organization; tombstoned workspaces are rejected.",
      "type": "string"
    }
  },
  "required": [
    "user_email",
    "workspace_id"
  ],
  "type": "object"
}

Show output JSON schema

{
  "description": "Result of durably updating the authenticated user's default workspace.",
  "properties": {
    "user_id": {
      "type": "string"
    },
    "user_email": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ]
    },
    "previous_default_workspace_id": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ]
    },
    "default_workspace_id": {
      "type": "string"
    },
    "default_workspace_name": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ]
    },
    "organization_id": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ]
    },
    "organization_name": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ]
    },
    "membership_basis": {
      "enum": [
        "workspace",
        "organization"
      ],
      "type": "string"
    },
    "message": {
      "type": "string"
    }
  },
  "required": [
    "user_id",
    "user_email",
    "previous_default_workspace_id",
    "default_workspace_id",
    "default_workspace_name",
    "organization_id",
    "organization_name",
    "membership_basis",
    "message"
  ],
  "type": "object"
}

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 Quartz cron expression defining when syncs should run. Must have 6 or 7 space-separated fields (seconds, minutes, hours, day-of-month, month, day-of-week[, year]), optionally followed by a timezone ID. Standard 5-field Unix cron expressions are rejected by the API, and schedules may run at most once per hour (seconds and minutes cannot be ''). Examples: '0 0 0 * ?' (daily at midnight UTC), '0 0 /6 * ?' (every 6 hours), '0 0 0 ? * SUN' (weekly on Sunday at midnight UTC), '0 0 9 ? * MON-FRI US/Pacific' (weekdays at 9am Pacific). 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 Quartz cron expression defining when syncs should run. Must have 6 or 7 space-separated fields (seconds, minutes, hours, day-of-month, month, day-of-week[, year]), optionally followed by a timezone ID. Standard 5-field Unix cron expressions are rejected by the API, and schedules may run at most once per hour (seconds and minutes cannot be '*'). Examples: '0 0 0 * * ?' (daily at midnight UTC), '0 0 */6 * * ?' (every 6 hours), '0 0 0 ? * SUN' (weekly on Sunday at midnight UTC), '0 0 9 ? * MON-FRI US/Pacific' (weekdays at 9am Pacific). 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 inline YAML or a file path. File paths are honored only when trusted execution is enabled; otherwise, the value is treated as inline YAML. Optional; omit to update only testing values."
        },
        {
          "type": "null"
        }
      ],
      "default": null
    },
    "workspace_id": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "description": "Workspace ID. Hosted MCP connections pass it via the `X-Airbyte-Workspace-Id` header; local or stdio connections use the `AIRBYTE_CLOUD_WORKSPACE_ID` environment variable."
    },
    "pre_validate": {
      "default": true,
      "description": "Whether to validate the manifest client-side before updating.",
      "type": "boolean"
    },
    "testing_values": {
      "anyOf": [
        {
          "additionalProperties": true,
          "type": "object"
        },
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "description": "Optional testing configuration values for the Builder UI. Can be provided as a JSON object or JSON string. Supports inline secret refs via 'secret_reference::ENV_VAR_NAME' syntax. If provided, these values replace any existing testing values for the connector builder project. The entire testing values object is overwritten, so pass the full set of values you want to persist."
    },
    "testing_values_secret_name": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "description": "Optional name of a secret containing testing configuration values in JSON or YAML format. The secret will be resolved by the MCP server and merged into testing_values, with secret values taking precedence. This lets the agent reference secrets without sending raw values as tool arguments."
    }
  },
  "required": [
    "definition_id"
  ],
  "type": "object"
}

Show output JSON schema

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

View Source

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