fix(mcp): guide Connector Builder agents through the credentials flow

airbyte/cloud/connector_builder/_util.py

@@ -146,7 +146,11 @@ def readiness_next_steps(
     pending_streams: list[str],
     not_attempted_streams: list[str],
 ) -> list[str]:
-    steps = [check.description for check in checks[:4] if check.result != "passed"]
+    steps = [
+        f"{check.description}: {check.detail}" if check.detail else check.description
+        for check in checks[:4]
+        if check.result != "passed"
+    ]
     failed = [stream.stream_name for stream in streams if stream.failed]
     capped = [
         stream.stream_name
@@ -211,8 +215,14 @@ def readiness_checks(
             result=_result(bool(draft.get("streams")) or bool(draft.get("dynamic_streams"))),
         ),
         ConnectorBuilderReadinessCheck(
-            description="Set testing values (enter secrets in the Connector Builder UI)",
-            result=_result(bool(testing_values) or takes_no_inputs),
+            description=(
+                "Ask the user to enter testing values, including secrets, in the Connector "
+                "Builder UI"
+            ),
+            result=_result(
+                takes_no_inputs
+                or (bool(testing_values) and not spec_missing_required(spec, testing_values))
+            ),
             detail=url,
         ),
         ConnectorBuilderReadinessCheck(
@@ -276,14 +286,51 @@ def spec_missing_required(
     schema: Any,  # noqa: ANN401
     config: dict[str, Any],
 ) -> bool:
-    """Return whether `config` lacks a required top-level field that has no default."""
-    if not isinstance(schema, dict):
+    """Return whether `config` lacks a required field that has no default.
+
+    Checks nested objects and `oneOf` options too.
+    """
+    if not isinstance(schema, dict) or not isinstance(config, dict):
         return False
-    properties = schema.get("properties") or {}
-    return any(
-        name not in config and "default" not in (properties.get(name) or {})
-        for name in schema.get("required") or []
-    )
+    properties = schema.get("properties")
+    if not isinstance(properties, dict):
+        properties = {}
+    for name in schema.get("required") or []:
+        property_schema = properties.get(name)
+        if name not in config and (
+            not isinstance(property_schema, dict) or "default" not in property_schema
+        ):
+            return True
+    for name, value in config.items():
+        property_schema = properties.get(name)
+        if (
+            isinstance(value, dict)
+            and isinstance(property_schema, dict)
+            and spec_missing_required(property_schema, value)
+        ):
+            return True
+    one_of = schema.get("oneOf")
+    if isinstance(one_of, list):
+        option_satisfied = False
+        for option in one_of:
+            if not isinstance(option, dict):
+                option_satisfied = True
+                break
+            option_properties = option.get("properties")
+            if isinstance(option_properties, dict) and any(
+                isinstance(property_schema, dict)
+                and "const" in property_schema
+                and name in config
+                and config[name] != property_schema["const"]
+                for name, property_schema in option_properties.items()
+            ):
+                continue
+            if not spec_missing_required(option, config):
+                option_satisfied = True
+                break
+        if not option_satisfied:
+            return True
+    return False
 
 
 def spec_has_secrets(schema: Any) -> bool:  # noqa: ANN401

airbyte/mcp/connector_builder.py

@@ -102,6 +102,8 @@ class ConnectorBuilderProjectCreateResult(BaseModel):
     """The test source ID, set only with `create_test_source`."""
     test_source_settings_url: str | None = None
     """Cloud settings page of the test source, set only with `create_test_source`."""
+    next_steps: list[str] = Field(default_factory=list)
+    """Credential-entry steps, or an empty list if no test source was created."""
 
 
 class ConnectorBuilderStreamTestReadToolResult(ConnectorBuilderStreamReadResult):
@@ -272,9 +274,8 @@ def create_connector_builder_project(
         Field(
             description=(
                 "Optional. Publish the initial manifest (which needs its spec) and create a "
-                "test source on it for a later full sync. Ask the user to enter their "
-                "credentials at both `connector_builder_project_url` (testing values) and "
-                "`test_source_settings_url` (then test and save) before you start building."
+                "test source on it for the full sync. Use this for connectors that need "
+                "credentials."
             ),
             default=False,
         ),
@@ -285,6 +286,17 @@ def create_connector_builder_project(
     The project holds the draft manifest and the testing values used for test reads.
     Nothing appears in the workspace's source list until the project is published with
     `publish_custom_source_definition`.
+
+    For a connector that needs credentials, include the spec in the initial manifest, with
+    secret fields marked `airbyte_secret`, and set `create_test_source`. Put the credentials
+    under one `credentials` object whose `oneOf` lists the auth options the API supports.
+    Each option has a constant `auth_type` and its own fields, for example `api_key`
+    (`api_key`), `access_token` (`access_token`), or `oauth2.0` (`client_id`,
+    `client_secret`, `refresh_token`). With more than one option, use a
+    `SelectiveAuthenticator` with `authenticator_selection_path: ["credentials", "auth_type"]`.
+    The test source is saved as a draft, so auth processes not tested yet don't block it.
+    Then ask the user to enter their credentials in both places listed in `next_steps`. Never
+    ask for secrets in the chat.
     """
     processed_manifest = resolve_manifest_yaml(
         manifest_yaml_text=manifest_yaml_text,
@@ -316,6 +328,12 @@ def create_connector_builder_project(
         result.definition_id = definition.definition_id
         result.test_source_id = test_source.connector_id
         result.test_source_settings_url = test_source.connector_url
+        result.next_steps = [
+            "Ask the user to enter their credentials in the Connector Builder testing "
+            f"values: {project.connector_builder_project_url}",
+            "Ask the user to enter the same credentials in the test source, then test and "
+            f"save it: {test_source.connector_url}",
+        ]
     return result
 
 

tests/unit_tests/test_cloud_connector_builder.py

@@ -30,6 +30,7 @@
     publish_spec,
     render_manifest_text,
     spec_has_secrets,
+    spec_missing_required,
 )
 from airbyte.cloud.workspaces import CloudWorkspace
 from airbyte.exceptions import (
@@ -48,6 +49,37 @@
     },
 }
 
+CREDENTIALS_ONE_OF_SPEC: dict[str, Any] = {
+    "type": "object",
+    "required": ["credentials"],
+    "properties": {
+        "credentials": {
+            "type": "object",
+            "oneOf": [
+                {
+                    "type": "object",
+                    "properties": {
+                        "auth_type": {"type": "string", "const": "api_key"},
+                        "api_key": {"type": "string", "airbyte_secret": True},
+                    },
+                    "required": ["auth_type", "api_key"],
+                },
+                {
+                    "type": "object",
+                    "properties": {
+                        "auth_type": {"type": "string", "const": "access_token"},
+                        "access_token": {
+                            "type": "string",
+                            "airbyte_secret": True,
+                        },
+                    },
+                    "required": ["auth_type", "access_token"],
+                },
+            ],
+        }
+    },
+}
+
 
 @pytest.mark.parametrize(
     ("edit_mode", "new_text", "old_text", "expected"),
@@ -912,33 +944,142 @@ def test_readiness_report_lists_caps_without_a_success_verdict(
     assert not any("ready" in step.lower() for step in full.next_steps)
 
 
+@pytest.mark.parametrize(
+    ("connection_specification", "testing_values", "expected_result"),
+    [
+        pytest.param(
+            {
+                "type": "object",
+                "required": ["api_key"],
+                "properties": {
+                    "api_key": {"type": "string", "airbyte_secret": True},
+                    "region": {"type": "string"},
+                },
+            },
+            {"region": "us-east-1"},
+            "failed",
+            id="required-value-missing",
+        ),
+        pytest.param(
+            {
+                "type": "object",
+                "required": ["api_key"],
+                "properties": {
+                    "api_key": {"type": "string", "airbyte_secret": True},
+                    "region": {"type": "string"},
+                },
+            },
+            {"api_key": "**********", "region": "us-east-1"},
+            "passed",
+            id="masked-secret-is-present",
+        ),
+        pytest.param(
+            {"type": "object", "properties": {}},
+            None,
+            "passed",
+            id="no-inputs",
+        ),
+        pytest.param(
+            {
+                "type": "object",
+                "required": ["region"],
+                "properties": {
+                    "region": {"type": "string", "default": "us-east-1"},
+                    "account": {"type": "string"},
+                },
+            },
+            {"account": "demo"},
+            "passed",
+            id="required-field-has-default",
+        ),
+        pytest.param(
+            CREDENTIALS_ONE_OF_SPEC,
+            {"credentials": {"auth_type": "api_key"}},
+            "failed",
+            id="one-of-required-secret-missing",
+        ),
+        pytest.param(
+            CREDENTIALS_ONE_OF_SPEC,
+            {"credentials": {"auth_type": "api_key", "api_key": "**********"}},
+            "passed",
+            id="one-of-masked-secret-is-present",
+        ),
+        pytest.param(
+            CREDENTIALS_ONE_OF_SPEC,
+            {"credentials": {"auth_type": "access_token", "api_key": "**********"}},
+            "failed",
+            id="one-of-matching-option-required-secret-missing",
+        ),
+        pytest.param(
+            {
+                "type": "object",
+                "required": ["credentials"],
+                "properties": {
+                    "credentials": {
+                        "type": "object",
+                        "required": ["region"],
+                        "properties": {
+                            "region": {"type": "string", "default": "us-east-1"}
+                        },
+                    }
+                },
+            },
+            {"credentials": {}},
+            "passed",
+            id="nested-required-field-has-default",
+        ),
+    ],
+)
 def test_readiness_checks_require_testing_values_when_spec_has_inputs(
     monkeypatch: pytest.MonkeyPatch,
+    connection_specification: dict[str, Any],
+    testing_values: dict[str, Any] | None,
+    expected_result: str,
 ) -> None:
     draft = {
         "version": "6.0.0",
-        "spec": {"connection_specification": {"properties": {"api_key": {}}}},
+        "spec": {"connection_specification": connection_specification},
         "streams": [{"name": "users"}],
     }
-    _patch_project_store(monkeypatch, draft=draft, source_definition_id="def")
+    _patch_project_store(
+        monkeypatch,
+        draft=draft,
+        source_definition_id="def",
+        testing_values=testing_values,
+    )
     project = CloudConnectorBuilderProject(
         workspace=_fake_workspace(), builder_project_id="bp"
     )
     _patch_reads(monkeypatch, project, {"users": _stream_read("users", record_count=1)})
 
     report = project._run_readiness_checks(["users"])
 
-    assert [check.result for check in report.checks] == [
-        "passed",
-        "passed",
-        "passed",
-        "failed",
-        "passed",
-        "passed",
-        "passed",
-        "unknown",
-    ]
-    assert report.failed_streams == []
+    testing_values_check = report.checks[3]
+    assert testing_values_check.description == (
+        "Ask the user to enter testing values, including secrets, in the Connector Builder UI"
+    )
+    assert testing_values_check.result == expected_result
+    assert testing_values_check.detail == project.connector_builder_project_url
+    if expected_result == "failed":
+        assert (
+            f"{testing_values_check.description}: {project.connector_builder_project_url}"
+            in report.next_steps
+        )
+
+
+def test_spec_missing_required_checks_nested_fields() -> None:
+    schema = {
+        "type": "object",
+        "properties": {
+            "credentials": {
+                "type": "object",
+                "properties": {"api_key": {"type": "string"}},
+                "required": ["api_key"],
+            }
+        },
+    }
+
+    assert spec_missing_required(schema, {"credentials": {}})
 
 
 def test_publish_spec_uses_api_field_names() -> None: