Refactor: extract shared action layer for CLI and MCP tools

nextmv/nextmv/cli/actions/__init__.py

@@ -0,0 +1,8 @@
+"""Shared business logic for CLI commands and MCP tools.
+
+Each submodule contains pure functions that:
+- Take a Client or Application as their first argument
+- Take plain-typed business parameters
+- Return data (dict, list, str, bool, None)
+- Have no side effects (no printing, no file I/O)
+"""

nextmv/nextmv/cli/actions/acceptance.py

@@ -0,0 +1,109 @@
+"""Core acceptance test actions.
+
+Pure functions that wrap SDK calls. No CLI or MCP presentation concerns.
+
+Acceptance tests compare a candidate instance against a baseline using a
+set of metrics. The ``get_acceptance_test_with_polling`` action blocks
+until the test completes (or the poll timeout expires).
+"""
+
+from typing import Any
+
+from nextmv.cli.framework.options import (
+    AcceptanceTestIdOption,
+    AppIdRequiredOption,
+    DescriptionOption,
+    NameOption,
+    OptionalAcceptanceTestIdOption,
+)
+from nextmv.cloud import Application, Client
+
+
+def list_acceptance_tests(
+    client: Client,
+    app_id: AppIdRequiredOption,
+) -> list[dict[str, Any]]:
+    """List all Nextmv Cloud acceptance tests for an application.
+
+    Returns a list of acceptance test dicts.
+    """
+    app = Application(client=client, id=app_id)
+    return [t.to_dict() for t in app.list_acceptance_tests()]
+
+
+def get_acceptance_test(
+    client: Client,
+    app_id: AppIdRequiredOption,
+    acceptance_test_id: AcceptanceTestIdOption,
+) -> dict[str, Any]:
+    """Get details and results of a Nextmv Cloud acceptance test.
+
+    Returns the test status, metric comparisons, and pass/fail outcome
+    (if the test has completed).
+    """
+    app = Application(client=client, id=app_id)
+    return app.acceptance_test(acceptance_test_id=acceptance_test_id).to_dict()
+
+
+def create_acceptance_test(
+    client: Client,
+    app_id: AppIdRequiredOption,
+    candidate_instance_id: str,
+    baseline_instance_id: str,
+    metrics: list[Any],
+    acceptance_test_id: OptionalAcceptanceTestIdOption = None,
+    name: NameOption = None,
+    input_set_id: str | None = None,
+    description: DescriptionOption = None,
+) -> dict[str, Any]:
+    """Create a new Nextmv Cloud acceptance test.
+
+    An acceptance test runs both the candidate and baseline instances
+    against the same input set and compares results using the specified
+    metrics to determine whether the candidate meets acceptance criteria.
+
+    Returns the created test dict.
+    """
+    app = Application(client=client, id=app_id)
+    return app.new_acceptance_test(
+        candidate_instance_id=candidate_instance_id,
+        baseline_instance_id=baseline_instance_id,
+        metrics=metrics,
+        id=acceptance_test_id,
+        name=name,
+        input_set_id=input_set_id,
+        description=description,
+    ).to_dict()
+
+
+def update_acceptance_test(
+    client: Client,
+    app_id: AppIdRequiredOption,
+    acceptance_test_id: AcceptanceTestIdOption,
+    name: NameOption = None,
+    description: DescriptionOption = None,
+) -> dict[str, Any]:
+    """Update a Nextmv Cloud acceptance test.
+
+    Only the provided fields are updated; omitted fields remain unchanged.
+    Returns the updated test dict.
+    """
+    app = Application(client=client, id=app_id)
+    return app.update_acceptance_test(
+        acceptance_test_id=acceptance_test_id,
+        name=name,
+        description=description,
+    ).to_dict()
+
+
+def delete_acceptance_test(
+    client: Client,
+    app_id: AppIdRequiredOption,
+    acceptance_test_id: AcceptanceTestIdOption,
+) -> None:
+    """Delete a Nextmv Cloud acceptance test permanently.
+
+    This action cannot be undone.
+    """
+    app = Application(client=client, id=app_id)
+    app.delete_acceptance_test(acceptance_test_id=acceptance_test_id)

nextmv/nextmv/cli/actions/account.py

@@ -0,0 +1,78 @@
+"""Core account management actions.
+
+Pure functions that wrap SDK calls. No CLI or MCP presentation concerns.
+
+Account management requires an SSO-enabled organization. Most operations
+take an account ID; create takes a name and a list of admin email
+addresses.
+"""
+
+from typing import Any
+
+from nextmv.cli.framework.options import AccountIdOption, NameOption
+from nextmv.cloud import Client
+from nextmv.cloud.account import Account
+
+
+def get_account(
+    client: Client,
+    account_id: AccountIdOption,
+) -> dict[str, Any]:
+    """Get details of a Nextmv Cloud account.
+
+    Returns the account dict including ID, name, and administrator list.
+    """
+    return Account.get(client=client, account_id=account_id).to_dict()
+
+
+def get_queue(
+    client: Client,
+    account_id: AccountIdOption,
+) -> dict[str, Any]:
+    """Get the run queue for a Nextmv Cloud account.
+
+    Returns the account's current queue state, including queued and
+    running jobs.
+    """
+    return Account.get(client=client, account_id=account_id).queue().to_dict()
+
+
+def create_account(
+    client: Client,
+    name: NameOption,
+    admins: list[str],
+) -> dict[str, Any]:
+    """Create a new Nextmv Cloud account in your organization.
+
+    To create managed accounts, SSO must be configured for your
+    organization. At least one administrator email address must be
+    provided.
+
+    Returns the created account dict.
+    """
+    return Account.new(client=client, name=name, admins=admins).to_dict()
+
+
+def update_account(
+    client: Client,
+    account_id: AccountIdOption,
+    name: NameOption,
+) -> dict[str, Any]:
+    """Update a Nextmv Cloud account's name.
+
+    Returns the updated account dict.
+    """
+    account = Account.get(client=client, account_id=account_id)
+    return account.update(name=name).to_dict()
+
+
+def delete_account(
+    client: Client,
+    account_id: AccountIdOption,
+) -> None:
+    """Delete a Nextmv Cloud account permanently.
+
+    You must have the ``administrator`` role on the account to delete
+    it. This action cannot be undone.
+    """
+    Account.get(client=client, account_id=account_id).delete()

