fix: preserve Horizon command contracts

docs/cli/overview.mdx

@@ -28,6 +28,9 @@ fastmcp --help
 | [`generate-cli`](/cli/generate-cli) | Scaffold a standalone typed CLI from a server's tool schemas |
 | [`project prepare`](/cli/running#pre-building-environments) | Pre-install dependencies into a reusable uv project |
 | [`auth cimd`](/cli/auth) | Create and validate CIMD documents for OAuth |
+| `login` | Sign in to Prefect Horizon with a browser device flow |
+| `whoami` | Show the current Horizon account |
+| `logout` | Revoke the current Horizon key and remove the local credential |
 | `version` | Print version info (`--copy` to copy to clipboard) |
 
 ## Server Targets
@@ -81,6 +84,43 @@ Run [`fastmcp discover`](/cli/client#discovering-configured-servers) to see what
 
 ## Authentication
 
+### Prefect Horizon Account
+
+Use the top-level account commands to manage the credential for Prefect Horizon.
+
+```bash
+fastmcp login
+fastmcp whoami
+fastmcp logout
+```
+
+`fastmcp login` first uses `HORIZON_API_KEY` or a valid stored key when one is available.
+When login needs a new key, it shows a verification URL and code.
+It opens a browser when the terminal supports it.
+If the browser does not open, use the shown URL and code on another device.
+To switch accounts, run `fastmcp logout` before you run `fastmcp login` again.
+
+Use `--host` to connect to another Horizon environment.
+The CLI saves the host and clears the stored key before a host change.
+
+```bash
+fastmcp login --host https://horizon.example.com
+```
+
+Login stores only the personal Horizon API key.
+It does not select or store a deployment organization.
+`fastmcp whoami` gets the current user from Horizon.
+`fastmcp logout` attempts to revoke the active key and always removes the local credential.
+
+Set `HORIZON_API_KEY` to use an environment credential instead.
+The CLI gives that value first precedence and never stores it.
+
+Use `--json` for stable command results.
+During JSON login, the verification challenge goes to stderr and the final result goes to stdout.
+JSON mode does not open a browser or ask a question.
+
+### MCP Server Authentication
+
 When targeting an HTTP URL, the CLI enables OAuth authentication by default. If the server requires it, you'll be guided through the flow (typically opening a browser). If it doesn't, the setup is a silent no-op.
 
 To skip authentication entirely — useful for local development servers — pass `--auth none`:

docs/deployment/prefect-horizon.mdx

@@ -13,6 +13,39 @@ Horizon includes a **free personal tier for FastMCP users**, making it the faste
 Horizon is free for personal projects. Enterprise governance features are available for teams deploying to thousands of users.
 </Info>
 
+## FastMCP CLI Account
+
+Sign in to Horizon from the FastMCP CLI with the device authorization flow.
+
+```bash
+fastmcp login
+```
+
+The command uses an environment key or a valid stored key when one is available.
+When login needs a new key, it shows a verification URL and code before it opens the browser.
+If the browser cannot open, visit the shown URL and enter the code.
+To switch accounts, run `fastmcp logout` before you run `fastmcp login` again.
+New users can register and create their first Horizon organization in the browser.
+Use `fastmcp login --host <url>` to save a different Horizon host.
+A host change clears the stored credential before login.
+
+Check the active account after login.
+
+```bash
+fastmcp whoami
+```
+
+Remove the local credential and revoke the active personal API key when possible.
+
+```bash
+fastmcp logout
+```
+
+Login does not select or store a deployment organization.
+
+For an agent or a CI process, set `HORIZON_API_KEY` instead of storing a key.
+The CLI never writes the environment value to its credential file.
+
 ## The Platform
 
 Horizon is organized into four integrated pillars:

fastmcp_slim/fastmcp/cli/cli.py

@@ -21,6 +21,7 @@ import fastmcp
 from fastmcp.cli import run as run_module
 from fastmcp.cli.auth import auth_app
 from fastmcp.cli.client import call_command, discover_command, list_command
+from fastmcp.cli.deploy.command import login, logout, whoami
 from fastmcp.cli.generate import generate_cli_command
 from fastmcp.cli.install import install_app
 from fastmcp.utilities.cli import is_already_in_uv_subprocess, load_and_merge_config
@@ -1134,6 +1135,11 @@ app.command(generate_cli_command, name="generate-cli")
 # Add auth subcommand group (includes CIMD commands)
 app.command(auth_app)
 
+# Add Prefect Horizon account commands
+app.command(login)
+app.command(logout)
+app.command(whoami)
+
 
 if __name__ == "__main__":
     app()

fastmcp_slim/fastmcp/cli/deploy/__init__.py

@@ -0,0 +1 @@
+"""Horizon deployment support for the FastMCP CLI."""

fastmcp_slim/fastmcp/cli/deploy/authentication.py

@@ -0,0 +1,101 @@
+"""Horizon device authorization workflow."""
+
+from __future__ import annotations
+
+import asyncio
+import time
+import webbrowser
+from collections.abc import Awaitable, Callable
+from contextlib import suppress
+
+from pydantic import SecretStr
+
+from fastmcp.cli.deploy.horizon_client import (
+    DeviceAuthorization,
+    DeviceMetadata,
+    HorizonClient,
+)
+
+
+class DeviceAuthorizationError(RuntimeError):
+    """Device authorization did not complete."""
+
+
+class DeviceAuthorizationDeniedError(DeviceAuthorizationError):
+    """The user denied the device authorization request."""
+
+
+class DeviceAuthorizationExpiredError(DeviceAuthorizationError):
+    """The device authorization request expired."""
+
+
+async def poll_device_authorization(
+    client: HorizonClient,
+    authorization: DeviceAuthorization,
+    *,
+    sleep: Callable[[float], Awaitable[None]] | None = None,
+    monotonic: Callable[[], float] = time.monotonic,
+) -> SecretStr:
+    """Poll at the server interval until the device request completes."""
+    sleep = asyncio.sleep if sleep is None else sleep
+    deadline = monotonic() + authorization.expires_in
+    interval = float(authorization.interval)
+
+    while True:
+        remaining = deadline - monotonic()
+        if remaining <= 0:
+            raise DeviceAuthorizationExpiredError(
+                "The device authorization request expired"
+            )
+
+        await sleep(min(interval, remaining))
+        if monotonic() >= deadline:
+            raise DeviceAuthorizationExpiredError(
+                "The device authorization request expired"
+            )
+
+        result = await client.exchange_device_authorization(authorization.device_code)
+        if result.access_token is not None:
+            return result.access_token
+        if result.error == "authorization_pending":
+            continue
+        if result.error == "slow_down":
+            interval += 5
+            continue
+        if result.error == "access_denied":
+            raise DeviceAuthorizationDeniedError(
+                "The device authorization request was denied"
+            )
+        if result.error == "expired_token":
+            raise DeviceAuthorizationExpiredError(
+                "The device authorization request expired"
+            )
+
+        raise DeviceAuthorizationError("Device authorization failed")
+
+
+async def authorize_device(
+    client: HorizonClient,
+    *,
+    metadata: DeviceMetadata | None = None,
+    on_challenge: Callable[[DeviceAuthorization], None] | None = None,
+    open_browser: bool = False,
+    browser_opener: Callable[[str], object] = webbrowser.open,
+    sleep: Callable[[float], Awaitable[None]] | None = None,
+    monotonic: Callable[[], float] = time.monotonic,
+) -> SecretStr:
+    """Create, present, and complete a Horizon device authorization."""
+    authorization = await client.create_device_authorization(metadata)
+    if on_challenge is not None:
+        on_challenge(authorization)
+
+    if open_browser:
+        with suppress(OSError, webbrowser.Error):
+            browser_opener(authorization.verification_uri_complete)
+
+    return await poll_device_authorization(
+        client,
+        authorization,
+        sleep=sleep,
+        monotonic=monotonic,
+    )

fastmcp_slim/fastmcp/cli/deploy/command.py

@@ -0,0 +1,358 @@
+"""Public Prefect Horizon authentication commands."""
+
+from __future__ import annotations
+
+import platform
+import sys
+import webbrowser
+from typing import Annotated, NoReturn
+
+from cyclopts import Parameter
+from rich.status import Status
+
+import fastmcp
+from fastmcp.cli.deploy.authentication import (
+    DeviceAuthorizationDeniedError,
+    DeviceAuthorizationError,
+    DeviceAuthorizationExpiredError,
+    authorize_device,
+)
+from fastmcp.cli.deploy.configuration import ConfigurationStore
+from fastmcp.cli.deploy.credentials import (
+    AuthenticationRequiredError,
+    CredentialStore,
+    ResolvedCredential,
+    resolve_credential,
+    revoke_and_clear_credential,
+)
+from fastmcp.cli.deploy.horizon_client import (
+    DeviceAuthorization,
+    DeviceMetadata,
+    HorizonClient,
+    HorizonResponseError,
+    HorizonUnauthorizedError,
+    HorizonUnavailableError,
+    HorizonUser,
+)
+from fastmcp.cli.deploy.output import (
+    CommandName,
+    ErrorCategory,
+    emit_device_challenge,
+    emit_error,
+    emit_identity,
+    emit_logout,
+    start_device_approval_status,
+    stop_device_approval_status,
+)
+from fastmcp.cli.deploy.state import StateFileError
+
+JsonOption = Annotated[
+    bool,
+    Parameter(
+        name="--json",
+        help="Write one final JSON result to stdout",
+        negative=(),
+    ),
+]
+HostOption = Annotated[
+    str | None,
+    Parameter(
+        name="--host",
+        help="Use and save a different Horizon host URL",
+    ),
+]
+
+
+def _can_open_browser() -> bool:
+    return sys.stdin.isatty() and sys.stdout.isatty()
+
+
+def _device_metadata() -> DeviceMetadata:
+    return DeviceMetadata(
+        device_name=platform.node() or None,
+        platform=platform.system().lower() or None,
+        architecture=platform.machine().lower() or None,
+        client_version=fastmcp.__version__,
+    )
+
+
+def _fail(
+    command: CommandName,
+    category: ErrorCategory,
+    message: str,
+    *,
+    json_output: bool,
+    details: dict[str, object] | None = None,
+) -> NoReturn:
+    emit_error(
+        command,
+        category,
+        message,
+        json_output=json_output,
+        details=details,
+    )
+    raise SystemExit(1)
+
+
+def _fail_for_expected_error(
+    command: CommandName,
+    error: Exception,
+    *,
+    json_output: bool,
+) -> NoReturn:
+    if isinstance(error, AuthenticationRequiredError):
+        _fail(
+            command,
+            "authentication_required",
+            "Run `fastmcp login` to sign in to Prefect Horizon.",
+            json_output=json_output,
+        )
+    if isinstance(error, HorizonUnauthorizedError):
+        _fail(
+            command,
+            "authentication_invalid",
+            "The Horizon credential is not valid. Run `fastmcp login` again.",
+            json_output=json_output,
+        )
+    if isinstance(error, DeviceAuthorizationDeniedError):
+        _fail(
+            command,
+            "authorization_denied",
+            "The device authorization request was denied.",
+            json_output=json_output,
+        )
+    if isinstance(error, DeviceAuthorizationExpiredError):
+        _fail(
+            command,
+            "authorization_expired",
+            "The device authorization request expired. Run the command again.",
+            json_output=json_output,
+        )
+    if isinstance(error, DeviceAuthorizationError):
+        _fail(
+            command,
+            "authorization_failed",
+            "The device authorization request failed. Run the command again.",
+            json_output=json_output,
+        )
+    if isinstance(error, HorizonUnavailableError):
+        _fail(
+            command,
+            "horizon_unavailable",
+            "The Horizon API is unavailable. Try again later.",
+            json_output=json_output,
+        )
+    if isinstance(error, HorizonResponseError):
+        _fail(
+            command,
+            "horizon_error",
+            "Horizon returned an unexpected response. Try again later.",
+            json_output=json_output,
+        )
+    if isinstance(error, StateFileError):
+        _fail(
+            command,
+            "state_error",
+            "The local Horizon state is invalid.",
+            json_output=json_output,
+        )
+    raise error
+
+
+async def _get_user(
+    api_origin: str,
+    credential: ResolvedCredential,
+) -> HorizonUser:
+    async with HorizonClient(api_origin, api_key=credential.api_key) as client:
+        return await client.get_current_user()
+
+
+async def login(
+    *,
+    host: HostOption = None,
+    json_output: JsonOption = False,
+) -> None:
+    """Sign in to Prefect Horizon."""
+    credentials = CredentialStore()
+
+    try:
+        configuration_store = ConfigurationStore()
+        if host is None:
+            configuration = configuration_store.load()
+        else:
+            try:
+                configuration = configuration_store.set_api_origin(
+                    host,
+                    credentials=credentials,
+                )
+            except ValueError:
+                _fail(
+                    "login",
+                    "invalid_host",
+                    "The Horizon host must be an HTTP origin.",
+                    json_output=json_output,
+                )
+
+        async def device_authorization():
+            approval_status: Status | None = None
+
+            def show_challenge(challenge: DeviceAuthorization) -> None:
+                nonlocal approval_status
+                emit_device_challenge(challenge, json_output=json_output)
+                approval_status = start_device_approval_status(json_output=json_output)
+
+            try:
+                async with HorizonClient(configuration.api_origin) as client:
+                    return await authorize_device(
+                        client,
+                        metadata=_device_metadata(),
+                        on_challenge=show_challenge,
+                        open_browser=not json_output and _can_open_browser(),
+                        browser_opener=webbrowser.open,
+                    )
+            finally:
+                stop_device_approval_status(approval_status)
+
+        credential = await resolve_credential(
+            credentials,
+            authorize=device_authorization,
+            expected_api_origin=configuration.api_origin,
+        )
+
+        try:
+            user = await _get_user(
+                configuration.api_origin,
+                credential,
+            )
+        except HorizonUnauthorizedError:
+            if credential.source == "interactive":
+                credentials.clear()
+                raise
+            if credential.source == "environment":
+                raise
+
+            credentials.clear()
+            credential = await resolve_credential(
+                credentials,
+                authorize=device_authorization,
+                expected_api_origin=configuration.api_origin,
+            )
+            try:
+                user = await _get_user(
+                    configuration.api_origin,
+                    credential,
+                )
+            except HorizonUnauthorizedError:
+                credentials.clear()
+                raise
+    except (
+        AuthenticationRequiredError,
+        DeviceAuthorizationError,
+        HorizonResponseError,
+        HorizonUnauthorizedError,
+        HorizonUnavailableError,
+        StateFileError,
+    ) as error:
+        _fail_for_expected_error("login", error, json_output=json_output)
+
+    emit_identity(
+        "login",
+        user,
+        json_output=json_output,
+    )
+
+
+async def whoami(
+    *,
+    json_output: JsonOption = False,
+) -> None:
+    """Show the current Prefect Horizon user."""
+    credentials = CredentialStore()
+    credential: ResolvedCredential | None = None
+
+    try:
+        configuration = ConfigurationStore().load()
+        credential = await resolve_credential(credentials)
+        user = await _get_user(
+            configuration.api_origin,
+            credential,
+        )
+    except HorizonUnauthorizedError as error:
+        if credential is not None and credential.source == "stored":
+            try:
+                credentials.clear()
+            except StateFileError as cleanup_error:
+                _fail_for_expected_error(
+                    "whoami",
+                    cleanup_error,
+                    json_output=json_output,
+                )
+        _fail_for_expected_error("whoami", error, json_output=json_output)
+    except (
+        AuthenticationRequiredError,
+        HorizonResponseError,
+        HorizonUnavailableError,
+        StateFileError,
+    ) as error:
+        _fail_for_expected_error("whoami", error, json_output=json_output)
+
+    emit_identity(
+        "whoami",
+        user,
+        json_output=json_output,
+    )
+
+
+async def logout(
+    *,
+    json_output: JsonOption = False,
+) -> None:
+    """Revoke the current Horizon key and remove the local credential."""
+    credentials = CredentialStore()
+
+    try:
+        configuration = ConfigurationStore().load()
+        credential = await resolve_credential(credentials)
+    except AuthenticationRequiredError:
+        emit_logout(remote_revoked=False, json_output=json_output)
+        return
+    except StateFileError:
+        try:
+            credentials.clear()
+        except StateFileError as error:
+            _fail_for_expected_error("logout", error, json_output=json_output)
+        _fail(
+            "logout",
+            "remote_revocation_failed",
+            "The local credential was removed, but the remote key can remain active.",
+            json_output=json_output,
+            details={
+                "localCredentialRemoved": True,
+                "remoteCredentialMayRemain": True,
+            },
+        )
+
+    try:
+        async with HorizonClient(
+            configuration.api_origin,
+            api_key=credential.api_key,
+        ) as client:
+            await revoke_and_clear_credential(client, credentials)
+    except HorizonUnauthorizedError:
+        emit_logout(remote_revoked=False, json_output=json_output)
+        return
+    except (HorizonResponseError, HorizonUnavailableError):
+        _fail(
+            "logout",
+            "remote_revocation_failed",
+            "The local credential was removed, but the remote key can remain active.",
+            json_output=json_output,
+            details={
+                "localCredentialRemoved": True,
+                "remoteCredentialMayRemain": True,
+            },
+        )
+    except StateFileError as error:
+        _fail_for_expected_error("logout", error, json_output=json_output)
+
+    emit_logout(remote_revoked=True, json_output=json_output)

fastmcp_slim/fastmcp/cli/deploy/configuration.py

@@ -0,0 +1,68 @@
+"""Global non-secret configuration for the FastMCP CLI."""
+
+from __future__ import annotations
+
+from pathlib import Path
+from typing import Literal
+
+from pydantic import BaseModel, ConfigDict, Field, field_validator
+
+from fastmcp.cli.deploy.credentials import CredentialStore
+from fastmcp.cli.deploy.horizon_client import (
+    DEFAULT_HORIZON_API_ORIGIN,
+    normalize_api_origin,
+)
+from fastmcp.cli.deploy.state import read_state, state_lock, write_state
+
+
+class HorizonConfiguration(BaseModel):
+    model_config = ConfigDict(extra="forbid", frozen=True, populate_by_name=True)
+
+    schema_version: Literal[1] = Field(alias="schemaVersion")
+    api_origin: str = Field(alias="apiOrigin")
+
+    @field_validator("api_origin")
+    @classmethod
+    def validate_api_origin(cls, value: str) -> str:
+        return normalize_api_origin(value)
+
+
+class ConfigurationStore:
+    """Persist the Horizon API origin without organization state."""
+
+    def __init__(self, state_directory: Path | None = None) -> None:
+        if state_directory is None:
+            import fastmcp
+
+            state_directory = fastmcp.settings.home / "cli"
+        self.path = state_directory / "config.json"
+
+    def load(self) -> HorizonConfiguration:
+        state = read_state(self.path, HorizonConfiguration)
+        if state is not None:
+            return state
+        return HorizonConfiguration(
+            schemaVersion=1,
+            apiOrigin=DEFAULT_HORIZON_API_ORIGIN,
+        )
+
+    def save(self, configuration: HorizonConfiguration) -> None:
+        write_state(
+            self.path,
+            configuration.model_dump(mode="json", by_alias=True),
+        )
+
+    def set_api_origin(
+        self,
+        api_origin: str,
+        *,
+        credentials: CredentialStore,
+    ) -> HorizonConfiguration:
+        """Set the origin and clear credentials before an origin change."""
+        with state_lock(self.path.parent):
+            current = self.load()
+            updated = HorizonConfiguration(schemaVersion=1, apiOrigin=api_origin)
+            if updated.api_origin != current.api_origin:
+                credentials.clear()
+            self.save(updated)
+            return updated

fastmcp_slim/fastmcp/cli/deploy/credentials.py

@@ -0,0 +1,142 @@
+"""Restricted Horizon credential storage and resolution."""
+
+from __future__ import annotations
+
+import os
+from collections.abc import Awaitable, Callable, Mapping
+from dataclasses import dataclass
+from pathlib import Path
+from typing import Literal
+
+from pydantic import (
+    BaseModel,
+    ConfigDict,
+    Field,
+    SecretStr,
+    ValidationError,
+    field_validator,
+)
+
+from fastmcp.cli.deploy.horizon_client import HorizonClient, normalize_api_origin
+from fastmcp.cli.deploy.state import (
+    StateFileError,
+    read_state,
+    remove_state,
+    state_lock,
+    write_state,
+)
+
+CredentialSource = Literal["environment", "stored", "interactive"]
+
+
+class AuthenticationRequiredError(RuntimeError):
+    """No Horizon credential is available without interactive authorization."""
+
+
+class AuthState(BaseModel):
+    model_config = ConfigDict(extra="forbid", frozen=True, populate_by_name=True)
+
+    schema_version: Literal[1] = Field(alias="schemaVersion")
+    api_key: SecretStr = Field(alias="apiKey")
+
+    @field_validator("api_key")
+    @classmethod
+    def require_nonempty_api_key(cls, value: SecretStr) -> SecretStr:
+        if not value.get_secret_value().strip():
+            raise ValueError("The API key is empty")
+        return value
+
+
+@dataclass(frozen=True)
+class ResolvedCredential:
+    api_key: SecretStr
+    source: CredentialSource
+
+
+class CredentialStore:
+    """Persist the active personal Horizon API key."""
+
+    def __init__(self, state_directory: Path | None = None) -> None:
+        if state_directory is None:
+            import fastmcp
+
+            state_directory = fastmcp.settings.home / "cli"
+        self.path = state_directory / "auth.json"
+
+    def load(self) -> SecretStr | None:
+        state = read_state(self.path, AuthState, secret=True)
+        return state.api_key if state is not None else None
+
+    def save(self, api_key: SecretStr | str) -> None:
+        try:
+            state = AuthState(schemaVersion=1, apiKey=api_key)
+        except ValidationError:
+            raise StateFileError("The Horizon API key is invalid") from None
+        write_state(
+            self.path,
+            {
+                "schemaVersion": state.schema_version,
+                "apiKey": state.api_key.get_secret_value(),
+            },
+        )
+
+    def save_for_origin(
+        self,
+        api_key: SecretStr | str,
+        *,
+        expected_api_origin: str,
+    ) -> None:
+        """Save a key only while its issuing Horizon origin is active."""
+        from fastmcp.cli.deploy.configuration import ConfigurationStore
+
+        expected_api_origin = normalize_api_origin(expected_api_origin)
+        with state_lock(self.path.parent):
+            active_api_origin = ConfigurationStore(self.path.parent).load().api_origin
+            if active_api_origin != expected_api_origin:
+                raise StateFileError("The Horizon host changed during login")
+            self.save(api_key)
+
+    def clear(self) -> None:
+        remove_state(self.path)
+
+
+async def resolve_credential(
+    store: CredentialStore,
+    *,
+    environ: Mapping[str, str] | None = None,
+    authorize: Callable[[], Awaitable[SecretStr]] | None = None,
+    expected_api_origin: str | None = None,
+) -> ResolvedCredential:
+    """Resolve environment, stored, then interactive credentials."""
+    environ = os.environ if environ is None else environ
+    environment_key = environ.get("HORIZON_API_KEY")
+    if environment_key:
+        return ResolvedCredential(
+            api_key=SecretStr(environment_key),
+            source="environment",
+        )
+
+    stored_key = store.load()
+    if stored_key is not None:
+        return ResolvedCredential(api_key=stored_key, source="stored")
+
+    if authorize is None:
+        raise AuthenticationRequiredError("Horizon authentication is required")
+
+    api_key = await authorize()
+    if expected_api_origin is None:
+        store.save(api_key)
+    else:
+        store.save_for_origin(api_key, expected_api_origin=expected_api_origin)
+    return ResolvedCredential(api_key=api_key, source="interactive")
+
+
+async def revoke_and_clear_credential(
+    client: HorizonClient,
+    store: CredentialStore,
+) -> None:
+    """Attempt remote revocation and always remove the stored credential."""
+    try:
+        await client.revoke_current_api_key()
+    finally:
+        store.clear()

fastmcp_slim/fastmcp/cli/deploy/horizon_client.py

@@ -0,0 +1,332 @@
+"""Typed HTTP client for the Horizon control plane."""
+
+from __future__ import annotations
+
+from collections.abc import Mapping
+from dataclasses import dataclass
+from types import TracebackType
+from typing import Annotated, Literal, TypeVar
+from urllib.parse import urlsplit, urlunsplit
+
+import httpx2
+from pydantic import (
+    BaseModel,
+    ConfigDict,
+    Field,
+    SecretStr,
+    ValidationError,
+    field_validator,
+)
+
+DEVICE_AUTH_CLIENT_ID = "fastmcp-cli"
+DEVICE_AUTH_GRANT_TYPE = "urn:ietf:params:oauth:grant-type:device_code"
+DEFAULT_HORIZON_API_ORIGIN = "https://horizon.prefect.io"
+
+DeviceTokenError = Literal[
+    "authorization_pending",
+    "slow_down",
+    "access_denied",
+    "expired_token",
+]
+
+
+class HorizonError(RuntimeError):
+    """A safe Horizon client error."""
+
+
+class HorizonUnavailableError(HorizonError):
+    """The Horizon API could not be reached."""
+
+
+class HorizonUnauthorizedError(HorizonError):
+    """The Horizon credential was rejected."""
+
+
+class HorizonResponseError(HorizonError):
+    """Horizon returned an unexpected response."""
+
+    def __init__(self, message: str, *, status_code: int | None = None) -> None:
+        super().__init__(message)
+        self.status_code = status_code
+
+
+class _ResponseModel(BaseModel):
+    model_config = ConfigDict(extra="ignore", frozen=True)
+
+
+ResponseModelT = TypeVar("ResponseModelT", bound=_ResponseModel)
+
+
+class DeviceAuthorization(_ResponseModel):
+    device_code: Annotated[str, Field(min_length=1)]
+    user_code: Annotated[str, Field(min_length=1)]
+    verification_uri: Annotated[str, Field(pattern=r"^https?://")]
+    verification_uri_complete: Annotated[str, Field(pattern=r"^https?://")]
+    expires_in: Annotated[int, Field(gt=0)]
+    interval: Annotated[int, Field(gt=0)]
+
+
+class DeviceAccessToken(_ResponseModel):
+    access_token: SecretStr
+    token_type: Literal["Bearer"]
+
+    @field_validator("access_token")
+    @classmethod
+    def require_nonempty_access_token(cls, value: SecretStr) -> SecretStr:
+        if not value.get_secret_value().strip():
+            raise ValueError("The access token is empty")
+        return value
+
+
+class _DeviceTokenErrorResponse(_ResponseModel):
+    error: DeviceTokenError
+
+
+class HorizonUser(_ResponseModel):
+    id: str
+    email: str
+    name: str | None
+
+
+class _CurrentUserResponse(_ResponseModel):
+    user: HorizonUser
+
+
+class HorizonOrganization(_ResponseModel):
+    id: str
+    name: str
+    slug: str
+
+
+class _PaginationMeta(_ResponseModel):
+    nextCursor: str | None
+    limit: int
+
+
+class _OrganizationsResponse(_ResponseModel):
+    items: tuple[HorizonOrganization, ...]
+    meta: _PaginationMeta
+
+
+@dataclass(frozen=True)
+class DeviceMetadata:
+    device_name: str | None = None
+    platform: str | None = None
+    architecture: str | None = None
+    client_version: str | None = None
+
+
+@dataclass(frozen=True)
+class DeviceTokenPoll:
+    access_token: SecretStr | None = None
+    error: DeviceTokenError | None = None
+
+    def __post_init__(self) -> None:
+        if (self.access_token is None) == (self.error is None):
+            raise ValueError("A device token poll must contain one result")
+
+
+def normalize_api_origin(value: str) -> str:
+    """Validate and normalize a Horizon API origin."""
+    parts = urlsplit(value)
+    try:
+        _ = parts.port
+    except ValueError:
+        raise ValueError("The Horizon API origin must be an HTTP origin") from None
+    if (
+        parts.scheme not in {"http", "https"}
+        or not parts.hostname
+        or parts.username is not None
+        or parts.password is not None
+        or parts.query
+        or parts.fragment
+        or parts.path not in {"", "/"}
+    ):
+        raise ValueError("The Horizon API origin must be an HTTP origin")
+
+    return urlunsplit((parts.scheme, parts.netloc, "", "", ""))
+
+
+class HorizonClient:
+    """Call the Horizon routes used by FastMCP CLI authentication."""
+
+    def __init__(
+        self,
+        api_origin: str = DEFAULT_HORIZON_API_ORIGIN,
+        *,
+        api_key: SecretStr | str | None = None,
+        transport: httpx2.AsyncBaseTransport | None = None,
+        timeout: float = 30.0,
+    ) -> None:
+        self.api_origin = normalize_api_origin(api_origin)
+        self._api_key = (
+            api_key
+            if isinstance(api_key, SecretStr)
+            else SecretStr(api_key)
+            if api_key is not None
+            else None
+        )
+        self._client = httpx2.AsyncClient(
+            base_url=self.api_origin,
+            follow_redirects=False,
+            timeout=timeout,
+            transport=transport,
+        )
+
+    async def __aenter__(self) -> HorizonClient:
+        await self._client.__aenter__()
+        return self
+
+    async def __aexit__(
+        self,
+        exc_type: type[BaseException] | None,
+        exc_value: BaseException | None,
+        traceback: TracebackType | None,
+    ) -> None:
+        await self._client.__aexit__(exc_type, exc_value, traceback)
+
+    async def aclose(self) -> None:
+        await self._client.aclose()
+
+    async def _request(
+        self,
+        method: str,
+        path: str,
+        *,
+        authenticated: bool = False,
+        data: Mapping[str, str] | None = None,
+        params: Mapping[str, str | int] | None = None,
+    ) -> httpx2.Response:
+        headers: dict[str, str] = {}
+        if authenticated:
+            if self._api_key is None:
+                raise HorizonUnauthorizedError("Horizon authentication is required")
+            headers["Authorization"] = f"Bearer {self._api_key.get_secret_value()}"
+
+        try:
+            response = await self._client.request(
+                method,
+                path,
+                headers=headers,
+                data=data,
+                params=params,
+            )
+        except httpx2.RequestError as exc:
+            raise HorizonUnavailableError("The Horizon API is unavailable") from exc
+
+        if authenticated and response.status_code == 401:
+            raise HorizonUnauthorizedError("The Horizon credential is not valid")
+        return response
+
+    @staticmethod
+    def _validate_response(
+        response: httpx2.Response,
+        model: type[ResponseModelT],
+    ) -> ResponseModelT:
+        try:
+            return model.model_validate_json(response.content)
+        except (ValidationError, ValueError):
+            raise HorizonResponseError(
+                "Horizon returned an invalid response",
+                status_code=response.status_code,
+            ) from None
+
+    @staticmethod
+    def _require_status(response: httpx2.Response, expected: int) -> None:
+        if response.status_code != expected:
+            raise HorizonResponseError(
+                "Horizon returned an unexpected status",
+                status_code=response.status_code,
+            )
+
+    async def create_device_authorization(
+        self,
+        metadata: DeviceMetadata | None = None,
+    ) -> DeviceAuthorization:
+        metadata = metadata or DeviceMetadata()
+        form = {
+            "client_id": DEVICE_AUTH_CLIENT_ID,
+            "device_name": metadata.device_name,
+            "platform": metadata.platform,
+            "architecture": metadata.architecture,
+            "client_version": metadata.client_version,
+        }
+        response = await self._request(
+            "POST",
+            "/api/v0/oauth/device/authorization",
+            data={key: value for key, value in form.items() if value is not None},
+        )
+        self._require_status(response, 200)
+        return self._validate_response(response, DeviceAuthorization)
+
+    async def exchange_device_authorization(
+        self,
+        device_code: str,
+    ) -> DeviceTokenPoll:
+        response = await self._request(
+            "POST",
+            "/api/v0/oauth/device/token",
+            data={
+                "grant_type": DEVICE_AUTH_GRANT_TYPE,
+                "client_id": DEVICE_AUTH_CLIENT_ID,
+                "device_code": device_code,
+            },
+        )
+
+        if response.status_code == 200:
+            result = self._validate_response(response, DeviceAccessToken)
+            return DeviceTokenPoll(access_token=result.access_token)
+
+        if response.status_code == 400:
+            result = self._validate_response(response, _DeviceTokenErrorResponse)
+            return DeviceTokenPoll(error=result.error)
+
+        self._require_status(response, 200)
+        raise AssertionError("unreachable")
+
+    async def get_current_user(self) -> HorizonUser:
+        response = await self._request(
+            "GET",
+            "/api/v0/me",
+            authenticated=True,
+        )
+        self._require_status(response, 200)
+        result = self._validate_response(response, _CurrentUserResponse)
+        return result.user
+
+    async def list_organizations(self) -> tuple[HorizonOrganization, ...]:
+        organizations: list[HorizonOrganization] = []
+        cursor: str | None = None
+        seen_cursors: set[str] = set()
+
+        while True:
+            params = {"limit": 100}
+            if cursor is not None:
+                params["cursor"] = cursor
+            response = await self._request(
+                "GET",
+                "/api/v0/me/organizations",
+                authenticated=True,
+                params=params,
+            )
+            self._require_status(response, 200)
+            result = self._validate_response(response, _OrganizationsResponse)
+            organizations.extend(result.items)
+
+            cursor = result.meta.nextCursor
+            if cursor is None:
+                return tuple(organizations)
+            if cursor in seen_cursors:
+                raise HorizonResponseError(
+                    "Horizon returned an invalid organization cursor",
+                    status_code=response.status_code,
+                )
+            seen_cursors.add(cursor)
+
+    async def revoke_current_api_key(self) -> None:
+        response = await self._request(
+            "DELETE",
+            "/api/v0/me/api-key",
+            authenticated=True,
+        )
+        self._require_status(response, 204)

fastmcp_slim/fastmcp/cli/deploy/output.py

@@ -0,0 +1,263 @@
+"""Stable terminal and JSON output for Horizon CLI commands."""
+
+from __future__ import annotations
+
+import json
+import sys
+from typing import Literal
+
+from rich import box
+from rich.align import Align
+from rich.console import Console, Group
+from rich.padding import Padding
+from rich.panel import Panel
+from rich.status import Status
+from rich.table import Table
+from rich.text import Text
+
+from fastmcp.cli.deploy.horizon_client import DeviceAuthorization, HorizonUser
+
+CommandName = Literal["login", "logout", "whoami"]
+ErrorCategory = Literal[
+    "authentication_invalid",
+    "authentication_required",
+    "authorization_denied",
+    "authorization_expired",
+    "authorization_failed",
+    "horizon_error",
+    "horizon_unavailable",
+    "invalid_host",
+    "remote_revocation_failed",
+    "state_error",
+]
+
+console = Console()
+error_console = Console(stderr=True)
+
+
+def _write_json(payload: object, *, stderr: bool = False) -> None:
+    stream = sys.stderr if stderr else sys.stdout
+    print(json.dumps(payload, separators=(",", ":")), file=stream, flush=True)
+
+
+def _banner(title: str, *, style: str) -> Panel:
+    return Panel(
+        Align.center(Text(title, style=f"bold {style}")),
+        box=box.ROUNDED,
+        border_style=style,
+        padding=(0, 1),
+        width=52,
+    )
+
+
+def _account_panel(
+    user: HorizonUser,
+    *,
+    title: str,
+    message: str,
+) -> Panel:
+    name = Text(user.name or user.email, style="bold")
+    details: list[Text] = [name]
+    if user.name:
+        details.append(Text(user.email, style="cyan"))
+    details.extend([Text(), Text(message, style="green")])
+    return Panel(
+        Group(*details),
+        title=Text(title, style="bold green"),
+        title_align="left",
+        box=box.ROUNDED,
+        border_style="green",
+        padding=(1, 2),
+        width=52,
+    )
+
+
+def _format_duration(seconds: int) -> str:
+    if seconds % 60 == 0:
+        minutes = seconds // 60
+        unit = "minute" if minutes == 1 else "minutes"
+        return f"{minutes} {unit}"
+    unit = "second" if seconds == 1 else "seconds"
+    return f"{seconds} {unit}"
+
+
+def emit_device_challenge(
+    authorization: DeviceAuthorization,
+    *,
+    json_output: bool,
+) -> None:
+    """Show a device challenge before polling starts."""
+    if json_output:
+        _write_json(
+            {
+                "event": "device_authorization",
+                "verificationUrl": authorization.verification_uri,
+                "verificationUrlComplete": authorization.verification_uri_complete,
+                "userCode": authorization.user_code,
+            },
+            stderr=True,
+        )
+        return
+
+    console.print()
+    console.print(_banner("Deploy FastMCP on Horizon", style="magenta"))
+    console.print()
+    console.print(Text("✓ Device authorization started", style="bold green"))
+    console.print()
+    console.print("  Open this URL in your browser:")
+    console.print()
+    console.print(
+        Padding(
+            Text(authorization.verification_uri_complete, style="cyan underline"),
+            (0, 2),
+        )
+    )
+    console.print()
+    console.print("  Confirm this code:")
+    console.print()
+    code = Table.grid()
+    code.add_column(justify="center", width=52)
+    code.add_row(Text(authorization.user_code, style="bold"))
+    console.print(code)
+    console.print()
+    expires_in = _format_duration(authorization.expires_in)
+    console.print(Text(f"The request expires in {expires_in}.", style="dim"))
+    console.print(Text("Press Ctrl-C to cancel.", style="dim"))
+    console.print()
+
+
+def start_device_approval_status(*, json_output: bool) -> Status | None:
+    """Start the terminal spinner while the browser approval is pending."""
+    if json_output:
+        return None
+    status = console.status(
+        "[cyan]Waiting for approval in your browser[/cyan]",
+        spinner="dots",
+        spinner_style="cyan",
+    )
+    status.start()
+    return status
+
+
+def stop_device_approval_status(status: Status | None) -> None:
+    """Stop a device approval spinner when one is active."""
+    if status is not None:
+        status.stop()
+
+
+def emit_identity(
+    command: Literal["login", "whoami"],
+    user: HorizonUser,
+    *,
+    json_output: bool,
+) -> None:
+    """Show the authenticated user."""
+    if json_output:
+        _write_json(
+            {
+                "ok": True,
+                "command": command,
+                "user": user.model_dump(mode="json"),
+            }
+        )
+        return
+
+    console.print()
+    if command == "login":
+        panel = _account_panel(
+            user,
+            title="Logged into Horizon",
+            message="You are signed in to FastMCP.",
+        )
+    else:
+        panel = _account_panel(
+            user,
+            title="Horizon Account",
+            message="● Signed in",
+        )
+    console.print(panel)
+    console.print()
+
+
+def emit_logout(
+    *,
+    remote_revoked: bool,
+    json_output: bool,
+) -> None:
+    """Show a successful local logout result."""
+    if json_output:
+        _write_json(
+            {
+                "ok": True,
+                "command": "logout",
+                "localCredentialRemoved": True,
+                "remoteRevoked": remote_revoked,
+            }
+        )
+        return
+
+    if remote_revoked:
+        title = "Logged out of Horizon"
+        message = "The Horizon credential was revoked and removed from this device."
+        style = "green"
+    else:
+        title = "Horizon Account"
+        message = "No active Horizon credential remains on this device."
+        style = "cyan"
+
+    console.print()
+    console.print(
+        Panel(
+            Text(message),
+            title=Text(title, style=f"bold {style}"),
+            title_align="left",
+            box=box.ROUNDED,
+            border_style=style,
+            padding=(1, 2),
+            width=60,
+        )
+    )
+    console.print()
+
+
+def emit_error(
+    command: CommandName,
+    category: ErrorCategory,
+    message: str,
+    *,
+    json_output: bool,
+    details: dict[str, object] | None = None,
+) -> None:
+    """Show a stable expected command failure."""
+    if json_output:
+        payload: dict[str, object] = {
+            "ok": False,
+            "command": command,
+            "error": {
+                "category": category,
+                "message": message,
+            },
+        }
+        if details:
+            payload.update(details)
+        _write_json(payload)
+        return
+
+    titles = {
+        "login": "✗ Sign in failed",
+        "logout": "✗ Sign out failed",
+        "whoami": "✗ Account lookup failed",
+    }
+    error_console.print()
+    error_console.print(
+        Panel(
+            Text(message),
+            title=Text(titles[command], style="bold red"),
+            title_align="left",
+            box=box.ROUNDED,
+            border_style="red",
+            padding=(1, 2),
+            width=60,
+        )
+    )
+    error_console.print()

fastmcp_slim/fastmcp/cli/deploy/state.py

@@ -0,0 +1,226 @@
+"""Versioned JSON state helpers for the FastMCP CLI."""
+
+from __future__ import annotations
+
+import errno
+import json
+import os
+import subprocess
+import tempfile
+from collections.abc import Iterator
+from contextlib import contextmanager, suppress
+from pathlib import Path
+from typing import Any, TypeVar
+
+from pydantic import BaseModel, ValidationError
+
+ModelT = TypeVar("ModelT", bound=BaseModel)
+
+
+class StateFileError(RuntimeError):
+    """A CLI state file could not be read or written safely."""
+
+
+_WINDOWS_ACL_SCRIPT = r"""
+$ErrorActionPreference = "Stop"
+$path = $env:FASTMCP_STATE_PATH
+$sid = [System.Security.Principal.WindowsIdentity]::GetCurrent().User
+$acl = Get-Acl -LiteralPath $path
+$acl.SetAccessRuleProtection($true, $false)
+foreach ($existingRule in @($acl.Access)) {
+    $acl.RemoveAccessRuleSpecific($existingRule)
+}
+
+if ([System.IO.Directory]::Exists($path)) {
+    $inheritance = [System.Security.AccessControl.InheritanceFlags]::ContainerInherit `
+        -bor [System.Security.AccessControl.InheritanceFlags]::ObjectInherit
+    $rule = [System.Security.AccessControl.FileSystemAccessRule]::new(
+        $sid,
+        [System.Security.AccessControl.FileSystemRights]::FullControl,
+        $inheritance,
+        [System.Security.AccessControl.PropagationFlags]::None,
+        [System.Security.AccessControl.AccessControlType]::Allow
+    )
+} else {
+    $rule = [System.Security.AccessControl.FileSystemAccessRule]::new(
+        $sid,
+        [System.Security.AccessControl.FileSystemRights]::FullControl,
+        [System.Security.AccessControl.AccessControlType]::Allow
+    )
+}
+
+$acl.AddAccessRule($rule)
+Set-Acl -LiteralPath $path -AclObject $acl
+"""
+
+
+def _restrict_windows_access(path: Path) -> None:
+    try:
+        subprocess.run(
+            [
+                "powershell.exe",
+                "-NoLogo",
+                "-NoProfile",
+                "-NonInteractive",
+                "-Command",
+                _WINDOWS_ACL_SCRIPT,
+            ],
+            check=True,
+            capture_output=True,
+            text=True,
+            env={**os.environ, "FASTMCP_STATE_PATH": str(path)},
+        )
+    except (OSError, subprocess.SubprocessError) as exc:
+        raise StateFileError("Could not restrict access to CLI state") from exc
+
+
+def _restrict_access(path: Path, *, directory: bool = False) -> None:
+    try:
+        if os.name == "nt":
+            _restrict_windows_access(path)
+        else:
+            path.chmod(0o700 if directory else 0o600)
+    except OSError as exc:
+        raise StateFileError("Could not restrict access to CLI state") from exc
+
+
+def _prepare_directory(path: Path) -> None:
+    try:
+        path.mkdir(parents=True, exist_ok=True)
+    except OSError as exc:
+        raise StateFileError("Could not create the CLI state directory") from exc
+    _restrict_access(path, directory=True)
+
+
+@contextmanager
+def state_lock(directory: Path) -> Iterator[None]:
+    """Lock related CLI state changes across processes."""
+    _prepare_directory(directory)
+    lock_path = directory / ".state.lock"
+    if lock_path.is_symlink():
+        raise StateFileError("The CLI state lock must not be a symbolic link")
+
+    lock_file = None
+    try:
+        lock_file = lock_path.open("a+b")
+        _restrict_access(lock_path)
+        if os.name == "nt":
+            import msvcrt
+
+            if lock_path.stat().st_size == 0:
+                lock_file.write(b"\0")
+                lock_file.flush()
+            lock_file.seek(0)
+            msvcrt.locking(lock_file.fileno(), msvcrt.LK_LOCK, 1)
+        else:
+            import fcntl
+
+            fcntl.flock(lock_file.fileno(), fcntl.LOCK_EX)
+    except (OSError, StateFileError) as exc:
+        if lock_file is not None:
+            with suppress(OSError):
+                lock_file.close()
+        if isinstance(exc, StateFileError):
+            raise
+        raise StateFileError("Could not lock CLI state") from exc
+
+    try:
+        yield
+    finally:
+        if os.name == "nt":
+            import msvcrt
+
+            with suppress(OSError):
+                lock_file.seek(0)
+                msvcrt.locking(lock_file.fileno(), msvcrt.LK_UNLCK, 1)
+        else:
+            import fcntl
+
+            with suppress(OSError):
+                fcntl.flock(lock_file.fileno(), fcntl.LOCK_UN)
+        with suppress(OSError):
+            lock_file.close()
+
+
+def read_state(
+    path: Path,
+    model: type[ModelT],
+    *,
+    secret: bool = False,
+) -> ModelT | None:
+    """Read and validate a versioned JSON state file."""
+    if not path.exists():
+        return None
+    if path.is_symlink():
+        raise StateFileError(f"CLI state must not be a symbolic link: {path.name}")
+
+    if secret:
+        _restrict_access(path.parent, directory=True)
+        _restrict_access(path)
+
+    try:
+        return model.model_validate_json(path.read_text(encoding="utf-8"))
+    except (ValidationError, ValueError):
+        raise StateFileError(f"CLI state is invalid: {path.name}") from None
+    except OSError as exc:
+        raise StateFileError(f"Could not read CLI state: {path.name}") from exc
+
+
+def write_state(path: Path, data: dict[str, Any]) -> None:
+    """Write JSON through a restricted temporary file and atomic replacement."""
+    _prepare_directory(path.parent)
+    payload = (json.dumps(data, indent=2, sort_keys=True) + "\n").encode()
+    descriptor: int | None = None
+    temporary_path: Path | None = None
+
+    try:
+        descriptor, temporary_name = tempfile.mkstemp(
+            dir=path.parent,
+            prefix=f".{path.name}.",
+            suffix=".tmp",
+        )
+        temporary_path = Path(temporary_name)
+        if os.name != "nt":
+            os.fchmod(descriptor, 0o600)
+
+        temporary_file = os.fdopen(descriptor, "wb")
+        descriptor = None
+        with temporary_file:
+            temporary_file.write(payload)
+            temporary_file.flush()
+            os.fsync(temporary_file.fileno())
+
+        _restrict_access(temporary_path)
+        os.replace(temporary_path, path)
+        temporary_path = None
+
+        if os.name != "nt":
+            directory_descriptor = os.open(path.parent, os.O_RDONLY)
+            try:
+                try:
+                    os.fsync(directory_descriptor)
+                except OSError as exc:
+                    unsupported = {errno.EINVAL, errno.ENOTSUP}
+                    if exc.errno not in unsupported:
+                        raise
+            finally:
+                os.close(directory_descriptor)
+    except StateFileError:
+        raise
+    except OSError as exc:
+        raise StateFileError(f"Could not write CLI state: {path.name}") from exc
+    finally:
+        if descriptor is not None:
+            with suppress(OSError):
+                os.close(descriptor)
+        if temporary_path is not None:
+            with suppress(OSError):
+                temporary_path.unlink(missing_ok=True)
+
+
+def remove_state(path: Path) -> None:
+    """Remove a state file when it exists."""
+    try:
+        path.unlink(missing_ok=True)
+    except OSError as exc:
+        raise StateFileError(f"Could not remove CLI state: {path.name}") from exc

tests/cli/deploy/test_authentication.py

@@ -0,0 +1,176 @@
+from __future__ import annotations
+
+import httpx2
+import pytest
+
+from fastmcp.cli.deploy.authentication import (
+    DeviceAuthorizationDeniedError,
+    DeviceAuthorizationExpiredError,
+    authorize_device,
+    poll_device_authorization,
+)
+from fastmcp.cli.deploy.horizon_client import DeviceAuthorization, HorizonClient
+
+
+class Clock:
+    def __init__(self) -> None:
+        self.now = 0.0
+        self.sleeps: list[float] = []
+
+    def monotonic(self) -> float:
+        return self.now
+
+    async def sleep(self, delay: float) -> None:
+        self.sleeps.append(delay)
+        self.now += delay
+
+
+def authorization(*, expires_in: int = 600, interval: int = 5) -> DeviceAuthorization:
+    return DeviceAuthorization(
+        device_code="device-secret",
+        user_code="BCDF-GHJK",
+        verification_uri="https://horizon.prefect.io/oauth/device",
+        verification_uri_complete=(
+            "https://horizon.prefect.io/oauth/device?user_code=BCDF-GHJK"
+        ),
+        expires_in=expires_in,
+        interval=interval,
+    )
+
+
+def sequenced_transport(
+    responses: list[httpx2.Response],
+) -> httpx2.MockTransport:
+    def handler(request: httpx2.Request) -> httpx2.Response:
+        return responses.pop(0)
+
+    return httpx2.MockTransport(handler)
+
+
+async def test_polling_handles_pending_slow_down_and_approval() -> None:
+    clock = Clock()
+    async with HorizonClient(
+        transport=sequenced_transport(
+            [
+                httpx2.Response(400, json={"error": "authorization_pending"}),
+                httpx2.Response(400, json={"error": "slow_down"}),
+                httpx2.Response(
+                    200,
+                    json={"access_token": "fmcp_secret", "token_type": "Bearer"},
+                ),
+            ]
+        )
+    ) as client:
+        api_key = await poll_device_authorization(
+            client,
+            authorization(),
+            sleep=clock.sleep,
+            monotonic=clock.monotonic,
+        )
+
+    assert api_key.get_secret_value() == "fmcp_secret"
+    assert clock.sleeps == [5, 5, 10]
+
+
+@pytest.mark.parametrize(
+    ("error", "exception"),
+    [
+        ("access_denied", DeviceAuthorizationDeniedError),
+        ("expired_token", DeviceAuthorizationExpiredError),
+    ],
+)
+async def test_polling_handles_terminal_errors(
+    error: str,
+    exception: type[Exception],
+) -> None:
+    clock = Clock()
+    async with HorizonClient(
+        transport=sequenced_transport([httpx2.Response(400, json={"error": error})])
+    ) as client:
+        with pytest.raises(exception):
+            await poll_device_authorization(
+                client,
+                authorization(),
+                sleep=clock.sleep,
+                monotonic=clock.monotonic,
+            )
+
+
+async def test_polling_stops_at_the_local_expiry_deadline() -> None:
+    clock = Clock()
+    requests: list[httpx2.Request] = []
+
+    def handler(request: httpx2.Request) -> httpx2.Response:
+        requests.append(request)
+        return httpx2.Response(400, json={"error": "authorization_pending"})
+
+    async with HorizonClient(transport=httpx2.MockTransport(handler)) as client:
+        with pytest.raises(DeviceAuthorizationExpiredError):
+            await poll_device_authorization(
+                client,
+                authorization(expires_in=5, interval=5),
+                sleep=clock.sleep,
+                monotonic=clock.monotonic,
+            )
+
+    assert requests == []
+
+
+async def test_authorize_device_presents_challenge_before_opening_browser() -> None:
+    events: list[str] = []
+    clock = Clock()
+
+    def handler(request: httpx2.Request) -> httpx2.Response:
+        if request.url.path.endswith("/authorization"):
+            return httpx2.Response(200, json=authorization().model_dump())
+        return httpx2.Response(
+            200,
+            json={"access_token": "fmcp_secret", "token_type": "Bearer"},
+        )
+
+    def present(challenge: DeviceAuthorization) -> None:
+        events.append(f"present:{challenge.user_code}")
+
+    def open_browser(url: str) -> None:
+        events.append(f"browser:{url}")
+
+    async with HorizonClient(transport=httpx2.MockTransport(handler)) as client:
+        await authorize_device(
+            client,
+            on_challenge=present,
+            open_browser=True,
+            browser_opener=open_browser,
+            sleep=clock.sleep,
+            monotonic=clock.monotonic,
+        )
+
+    assert events == [
+        "present:BCDF-GHJK",
+        "browser:https://horizon.prefect.io/oauth/device?user_code=BCDF-GHJK",
+    ]
+
+
+async def test_browser_failure_does_not_stop_remote_login() -> None:
+    clock = Clock()
+
+    def handler(request: httpx2.Request) -> httpx2.Response:
+        if request.url.path.endswith("/authorization"):
+            return httpx2.Response(200, json=authorization().model_dump())
+        return httpx2.Response(
+            200,
+            json={"access_token": "fmcp_secret", "token_type": "Bearer"},
+        )
+
+    def fail_to_open(url: str) -> None:
+        raise OSError("no browser")
+
+    async with HorizonClient(transport=httpx2.MockTransport(handler)) as client:
+        api_key = await authorize_device(
+            client,
+            open_browser=True,
+            browser_opener=fail_to_open,
+            sleep=clock.sleep,
+            monotonic=clock.monotonic,
+        )
+
+    assert api_key.get_secret_value() == "fmcp_secret"

tests/cli/deploy/test_command.py

@@ -0,0 +1,358 @@
+import json
+from collections.abc import Callable
+from unittest.mock import Mock
+from urllib.parse import parse_qs
+
+import httpx2
+import pytest
+from pydantic import SecretStr
+
+import fastmcp
+import fastmcp.cli.deploy.authentication as authentication_module
+import fastmcp.cli.deploy.command as command_module
+from fastmcp.cli.deploy.command import login, logout, whoami
+from fastmcp.cli.deploy.credentials import CredentialStore
+from fastmcp.cli.deploy.horizon_client import HorizonClient
+from fastmcp.cli.deploy.state import StateFileError
+
+
+class HorizonAuthAPI:
+    def __init__(
+        self,
+        *,
+        token_error: str | None = None,
+        revoke_status: int = 204,
+        invalid_api_key: str | None = None,
+    ) -> None:
+        self.token_error = token_error
+        self.revoke_status = revoke_status
+        self.invalid_api_key = invalid_api_key
+        self.requests: list[httpx2.Request] = []
+
+    def __call__(self, request: httpx2.Request) -> httpx2.Response:
+        self.requests.append(request)
+        path = request.url.path
+        if path == "/api/v0/oauth/device/authorization":
+            return httpx2.Response(
+                200,
+                json={
+                    "device_code": "device-secret",
+                    "user_code": "ABCD-EFGH",
+                    "verification_uri": "https://horizon.prefect.io/oauth/device",
+                    "verification_uri_complete": (
+                        "https://horizon.prefect.io/oauth/device?user_code=ABCD-EFGH"
+                    ),
+                    "expires_in": 600,
+                    "interval": 1,
+                },
+            )
+        if path == "/api/v0/oauth/device/token":
+            if self.token_error is not None:
+                return httpx2.Response(400, json={"error": self.token_error})
+            return httpx2.Response(
+                200,
+                json={"access_token": "fmcp_device_key", "token_type": "Bearer"},
+            )
+        if path == "/api/v0/me":
+            if request.headers.get("Authorization") == (
+                f"Bearer {self.invalid_api_key}"
+            ):
+                return httpx2.Response(401)
+            return httpx2.Response(
+                200,
+                json={
+                    "user": {
+                        "id": "user-1",
+                        "email": "ada@example.com",
+                        "name": "Ada",
+                    }
+                },
+            )
+        if path == "/api/v0/me/api-key":
+            return httpx2.Response(self.revoke_status)
+        raise AssertionError(f"Unexpected request: {request.method} {path}")
+
+
+@pytest.fixture
+def use_horizon_api(
+    monkeypatch: pytest.MonkeyPatch,
+) -> Callable[[HorizonAuthAPI], None]:
+    def use(api: HorizonAuthAPI) -> None:
+        transport = httpx2.MockTransport(api)
+
+        def client(
+            api_origin: str,
+            *,
+            api_key: SecretStr | str | None = None,
+        ) -> HorizonClient:
+            return HorizonClient(
+                api_origin,
+                api_key=api_key,
+                transport=transport,
+            )
+
+        monkeypatch.setattr(command_module, "HorizonClient", client)
+
+    return use
+
+
+@pytest.fixture(autouse=True)
+def no_device_poll_delay(monkeypatch: pytest.MonkeyPatch) -> None:
+    async def sleep(_: float) -> None:
+        return None
+
+    monkeypatch.setattr(authentication_module.asyncio, "sleep", sleep)
+
+
+async def test_json_login_writes_one_result_and_challenge_to_stderr(
+    use_horizon_api: Callable[[HorizonAuthAPI], None],
+    capsys: pytest.CaptureFixture[str],
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    api = HorizonAuthAPI()
+    use_horizon_api(api)
+    browser_open = Mock()
+    monkeypatch.setattr(command_module.webbrowser, "open", browser_open)
+    monkeypatch.setattr(command_module.platform, "node", lambda: "Avery's laptop")
+    monkeypatch.setattr(command_module.platform, "system", lambda: "Darwin")
+    monkeypatch.setattr(command_module.platform, "machine", lambda: "arm64")
+    monkeypatch.setattr(command_module.fastmcp, "__version__", "4.0.0")
+
+    await login(json_output=True)
+
+    captured = capsys.readouterr()
+    stdout_lines = captured.out.strip().splitlines()
+    assert len(stdout_lines) == 1
+    assert json.loads(stdout_lines[0]) == {
+        "ok": True,
+        "command": "login",
+        "user": {
+            "id": "user-1",
+            "email": "ada@example.com",
+            "name": "Ada",
+        },
+    }
+    assert json.loads(captured.err) == {
+        "event": "device_authorization",
+        "verificationUrl": "https://horizon.prefect.io/oauth/device",
+        "verificationUrlComplete": (
+            "https://horizon.prefect.io/oauth/device?user_code=ABCD-EFGH"
+        ),
+        "userCode": "ABCD-EFGH",
+    }
+    browser_open.assert_not_called()
+    authorization_request = next(
+        request
+        for request in api.requests
+        if request.url.path == "/api/v0/oauth/device/authorization"
+    )
+    assert parse_qs(authorization_request.content.decode()) == {
+        "client_id": ["fastmcp-cli"],
+        "device_name": ["Avery's laptop"],
+        "platform": ["darwin"],
+        "architecture": ["arm64"],
+        "client_version": ["4.0.0"],
+    }
+
+    state = json.loads(CredentialStore().path.read_text())
+    assert state == {"schemaVersion": 1, "apiKey": "fmcp_device_key"}
+    assert not (fastmcp.settings.home / "cli" / "config.json").exists()
+
+
+async def test_login_host_is_saved_before_device_authorization(
+    use_horizon_api: Callable[[HorizonAuthAPI], None],
+    capsys: pytest.CaptureFixture[str],
+) -> None:
+    api = HorizonAuthAPI()
+    use_horizon_api(api)
+
+    await login(host="https://dev.horizon.prefect.io/", json_output=True)
+
+    assert json.loads(capsys.readouterr().out)["ok"] is True
+    configuration_path = fastmcp.settings.home / "cli" / "config.json"
+    assert json.loads(configuration_path.read_text()) == {
+        "schemaVersion": 1,
+        "apiOrigin": "https://dev.horizon.prefect.io",
+    }
+    assert {request.url.host for request in api.requests} == {"dev.horizon.prefect.io"}
+
+
+async def test_login_rejects_an_invalid_host(
+    capsys: pytest.CaptureFixture[str],
+) -> None:
+    with pytest.raises(SystemExit, match="1"):
+        await login(host="https://horizon.prefect.io/path", json_output=True)
+
+    result = json.loads(capsys.readouterr().out)
+    assert result["error"]["category"] == "invalid_host"
+    assert CredentialStore().path.exists() is False
+
+
+async def test_tty_login_survives_browser_open_failure(
+    use_horizon_api: Callable[[HorizonAuthAPI], None],
+    capsys: pytest.CaptureFixture[str],
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    use_horizon_api(HorizonAuthAPI())
+    browser_open = Mock(side_effect=OSError("No browser"))
+    monkeypatch.setattr(command_module, "_can_open_browser", lambda: True)
+    monkeypatch.setattr(command_module.webbrowser, "open", browser_open)
+
+    await login()
+
+    output = capsys.readouterr().out
+    assert "https://horizon.prefect.io/oauth/device" in output
+    assert "ABCD-EFGH" in output
+    assert "Logged into Horizon" in output
+    assert "Ada" in output
+    assert "ada@example.com" in output
+    assert "Organization" not in output
+    browser_open.assert_called_once()
+
+
+async def test_whoami_uses_the_stored_key_after_a_restart(
+    use_horizon_api: Callable[[HorizonAuthAPI], None],
+    capsys: pytest.CaptureFixture[str],
+) -> None:
+    api = HorizonAuthAPI()
+    use_horizon_api(api)
+    await login(json_output=True)
+    capsys.readouterr()
+
+    await whoami(json_output=True)
+
+    result = json.loads(capsys.readouterr().out)
+    assert result["command"] == "whoami"
+    assert result["user"]["email"] == "ada@example.com"
+    assert [request.url.path for request in api.requests].count("/api/v0/me") == 2
+
+
+async def test_login_replaces_an_invalid_stored_key(
+    use_horizon_api: Callable[[HorizonAuthAPI], None],
+    capsys: pytest.CaptureFixture[str],
+) -> None:
+    use_horizon_api(HorizonAuthAPI(invalid_api_key="fmcp_stale_key"))
+    CredentialStore().save("fmcp_stale_key")
+
+    await login(json_output=True)
+
+    captured = capsys.readouterr()
+    assert json.loads(captured.out)["ok"] is True
+    assert json.loads(captured.err)["event"] == "device_authorization"
+    stored_key = CredentialStore().load()
+    assert stored_key is not None
+    assert stored_key.get_secret_value() == "fmcp_device_key"
+
+
+async def test_login_never_persists_an_environment_key(
+    use_horizon_api: Callable[[HorizonAuthAPI], None],
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    api = HorizonAuthAPI()
+    use_horizon_api(api)
+    monkeypatch.setenv("HORIZON_API_KEY", "fmcp_environment_key")
+
+    await login(json_output=True)
+
+    assert CredentialStore().path.exists() is False
+    assert not any(
+        request.url.path.startswith("/api/v0/oauth/device") for request in api.requests
+    )
+
+
+async def test_json_whoami_reports_a_failed_rejected_key_cleanup(
+    use_horizon_api: Callable[[HorizonAuthAPI], None],
+    capsys: pytest.CaptureFixture[str],
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    use_horizon_api(HorizonAuthAPI(invalid_api_key="fmcp_stale_key"))
+    CredentialStore().save("fmcp_stale_key")
+
+    def fail_clear(store: CredentialStore) -> None:
+        raise StateFileError("cleanup failed")
+
+    monkeypatch.setattr(CredentialStore, "clear", fail_clear)
+
+    with pytest.raises(SystemExit, match="1"):
+        await whoami(json_output=True)
+
+    result = json.loads(capsys.readouterr().out)
+    assert result["error"]["category"] == "state_error"
+
+
+async def test_json_whoami_does_not_start_device_authorization(
+    capsys: pytest.CaptureFixture[str],
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    browser_open = Mock()
+    monkeypatch.setattr(command_module.webbrowser, "open", browser_open)
+
+    with pytest.raises(SystemExit, match="1"):
+        await whoami(json_output=True)
+
+    result = json.loads(capsys.readouterr().out)
+    assert result["error"]["category"] == "authentication_required"
+    browser_open.assert_not_called()
+
+
+@pytest.mark.parametrize(
+    ("token_error", "category"),
+    [
+        ("access_denied", "authorization_denied"),
+        ("expired_token", "authorization_expired"),
+    ],
+)
+async def test_json_login_reports_stable_device_failures(
+    token_error: str,
+    category: str,
+    use_horizon_api: Callable[[HorizonAuthAPI], None],
+    capsys: pytest.CaptureFixture[str],
+) -> None:
+    use_horizon_api(HorizonAuthAPI(token_error=token_error))
+
+    with pytest.raises(SystemExit, match="1"):
+        await login(json_output=True)
+
+    result = json.loads(capsys.readouterr().out)
+    assert result["error"]["category"] == category
+    assert CredentialStore().path.exists() is False
+
+
+async def test_logout_revokes_the_remote_key_and_clears_local_state(
+    use_horizon_api: Callable[[HorizonAuthAPI], None],
+    capsys: pytest.CaptureFixture[str],
+) -> None:
+    api = HorizonAuthAPI()
+    use_horizon_api(api)
+    CredentialStore().save("fmcp_stored_key")
+
+    await logout(json_output=True)
+
+    assert json.loads(capsys.readouterr().out) == {
+        "ok": True,
+        "command": "logout",
+        "localCredentialRemoved": True,
+        "remoteRevoked": True,
+    }
+    assert CredentialStore().path.exists() is False
+    assert any(
+        request.method == "DELETE" and request.url.path == "/api/v0/me/api-key"
+        for request in api.requests
+    )
+
+
+async def test_logout_clears_local_state_when_remote_revocation_fails(
+    use_horizon_api: Callable[[HorizonAuthAPI], None],
+    capsys: pytest.CaptureFixture[str],
+) -> None:
+    use_horizon_api(HorizonAuthAPI(revoke_status=503))
+    CredentialStore().save("fmcp_stored_key")
+
+    with pytest.raises(SystemExit, match="1"):
+        await logout(json_output=True)
+
+    result = json.loads(capsys.readouterr().out)
+    assert result["error"]["category"] == "remote_revocation_failed"
+    assert result["localCredentialRemoved"] is True
+    assert result["remoteCredentialMayRemain"] is True
+    assert CredentialStore().path.exists() is False

tests/cli/deploy/test_configuration.py

@@ -0,0 +1,131 @@
+from __future__ import annotations
+
+import json
+from pathlib import Path
+
+import pytest
+
+from fastmcp.cli.deploy.configuration import (
+    ConfigurationStore,
+    HorizonConfiguration,
+)
+from fastmcp.cli.deploy.credentials import CredentialStore
+from fastmcp.cli.deploy.horizon_client import DEFAULT_HORIZON_API_ORIGIN
+from fastmcp.cli.deploy.state import StateFileError
+
+
+def test_configuration_defaults_to_the_production_origin(tmp_path: Path) -> None:
+    store = ConfigurationStore(tmp_path)
+
+    configuration = store.load()
+
+    assert configuration.api_origin == DEFAULT_HORIZON_API_ORIGIN
+    assert store.path.exists() is False
+
+
+def test_configuration_stores_only_schema_and_api_origin(tmp_path: Path) -> None:
+    store = ConfigurationStore(tmp_path)
+    configuration = HorizonConfiguration(
+        schemaVersion=1,
+        apiOrigin="https://example.com/",
+    )
+
+    store.save(configuration)
+
+    assert json.loads(store.path.read_text()) == {
+        "schemaVersion": 1,
+        "apiOrigin": "https://example.com",
+    }
+    assert store.load() == configuration
+
+
+def test_configuration_rejects_organization_state(tmp_path: Path) -> None:
+    store = ConfigurationStore(tmp_path)
+    store.path.write_text(
+        json.dumps(
+            {
+                "schemaVersion": 1,
+                "apiOrigin": DEFAULT_HORIZON_API_ORIGIN,
+                "currentOrganizationId": "org-id",
+            }
+        )
+    )
+
+    with pytest.raises(StateFileError):
+        store.load()
+
+
+def test_origin_change_clears_credentials_before_writing_configuration(
+    tmp_path: Path,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    credentials = CredentialStore(tmp_path)
+    credentials.save("fmcp_secret")
+    configuration = ConfigurationStore(tmp_path)
+    configuration.save(
+        HorizonConfiguration(
+            schemaVersion=1,
+            apiOrigin=DEFAULT_HORIZON_API_ORIGIN,
+        )
+    )
+    events: list[str] = []
+    original_clear = credentials.clear
+    original_save = configuration.save
+
+    def clear() -> None:
+        events.append("clear")
+        original_clear()
+
+    def save(value: HorizonConfiguration) -> None:
+        events.append("save")
+        original_save(value)
+
+    monkeypatch.setattr(credentials, "clear", clear)
+    monkeypatch.setattr(configuration, "save", save)
+
+    result = configuration.set_api_origin(
+        "https://dev.horizon.prefect.io",
+        credentials=credentials,
+    )
+
+    assert events == ["clear", "save"]
+    assert credentials.load() is None
+    assert result.api_origin == "https://dev.horizon.prefect.io"
+
+
+def test_same_origin_does_not_clear_credentials(tmp_path: Path) -> None:
+    credentials = CredentialStore(tmp_path)
+    credentials.save("fmcp_secret")
+    configuration = ConfigurationStore(tmp_path)
+
+    configuration.set_api_origin(
+        DEFAULT_HORIZON_API_ORIGIN,
+        credentials=credentials,
+    )
+
+    stored = credentials.load()
+    assert stored is not None
+    assert stored.get_secret_value() == "fmcp_secret"
+
+
+def test_failed_origin_write_leaves_no_cross_origin_credential(
+    tmp_path: Path,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    credentials = CredentialStore(tmp_path)
+    credentials.save("fmcp_secret")
+    configuration = ConfigurationStore(tmp_path)
+
+    def fail_save(value: HorizonConfiguration) -> None:
+        raise StateFileError("write failed")
+
+    monkeypatch.setattr(configuration, "save", fail_save)
+
+    with pytest.raises(StateFileError):
+        configuration.set_api_origin(
+            "https://dev.horizon.prefect.io",
+            credentials=credentials,
+        )
+
+    assert credentials.load() is None
+    assert configuration.load().api_origin == DEFAULT_HORIZON_API_ORIGIN

tests/cli/deploy/test_credentials.py

@@ -0,0 +1,309 @@
+from __future__ import annotations
+
+import errno
+import json
+import os
+import stat
+import subprocess
+import traceback
+from pathlib import Path
+from typing import cast
+
+import httpx2
+import pytest
+from pydantic import SecretStr
+
+from fastmcp.cli.deploy.configuration import (
+    ConfigurationStore,
+    HorizonConfiguration,
+)
+from fastmcp.cli.deploy.credentials import (
+    AuthenticationRequiredError,
+    CredentialStore,
+    resolve_credential,
+    revoke_and_clear_credential,
+)
+from fastmcp.cli.deploy.horizon_client import HorizonClient, HorizonUnavailableError
+from fastmcp.cli.deploy.state import (
+    StateFileError,
+    _restrict_windows_access,
+)
+
+
+def load_secret(store: CredentialStore) -> SecretStr:
+    secret = store.load()
+    assert secret is not None
+    return secret
+
+
+def test_credential_store_writes_only_the_approved_contract(tmp_path: Path) -> None:
+    store = CredentialStore(tmp_path)
+    store.save("fmcp_secret")
+
+    assert json.loads(store.path.read_text()) == {
+        "schemaVersion": 1,
+        "apiKey": "fmcp_secret",
+    }
+    assert load_secret(store).get_secret_value() == "fmcp_secret"
+    assert "user" not in store.path.read_text()
+
+
+@pytest.mark.skipif(os.name == "nt", reason="POSIX permission bits")
+def test_credential_store_restricts_file_and_directory_modes(tmp_path: Path) -> None:
+    state_directory = tmp_path / "cli"
+    store = CredentialStore(state_directory)
+    store.save("fmcp_secret")
+
+    assert store.path.stat().st_mode & 0o777 == 0o600
+    assert state_directory.stat().st_mode & 0o777 == 0o700
+
+
+def test_credential_store_restricts_an_existing_secret_file(tmp_path: Path) -> None:
+    store = CredentialStore(tmp_path)
+    store.path.write_text('{"schemaVersion": 1, "apiKey": "fmcp_secret"}')
+    if os.name != "nt":
+        store.path.chmod(0o644)
+
+    assert load_secret(store).get_secret_value() == "fmcp_secret"
+    if os.name != "nt":
+        assert store.path.stat().st_mode & 0o777 == 0o600
+
+
+@pytest.mark.skipif(os.name == "nt", reason="POSIX directory fsync")
+def test_atomic_write_ignores_unsupported_directory_fsync(
+    tmp_path: Path,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    store = CredentialStore(tmp_path)
+    original_fsync = os.fsync
+
+    def fsync(descriptor: int) -> None:
+        if stat.S_ISDIR(os.fstat(descriptor).st_mode):
+            raise OSError(errno.EINVAL, "directory sync is not supported")
+        original_fsync(descriptor)
+
+    monkeypatch.setattr("fastmcp.cli.deploy.state.os.fsync", fsync)
+
+    store.save("fmcp_secret")
+
+    assert load_secret(store).get_secret_value() == "fmcp_secret"
+
+
+def test_atomic_write_preserves_previous_state_on_replace_failure(
+    tmp_path: Path,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    store = CredentialStore(tmp_path)
+    store.save("fmcp_original")
+
+    def fail_replace(source: Path, destination: Path) -> None:
+        raise OSError("replace failed")
+
+    monkeypatch.setattr("fastmcp.cli.deploy.state.os.replace", fail_replace)
+    with pytest.raises(StateFileError):
+        store.save("fmcp_new")
+
+    assert json.loads(store.path.read_text())["apiKey"] == "fmcp_original"
+    assert list(tmp_path.glob(".*.tmp")) == []
+
+
+async def test_environment_credential_takes_precedence_and_is_not_stored(
+    tmp_path: Path,
+) -> None:
+    store = CredentialStore(tmp_path)
+    store.save("fmcp_stored")
+    authorize_called = False
+
+    async def authorize() -> SecretStr:
+        nonlocal authorize_called
+        authorize_called = True
+        return SecretStr("fmcp_interactive")
+
+    result = await resolve_credential(
+        store,
+        environ={"HORIZON_API_KEY": "fmcp_environment"},
+        authorize=authorize,
+    )
+
+    assert result.source == "environment"
+    assert result.api_key.get_secret_value() == "fmcp_environment"
+    assert load_secret(store).get_secret_value() == "fmcp_stored"
+    assert authorize_called is False
+
+
+async def test_stored_credential_precedes_interactive_authorization(
+    tmp_path: Path,
+) -> None:
+    store = CredentialStore(tmp_path)
+    store.save("fmcp_stored")
+
+    async def authorize() -> SecretStr:
+        raise AssertionError("interactive authorization must not run")
+
+    result = await resolve_credential(store, environ={}, authorize=authorize)
+
+    assert result.source == "stored"
+    assert result.api_key.get_secret_value() == "fmcp_stored"
+
+
+async def test_interactive_credential_is_persisted(tmp_path: Path) -> None:
+    store = CredentialStore(tmp_path)
+
+    async def authorize() -> SecretStr:
+        return SecretStr("fmcp_interactive")
+
+    result = await resolve_credential(store, environ={}, authorize=authorize)
+
+    assert result.source == "interactive"
+    assert load_secret(store).get_secret_value() == "fmcp_interactive"
+
+
+async def test_interactive_credential_rejects_an_origin_change(
+    tmp_path: Path,
+) -> None:
+    store = CredentialStore(tmp_path)
+    ConfigurationStore(tmp_path).save(
+        HorizonConfiguration(
+            schemaVersion=1,
+            apiOrigin="https://dev.horizon.prefect.io",
+        )
+    )
+
+    async def authorize() -> SecretStr:
+        return SecretStr("fmcp_old_origin")
+
+    with pytest.raises(StateFileError, match="host changed"):
+        await resolve_credential(
+            store,
+            environ={},
+            authorize=authorize,
+            expected_api_origin="https://horizon.prefect.io",
+        )
+
+    assert store.load() is None
+
+
+async def test_missing_noninteractive_credential_is_explicit(tmp_path: Path) -> None:
+    with pytest.raises(AuthenticationRequiredError):
+        await resolve_credential(CredentialStore(tmp_path), environ={})
+
+
+async def test_remote_revoke_always_removes_the_local_credential(
+    tmp_path: Path,
+) -> None:
+    store = CredentialStore(tmp_path)
+    store.save("fmcp_stored")
+
+    def unavailable(request: httpx2.Request) -> httpx2.Response:
+        raise httpx2.ConnectError("offline", request=request)
+
+    async with HorizonClient(
+        api_key="fmcp_stored",
+        transport=httpx2.MockTransport(unavailable),
+    ) as client:
+        with pytest.raises(HorizonUnavailableError):
+            await revoke_and_clear_credential(client, store)
+
+    assert store.load() is None
+
+
+def test_windows_acl_replaces_the_existing_access_list(
+    tmp_path: Path,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    path = tmp_path / "auth.json"
+    path.write_text("{}")
+    calls: list[list[str]] = []
+    state_paths: list[str] = []
+
+    def run(command: list[str], **kwargs: object) -> subprocess.CompletedProcess[str]:
+        calls.append(command)
+        environment = cast(dict[str, str], kwargs["env"])
+        state_paths.append(environment["FASTMCP_STATE_PATH"])
+        return subprocess.CompletedProcess(command, 0, "", "")
+
+    monkeypatch.setattr("fastmcp.cli.deploy.state.subprocess.run", run)
+    _restrict_windows_access(path)
+
+    assert calls == [
+        [
+            "powershell.exe",
+            "-NoLogo",
+            "-NoProfile",
+            "-NonInteractive",
+            "-Command",
+            calls[0][5],
+        ]
+    ]
+    assert state_paths == [str(path)]
+    assert "$path = $env:FASTMCP_STATE_PATH" in calls[0][5]
+    assert "Get-Acl -LiteralPath $path" in calls[0][5]
+    assert "SetAccessRuleProtection($true, $false)" in calls[0][5]
+    assert "RemoveAccessRuleSpecific($existingRule)" in calls[0][5]
+
+
+@pytest.mark.skipif(os.name != "nt", reason="Windows ACL inspection")
+def test_windows_credential_state_allows_only_the_current_user(tmp_path: Path) -> None:
+    state_directory = tmp_path / "cli"
+    store = CredentialStore(state_directory)
+    store.save("fmcp_secret")
+    inspect_acl = r"""
+$acl = Get-Acl -LiteralPath $env:FASTMCP_STATE_PATH
+$current = [System.Security.Principal.WindowsIdentity]::GetCurrent().User.Value
+$access = @($acl.Access | ForEach-Object {
+    $_.IdentityReference.Translate(
+        [System.Security.Principal.SecurityIdentifier]
+    ).Value
+})
+[pscustomobject]@{
+    current = $current
+    access = $access
+    protected = $acl.AreAccessRulesProtected
+    inherited = @($acl.Access | ForEach-Object { $_.IsInherited })
+} | ConvertTo-Json -Compress
+"""
+
+    for path in (state_directory, store.path):
+        result = subprocess.run(
+            [
+                "powershell.exe",
+                "-NoLogo",
+                "-NoProfile",
+                "-NonInteractive",
+                "-Command",
+                inspect_acl,
+            ],
+            check=True,
+            capture_output=True,
+            text=True,
+            env={**os.environ, "FASTMCP_STATE_PATH": str(path)},
+        )
+        acl = json.loads(result.stdout)
+        assert set(acl["access"]) == {acl["current"]}
+        assert acl["protected"] is True
+        assert not any(acl["inherited"])
+
+
+def test_credential_store_rejects_empty_api_keys(tmp_path: Path) -> None:
+    store = CredentialStore(tmp_path)
+
+    for api_key in ("", "   "):
+        with pytest.raises(StateFileError):
+            store.save(api_key)
+
+    assert store.path.exists() is False
+
+
+def test_malformed_credential_state_has_a_safe_error(tmp_path: Path) -> None:
+    store = CredentialStore(tmp_path)
+    store.path.write_text(
+        '{"schemaVersion": 1, "apiKey": "fmcp_valid", "metadata": "fmcp_secret"}'
+    )
+
+    with pytest.raises(StateFileError) as exc_info:
+        store.load()
+
+    formatted_exception = "".join(traceback.format_exception(exc_info.value))
+    assert "fmcp_secret" not in formatted_exception
+    assert exc_info.value.__cause__ is None
+    assert exc_info.value.__suppress_context__ is True

tests/cli/deploy/test_horizon_client.py

@@ -0,0 +1,265 @@
+from __future__ import annotations
+
+from collections.abc import Callable
+from urllib.parse import parse_qs
+
+import httpx2
+import pytest
+from pydantic import SecretStr
+
+from fastmcp.cli.deploy.horizon_client import (
+    DEVICE_AUTH_CLIENT_ID,
+    DEVICE_AUTH_GRANT_TYPE,
+    DeviceMetadata,
+    HorizonClient,
+    HorizonResponseError,
+    HorizonUnauthorizedError,
+    normalize_api_origin,
+)
+
+
+def mock_transport(
+    handler: Callable[[httpx2.Request], httpx2.Response],
+) -> httpx2.MockTransport:
+    return httpx2.MockTransport(handler)
+
+
+async def test_device_authorization_uses_the_oauth_form_contract() -> None:
+    def handler(request: httpx2.Request) -> httpx2.Response:
+        assert request.url.path == "/api/v0/oauth/device/authorization"
+        assert request.headers["content-type"].startswith(
+            "application/x-www-form-urlencoded"
+        )
+        assert "authorization" not in request.headers
+        assert parse_qs(request.content.decode()) == {
+            "client_id": [DEVICE_AUTH_CLIENT_ID],
+            "device_name": ["Avery's laptop"],
+            "platform": ["darwin"],
+            "architecture": ["arm64"],
+            "client_version": ["4.0.0"],
+        }
+        return httpx2.Response(
+            200,
+            json={
+                "device_code": "device-secret",
+                "user_code": "BCDF-GHJK",
+                "verification_uri": "https://horizon.prefect.io/oauth/device",
+                "verification_uri_complete": "https://horizon.prefect.io/oauth/device?user_code=BCDF-GHJK",
+                "expires_in": 600,
+                "interval": 5,
+            },
+        )
+
+    async with HorizonClient(
+        transport=mock_transport(handler),
+    ) as client:
+        result = await client.create_device_authorization(
+            DeviceMetadata(
+                device_name="Avery's laptop",
+                platform="darwin",
+                architecture="arm64",
+                client_version="4.0.0",
+            )
+        )
+
+    assert result.user_code == "BCDF-GHJK"
+    assert result.interval == 5
+
+
+@pytest.mark.parametrize(
+    "error",
+    ["authorization_pending", "slow_down", "access_denied", "expired_token"],
+)
+async def test_device_token_exchange_returns_expected_poll_errors(error: str) -> None:
+    def handler(request: httpx2.Request) -> httpx2.Response:
+        assert parse_qs(request.content.decode()) == {
+            "grant_type": [DEVICE_AUTH_GRANT_TYPE],
+            "client_id": [DEVICE_AUTH_CLIENT_ID],
+            "device_code": ["device-secret"],
+        }
+        return httpx2.Response(400, json={"error": error})
+
+    async with HorizonClient(transport=mock_transport(handler)) as client:
+        result = await client.exchange_device_authorization("device-secret")
+
+    assert result.error == error
+    assert result.access_token is None
+
+
+@pytest.mark.parametrize("access_token", ["", "   "])
+async def test_device_token_exchange_rejects_empty_access_tokens(
+    access_token: str,
+) -> None:
+    async with HorizonClient(
+        transport=mock_transport(
+            lambda request: httpx2.Response(
+                200,
+                json={"access_token": access_token, "token_type": "Bearer"},
+            )
+        )
+    ) as client:
+        with pytest.raises(HorizonResponseError):
+            await client.exchange_device_authorization("device-secret")
+
+
+async def test_device_token_exchange_keeps_the_api_key_secret() -> None:
+    async with HorizonClient(
+        transport=mock_transport(
+            lambda request: httpx2.Response(
+                200,
+                json={"access_token": "fmcp_secret", "token_type": "Bearer"},
+            )
+        )
+    ) as client:
+        result = await client.exchange_device_authorization("device-secret")
+
+    assert isinstance(result.access_token, SecretStr)
+    assert result.access_token.get_secret_value() == "fmcp_secret"
+    assert "fmcp_secret" not in repr(result)
+
+
+async def test_authenticated_routes_use_the_current_key_and_paginate() -> None:
+    cursors: list[str | None] = []
+
+    def handler(request: httpx2.Request) -> httpx2.Response:
+        assert request.headers["authorization"] == "Bearer fmcp_secret"
+        if request.url.path == "/api/v0/me":
+            return httpx2.Response(
+                200,
+                json={
+                    "user": {
+                        "id": "user-id",
+                        "email": "avery@example.com",
+                        "name": "Avery",
+                        "workosUserId": "workos-id",
+                        "createdAt": "2026-08-08T00:00:00Z",
+                    }
+                },
+            )
+
+        assert request.url.path == "/api/v0/me/organizations"
+        cursor = request.url.params.get("cursor")
+        cursors.append(cursor)
+        if cursor is None:
+            return httpx2.Response(
+                200,
+                json={
+                    "items": [{"id": "org-1", "name": "First", "slug": "first"}],
+                    "meta": {"nextCursor": "next-page", "limit": 100},
+                },
+            )
+        return httpx2.Response(
+            200,
+            json={
+                "items": [{"id": "org-2", "name": "Second", "slug": "second"}],
+                "meta": {"nextCursor": None, "limit": 100},
+            },
+        )
+
+    async with HorizonClient(
+        api_key="fmcp_secret",
+        transport=mock_transport(handler),
+    ) as client:
+        user = await client.get_current_user()
+        organizations = await client.list_organizations()
+
+    assert user.email == "avery@example.com"
+    assert [organization.slug for organization in organizations] == ["first", "second"]
+    assert cursors == [None, "next-page"]
+
+
+@pytest.mark.parametrize("count", [0, 1, 3])
+async def test_organization_memberships_preserve_zero_one_and_many(count: int) -> None:
+    organizations = [
+        {"id": f"org-{index}", "name": f"Org {index}", "slug": f"org-{index}"}
+        for index in range(count)
+    ]
+    async with HorizonClient(
+        api_key="fmcp_secret",
+        transport=mock_transport(
+            lambda request: httpx2.Response(
+                200,
+                json={
+                    "items": organizations,
+                    "meta": {"nextCursor": None, "limit": 100},
+                },
+            )
+        ),
+    ) as client:
+        result = await client.list_organizations()
+
+    assert len(result) == count
+
+
+async def test_revoke_uses_the_current_authenticated_key() -> None:
+    def handler(request: httpx2.Request) -> httpx2.Response:
+        assert request.method == "DELETE"
+        assert request.url.path == "/api/v0/me/api-key"
+        assert request.headers["authorization"] == "Bearer fmcp_current"
+        return httpx2.Response(204)
+
+    async with HorizonClient(
+        api_key="fmcp_current",
+        transport=mock_transport(handler),
+    ) as client:
+        await client.revoke_current_api_key()
+
+
+async def test_protected_routes_require_a_credential() -> None:
+    async with HorizonClient(
+        transport=mock_transport(lambda request: httpx2.Response(200))
+    ) as client:
+        with pytest.raises(HorizonUnauthorizedError):
+            await client.get_current_user()
+
+
+async def test_protected_routes_report_a_rejected_credential() -> None:
+    async with HorizonClient(
+        api_key="fmcp_invalid",
+        transport=mock_transport(lambda request: httpx2.Response(401)),
+    ) as client:
+        with pytest.raises(HorizonUnauthorizedError):
+            await client.get_current_user()
+
+
+async def test_public_routes_do_not_report_a_missing_credential() -> None:
+    async with HorizonClient(
+        transport=mock_transport(lambda request: httpx2.Response(401))
+    ) as client:
+        with pytest.raises(HorizonResponseError):
+            await client.create_device_authorization()
+
+
+async def test_invalid_responses_do_not_include_response_bodies() -> None:
+    secret_body = "fmcp_response_secret"
+    async with HorizonClient(
+        transport=mock_transport(lambda request: httpx2.Response(500, text=secret_body))
+    ) as client:
+        with pytest.raises(HorizonResponseError) as exc_info:
+            await client.create_device_authorization()
+
+    assert exc_info.value.status_code == 500
+    assert secret_body not in str(exc_info.value)
+
+
+@pytest.mark.parametrize(
+    "value",
+    [
+        "ftp://horizon.prefect.io",
+        "https://user@example.com",
+        "https://horizon.prefect.io/path",
+        "https://horizon.prefect.io?query=value",
+        "https://horizon.prefect.io:abc",
+        "https://horizon.prefect.io:99999",
+    ],
+)
+def test_api_origin_rejects_values_that_are_not_origins(value: str) -> None:
+    with pytest.raises(ValueError):
+        normalize_api_origin(value)
+
+
+def test_api_origin_normalizes_one_trailing_slash() -> None:
+    assert (
+        normalize_api_origin("https://horizon.prefect.io/")
+        == "https://horizon.prefect.io"
+    )

tests/cli/deploy/test_output.py

@@ -0,0 +1,140 @@
+import json
+
+import pytest
+
+from fastmcp.cli.deploy.horizon_client import DeviceAuthorization, HorizonUser
+from fastmcp.cli.deploy.output import (
+    emit_device_challenge,
+    emit_error,
+    emit_identity,
+    emit_logout,
+)
+
+
+def authorization() -> DeviceAuthorization:
+    return DeviceAuthorization(
+        device_code="device-secret",
+        user_code="ABCD-EFGH",
+        verification_uri="https://horizon.prefect.io/oauth/device",
+        verification_uri_complete=(
+            "https://horizon.prefect.io/oauth/device?user_code=ABCD-EFGH"
+        ),
+        expires_in=600,
+        interval=5,
+    )
+
+
+def user() -> HorizonUser:
+    return HorizonUser(id="user-1", email="ada@example.com", name="Ada")
+
+
+def test_json_device_challenge_uses_only_stderr(
+    capsys: pytest.CaptureFixture[str],
+) -> None:
+    emit_device_challenge(authorization(), json_output=True)
+
+    captured = capsys.readouterr()
+    assert captured.out == ""
+    assert json.loads(captured.err) == {
+        "event": "device_authorization",
+        "verificationUrl": "https://horizon.prefect.io/oauth/device",
+        "verificationUrlComplete": (
+            "https://horizon.prefect.io/oauth/device?user_code=ABCD-EFGH"
+        ),
+        "userCode": "ABCD-EFGH",
+    }
+
+
+def test_tty_device_challenge_uses_the_sign_in_layout(
+    capsys: pytest.CaptureFixture[str],
+) -> None:
+    emit_device_challenge(authorization(), json_output=False)
+
+    output = capsys.readouterr().out
+    assert "│" in output
+    assert "Deploy FastMCP on Horizon" in output
+    assert "✓ Device authorization started" in output
+    assert "https://horizon.prefect.io/oauth/device?user_code=ABCD-EFGH" in output
+    assert "ABCD-EFGH" in output
+    assert "The request expires in 10 minutes." in output
+
+
+def test_json_identity_has_stable_fields(
+    capsys: pytest.CaptureFixture[str],
+) -> None:
+    emit_identity("login", user(), json_output=True)
+
+    result = json.loads(capsys.readouterr().out)
+    assert result == {
+        "ok": True,
+        "command": "login",
+        "user": {
+            "id": "user-1",
+            "email": "ada@example.com",
+            "name": "Ada",
+        },
+    }
+
+
+def test_tty_identity_uses_an_account_panel(
+    capsys: pytest.CaptureFixture[str],
+) -> None:
+    emit_identity("whoami", user(), json_output=False)
+
+    output = capsys.readouterr().out
+    assert "│" in output
+    assert "Horizon Account" in output
+    assert "Ada" in output
+    assert "ada@example.com" in output
+    assert "● Signed in" in output
+    assert "Organization" not in output
+
+
+def test_json_error_has_stable_fields(
+    capsys: pytest.CaptureFixture[str],
+) -> None:
+    emit_error(
+        "logout",
+        "remote_revocation_failed",
+        "The remote key can remain active.",
+        json_output=True,
+        details={
+            "localCredentialRemoved": True,
+            "remoteCredentialMayRemain": True,
+        },
+    )
+
+    result = json.loads(capsys.readouterr().out)
+    assert result == {
+        "ok": False,
+        "command": "logout",
+        "error": {
+            "category": "remote_revocation_failed",
+            "message": "The remote key can remain active.",
+        },
+        "localCredentialRemoved": True,
+        "remoteCredentialMayRemain": True,
+    }
+
+
+def test_tty_logout_uses_the_horizon_header(
+    capsys: pytest.CaptureFixture[str],
+) -> None:
+    emit_logout(remote_revoked=True, json_output=False)
+
+    output = capsys.readouterr().out
+    assert "Logged out of Horizon" in output
+    assert "│" in output
+
+
+def test_json_logout_has_stable_fields(
+    capsys: pytest.CaptureFixture[str],
+) -> None:
+    emit_logout(remote_revoked=True, json_output=True)
+
+    assert json.loads(capsys.readouterr().out) == {
+        "ok": True,
+        "command": "logout",
+        "localCredentialRemoved": True,
+        "remoteRevoked": True,
+    }

tests/cli/test_cli.py

@@ -35,6 +35,20 @@ class TestMainCLI:
         assert isinstance(exc_info.value, SystemExit)
         assert exc_info.value.code == 1
 
+    @pytest.mark.parametrize("name", ["login", "logout", "whoami"])
+    def test_horizon_account_commands_are_top_level(self, name: str):
+        command, bound, _ = app.parse_args([name, "--json"])
+
+        assert command.__name__ == name  # type: ignore[attr-defined]  # ty:ignore[unresolved-attribute]
+        assert bound.arguments == {"json_output": True}
+
+    def test_login_accepts_a_horizon_host(self):
+        _, bound, _ = app.parse_args(
+            ["login", "--host", "https://dev.horizon.prefect.io"]
+        )
+
+        assert bound.arguments == {"host": "https://dev.horizon.prefect.io"}
+
 
 class TestVersionCommand:
     """Test the version command."""