feat: add opt-in execute_command tool

README.md

@@ -42,15 +42,16 @@ arguments.
 
 The server exposes the following tools for interacting with the Enapter EMS:
 
-| Tool                        | Description                                                      |
-| --------------------------- | ---------------------------------------------------------------- |
-| `search_sites`              | Search among all sites with name and timezone regex filtering    |
-| `search_devices`            | Search devices by site, type, and name regex filtering           |
-| `search_command_executions` | Search the history of command executions                         |
-| `read_blueprint`            | Access device blueprint sections (properties, telemetry, alerts) |
-| `get_historical_telemetry`  | Retrieve time-series telemetry with configurable granularity     |
-| `search_rules`              | Search for automation rules within a specific site               |
-| `read_rule`                 | Read the paginated lines of a rule's Lua script                  |
+| Tool                        | Description                                                      | Access     | Default  |
+| --------------------------- | ---------------------------------------------------------------- | ---------- | -------- |
+| `search_sites`              | Search among all sites with name and timezone regex filtering    | Read-only  | Enabled  |
+| `search_devices`            | Search devices by site, type, and name regex filtering           | Read-only  | Enabled  |
+| `search_command_executions` | Search the history of command executions                         | Read-only  | Enabled  |
+| `read_blueprint`            | Access device blueprint sections (properties, telemetry, alerts) | Read-only  | Enabled  |
+| `get_historical_telemetry`  | Retrieve time-series telemetry with configurable granularity     | Read-only  | Enabled  |
+| `search_rules`              | Search for automation rules within a specific site               | Read-only  | Enabled  |
+| `read_rule`                 | Read the paginated lines of a rule's Lua script                  | Read-only  | Enabled  |
+| `execute_command`           | Execute a command on a device                                    | Read-write | Disabled |
 
 ## Usage Examples
 
@@ -114,6 +115,36 @@ Here are realistic examples of how you can interact with your Enapter devices us
 - Retrieves the rule's Lua script using `read_rule`
 - Analyzes the logic and confirms the exact threshold and conditions that trigger the electrolyser
 
+### Example 5: Executing a Command (Human Confirmation Required)
+
+> ⚠️ `execute_command` is **destructive** — it acts on real physical hardware
+> (pumps, electrolysers, valves, inverters). It is **disabled by default**. Enable
+> it with `--command-execution-enabled` on the command line or by setting
+> `ENAPTER_COMMAND_EXECUTION_ENABLED=1`.
+
+**User prompt:**
+
+> The electrolyser at the Alpha site has been running for a long time. Please
+> reboot it for me.
+
+**What happens:**
+
+- Server locates the electrolyser device using `search_devices`
+- Reads the device blueprint with `read_blueprint(section="commands")` to
+  discover the `reboot` command and checks whether it declares a `confirmation`
+  block
+- The `reboot` command declares a `confirmation` with a `title` and
+  `description` (e.g. _"Reboot the electrolyser"_ / _"This will restart the
+  device and interrupt production."_), so the assistant **presents these to the
+  human and waits for explicit approval** — it does not act on its own initiative
+- Only after the human confirms does the assistant call `execute_command` with
+  `human_confirmed_this_action=True`
+- The device runs the command and the tool returns the resulting
+  `CommandExecution`, whose `state` field reports the outcome
+  (`success`/`error`/`timeout`/`unsync`)
+- The returned execution `id` can later be referenced or audited via
+  `search_command_executions`
+
 ## Support
 
 For issues, questions, or contributions, please:

src/enapter_mcp_server/cli/serve_command.py

@@ -38,6 +38,7 @@
 )
 ENAPTER_OAUTH_PROXY_JWT_SIGNING_KEY = os.getenv("ENAPTER_OAUTH_PROXY_JWT_SIGNING_KEY")
 ENAPTER_CORS_ALLOW_ORIGINS = os.getenv("ENAPTER_CORS_ALLOW_ORIGINS")
+ENAPTER_COMMAND_EXECUTION_ENABLED = os.getenv("ENAPTER_COMMAND_EXECUTION_ENABLED", "0")
 
 
 class ServeCommand(Command):
@@ -125,6 +126,13 @@ def register(parent: Subparsers) -> None:
             default=ENAPTER_OAUTH_PROXY_JWT_SIGNING_KEY,
             help="Signing key for JWTs issued by OAuth proxy. Required if OAuth proxy is enabled",
         )
+        parser.add_argument(
+            "--command-execution-enabled",
+            choices=["0", "1"],
+            default=ENAPTER_COMMAND_EXECUTION_ENABLED,
+            help="Enable the destructive `execute_command` tool (kill switch). "
+            "When disabled (the default), the tool is not registered at all",
+        )
 
     @staticmethod
     async def run(args: argparse.Namespace) -> None:
@@ -165,6 +173,7 @@ async def run(args: argparse.Namespace) -> None:
             oauth_proxy_config=oauth_proxy_config,
             logo_url=args.logo_url,
             cors_allow_origins=cors_allow_origins,