nextmv/nextmv/cli/actions/app.py

@@ -0,0 +1,136 @@
+"""Core application management actions.
+
+Pure functions that wrap SDK calls. No CLI or MCP presentation concerns.
+
+Each action takes ``client: Client`` as its first parameter so the CLI and
+MCP frameworks can inject a client at call time. Remaining parameters use
+dual-purpose ``Annotated`` aliases from ``nextmv.cli.framework.options`` so
+the same metadata drives both Typer's ``--help`` output and FastMCP's tool
+input schema.
+
+Docstrings are multi-paragraph: the first line is a short description used
+by both frontends as the primary tool/command description, and subsequent
+paragraphs provide additional guidance that reads naturally in both
+contexts.
+"""
+
+from typing import Any
+
+from nextmv.cli.framework.options import (
+    AppDirOption,
+    AppIdOption,
+    AppIdRequiredOption,
+    DefaultExperimentInstanceOption,
+    DefaultInstanceIdOption,
+    DescriptionOption,
+    ExistOkOption,
+    IsWorkflowOption,
+    NameOption,
+)
+from nextmv.cloud import Application, Client, list_applications
+
+
+def list_apps(client: Client) -> list[dict[str, Any]]:
+    """List all Nextmv Cloud applications in the current account.
+
+    Returns a list of application dictionaries containing each application's
+    ID, name, description, and default instance.
+    """
+    apps = list_applications(client)
+    return [a.to_dict() for a in apps]
+
+
+def create_app(
+    client: Client,
+    name: NameOption = None,
+    app_id: AppIdOption = None,
+    description: DescriptionOption = None,
+    is_workflow: IsWorkflowOption = False,
+    exist_ok: ExistOkOption = False,
+    default_instance_id: DefaultInstanceIdOption = None,
+    default_experiment_instance: DefaultExperimentInstanceOption = None,
+) -> dict[str, Any]:
+    """Create a new Nextmv Cloud application.
+
+    Set ``exist_ok`` to avoid errors when an application with the given ID
+    already exists; the existing application is returned instead.
+
+    An application can be marked as a workflow via ``is_workflow``. Workflows
+    leverage Nextpipe to orchestrate multiple decision models.
+
+    Returns the created (or existing) application object. A version and
+    default instance are automatically provisioned for new applications.
+    """
+    return Application.new(
+        client=client,
+        name=name,
+        id=app_id,
+        description=description,
+        is_workflow=is_workflow,
+        exist_ok=exist_ok,
+        default_instance_id=default_instance_id,
+        default_experiment_instance=default_experiment_instance,
+    ).to_dict()
+
+
+def get_app(client: Client, app_id: AppIdRequiredOption) -> dict[str, Any]:
+    """Get details of a specific Nextmv Cloud application.
+
+    Returns the full application object including its name, description,
+    default instance, and creation timestamp.
+    """
+    return Application.get(client=client, id=app_id).to_dict()
+
+
+def delete_app(client: Client, app_id: AppIdRequiredOption) -> None:
+    """Delete a Nextmv Cloud application permanently.
+
+    This action cannot be undone. All versions, instances, and run history
+    associated with the application will be removed.
+    """
+    Application(client=client, id=app_id).delete()
+
+
+def app_exists(client: Client, app_id: AppIdRequiredOption) -> bool:
+    """Check whether a Nextmv Cloud application exists.
+
+    Returns ``True`` if an application with the given ID exists in the
+    current account, ``False`` otherwise.
+    """
+    return Application.exists(client=client, id=app_id)
+
+
+def update_app(
+    client: Client,
+    app_id: AppIdRequiredOption,
+    name: NameOption = None,
+    description: DescriptionOption = None,
+    default_instance_id: DefaultInstanceIdOption = None,
+    default_experiment_instance: DefaultExperimentInstanceOption = None,
+) -> dict[str, Any]:
+    """Update attributes of a Nextmv Cloud application.
+
+    Only the provided fields are updated; omitted fields remain unchanged.
+    Returns the updated application object.
+    """
+    app = Application(client=client, id=app_id)
+    return app.update(
+        name=name,
+        description=description,
+        default_instance_id=default_instance_id,
+        default_experiment_instance=default_experiment_instance,
+    ).to_dict()
+
+
+def push_app(
+    client: Client,
+    app_id: AppIdRequiredOption,
+    app_dir: AppDirOption,
+) -> None:
+    """Push local application code to a Nextmv Cloud application.
+
+    Uploads the contents of a local directory as a new version of the
+    application. The directory must contain an ``app.yaml`` manifest.
+    """
+    app = Application(client=client, id=app_id)
+    app.push(app_dir=app_dir)