feat: Add capability to set testing values on custom connector projects in Builder UI

airbyte/_util/api_util.py

@@ -1564,6 +1564,51 @@ def get_connector_builder_project_for_definition_id(
     return json_result.get("builderProjectId")
 
 
+def update_connector_builder_project_testing_values(
+    *,
+    workspace_id: str,
+    builder_project_id: str,
+    testing_values: dict[str, Any],
+    spec: dict[str, Any],
+    api_root: str,
+    client_id: SecretString,
+    client_secret: SecretString,
+) -> dict[str, Any]:
+    """Update the testing values for a connector builder project.
+
+    This call replaces the entire testing values object stored for the project.
+    Any keys not included in `testing_values` will be removed.
+
+    Uses the Config API endpoint:
+    /v1/connector_builder_projects/update_testing_values
+
+    Args:
+        workspace_id: The workspace ID
+        builder_project_id: The connector builder project ID
+        testing_values: The testing values (config blob) to persist. This replaces
+            any existing testing values entirely.
+        spec: The source definition specification (connector spec)
+        api_root: The API root URL
+        client_id: OAuth client ID
+        client_secret: OAuth client secret
+
+    Returns:
+        The updated testing values from the API response
+    """
+    return _make_config_api_request(
+        path="/connector_builder_projects/update_testing_values",
+        json={
+            "workspaceId": workspace_id,
+            "builderProjectId": builder_project_id,
+            "testingValues": testing_values,
+            "spec": spec,
+        },
+        api_root=api_root,
+        client_id=client_id,
+        client_secret=client_secret,
+    )
+
+
 # Organization and workspace listing
 
 

airbyte/cloud/connectors.py

@@ -690,3 +690,70 @@ def deploy_source(
             workspace=self.workspace,
             source_response=result,
         )
+
+    def set_testing_values(
+        self,
+        testing_values: dict[str, Any],
+    ) -> CustomCloudSourceDefinition:
+        """Set the testing values for this custom source definition's connector builder project.
+
+        Testing values are the input configuration values used when testing the connector
+        in the Connector Builder UI. Setting these values allows users to immediately
+        run test read operations after deploying a custom source to the Builder UI.
+
+        This method replaces any existing testing values with the provided dictionary.
+        Pass the full set of values you want to persist, not just the fields you're changing.
+
+        Args:
+            testing_values: A dictionary containing the configuration values to use for testing.
+                This should match the connector's spec schema. Replaces any existing values.
+
+        Returns:
+            This `CustomCloudSourceDefinition` object (for method chaining).
+
+        Raises:
+            NotImplementedError: If this is not a YAML custom source definition.
+            PyAirbyteInputError: If the connector builder project ID cannot be found.
+        """
+        if self.definition_type != "yaml":
+            raise NotImplementedError(
+                "Testing values can only be set for YAML custom source definitions. "
+                "Docker custom sources are not yet supported."
+            )
+
+        builder_project_id = self.connector_builder_project_id
+        if not builder_project_id:
+            raise exc.PyAirbyteInputError(
+                message="Could not find connector builder project ID for this definition.",
+                context={
+                    "definition_id": self.definition_id,
+                    "workspace_id": self.workspace.workspace_id,
+                },
+            )
+
+        # Get the spec from the definition info
+        if not self._definition_info:
+            self._definition_info = self._fetch_definition_info()
+
+        # Build the spec object from the manifest
+        spec: dict[str, Any] = {}
+        if self._definition_info.manifest:
+            manifest_spec = self._definition_info.manifest.get("spec", {})
+            if manifest_spec:
+                spec = {
+                    "connectionSpecification": manifest_spec.get(
+                        "connection_specification", manifest_spec
+                    ),
+                }
+
+        api_util.update_connector_builder_project_testing_values(
+            workspace_id=self.workspace.workspace_id,
+            builder_project_id=builder_project_id,
+            testing_values=testing_values,
+            spec=spec,
+            api_root=self.workspace.api_root,
+            client_id=self.workspace.client_id,
+            client_secret=self.workspace.client_secret,
+        )
+
+        return self

airbyte/cloud/workspaces.py

@@ -506,6 +506,7 @@ def publish_custom_source_definition(
         docker_tag: str | None = None,
         unique: bool = True,
         pre_validate: bool = True,
+        testing_values: dict[str, Any] | None = None,
     ) -> CustomCloudSourceDefinition:
         """Publish a custom source connector definition.
 
@@ -519,6 +520,10 @@ def publish_custom_source_definition(
             docker_tag: Docker image tag (e.g., '1.0.0')
             unique: Whether to enforce name uniqueness
             pre_validate: Whether to validate manifest client-side (YAML only)
+            testing_values: Optional configuration values to use for testing in the
+                Connector Builder UI. If provided, these values are stored as the complete
+                testing values object for the connector builder project (replaces any existing
+                values), allowing immediate test read operations.
 
         Returns:
             CustomCloudSourceDefinition object representing the created definition
@@ -583,7 +588,15 @@ def publish_custom_source_definition(
                 client_id=self.client_id,
                 client_secret=self.client_secret,
             )
-            return CustomCloudSourceDefinition._from_yaml_response(self, result)  # noqa: SLF001
+            custom_definition = CustomCloudSourceDefinition._from_yaml_response(  # noqa: SLF001
+                self, result
+            )
+
+            # Set testing values if provided
+            if testing_values is not None:
+                custom_definition.set_testing_values(testing_values)
+
+            return custom_definition
 
         raise NotImplementedError(
             "Docker custom source definitions are not yet supported. "

airbyte/mcp/cloud_ops.py

@@ -1356,6 +1356,31 @@ def publish_custom_source_definition(
             default=True,
         ),
     ] = True,
+    testing_values: Annotated[
+        dict | str | None,
+        Field(
+            description=(
+                "Optional testing configuration values for the Builder UI. "
+                "Can be provided as a JSON object or JSON string. "
+                "If provided, these values replace any existing testing values "
+                "for the connector builder project, allowing immediate test read operations."
+            ),
+            default=None,
+        ),
+    ] = None,
+    testing_values_secret_name: Annotated[
+        str | None,
+        Field(
+            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."
+            ),
+            default=None,
+        ),
+    ] = None,
 ) -> str:
     """Publish a custom YAML source connector definition to Airbyte Cloud.
 
@@ -1366,12 +1391,24 @@ def publish_custom_source_definition(
     if isinstance(manifest_yaml, str) and "\n" not in manifest_yaml:
         processed_manifest = Path(manifest_yaml)
 
+    # Resolve testing values from inline config and/or secret
+    testing_values_dict: dict[str, Any] | None = None
+    if testing_values is not None or testing_values_secret_name is not None:
+        testing_values_dict = (
+            resolve_config(
+                config=testing_values,
+                config_secret_name=testing_values_secret_name,
+            )
+            or None
+        )
+
     workspace: CloudWorkspace = _get_cloud_workspace(workspace_id)
     custom_source = workspace.publish_custom_source_definition(
         name=name,
         manifest_yaml=processed_manifest,
         unique=unique,
         pre_validate=pre_validate,
+        testing_values=testing_values_dict,
     )
     register_guid_created_in_session(custom_source.definition_id)
     return (
@@ -1431,11 +1468,15 @@ def update_custom_source_definition(
         Field(description="The ID of the definition to update."),
     ],
     manifest_yaml: Annotated[
-        str | Path,
+        str | Path | None,
         Field(
-            description="New manifest as YAML string or file path.",
+            description=(
+                "New manifest as YAML string or file path. "
+                "Optional; omit to update only testing values."
+            ),
+            default=None,
         ),
-    ],
+    ] = None,
     *,
     workspace_id: Annotated[
         str | None,
@@ -1451,26 +1492,84 @@ def update_custom_source_definition(
             default=True,
         ),
     ] = True,
+    testing_values: Annotated[
+        dict | str | None,
+        Field(
+            description=(
+                "Optional testing configuration values for the Builder UI. "
+                "Can be provided as a JSON object or JSON string. "
+                "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."
+            ),
+            default=None,
+        ),
+    ] = None,
+    testing_values_secret_name: Annotated[
+        str | None,
+        Field(
+            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."
+            ),
+            default=None,
+        ),
+    ] = None,
 ) -> str:
     """Update a custom YAML source definition in Airbyte Cloud.
 
-    Note: Only YAML (declarative) connectors are currently supported.
-    Docker-based custom sources are not yet available.
+    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.
     """
     check_guid_created_in_session(definition_id)
