airbyte.mcp.cloud

Airbyte Cloud MCP operations.

cloud module

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

Tools (42)

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_destination

Hints: read-only · idempotent · open-world

Check the configuration and credentials of a deployed destination connector.

Parameters

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

Show input JSON schema

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

Show output JSON schema

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

check_cloud_source

Hints: read-only · idempotent · open-world

Check the configuration and credentials of a deployed source connector.

Parameters

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

Show input JSON schema

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

Show output JSON schema

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

create_connection_on_cloud

Hints: open-world

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

Parameters

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

Show input JSON schema

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

Show output JSON schema

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

deploy_destination_to_cloud

Hints: open-world

Deploy a destination connector to Airbyte Cloud.

Parameters

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

Show input JSON schema

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

Show output JSON schema

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

deploy_noop_destination_to_cloud

Hints: open-world

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

Parameters

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

Show input JSON schema

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

Show output JSON schema

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

deploy_source_to_cloud

Hints: open-world

Deploy a source connector to Airbyte Cloud.

Parameters

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

Show input JSON schema

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

Show output JSON schema

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

describe_cloud_connection

Hints: read-only · idempotent · open-world

Get detailed information about a specific deployed connection.

Parameters

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

Show input JSON schema

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

Show output JSON schema

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

describe_cloud_destination

Hints: read-only · idempotent · open-world

Get detailed information about a specific deployed destination connector.

Parameters

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

Show input JSON schema

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

Show output JSON schema

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

describe_cloud_organization

Hints: read-only · idempotent · open-world

Get 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.

Show input JSON schema

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

Show output JSON schema

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

permanently_delete_cloud_destination

Hints: destructive · open-world

Permanently delete a deployed destination connector from Airbyte Cloud.

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

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

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

Parameters

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

Show input JSON schema

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

Show output JSON schema

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

permanently_delete_cloud_source

Hints: destructive · open-world

Permanently delete a deployed source connector from Airbyte Cloud.

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

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

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

Parameters

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

Show input JSON schema

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

Show output JSON schema

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

permanently_delete_custom_source_definition

Hints: destructive · open-world

Permanently delete a custom YAML source definition from Airbyte Cloud.

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

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

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

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

Parameters

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

Show input JSON schema

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

Show output JSON schema

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

publish_custom_source_definition

Hints: open-world

Publish a custom YAML source connector definition to Airbyte Cloud.

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

Parameters

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

Show input JSON schema

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

Show output JSON schema

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

rename_cloud_connection

Hints: open-world

Rename a connection on Airbyte Cloud.

Parameters

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

Show input JSON schema

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

Show output JSON schema

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

rename_cloud_destination

Hints: open-world

Rename a deployed destination connector on Airbyte Cloud.

Parameters

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

Show input JSON schema

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

Show output JSON schema

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

rename_cloud_source

Hints: open-world

Rename a deployed source connector on Airbyte Cloud.

Parameters

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

Show input JSON schema

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

Show output JSON schema

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

run_cloud_sync

Hints: open-world

Run a sync job on Airbyte Cloud.

Parameters

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

Show input JSON schema

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

Show output JSON schema

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

set_cloud_connection_selected_streams

Hints: destructive · open-world

Set the selected streams for a connection on Airbyte Cloud.

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

Parameters

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

Show input JSON schema

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

Show output JSON schema

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

set_cloud_connection_table_prefix

Hints: destructive · open-world

Set the table prefix for a connection on Airbyte Cloud.

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

Parameters

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

Show input JSON schema

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

Show output JSON schema

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

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 cron expression defining when syncs should run. Examples: '0 0 * * ' (daily at midnight UTC), '0 /6 * * ' (every 6 hours), '0 0 * 0' (weekly on Sunday at midnight UTC). Leave unset to keep the current schedule. Cannot be used together with 'manual_schedule'.
`manual_schedule`	`boolean \| null`	no	`null`	Set to True to disable automatic syncs (manual scheduling only). Syncs will only run when manually triggered. Cannot be used together with 'cron_expression'.
`workspace_id`	`string \| null`	no	`null`	Workspace ID. Hosted MCP connections pass it via the `X-Airbyte-Workspace-Id` header; local or stdio connections use the `AIRBYTE_CLOUD_WORKSPACE_ID` environment variable.

Show input JSON schema

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

Show output JSON schema

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

update_cloud_destination_config

Hints: destructive · open-world

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

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

Parameters

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

Show input JSON schema

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

Show output JSON schema

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

update_cloud_source_config

Hints: destructive · open-world

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

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

Parameters

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

Show input JSON schema

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

Show output JSON schema

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

update_custom_source_definition

Hints: destructive · open-world

Update a custom YAML source definition in Airbyte Cloud.

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

Parameters

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

Show input JSON schema

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

Show output JSON schema

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

View Source

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