feat(onboarding): scaffold onboarding package with init and state tools

Adds the foundation for AI-driven dbt platform onboarding. An LLM agent
can now start or resume an onboarding session and check its progress —
all without touching any UI.

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

Author: joaodaher

.changes/unreleased/Enhancement or New Feature-20260608-120113.yaml

@@ -0,0 +1,3 @@
+kind: Enhancement or New Feature
+body: Add onboarding package scaffold and init/state tools (CC-3674, CC-3675)
+time: 2026-06-08T12:01:13.725527-03:00

README.md

@@ -82,6 +82,8 @@ Allowing your client to utilize dbt commands through the MCP tooling could modif
 
 To learn more about the dbt Administrative API, click [here](https://docs.getdbt.com/docs/dbt-cloud-apis/admin-cloud-api).
 - `cancel_job_run`: Cancels a running job.
+- `dbt_admin_onboarding_init`: Starts or resumes an onboarding session; returns session ID, current phase, and decision points.
+- `dbt_admin_onboarding_state`: Returns the current onboarding session phase and server-side applied resource progress.
 - `get_job_details`: Gets job configuration including triggers, schedule, and dbt commands.
 - `get_job_run_details`: Gets run details including status, timing, steps, and artifacts.
 - `get_job_run_error`: Gets error and/or warning details for a job run; option to include or show warnings only.

src/dbt_mcp/dbt_admin/onboarding/__init__.py

@@ -0,0 +1,56 @@
+from __future__ import annotations
+
+import logging
+from typing import Any
+
+import httpx
+
+from dbt_mcp.config.config_providers import AdminApiConfig, ConfigProvider
+from dbt_mcp.errors import AdminAPIError, InvalidParameterError
+
+logger = logging.getLogger(__name__)
+
+
+class OnboardingClient:
+    """HTTP client for the dbt platform onboarding API.
+
+    Reuses AdminApiConfig (same base URL + auth headers) since the onboarding
+    endpoints live under /api/v3/accounts/{account_id}/onboarding/.
+    """
+
+    def __init__(self, config_provider: ConfigProvider[AdminApiConfig]) -> None:
+        self.config_provider = config_provider
+
+    async def get_state(self, account_id: int) -> dict[str, Any]:
+        """Fetch the server-side onboarding state for the account.
+
+        Returns the OnboardingSession row: applied resource IDs and status.
+        """
+        config = await self.config_provider.get_config()
+        url = f"{config.url}/api/v3/accounts/{account_id}/onboarding/state/"
+        headers = {
+            "Content-Type": "application/json",
+            "Accept": "application/json",
+        } | config.headers_provider.get_headers()
+
+        try:
+            async with httpx.AsyncClient() as client:
+                response = await client.get(
+                    url, headers=headers, follow_redirects=True
+                )
+                response.raise_for_status()
+                return response.json()
+        except httpx.HTTPStatusError as e:
+            if 400 <= e.response.status_code < 500:
+                raise InvalidParameterError(
+                    f"Onboarding state request failed ({e.response.status_code}) "
+                    f"for account {account_id}"
+                ) from e
+            raise AdminAPIError(
+                f"Onboarding state request failed ({e.response.status_code}) "
+                f"for account {account_id}"
+            ) from e
+        except httpx.HTTPError as e:
+            raise AdminAPIError(
+                f"Onboarding state request failed for account {account_id}"
+            ) from e

src/dbt_mcp/dbt_admin/onboarding/client.py

@@ -0,0 +1,29 @@
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+
+
+@dataclass
+class DecisionPoint:
+    """A single decision the user must make during onboarding.
+
+    Populated in Phase 0+ tickets as each resource type is added to the flow.
+    The LLM proposes values; the user confirms every one.
+    """
+
+    key: str
+    label: str
+    description: str
+    sensitive: bool = False
+    rationale_hint: str = ""
+    examples: list[str] = field(default_factory=list)
+
+
+def get_decision_points() -> list[DecisionPoint]:
+    """Return the current set of decision points for the onboarding flow.
+
+    Empty in the Foundation phase (CC-3674/CC-3675). Decision points are added
+    incrementally in Phase 0 (account), Phase 1 (project), Phase 2 (environment),
+    and Phase 3 (job) tickets.
+    """
+    return []

src/dbt_mcp/dbt_admin/onboarding/decision_points.py

@@ -0,0 +1,42 @@
+from typing import Any
+
+from pydantic import BaseModel
+
+from dbt_mcp.dbt_admin.onboarding.decision_points import DecisionPoint
+
+
+class OnboardingInitResult(BaseModel):
+    session_id: str
+    phase: str
+    account_id: int
+    decision_points: list[dict[str, Any]]
+
+
+class ServerOnboardingState(BaseModel):
+    """Mirrors the GET /api/v3/accounts/{id}/onboarding/state/ response.
+
+    Resource FK fields are added incrementally as each phase ships.
+    """
+
+    status: str
+    data: dict[str, Any] = {}
+
+
+class OnboardingStateResult(BaseModel):
+    phase: str | None
+    session_id: str | None
+    server_state: ServerOnboardingState | None
+
+
+def decision_points_to_dicts(points: list[DecisionPoint]) -> list[dict[str, Any]]:
+    return [
+        {
+            "key": p.key,
+            "label": p.label,
+            "description": p.description,
+            "sensitive": p.sensitive,
+            "rationale_hint": p.rationale_hint,
+            "examples": p.examples,
+        }
+        for p in points
+    ]

src/dbt_mcp/dbt_admin/onboarding/models.py

@@ -0,0 +1,63 @@
+from __future__ import annotations
+
+import uuid
+from dataclasses import dataclass, field
+from enum import Enum
+from typing import Any
+
+
+class SessionPhase(str, Enum):
+    DECIDING = "deciding"
+    REVIEWING = "reviewing"
+    APPLYING = "applying"
+    APPLY_FAILED = "apply_failed"
+    AWAITING_VALIDATION = "awaiting_validation"
+    VALIDATING = "validating"
+    COMPLETE = "complete"
+    VALIDATION_FAILED = "validation_failed"
+    APPLIED_UNVALIDATED = "applied_unvalidated"
+
+
+# Phases where server-side state should be fetched alongside local session phase
+_PHASES_WITH_SERVER_STATE = {
+    SessionPhase.APPLYING,
+    SessionPhase.APPLY_FAILED,
+    SessionPhase.AWAITING_VALIDATION,
+    SessionPhase.VALIDATING,
+    SessionPhase.COMPLETE,
+    SessionPhase.VALIDATION_FAILED,
+    SessionPhase.APPLIED_UNVALIDATED,
+}
+
+
+@dataclass
+class OnboardingSession:
+    session_id: str = field(default_factory=lambda: str(uuid.uuid4()))
+    account_id: int = 0
+    phase: SessionPhase = SessionPhase.DECIDING
+    decisions: dict[str, Any] = field(default_factory=dict)
+
+    def has_server_state(self) -> bool:
+        return self.phase in _PHASES_WITH_SERVER_STATE
+
+
+class InMemorySessionStore:
+    """In-memory session store keyed by account_id.
+
+    Per-process — survives the lifetime of the MCP server process but not restarts.
+    Sufficient for stdio transport. Remote MCP persistence is a future concern.
+    """
+
+    def __init__(self) -> None:
+        self._sessions: dict[int, OnboardingSession] = {}
+
+    def get_or_create(self, account_id: int) -> OnboardingSession:
+        if account_id not in self._sessions:
+            self._sessions[account_id] = OnboardingSession(account_id=account_id)
+        return self._sessions[account_id]
+
+    def get(self, account_id: int) -> OnboardingSession | None:
+        return self._sessions.get(account_id)
+
+    def update(self, session: OnboardingSession) -> None:
+        self._sessions[session.account_id] = session

src/dbt_mcp/dbt_admin/onboarding/session.py

@@ -0,0 +1,124 @@
+from dataclasses import dataclass
+
+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.decision_points import get_decision_points
+from dbt_mcp.dbt_admin.onboarding.models import (
+    OnboardingInitResult,
+    OnboardingStateResult,
+    ServerOnboardingState,
+    decision_points_to_dicts,
+)
+from dbt_mcp.dbt_admin.onboarding.session import InMemorySessionStore
+from dbt_mcp.prompts.prompts import get_prompt
+from dbt_mcp.tools.definitions import dbt_mcp_tool
+from dbt_mcp.tools.register import register_tools
+from dbt_mcp.tools.tool_names import ToolName
+from dbt_mcp.tools.toolsets import Toolset
+
+
+@dataclass
+class OnboardingToolContext:
+    admin_api_config_provider: ConfigProvider[AdminApiConfig]
+    session_store: InMemorySessionStore
+    onboarding_client: OnboardingClient
+
+
+@dbt_mcp_tool(
+    description=get_prompt("admin_api/onboarding_init"),
+    title="Start or Resume Onboarding",
+    read_only_hint=False,
+    destructive_hint=False,
+    idempotent_hint=True,
+)
+async def dbt_admin_onboarding_init(
+    context: OnboardingToolContext,
+) -> OnboardingInitResult:
+    """Start or resume an onboarding session for the account."""
+    config = await context.admin_api_config_provider.get_config()
+    account_id = config.account_id
+
+    session = context.session_store.get_or_create(account_id)
+
+    return OnboardingInitResult(
+        session_id=session.session_id,
+        phase=session.phase.value,
+        account_id=account_id,
+        decision_points=decision_points_to_dicts(get_decision_points()),
+    )
+
+
+@dbt_mcp_tool(
+    description=get_prompt("admin_api/onboarding_state"),
+    title="Get Onboarding State",
+    read_only_hint=True,
+    destructive_hint=False,
+    idempotent_hint=True,
+)
+async def dbt_admin_onboarding_state(
+    context: OnboardingToolContext,
+) -> OnboardingStateResult:
+    """Return the current onboarding session state for the account."""
+    config = await context.admin_api_config_provider.get_config()
+    account_id = config.account_id
+
+    session = context.session_store.get(account_id)
+
+    if session is None:
+        return OnboardingStateResult(
+            phase=None,
+            session_id=None,
+            server_state=None,
+        )
+
+    server_state: ServerOnboardingState | None = None
+    if session.has_server_state():
+        raw = await context.onboarding_client.get_state(account_id)
+        server_state = ServerOnboardingState(
+            status=raw.get("data", {}).get("status", raw.get("status", "unknown")),
+            data=raw.get("data", raw),
+        )
+
+    return OnboardingStateResult(
+        phase=session.phase.value,
+        session_id=session.session_id,
+        server_state=server_state,
+    )
+
+
+ONBOARDING_TOOLS = [
+    dbt_admin_onboarding_init,
+    dbt_admin_onboarding_state,
+]
+
+
+def register_onboarding_tools(
+    dbt_mcp: FastMCP,
+    admin_api_config_provider: ConfigProvider[AdminApiConfig],
+    *,
+    disabled_tools: set[ToolName],
+    enabled_tools: set[ToolName] | None,
+    enabled_toolsets: set[Toolset],
+    disabled_toolsets: set[Toolset],
+) -> None:
+    """Register dbt onboarding tools."""
+    session_store = InMemorySessionStore()
+    onboarding_client = OnboardingClient(admin_api_config_provider)
+
+    def bind_context() -> OnboardingToolContext:
+        return OnboardingToolContext(
+            admin_api_config_provider=admin_api_config_provider,
+            session_store=session_store,
+            onboarding_client=onboarding_client,
+        )
+
+    register_tools(
+        dbt_mcp,
+        tool_definitions=[tool.adapt_context(bind_context) for tool in ONBOARDING_TOOLS],
+        disabled_tools=disabled_tools,
+        enabled_tools=enabled_tools,
+        enabled_toolsets=enabled_toolsets,
+        disabled_toolsets=disabled_toolsets,
+    )

src/dbt_mcp/dbt_admin/onboarding/tools.py

@@ -12,6 +12,7 @@
 from mcp.types import ContentBlock, Tool
 
 from dbt_mcp.config.config import Config
+from dbt_mcp.dbt_admin.onboarding.tools import register_onboarding_tools
 from dbt_mcp.dbt_admin.tools import register_admin_api_tools
 from dbt_mcp.dbt_cli.tools import register_dbt_cli_tools
 from dbt_mcp.dbt_codegen.tools import register_dbt_codegen_tools
@@ -232,6 +233,15 @@ async def register_multi_project_dbt_mcp(dbt_mcp: FastMCP, config: Config) -> No
             enabled_toolsets=enabled_toolsets,
             disabled_toolsets=disabled_toolsets,
         )