-    processed_manifest = manifest_yaml
+
+    workspace: CloudWorkspace = _get_cloud_workspace(workspace_id)
+
+    if manifest_yaml is None and testing_values is None and testing_values_secret_name is None:
+        raise PyAirbyteInputError(
+            message=(
+                "At least one of manifest_yaml, testing_values, or testing_values_secret_name "
+                "must be provided to update a custom source definition."
+            ),
+            context={
+                "definition_id": definition_id,
+                "workspace_id": workspace.workspace_id,
+            },
+        )
+
+    processed_manifest: str | Path | None = manifest_yaml
     if isinstance(manifest_yaml, str) and "\n" not in manifest_yaml:
         processed_manifest = Path(manifest_yaml)
 
-    workspace: CloudWorkspace = _get_cloud_workspace(workspace_id)
+    # Resolve testing values from inline config and/or secret
+    testing_values_dict: dict[str, Any] | None = None
+    if testing_values is not None or testing_values_secret_name is not None:
+        testing_values_dict = (
+            resolve_config(
+                config=testing_values,
+                config_secret_name=testing_values_secret_name,
+            )
+            or None
+        )
+
     definition = workspace.get_custom_source_definition(
         definition_id=definition_id,
         definition_type="yaml",
     )
-    custom_source: CustomCloudSourceDefinition = definition.update_definition(
-        manifest_yaml=processed_manifest,
-        pre_validate=pre_validate,
-    )
+    custom_source: CustomCloudSourceDefinition = definition
+
+    if processed_manifest is not None:
+        custom_source = definition.update_definition(
+            manifest_yaml=processed_manifest,
+            pre_validate=pre_validate,
+        )
+
+    if testing_values_dict is not None:
+        custom_source.set_testing_values(testing_values_dict)
+
     return (
         "Successfully updated custom YAML source definition:\n"
         + _get_custom_source_definition_description(