feat(onboarding): reshape to get/validate/apply tools

Replace init/state with three purpose-built tools:
- dbt_admin_onboarding_get: GET /onboarding/ — check current status
- dbt_admin_onboarding_validate: POST /onboarding/validate/ — check data without applying
- dbt_admin_onboarding_apply: POST /onboarding/ — submit data incrementally (idempotent)

All three verified against the warm PR env. validate returns a clean error
until the backend implements the /validate/ endpoint.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

Author: joaodaher

src/dbt_mcp/dbt_admin/onboarding/client.py

@@ -31,9 +31,9 @@ async def _headers(self, config: AdminApiConfig) -> dict[str, str]:
         } | config.headers_provider.get_headers()
 
     async def get(self, account_id: int) -> dict[str, Any] | None:
-        """Fetch the current onboarding model for the account.
+        """Fetch the current onboarding record for the account.
 
-        Returns the raw data dict, or None if no onboarding exists yet.
+        Returns the raw data dict, or None if no onboarding exists yet (404).
         """
         config = await self.config_provider.get_config()
         url = self._base_url(config, account_id)
@@ -49,41 +49,72 @@ async def get(self, account_id: int) -> dict[str, Any] | None:
         except httpx.HTTPStatusError as e:
             if 400 <= e.response.status_code < 500:
                 raise InvalidParameterError(
-                    f"Onboarding request failed ({e.response.status_code}) for account {account_id}"
+                    f"Onboarding get failed ({e.response.status_code}) for account {account_id}"
                 ) from e
             raise AdminAPIError(
-                f"Onboarding request failed ({e.response.status_code}) for account {account_id}"
+                f"Onboarding get failed ({e.response.status_code}) for account {account_id}"
             ) from e
         except httpx.HTTPError as e:
             raise AdminAPIError(
-                f"Onboarding request failed for account {account_id}"
+                f"Onboarding get failed for account {account_id}"
             ) from e
 
-    async def create_or_get(self, account_id: int) -> dict[str, Any]:
-        """Create the onboarding model if it doesn't exist, or return the existing one.
+    async def validate(self, account_id: int, data: dict[str, Any]) -> dict[str, Any]:
+        """Validate onboarding data without applying it.
 
-        POST is idempotent on the backend — returns the existing record if one is present.
+        Returns a dict with 'valid' bool and 'errors' list.
         """
         config = await self.config_provider.get_config()
+        url = self._base_url(config, account_id) + "validate/"
+        headers = await self._headers(config)
+
+        try:
+            async with httpx.AsyncClient() as client:
+                response = await client.post(
+                    url, headers=headers, json=data, follow_redirects=True
+                )
+                response.raise_for_status()
+                return response.json()
+        except httpx.HTTPStatusError as e:
+            if 400 <= e.response.status_code < 500:
+                # Validation errors are expected — return them as structured data
+                try:
+                    return response.json()
+                except Exception:
+                    pass
+                raise InvalidParameterError(
+                    f"Onboarding validate failed ({e.response.status_code}) for account {account_id}"
+                ) from e
+            raise AdminAPIError(
+                f"Onboarding validate failed ({e.response.status_code}) for account {account_id}"
+            ) from e
+        except httpx.HTTPError as e:
+            raise AdminAPIError(
+                f"Onboarding validate failed for account {account_id}"
+            ) from e
+
+    async def apply(self, account_id: int, data: dict[str, Any]) -> dict[str, Any]:
+        """Submit onboarding data. Idempotent — safe to call multiple times as data is gathered."""
+        config = await self.config_provider.get_config()
         url = self._base_url(config, account_id)
         headers = await self._headers(config)
 
         try:
             async with httpx.AsyncClient() as client:
                 response = await client.post(
-                    url, headers=headers, json={}, follow_redirects=True
+                    url, headers=headers, json=data, follow_redirects=True
                 )
                 response.raise_for_status()
                 return response.json().get("data", {})
         except httpx.HTTPStatusError as e:
             if 400 <= e.response.status_code < 500:
                 raise InvalidParameterError(
-                    f"Onboarding create failed ({e.response.status_code}) for account {account_id}"
+                    f"Onboarding apply failed ({e.response.status_code}) for account {account_id}"
                 ) from e
             raise AdminAPIError(
-                f"Onboarding create failed ({e.response.status_code}) for account {account_id}"
+                f"Onboarding apply failed ({e.response.status_code}) for account {account_id}"
             ) from e
         except httpx.HTTPError as e:
             raise AdminAPIError(
-                f"Onboarding create failed for account {account_id}"
+                f"Onboarding apply failed for account {account_id}"
             ) from e