+        logger.info("Registering onboarding tools for multi-project")
+        register_onboarding_tools(
+            dbt_mcp,
+            config.admin_api_config_provider,
+            disabled_tools=disabled_tools,
+            enabled_tools=enabled_tools,
+            enabled_toolsets=enabled_toolsets,
+            disabled_toolsets=disabled_toolsets,
+        )
 
     # Product docs and MCP server metadata are globally available (not project-specific)
     logger.info("Registering product docs tools for multi-project")
@@ -336,6 +346,16 @@ async def register_dbt_mcp_tools(dbt_mcp: FastMCP, config: Config) -> None:
         disabled_toolsets=disabled_toolsets,
     )
 
+    logger.info("Registering onboarding tools")
+    register_onboarding_tools(
+        dbt_mcp,
+        config.admin_api_config_provider,
+        disabled_tools=disabled_tools,
+        enabled_tools=enabled_tools,
+        enabled_toolsets=enabled_toolsets,
+        disabled_toolsets=disabled_toolsets,
+    )
+
     if config.lsp_config:
         logger.info("Registering LSP tools")
         await register_lsp_tools(

src/dbt_mcp/mcp/server.py

@@ -0,0 +1,13 @@
+Start or resume an onboarding session for this dbt platform account.
+
+Call this at the beginning of any onboarding conversation. It issues a `session_id` and returns the full decision tree — the set of values the user needs to confirm before the platform can be configured.
+
+Returns:
+- **session_id**: A unique identifier for this onboarding session. Pass it to subsequent onboarding tools.
+- **phase**: The current session phase. Starts as `deciding` and progresses through `reviewing → applying → awaiting_validation → complete`.
+- **account_id**: The account being onboarded.
+- **decision_points**: The list of decisions the user must make, each with a `key`, `label`, `description`, optional `rationale_hint` for how to suggest a value, and `examples`. Empty in the Foundation phase; populated as account/project/environment/job phases ship.
+
+If a session already exists for this account, it is resumed from the current phase — the user does not need to restart.
+
+Suggest values for non-`sensitive` decision points based on detected project context. Never invent values for `sensitive: true` keys — ask the user explicitly.