+            command_execution_enabled=args.command_execution_enabled == "1",
         )
         async with asyncio.TaskGroup() as task_group:
             async with http.EnapterAPI(

src/enapter_mcp_server/core/__init__.py

@@ -5,6 +5,8 @@
 from .device_search_query import DeviceSearchQuery
 from .enapter_api import EnapterAPI
 from .errors import (
+    CommandNotFound,
+    ConfirmationRequired,
     GatewayUnavailable,
     LatestTelemetryUnavailable,
     SearchQueryTooBroad,
@@ -19,6 +21,8 @@
     "ApplicationServer",
     "AuthConfig",
     "CommandExecutionSearchQuery",
+    "CommandNotFound",
+    "ConfirmationRequired",
     "DeviceDTO",
     "DeviceSearchQuery",
     "EnapterAPI",

src/enapter_mcp_server/core/application_server.py

@@ -1,6 +1,7 @@
 import asyncio
 import datetime
 import re
+from typing import Any
 
 from enapter_mcp_server import domain
 
@@ -9,7 +10,12 @@
 from .device_dto import DeviceDTO
 from .device_search_query import DeviceSearchQuery
 from .enapter_api import EnapterAPI
-from .errors import GatewayUnavailable, SearchQueryTooBroad
+from .errors import (
+    CommandNotFound,
+    ConfirmationRequired,
+    GatewayUnavailable,
+    SearchQueryTooBroad,
+)
 from .rule_search_query import RuleSearchQuery
 from .site_dto import SiteDTO
 from .site_search_query import SiteSearchQuery
@@ -310,6 +316,44 @@ async def read_blueprint(
         entities.sort(key=key)
         return entities[offset : offset + limit]
 
+    async def execute_command(
+        self,
+        auth: AuthConfig,
+        device_id: str,
+        command_name: str,
+        arguments: dict[str, Any] | None = None,
+        human_confirmed_this_action: bool = False,
+    ) -> domain.CommandExecution:
+        commands = await self._resolve_manifest_commands(auth, device_id)
+        declaration = commands.get(command_name)
+        if declaration is None:
+            available = ", ".join(sorted(commands)) or "(none)"
+            raise CommandNotFound(
+                f"Command {command_name!r} is not declared in the manifest of"
+                f" device {device_id!r}. Available commands: {available}."
+            )
+        if declaration.confirmation is not None and not human_confirmed_this_action:
+            confirmation = declaration.confirmation
+            title = confirmation.title or declaration.display_name
+            description = confirmation.description
+            raise ConfirmationRequired(
+                f"Command {command_name!r} on device {device_id!r} requires human"
+                f" confirmation before execution. Title: {title}."
+                f" Description: {description}."
+            )
+        return await self._enapter_api.execute_command(
+            auth, device_id, command_name, arguments
+        )
+
+    async def _resolve_manifest_commands(
+        self, auth: AuthConfig, device_id: str
+    ) -> dict[str, domain.CommandDeclaration]:
+        device_dto = await self._enapter_api.get_device(
+            auth, device_id, expand_manifest=True
+        )
+        assert device_dto.manifest is not None
+        return device_dto.manifest.commands
+
     async def get_historical_telemetry(
         self,
         auth: AuthConfig,

src/enapter_mcp_server/core/enapter_api.py

@@ -53,6 +53,14 @@ async def get_device(
         expand_active_alerts: bool = False,
     ) -> DeviceDTO: ...
 
+    async def execute_command(
+        self,
+        auth: AuthConfig,
+        device_id: str,
+        command_name: str,
+        arguments: dict[str, Any] | None,
+    ) -> domain.CommandExecution: ...
+
     @enapter.async_.generator
     async def list_command_executions(
         self,

src/enapter_mcp_server/core/errors.py

@@ -8,3 +8,11 @@ class SearchQueryTooBroad(Exception):
 
 class GatewayUnavailable(Exception):
     pass
+
+
+class ConfirmationRequired(Exception):
+    pass
+
+
+class CommandNotFound(Exception):
+    pass

src/enapter_mcp_server/domain/__init__.py

@@ -5,6 +5,7 @@
 from .blueprint_section import BlueprintSection
 from .blueprint_summary import BlueprintSummary
 from .command_argument_declaration import CommandArgumentDeclaration
+from .command_confirmation import CommandConfirmation
 from .command_declaration import CommandDeclaration
 from .command_execution import CommandExecution
 from .command_execution_state import CommandExecutionState
@@ -33,6 +34,7 @@
     "BlueprintSection",
     "BlueprintSummary",
     "CommandArgumentDeclaration",
+    "CommandConfirmation",
     "CommandDeclaration",
     "CommandExecution",
     "CommandExecutionState",

src/enapter_mcp_server/domain/command_confirmation.py

@@ -0,0 +1,8 @@
+import dataclasses
+
+
+@dataclasses.dataclass(frozen=True, kw_only=True)
+class CommandConfirmation:
+    severity: str | None = None
+    title: str | None = None
+    description: str | None = None

src/enapter_mcp_server/domain/command_declaration.py

@@ -2,6 +2,7 @@
 
 from .access_role import AccessRole
 from .command_argument_declaration import CommandArgumentDeclaration
+from .command_confirmation import CommandConfirmation
 
 
 @dataclasses.dataclass(frozen=True, kw_only=True)
@@ -12,3 +13,4 @@ class CommandDeclaration:
     description: str | None
     arguments: list[CommandArgumentDeclaration]
     implements: list[str]
+    confirmation: CommandConfirmation | None = None

src/enapter_mcp_server/http/enapter_api.py

@@ -119,6 +119,19 @@ async def get_device(
             )
             return self._data_mapper.to_device_dto(device)
 
+    async def execute_command(
+        self,
+        auth: core.AuthConfig,
+        device_id: str,
+        command_name: str,
+        arguments: dict[str, Any] | None,
+    ) -> domain.CommandExecution:
+        async with self._new_client(auth) as client:
+            execution = await client.commands.execute(
+                device_id, command_name, arguments
+            )
+            return self._data_mapper.to_command_execution(execution)
+
     @enapter.async_.generator
     async def list_command_executions(
         self,

src/enapter_mcp_server/http/enapter_data_mapper.py

@@ -124,6 +124,18 @@ def to_command_declaration(
                 for arg_name, arg_dto in (dto.get("arguments") or {}).items()
             ],
             implements=dto.get("implements") or [],
+            confirmation=self.to_command_confirmation(dto.get("confirmation")),
+        )
+
+    def to_command_confirmation(
+        self, dto: dict[str, Any] | None
+    ) -> domain.CommandConfirmation | None:
+        if dto is None:
+            return None
+        return domain.CommandConfirmation(
+            severity=dto.get("severity"),
+            title=dto.get("title"),
+            description=dto.get("description"),
         )
 
     def to_command_argument_declaration(

src/enapter_mcp_server/mcp/models/__init__.py

@@ -5,6 +5,7 @@
 from .blueprint_section import BlueprintSection
 from .blueprint_summary import BlueprintSummary
 from .command_argument_declaration import CommandArgumentDeclaration
+from .command_confirmation import CommandConfirmation
 from .command_declaration import CommandDeclaration
 from .command_execution import CommandExecution
 from .command_execution_state import CommandExecutionState
@@ -32,6 +33,7 @@
     "BlueprintSection",
     "BlueprintSummary",
     "CommandArgumentDeclaration",
+    "CommandConfirmation",
     "CommandDeclaration",
     "CommandExecution",
     "CommandExecutionState",

src/enapter_mcp_server/mcp/models/command_confirmation.py

@@ -0,0 +1,21 @@
+from typing import Self
+
+import pydantic
+
+from enapter_mcp_server import domain
+
+
+class CommandConfirmation(pydantic.BaseModel):
+    """Vendor-declared confirmation block for a consequential command."""
+
+    severity: str | None = None
+    title: str | None = None
+    description: str | None = None
+
+    @classmethod
+    def from_domain(cls, confirmation: domain.CommandConfirmation) -> Self:
+        return cls(
+            severity=confirmation.severity,
+            title=confirmation.title,
+            description=confirmation.description,
+        )

src/enapter_mcp_server/mcp/models/command_declaration.py

@@ -6,14 +6,15 @@
 
 from .access_role import AccessRole
 from .command_argument_declaration import CommandArgumentDeclaration
+from .command_confirmation import CommandConfirmation
 
 
 class CommandDeclaration(pydantic.BaseModel):
     """A declaration of a device command.
 
-    The `access_level` field defines the minimum role required to execute
-    this command. A user can execute this command only if their
-    `authorized_role` for the device is at or after this `access_level`.
+    The `access_level` field defines the minimum role required to execute this command. A user can execute this command only if their `authorized_role` for the device is at or after this `access_level`.
+
+    The `confirmation` field, when present, marks the command as consequential per the vendor's declaration.
     """
 
     name: str
@@ -22,6 +23,7 @@ class CommandDeclaration(pydantic.BaseModel):
     description: str | None
     arguments: list[CommandArgumentDeclaration]
     implements: list[str]
+    confirmation: CommandConfirmation | None = None
 
     @classmethod
     def from_domain(cls, declaration: domain.CommandDeclaration) -> Self:
@@ -34,4 +36,9 @@ def from_domain(cls, declaration: domain.CommandDeclaration) -> Self:
                 CommandArgumentDeclaration.from_domain(a) for a in declaration.arguments
             ],
             implements=declaration.implements,
+            confirmation=(
+                CommandConfirmation.from_domain(declaration.confirmation)
+                if declaration.confirmation is not None
+                else None
+            ),
         )