src/dbt_mcp/dbt_admin/onboarding/models.py

@@ -35,11 +35,15 @@ def from_api(cls, data: dict[str, Any]) -> "OnboardingModel":
         )
 
 
-class OnboardingInitResult(BaseModel):
-    onboarding: OnboardingModel
-    created: bool
+class OnboardingGetResult(BaseModel):
+    onboarding: OnboardingModel | None
 
 
-class OnboardingStateResult(BaseModel):
-    onboarding: OnboardingModel | None
-    decision_points: list[dict[str, Any]] = Field(default_factory=list)
+class OnboardingValidateResult(BaseModel):
+    valid: bool
+    errors: list[str] = Field(default_factory=list)
+    missing_fields: list[str] = Field(default_factory=list)
+
+
+class OnboardingApplyResult(BaseModel):
+    onboarding: OnboardingModel

src/dbt_mcp/dbt_admin/onboarding/tools.py

@@ -1,13 +1,15 @@
 from dataclasses import dataclass
+from typing import Any
 
 from mcp.server.fastmcp import FastMCP
 
 from dbt_mcp.config.config_providers import AdminApiConfig, ConfigProvider
 from dbt_mcp.dbt_admin.onboarding.client import OnboardingClient
 from dbt_mcp.dbt_admin.onboarding.models import (
-    OnboardingInitResult,
+    OnboardingApplyResult,
+    OnboardingGetResult,
     OnboardingModel,
-    OnboardingStateResult,
+    OnboardingValidateResult,
 )
 from dbt_mcp.prompts.prompts import get_prompt
 from dbt_mcp.tools.definitions import dbt_mcp_tool
@@ -23,57 +25,66 @@ class OnboardingToolContext:
 
 
 @dbt_mcp_tool(
-    description=get_prompt("admin_api/onboarding_init"),
-    title="Start or Resume Onboarding",
-    read_only_hint=False,
+    description=get_prompt("admin_api/onboarding_get"),
+    title="Get Onboarding Status",
+    read_only_hint=True,
     destructive_hint=False,
     idempotent_hint=True,
 )
-async def dbt_admin_onboarding_init(
+async def dbt_admin_onboarding_get(
     context: OnboardingToolContext,
-) -> OnboardingInitResult:
-    """Start or resume an onboarding session for the account."""
+) -> OnboardingGetResult:
+    """Return the current onboarding record, or null if none has been started."""
     config = await context.admin_api_config_provider.get_config()
