feat: add Horizon account commands

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 user and organization memberships |
+| `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,34 @@ 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` always 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.
+
+Login stores only the personal Horizon API key.
+It does not select or store a deployment organization.
+`fastmcp whoami` gets the current user and organization memberships 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,35 @@ 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 shows a verification URL and code before it opens the browser.
+If the browser cannot open, visit the shown URL and enter the code.
+New users can register and create their first Horizon organization in the browser.
+
+Check the active account and its current organization memberships 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/command.py

@@ -0,0 +1,307 @@
+"""Public Prefect Horizon authentication commands."""
+
+from __future__ import annotations
+
+import sys
+import webbrowser
+from typing import Annotated, NoReturn
+
+from cyclopts import Parameter
+
+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 (
+    HorizonClient,
+    HorizonOrganization,
+    HorizonResponseError,
+    HorizonUnauthorizedError,
+    HorizonUnavailableError,
+    HorizonUser,
+)
+from fastmcp.cli.deploy.output import (
+    CommandName,
+    ErrorCategory,
+    emit_device_challenge,
+    emit_error,
+    emit_identity,
+    emit_logout,
+)
+from fastmcp.cli.deploy.state import StateFileError
+
+JsonOption = Annotated[
+    bool,
+    Parameter(
+        name="--json",
+        help="Write one final JSON result to stdout",
+        negative=(),
+    ),
+]
+
+
+def _can_open_browser() -> bool:
+    return sys.stdin.isatty() and sys.stdout.isatty()
+
+
+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_identity(
+    api_origin: str,
+    credential: ResolvedCredential,
+) -> tuple[HorizonUser, tuple[HorizonOrganization, ...]]:
+    async with HorizonClient(api_origin, api_key=credential.api_key) as client:
+        user = await client.get_current_user()
+        organizations = await client.list_organizations()
+    return user, organizations
+
+
+async def login(
+    *,
+    json_output: JsonOption = False,
+) -> None:
+    """Sign in to Prefect Horizon."""
+    credentials = CredentialStore()
+
+    try:
+        configuration = ConfigurationStore().load()
+
+        async def device_authorization():
+            async with HorizonClient(configuration.api_origin) as client:
+                return await authorize_device(
+                    client,
+                    on_challenge=lambda challenge: emit_device_challenge(
+                        challenge,
+                        json_output=json_output,
+                    ),
+                    open_browser=not json_output and _can_open_browser(),
+                    browser_opener=webbrowser.open,
+                )
+
+        credential = await resolve_credential(
+            credentials,
+            authorize=device_authorization,
+        )
+
+        try:
+            user, organizations = await _get_identity(
+                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,
+            )
+            try:
+                user, organizations = await _get_identity(
+                    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,
+        organizations,
+        json_output=json_output,
+    )
+
+
+async def whoami(
+    *,
+    json_output: JsonOption = False,
+) -> None:
+    """Show the current Prefect Horizon user and organization memberships."""
+    credentials = CredentialStore()
+    credential: ResolvedCredential | None = None
+
+    try:
+        configuration = ConfigurationStore().load()
+        credential = await resolve_credential(credentials)
+        user, organizations = await _get_identity(
+            configuration.api_origin,
+            credential,
+        )
+    except HorizonUnauthorizedError as error:
+        if credential is not None and credential.source == "stored":
+            credentials.clear()
+        _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,
+        organizations,
+        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/output.py

