CC-3677: account-create tool + owner-token stash

Phase 0 (MCP side) of agent-driven onboarding. Adds the low-level
`dbt_admin_account_create(name, owner_email, created_via?)` tool that calls
the public dbt-cloud POST /api/v3/accounts/ endpoint via a token-less,
host-only client (account creation happens before any account or token
exists, so it must not trigger the OAuth login flow).

On success it stashes the returned account_id + owner token into the shared
CredentialsProvider, so the existing account-scoped admin/onboarding tools
authenticate automatically for the rest of the session.

Gated under the admin_api toolset; billable so destructive-confirm applies.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

Author: joaodaher

README.md

@@ -82,6 +82,7 @@ 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_account_create`: Bootstraps a brand-new trial account and owner token (billable); the token is stashed for the rest of the session.
 - `dbt_admin_onboarding_apply`: Submits collected onboarding data to the platform; call incrementally as each piece of data is gathered.
 - `dbt_admin_onboarding_get`: Returns the current onboarding record and progress; null if no onboarding has been started.
 - `dbt_admin_onboarding_validate`: Validates the collected onboarding data and returns what is missing or invalid.

src/dbt_mcp/dbt_admin/onboarding/account_client.py

@@ -0,0 +1,79 @@
+from __future__ import annotations
+
+import logging
+from typing import TYPE_CHECKING, Any
+
+import httpx
+
+from dbt_mcp.errors import AdminAPIError, InvalidParameterError
+
+if TYPE_CHECKING:
+    from dbt_mcp.config.credentials import CredentialsProvider
+
+logger = logging.getLogger(__name__)
+
+
+class AccountClient:
+    """HTTP client for the public, unauthenticated account-create endpoint.
+
+    Account creation happens *before* any account or token exists, so this
+    client only needs the host — it deliberately does NOT call
+    ``credentials_provider.get_credentials()`` (which would trigger the OAuth
+    login flow). It reads the host straight off the configured settings.
+    """
+
+    def __init__(self, credentials_provider: CredentialsProvider) -> None:
+        self.credentials_provider = credentials_provider
+
+    def _url(self) -> str:
+        settings = self.credentials_provider.settings
+        host = settings.actual_host
+        if not host:
+            raise InvalidParameterError(
+                "DBT_HOST is required to create an account but is not configured."
+            )
+        if settings.actual_host_prefix:
+            base = f"https://{settings.actual_host_prefix}.{settings.base_host}"
+        else:
+            base = f"https://{host}"
+        return f"{base}/api/v3/accounts/"
+
+    async def create(
+        self,
+        *,
+        name: str,
+        owner_email: str,
+        created_via: str | None = None,
+    ) -> dict[str, Any]:
+        """Create a trial account + owner token. Returns the response ``data`` dict."""
+        url = self._url()
+        body: dict[str, Any] = {"name": name, "owner_email": owner_email}
+        if created_via is not None:
+            body["created_via"] = created_via
+
+        try:
+            async with httpx.AsyncClient() as client:
+                response = await client.post(
+                    url,
+                    headers={
+                        "Content-Type": "application/json",
+                        "Accept": "application/json",
+                    },
+                    json=body,
+                    follow_redirects=True,
+                )
+                response.raise_for_status()
+                data = response.json().get("data")
+                if not data:
+                    raise AdminAPIError("Account create returned no data")
+                return data
+        except httpx.HTTPStatusError as e:
+            if 400 <= e.response.status_code < 500:
+                raise InvalidParameterError(
+                    f"Account create failed ({e.response.status_code})"
+                ) from e
+            raise AdminAPIError(
+                f"Account create failed ({e.response.status_code})"
+            ) from e
+        except httpx.HTTPError as e:
+            raise AdminAPIError("Account create failed") from e

src/dbt_mcp/dbt_admin/onboarding/models.py

@@ -35,6 +35,13 @@ def from_api(cls, data: dict[str, Any]) -> "OnboardingModel":
         )
 
 
+class AccountCreateResult(BaseModel):
+    """Result of bootstrapping a brand-new account via the public endpoint."""
+
+    account_id: int
+    owner_token: str
+
+
 class OnboardingGetResult(BaseModel):
     onboarding: OnboardingModel | None
 

src/dbt_mcp/dbt_admin/onboarding/tools.py

@@ -1,27 +1,69 @@
 from dataclasses import dataclass
-from typing import Any
+from typing import TYPE_CHECKING, Any
 
 from mcp.server.fastmcp import FastMCP
 
 from dbt_mcp.config.config_providers import AdminApiConfig, ConfigProvider
+from dbt_mcp.dbt_admin.onboarding.account_client import AccountClient
 from dbt_mcp.dbt_admin.onboarding.client import OnboardingClient
 from dbt_mcp.dbt_admin.onboarding.models import (
+    AccountCreateResult,
     OnboardingApplyResult,
     OnboardingGetResult,
     OnboardingModel,
     OnboardingValidateResult,
 )
+from dbt_mcp.oauth.token_provider import StaticTokenProvider
 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
 
+if TYPE_CHECKING:
+    from dbt_mcp.config.credentials import CredentialsProvider
+
 
 @dataclass
 class OnboardingToolContext:
     admin_api_config_provider: ConfigProvider[AdminApiConfig]
     onboarding_client: OnboardingClient
+    credentials_provider: "CredentialsProvider"
+    account_client: AccountClient
+
+
+@dbt_mcp_tool(
+    description=get_prompt("admin_api/account_create"),
+    title="Create dbt Platform Account",
+    read_only_hint=False,
+    destructive_hint=False,
+    idempotent_hint=False,
+)
+async def dbt_admin_account_create(
+    context: OnboardingToolContext,
+    name: str,
+    owner_email: str,
+    created_via: str | None = None,
+) -> AccountCreateResult:
+    """Bootstrap a brand-new dbt platform account and owner token.
+
+    Use this only when the user has no account yet — it is billable and creates
+    a trial account. The returned owner token is stashed for the rest of this
+    session, so subsequent admin/onboarding tools authenticate automatically.
+    """
+    data = await context.account_client.create(
+        name=name,
+        owner_email=owner_email,
+        created_via=created_via,
+    )
+    account_id = int(data["account_id"])
+    owner_token = data["owner_token"]
+
+    # Stash the new identity so later account-scoped tools authenticate.
+    context.credentials_provider.settings.dbt_account_id = account_id
+    context.credentials_provider.token_provider = StaticTokenProvider(token=owner_token)
+
+    return AccountCreateResult(account_id=account_id, owner_token=owner_token)
 
 
 @dbt_mcp_tool(
@@ -82,6 +124,7 @@ async def dbt_admin_onboarding_apply(
 
 
 ONBOARDING_TOOLS = [
+    dbt_admin_account_create,
     dbt_admin_onboarding_get,
     dbt_admin_onboarding_validate,
     dbt_admin_onboarding_apply,
@@ -91,6 +134,7 @@ async def dbt_admin_onboarding_apply(
 def register_onboarding_tools(
     dbt_mcp: FastMCP,
     admin_api_config_provider: ConfigProvider[AdminApiConfig],
+    credentials_provider: "CredentialsProvider",
     *,
     disabled_tools: set[ToolName],
     enabled_tools: set[ToolName] | None,
@@ -99,11 +143,14 @@ def register_onboarding_tools(
 ) -> None:
     """Register dbt onboarding tools."""
     onboarding_client = OnboardingClient(admin_api_config_provider)
+    account_client = AccountClient(credentials_provider)
 
     def bind_context() -> OnboardingToolContext:
         return OnboardingToolContext(
             admin_api_config_provider=admin_api_config_provider,
             onboarding_client=onboarding_client,
+            credentials_provider=credentials_provider,
+            account_client=account_client,
         )
 
     register_tools(

src/dbt_mcp/mcp/server.py

@@ -237,6 +237,7 @@ async def register_multi_project_dbt_mcp(dbt_mcp: FastMCP, config: Config) -> No
         register_onboarding_tools(
             dbt_mcp,
             config.admin_api_config_provider,
+            config.credentials_provider.inner_provider,
             disabled_tools=disabled_tools,
             enabled_tools=enabled_tools,
             enabled_toolsets=enabled_toolsets,
@@ -350,6 +351,7 @@ async def register_dbt_mcp_tools(dbt_mcp: FastMCP, config: Config) -> None:
     register_onboarding_tools(
         dbt_mcp,
         config.admin_api_config_provider,
+        config.credentials_provider.inner_provider,
         disabled_tools=disabled_tools,
         enabled_tools=enabled_tools,
         enabled_toolsets=enabled_toolsets,

src/dbt_mcp/prompts/admin_api/account_create.md

@@ -0,0 +1,7 @@
+Creates a brand-new dbt platform account and returns an owner API token.
+
+Use this only when the user does not have an account yet. It is billable — it spins up a trial account — so confirm with the user before calling. If the user already has an account, do not call this; use their existing account instead.
+
+Provide the desired account `name` and the owner's `owner_email`. Optionally pass `created_via` (e.g. `onboarding_api`) for funnel attribution; leave it unset if unsure.
+
+The returned owner token is stored for the rest of this session, so the account-scoped admin and onboarding tools will authenticate automatically afterwards — you do not need to ask the user for a token.

src/dbt_mcp/tools/readme_mappings.py

@@ -60,6 +60,7 @@
     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.ACCOUNT_CREATE: "Bootstraps a brand-new trial account and owner token (billable); the token is stashed for the rest of the session.",
     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.",

src/dbt_mcp/tools/tool_names.py

@@ -65,6 +65,7 @@ class ToolName(Enum):
     GET_JOB_RUN_ERROR = "get_job_run_error"
 
     # Onboarding tools
+    ACCOUNT_CREATE = "dbt_admin_account_create"
     ONBOARDING_GET = "dbt_admin_onboarding_get"
     ONBOARDING_VALIDATE = "dbt_admin_onboarding_validate"
     ONBOARDING_APPLY = "dbt_admin_onboarding_apply"

src/dbt_mcp/tools/toolsets.py

@@ -103,6 +103,7 @@ class Toolset(Enum):
         ToolName.RETRY_JOB_RUN,
         ToolName.LIST_JOB_RUN_ARTIFACTS,
         ToolName.GET_JOB_RUN_ERROR,
+        ToolName.ACCOUNT_CREATE,
         ToolName.ONBOARDING_GET,
         ToolName.ONBOARDING_VALIDATE,
         ToolName.ONBOARDING_APPLY,

tests/unit/dbt_admin/onboarding/test_account_client.py

@@ -0,0 +1,80 @@
+from types import SimpleNamespace
+from unittest.mock import AsyncMock, MagicMock, patch
+
+import pytest
+
+from dbt_mcp.dbt_admin.onboarding.account_client import AccountClient
+from dbt_mcp.errors import InvalidParameterError
+
+
+def _creds(actual_host, actual_host_prefix=None, base_host=None):
+    return SimpleNamespace(
+        settings=SimpleNamespace(
+            actual_host=actual_host,
+            actual_host_prefix=actual_host_prefix,
+            base_host=base_host or actual_host,
+        )
+    )
+
+
+def _mock_httpx_client(mock_response):
+    mock_client = AsyncMock()
+    mock_client.post = AsyncMock(return_value=mock_response)
+    mock_client.__aenter__ = AsyncMock(return_value=mock_client)
+    mock_client.__aexit__ = AsyncMock(return_value=None)
+    return mock_client
+
+
+async def test_create_posts_unauthenticated_to_accounts_url():
+    client = AccountClient(_creds("cloud.getdbt.com"))
+    mock_response = MagicMock()
+    mock_response.json.return_value = {"data": {"account_id": 7, "owner_token": "tok"}}
+    mock_response.raise_for_status.return_value = None
+    mock_client = _mock_httpx_client(mock_response)
+
+    with patch("httpx.AsyncClient", return_value=mock_client):
+        data = await client.create(name="acme", owner_email="o@acme.example")
+
+    assert data == {"account_id": 7, "owner_token": "tok"}
+    args, kwargs = mock_client.post.call_args
+    assert args[0] == "https://cloud.getdbt.com/api/v3/accounts/"
+    assert kwargs["json"] == {"name": "acme", "owner_email": "o@acme.example"}
+    # Public endpoint — no Authorization header is sent.
+    assert "Authorization" not in kwargs["headers"]
+
+
+async def test_create_includes_created_via_when_provided():
+    client = AccountClient(_creds("cloud.getdbt.com"))
+    mock_response = MagicMock()
+    mock_response.json.return_value = {"data": {"account_id": 7, "owner_token": "tok"}}
+    mock_response.raise_for_status.return_value = None
+    mock_client = _mock_httpx_client(mock_response)
+
+    with patch("httpx.AsyncClient", return_value=mock_client):
+        await client.create(
+            name="acme", owner_email="o@acme.example", created_via="onboarding_api"
+        )
+
+    assert mock_client.post.call_args.kwargs["json"]["created_via"] == "onboarding_api"
+
+
+async def test_create_uses_host_prefix():
+    client = AccountClient(_creds("cloud.getdbt.com", actual_host_prefix="ab123"))
+    mock_response = MagicMock()
+    mock_response.json.return_value = {"data": {"account_id": 1, "owner_token": "t"}}
+    mock_response.raise_for_status.return_value = None
+    mock_client = _mock_httpx_client(mock_response)
+
+    with patch("httpx.AsyncClient", return_value=mock_client):
+        await client.create(name="a", owner_email="o@a.example")
+
+    assert (
+        mock_client.post.call_args.args[0]
+        == "https://ab123.cloud.getdbt.com/api/v3/accounts/"
+    )
+
+
+async def test_create_without_host_raises():
+    client = AccountClient(_creds(None))
+    with pytest.raises(InvalidParameterError):
+        await client.create(name="a", owner_email="o@a.example")