-    account_id = config.account_id
-
-    existing = await context.onboarding_client.get(account_id)
-    if existing is not None:
-        return OnboardingInitResult(
-            onboarding=OnboardingModel.from_api(existing),
-            created=False,
-        )
-
-    data = await context.onboarding_client.create_or_get(account_id)
-    return OnboardingInitResult(
-        onboarding=OnboardingModel.from_api(data),
-        created=True,
+    data = await context.onboarding_client.get(config.account_id)
+    return OnboardingGetResult(
+        onboarding=OnboardingModel.from_api(data) if data is not None else None
     )
 
 
 @dbt_mcp_tool(
-    description=get_prompt("admin_api/onboarding_state"),
-    title="Get Onboarding State",
+    description=get_prompt("admin_api/onboarding_validate"),
+    title="Validate Onboarding Data",
     read_only_hint=True,
     destructive_hint=False,
     idempotent_hint=True,
 )
-async def dbt_admin_onboarding_state(
+async def dbt_admin_onboarding_validate(
     context: OnboardingToolContext,
-) -> OnboardingStateResult:
-    """Return the current onboarding state for the account."""
+    data: dict[str, Any],
+) -> OnboardingValidateResult:
+    """Validate onboarding data without applying it; returns what is missing or invalid."""
     config = await context.admin_api_config_provider.get_config()
-    account_id = config.account_id
+    raw = await context.onboarding_client.validate(config.account_id, data)
+    status = raw.get("status", {})
+    if status.get("is_success"):
+        return OnboardingValidateResult(valid=True)
+    developer_message = status.get("developer_message", "")
+    errors = [developer_message] if developer_message else []
+    return OnboardingValidateResult(valid=False, errors=errors)
 
-    data = await context.onboarding_client.get(account_id)
-    if data is None:
-        return OnboardingStateResult(onboarding=None)
 
-    return OnboardingStateResult(onboarding=OnboardingModel.from_api(data))
+@dbt_mcp_tool(
+    description=get_prompt("admin_api/onboarding_apply"),
+    title="Apply Onboarding Data",
+    read_only_hint=False,
+    destructive_hint=False,
+    idempotent_hint=True,
+)
+async def dbt_admin_onboarding_apply(
+    context: OnboardingToolContext,
+    data: dict[str, Any],
+) -> OnboardingApplyResult:
+    """Submit collected onboarding data; safe to call multiple times with partial data."""
+    config = await context.admin_api_config_provider.get_config()
+    raw = await context.onboarding_client.apply(config.account_id, data)
+    return OnboardingApplyResult(onboarding=OnboardingModel.from_api(raw))
 
 
 ONBOARDING_TOOLS = [
-    dbt_admin_onboarding_init,
-    dbt_admin_onboarding_state,
+    dbt_admin_onboarding_get,
+    dbt_admin_onboarding_validate,
+    dbt_admin_onboarding_apply,
 ]
 
 

src/dbt_mcp/tools/readme_mappings.py

@@ -60,8 +60,9 @@
     ToolName.LIST_JOB_RUN_ARTIFACTS: "Lists available artifacts from a job run.",
     ToolName.GET_JOB_RUN_ERROR: "Gets error and/or warning details for a job run; option to include or show warnings only.",
     # Onboarding tools
-    ToolName.ONBOARDING_INIT: "Starts or resumes an onboarding session; returns session ID, current phase, and decision points.",
-    ToolName.ONBOARDING_STATE: "Returns the current onboarding session phase and server-side applied resource progress.",
+    ToolName.ONBOARDING_GET: "Returns the current onboarding record and progress; null if no onboarding has been started.",
+    ToolName.ONBOARDING_VALIDATE: "Validates the collected onboarding data and returns what is missing or invalid.",
+    ToolName.ONBOARDING_APPLY: "Submits collected onboarding data to the platform; call incrementally as each piece of data is gathered.",
     # dbt-codegen tools
     ToolName.GENERATE_SOURCE: "Generates source YAML by introspecting database schemas; option to include columns.",
     ToolName.GENERATE_MODEL_YAML: "Generates model YAML with columns; option to inherit upstream descriptions.",

src/dbt_mcp/tools/tool_names.py

@@ -65,8 +65,9 @@ class ToolName(Enum):
     GET_JOB_RUN_ERROR = "get_job_run_error"
 
     # Onboarding tools
-    ONBOARDING_INIT = "dbt_admin_onboarding_init"
-    ONBOARDING_STATE = "dbt_admin_onboarding_state"
+    ONBOARDING_GET = "dbt_admin_onboarding_get"
+    ONBOARDING_VALIDATE = "dbt_admin_onboarding_validate"
+    ONBOARDING_APPLY = "dbt_admin_onboarding_apply"
 
     # dbt-codegen tools
     GENERATE_SOURCE = "generate_source"

src/dbt_mcp/tools/toolsets.py

@@ -103,8 +103,9 @@ class Toolset(Enum):
         ToolName.RETRY_JOB_RUN,
         ToolName.LIST_JOB_RUN_ARTIFACTS,
         ToolName.GET_JOB_RUN_ERROR,
-        ToolName.ONBOARDING_INIT,
-        ToolName.ONBOARDING_STATE,
+        ToolName.ONBOARDING_GET,
+        ToolName.ONBOARDING_VALIDATE,
+        ToolName.ONBOARDING_APPLY,
     },
     Toolset.DBT_CODEGEN: {
         ToolName.GENERATE_SOURCE,