@@ -0,0 +1,143 @@
+"""Stable terminal and JSON output for Horizon CLI commands."""
+
+from __future__ import annotations
+
+import json
+import sys
+from typing import Literal
+
+from rich.console import Console
+
+from fastmcp.cli.deploy.horizon_client import (
+    DeviceAuthorization,
+    HorizonOrganization,
+    HorizonUser,
+)
+
+CommandName = Literal["login", "logout", "whoami"]
+ErrorCategory = Literal[
+    "authentication_invalid",
+    "authentication_required",
+    "authorization_denied",
+    "authorization_expired",
+    "authorization_failed",
+    "horizon_error",
+    "horizon_unavailable",
+    "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 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("Open this URL to sign in to Prefect Horizon:")
+    console.print(authorization.verification_uri)
+    console.print(f"Enter code: {authorization.user_code}")
+    console.print("Waiting for approval...")
+
+
+def emit_identity(
+    command: Literal["login", "whoami"],
+    user: HorizonUser,
+    organizations: tuple[HorizonOrganization, ...],
+    *,
+    json_output: bool,
+) -> None:
+    """Show the authenticated user and current organization memberships."""
+    if json_output:
+        _write_json(
+            {
+                "ok": True,
+                "command": command,
+                "user": user.model_dump(mode="json"),
+                "organizations": [
+                    organization.model_dump(mode="json")
+                    for organization in organizations
+                ],
+            }
+        )
+        return
+
+    prefix = "Signed in" if command == "login" else "Authenticated"
+    display_name = f"{user.name} <{user.email}>" if user.name else user.email
+    console.print(f"{prefix} as {display_name}.", markup=False)
+    if not organizations:
+        console.print("Organization memberships: none")
+        return
+
+    console.print("Organization memberships:")
+    for organization in organizations:
+        console.print(f"- {organization.name} ({organization.slug})", markup=False)
+
+
+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:
+        console.print("Signed out of Prefect Horizon.")
+    else:
+        console.print("No active Horizon credential remains on this device.")
+
+
+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
+
+    error_console.print(f"Error: {message}", markup=False)

tests/cli/deploy/test_command.py

@@ -0,0 +1,302 @@
+import json
+from collections.abc import Callable
+from unittest.mock import Mock
+
+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
+
+
+class HorizonAuthAPI:
+    def __init__(
+        self,
+        *,
+        token_error: str | None = None,
+        revoke_status: int = 204,
+        organizations: list[dict[str, str]] | None = None,
+        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.organizations = organizations or []
+        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/organizations":
+            return httpx2.Response(
+                200,
+                json={
+                    "items": self.organizations,
+                    "meta": {"nextCursor": None, "limit": 100},
+                },
+            )
+        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(
+        organizations=[{"id": "org-1", "name": "Acme", "slug": "acme"}]
+    )
+    use_horizon_api(api)
+    browser_open = Mock()
+    monkeypatch.setattr(command_module.webbrowser, "open", browser_open)
+
+    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",
+        },
+        "organizations": [{"id": "org-1", "name": "Acme", "slug": "acme"}],
+    }
+    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()
+
+    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_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 "Signed in as Ada <ada@example.com>." in output
+    assert "Organization memberships: none" 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_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_output.py

@@ -0,0 +1,127 @@
+import json
+
+import pytest
+
+from fastmcp.cli.deploy.horizon_client import (
+    DeviceAuthorization,
+    HorizonOrganization,
+    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 organizations() -> tuple[HorizonOrganization, ...]:
+    return (
+        HorizonOrganization(id="org-1", name="Acme", slug="acme"),
+        HorizonOrganization(id="org-2", name="Research", slug="research"),
+    )
+
+
+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_json_identity_has_stable_fields(
+    capsys: pytest.CaptureFixture[str],
+) -> None:
+    emit_identity("login", user(), organizations(), 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",
+        },
+        "organizations": [
+            {"id": "org-1", "name": "Acme", "slug": "acme"},
+            {"id": "org-2", "name": "Research", "slug": "research"},
+        ],
+    }
+
+
+def test_tty_identity_handles_no_organizations(
+    capsys: pytest.CaptureFixture[str],
+) -> None:
+    emit_identity("whoami", user(), (), json_output=False)
+
+    output = capsys.readouterr().out
+    assert "Authenticated as Ada <ada@example.com>." in output
+    assert "Organization memberships: none" 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_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,13 @@ 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}
+
 
 class TestVersionCommand:
     """Test the version command."""