From 55052615aaaa5ae3b637b0bf39bf7b7d9713254d Mon Sep 17 00:00:00 2001 From: caballeto Date: Wed, 6 May 2026 12:16:58 +0200 Subject: [PATCH 1/2] feat: add maintenance_window MCP tools MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Five new tools so AI coding assistants can suppress alerts around risky operations: - list_maintenance_windows — inspect active / upcoming windows - get_maintenance_window — fetch one window with full details - create_maintenance_window — schedule downtime BEFORE a deploy - update_maintenance_window — extend ``endsAt`` when a deploy runs long - cancel_maintenance_window — clear suppression after success The tools wrap ``/api/v1/maintenance-windows`` directly through the SDK's low-level helpers because the parallel ``client.maintenance_windows`` SDK PR hasn't shipped yet. Once it does, every tool body collapses to a one-liner against the resource without changing the public tool surface. Bumps the locked ``devhelm`` SDK to 0.6.3 to pick up the generated ``CreateMaintenanceWindowRequest`` / ``MaintenanceWindowDto`` / ``UpdateMaintenanceWindowRequest`` Pydantic models. The pyproject pin (``devhelm>=0.6.0``) is unchanged; this is a lock-only bump. Co-authored-by: Cursor --- src/devhelm_mcp/server.py | 7 +- src/devhelm_mcp/tools/maintenance_windows.py | 266 +++++++++++++++++++ uv.lock | 6 +- 3 files changed, 274 insertions(+), 5 deletions(-) create mode 100644 src/devhelm_mcp/tools/maintenance_windows.py diff --git a/src/devhelm_mcp/server.py b/src/devhelm_mcp/server.py index 87a1cea..aae51cd 100644 --- a/src/devhelm_mcp/server.py +++ b/src/devhelm_mcp/server.py @@ -35,6 +35,7 @@ environments, forensics, incidents, + maintenance_windows, monitors, notification_policies, resource_groups, @@ -76,8 +77,9 @@ def _package_version() -> str: "DevHelm MCP server for monitoring infrastructure. " "Use these tools to manage uptime monitors, incidents, alert channels, " "notification policies, environments, secrets, tags, resource groups, " - "webhooks, API keys, service dependencies, deploy locks, status pages, " - "and view dashboard status. All operations require a valid DevHelm API token." + "webhooks, API keys, service dependencies, deploy locks, maintenance " + "windows, status pages, and view dashboard status. All operations " + "require a valid DevHelm API token." ), ) @@ -95,6 +97,7 @@ def _package_version() -> str: api_keys, dependencies, deploy_lock, + maintenance_windows, status, status_pages, ] diff --git a/src/devhelm_mcp/tools/maintenance_windows.py b/src/devhelm_mcp/tools/maintenance_windows.py new file mode 100644 index 0000000..90e01a7 --- /dev/null +++ b/src/devhelm_mcp/tools/maintenance_windows.py @@ -0,0 +1,266 @@ +"""Maintenance window tools — schedule downtime so deploys don't page on-call. + +These tools let an AI agent proactively suppress alerts before running a +risky operation (a database migration, a deploy, a third-party API +maintenance) and then clear the suppression once the operation succeeds. +The flow that ships value to users looks like: + + 1. agent calls ``create_maintenance_window(...)`` before kicking off + a deploy script + 2. the deploy runs (monitors may briefly fail — alerts stay silent) + 3. on success the agent calls ``cancel_maintenance_window(...)`` so + new failures page on-call again + 4. on a runaway deploy the agent calls ``update_maintenance_window`` + to push the end time back instead of letting the window expire + and pages flood the channel + +The underlying ``client.maintenance_windows`` resource is being added to +the Python SDK in a parallel PR. Until that lands and a new SDK release +is published, the tools below call the generated request/response models +through the SDK's existing low-level HTTP helpers — same wire contract, +same telemetry headers, same auth resolution. Once +``client.maintenance_windows.create(...)`` ships we can swap the bodies +of these tools to use it; the public tool surface stays unchanged. +""" + +from __future__ import annotations + +from typing import Any + +import httpx +from devhelm import DevhelmError + +# The maintenance-window models live in ``devhelm._generated`` — they +# haven't been re-exported through ``devhelm.types`` yet (that's part of +# the SDK PR that adds ``client.maintenance_windows``). Reaching into +# ``_generated`` is the documented workaround for surfaces that need a +# resource ahead of its SDK release. Once the SDK ships the public +# re-exports, switch these imports to ``from devhelm.types import ...``. +from devhelm._generated import ( + CreateMaintenanceWindowRequest, + MaintenanceWindowDto, + UpdateMaintenanceWindowRequest, +) +from devhelm._http import api_delete, api_get, api_post, api_put, path_param +from devhelm._validation import parse_list, parse_single +from fastmcp import FastMCP + +from devhelm_mcp.client import ( + ToolResult, + get_client, + raise_tool_error, + serialize, +) + +_BASE_PATH = "/api/v1/maintenance-windows" + + +def _http_client(api_token: str | None) -> httpx.Client: + """Resolve the SDK's configured ``httpx.Client``. + + ``Devhelm.__init__`` builds exactly one ``httpx.Client`` (auth + header, tenant headers, telemetry headers, base URL, timeout) and + shares it across every resource. Reaching into + ``client.monitors._client`` reuses that same instance instead of + duplicating the env-resolution + header-building logic on the MCP + side. This is a deliberate, narrow workaround until the SDK ships + a public ``client.maintenance_windows`` resource — at which point + every tool here collapses to a one-liner against that resource. + """ + return get_client(api_token).monitors._client + + +def register(mcp: FastMCP) -> None: + @mcp.tool() + def list_maintenance_windows( + monitor_id: str | None = None, + status: str | None = None, + api_token: str | None = None, + ) -> ToolResult: + """List maintenance windows for the workspace. + + Use this BEFORE creating a new window to check whether someone + else (or an earlier agent run) already scheduled overlap, or + AFTER a deploy to confirm the window you opened is still + active. + + Filters (all optional; combine freely): + - ``monitor_id``: UUID of a monitor — only windows attached + to that single monitor (org-wide windows are excluded). + - ``status``: ``"active"`` for windows currently in + progress, or ``"upcoming"`` for windows scheduled in the + future. Past / cancelled windows are not returned by the + API today; omit ``status`` for the broadest result. + """ + try: + params: dict[str, Any] = {} + if monitor_id: + params["monitorId"] = monitor_id + if status: + params["filter"] = status + data = api_get( + _http_client(api_token), + _BASE_PATH, + params=params or None, + ) + # The API returns a ``TableValueResult`` + # envelope (``{data: [...], hasNext, hasPrev, …}``). The list + # is small enough in practice (per-org, active + upcoming + # only) that the controller doesn't paginate today, so we + # surface the ``data`` array directly. If the controller + # ever starts honouring ``page``/``size``, swap this for + # ``fetch_all_pages`` once the SDK ships + # ``client.maintenance_windows``. + items = data.get("data", []) if isinstance(data, dict) else (data or []) + windows = parse_list(MaintenanceWindowDto, items, f"GET {_BASE_PATH}") + return serialize(windows) + except DevhelmError as e: + raise_tool_error(e) + + @mcp.tool() + def get_maintenance_window( + window_id: str, api_token: str | None = None + ) -> ToolResult: + """Get a single maintenance window by ID with full details.""" + try: + data = api_get( + _http_client(api_token), + f"{_BASE_PATH}/{path_param(window_id)}", + ) + return serialize( + parse_single( + MaintenanceWindowDto, + data, + f"GET {_BASE_PATH}/{window_id}", + ) + ) + except DevhelmError as e: + raise_tool_error(e) + + @mcp.tool() + def create_maintenance_window( + body: CreateMaintenanceWindowRequest, api_token: str | None = None + ) -> ToolResult: + """Schedule a maintenance window to suppress alerts during planned work. + + Call this BEFORE running an operation that may legitimately + cause monitors to fail — a deploy, a database migration, a + third-party service's announced downtime — so the on-call + rotation isn't paged for known-expected failures. Always + pair every successful create with a follow-up + ``cancel_maintenance_window`` once the operation finishes; + if the operation runs long, call ``update_maintenance_window`` + to push the end time back rather than letting the window + lapse early. + + Time fields use ISO 8601 / RFC 3339 timestamps with explicit + timezone — UTC strongly preferred. Example: + ``"2026-05-15T14:00:00Z"``. Naive timestamps (no timezone) + are rejected by the API. + + Body fields: + - ``startsAt`` (required): when the window opens. + - ``endsAt`` (required): when the window closes; must be + strictly after ``startsAt``. + - ``monitorId`` (optional): UUID of a single monitor to + scope the window to. Omit (or set null) to make this an + **org-wide window** that suppresses alerts on every + monitor in the workspace — the right choice for a deploy + or migration that touches the whole platform. + - ``reason`` (optional): human-readable explanation + ("v0.7.3 deploy", "Postgres major upgrade"). Surfaces in + the dashboard and on-call channel; keep it specific. + - ``repeatRule`` (optional): iCal RRULE string for + recurring windows (max 100 chars), e.g. + ``FREQ=WEEKLY;BYDAY=SU`` for weekly Sunday maintenance. + Omit for one-time windows. + - ``suppressAlerts`` (optional): whether the window + actually silences alerts. Default ``true``; set + ``false`` to record a maintenance window for audit + without changing alerting behavior. + """ + try: + # Pass the validated Pydantic model straight to ``api_post``; + # the SDK's ``_serialize_body`` does ``model_dump(mode="json", + # by_alias=True, exclude_none=True)`` which is the canonical + # camelCase / RFC-3339 wire shape. Dicts are intentionally + # rejected by ``api_post`` (P5: no raw-dict request bodies), + # so handing it the model also catches a future regression + # where ``body`` is downgraded to ``dict[str, Any]``. + data = api_post(_http_client(api_token), _BASE_PATH, body) + return serialize( + parse_single( + MaintenanceWindowDto, + data, + f"POST {_BASE_PATH}", + ) + ) + except DevhelmError as e: + raise_tool_error(e) + + @mcp.tool() + def update_maintenance_window( + window_id: str, + body: UpdateMaintenanceWindowRequest, + api_token: str | None = None, + ) -> ToolResult: + """Update an in-flight or scheduled maintenance window. + + The most common use is **extending** an active window when a + deploy runs longer than expected — call this with the new + ``endsAt`` to keep alerts suppressed past the original + deadline. The endpoint is a full replacement (PUT, not + PATCH): pass the complete intended state, not a delta. + Any field omitted falls back to the underlying model's + default rather than preserving the existing value. + + Time fields use ISO 8601 / RFC 3339 timestamps with explicit + timezone (UTC preferred), e.g. ``"2026-05-15T16:30:00Z"``. + + Body fields (same schema as create): + - ``startsAt`` (required) + - ``endsAt`` (required) + - ``monitorId`` (optional; null = org-wide) + - ``reason`` (optional; null clears) + - ``repeatRule`` (optional; null clears the recurrence) + - ``suppressAlerts`` (optional) + """ + try: + data = api_put( + _http_client(api_token), + f"{_BASE_PATH}/{path_param(window_id)}", + body, + ) + return serialize( + parse_single( + MaintenanceWindowDto, + data, + f"PUT {_BASE_PATH}/{window_id}", + ) + ) + except DevhelmError as e: + raise_tool_error(e) + + @mcp.tool() + def cancel_maintenance_window(window_id: str, api_token: str | None = None) -> str: + """Cancel a maintenance window — alerts resume immediately. + + Call this AFTER a deploy or maintenance operation completes + successfully so any new monitor failures surface as real + incidents instead of being silently absorbed. If the window + was scheduled but not yet started, this prevents it from + ever opening. + + The window record is removed; the audit log preserves the + historical fact that the window existed. There is no + "uncancel" — schedule a new window if you need to restore + suppression. + """ + try: + api_delete( + _http_client(api_token), + f"{_BASE_PATH}/{path_param(window_id)}", + ) + return "Maintenance window cancelled." + except DevhelmError as e: + raise_tool_error(e) diff --git a/uv.lock b/uv.lock index 825c837..648dd9f 100644 --- a/uv.lock +++ b/uv.lock @@ -388,15 +388,15 @@ wheels = [ [[package]] name = "devhelm" -version = "0.6.1" +version = "0.6.3" source = { registry = "https://pypi.org/simple" } dependencies = [ { name = "httpx" }, { name = "pydantic", extra = ["email"] }, ] -sdist = { url = "https://files.pythonhosted.org/packages/68/97/8b65fac3e60668fda443cac606f5f87b2c2cd3df72795d8a0d819f68d18b/devhelm-0.6.1.tar.gz", hash = "sha256:a5cfd34fb71512a17823df5082cf2ba5ddf96431a0e80b8cdb7be2e47be5e9d9", size = 233744, upload-time = "2026-05-05T17:56:43.576Z" } +sdist = { url = "https://files.pythonhosted.org/packages/24/24/fe798fa4e7d610a1ec2ef5cce55ddb2c5c05a1540389272711c571e6cbab/devhelm-0.6.3.tar.gz", hash = "sha256:78dafbff94b8b6fa17b9750de22a1af46b78d2f62081e190e2dc74b8e4bc0d1b", size = 239367, upload-time = "2026-05-06T09:45:23.356Z" } wheels = [ - { url = "https://files.pythonhosted.org/packages/15/ad/dc8105f5b165f15a51603dd353af613ded1be342d3774d686f3a292ba4fe/devhelm-0.6.1-py3-none-any.whl", hash = "sha256:97fdd1165ca8c2491c10c6e67b6165b3ab743f7fb78b1c2cb3470b24a5a584c6", size = 73318, upload-time = "2026-05-05T17:56:42.031Z" }, + { url = "https://files.pythonhosted.org/packages/a5/ec/466185ea760970790208bb89a76e37fecde05fa98ef53f6e8a7976db5df7/devhelm-0.6.3-py3-none-any.whl", hash = "sha256:e4f900a940b4d9602dc65f67c5f9c08ddf74cfe32fa969213e533e034750e0be", size = 75053, upload-time = "2026-05-06T09:45:22.037Z" }, ] [[package]] From d9280af4c42d9b6879efdc01a449f99f8cefc52f Mon Sep 17 00:00:00 2001 From: caballeto Date: Wed, 6 May 2026 12:17:10 +0200 Subject: [PATCH 2/2] test: pin maintenance_window tool schema and HTTP wire contract MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Adds 25 tests across five concerns: - registration: every new tool surfaces in ``mcp.list_tools`` with a non-empty description (LLM docs) - HTTP contract: each tool builds the right path / method / body via patched ``api_get`` / ``api_post`` / ``api_put`` / ``api_delete``, including the camelCase aliases on the wire (``startsAt``, ``endsAt``, ``monitorId``) and the ``filter`` / ``monitorId`` query keys for the list endpoint - schema hygiene: regression that ``api_token`` never leaks into the LLM-facing ``inputSchema`` (P2.Bug7 contract) and that the ``CreateMaintenanceWindowRequest`` / ``UpdateMaintenanceWindowRequest`` body schemas don't expose ``managedBy`` — pinning now means a future SDK regen that bolts the field on can't silently surface it - error surfacing: upstream ``DevhelmApiError`` propagates to the client as ``isError=True`` with the formatted ApiError envelope (P1.Bug3 contract) - expected-tools list: keeps ``test_tools.py`` in sync so the count assertion reflects the new five tools Total suite is now 135 tests (110 baseline + 25 new), all green on Python 3.11 and 3.13. Co-authored-by: Cursor --- tests/test_maintenance_windows.py | 492 ++++++++++++++++++++++++++++++ tests/test_tools.py | 5 + 2 files changed, 497 insertions(+) create mode 100644 tests/test_maintenance_windows.py diff --git a/tests/test_maintenance_windows.py b/tests/test_maintenance_windows.py new file mode 100644 index 0000000..0c5f472 --- /dev/null +++ b/tests/test_maintenance_windows.py @@ -0,0 +1,492 @@ +"""Tests for maintenance window MCP tools. + +Asserts each tool builds the right HTTP request to the right path, that +the create/update tools forward Pydantic-validated bodies, and that the +LLM-facing input schema does not leak ``api_token`` or ``managedBy``. + +The maintenance-window resource doesn't ship in the pinned ``devhelm`` +SDK release yet (the parallel SDK PR adds ``client.maintenance_windows`` +later); the tools here call the SDK's low-level ``api_get`` / ``api_post`` +/ ``api_put`` / ``api_delete`` helpers directly. To exercise them +without booting the API, each test patches those helpers in the tool +module's namespace and captures the call args. +""" + +from __future__ import annotations + +import asyncio +from typing import Any +from unittest.mock import MagicMock, patch + +import pytest +from devhelm import DevhelmApiError + +from devhelm_mcp.server import _strip_internal_schema_fields, mcp + +RegisteredTools = dict[str, Any] + + +@pytest.fixture(scope="module") +def registered_tools() -> RegisteredTools: + asyncio.run(_strip_internal_schema_fields()) + tools = asyncio.run(mcp.list_tools()) + return {t.name: t for t in tools} + + +# Sample DTO shape returned by the API (camelCase, MaintenanceWindowDto). +_SAMPLE_WINDOW: dict[str, Any] = { + "id": "11111111-1111-1111-1111-111111111111", + "monitorId": None, + "organizationId": 1, + "startsAt": "2026-05-15T14:00:00Z", + "endsAt": "2026-05-15T15:00:00Z", + "repeatRule": None, + "reason": "deploy v0.7.3", + "suppressAlerts": True, + "createdAt": "2026-05-06T10:00:00Z", +} + +_SAMPLE_ENVELOPE_SINGLE: dict[str, Any] = {"data": _SAMPLE_WINDOW} + +_SAMPLE_ENVELOPE_LIST: dict[str, Any] = { + "data": [_SAMPLE_WINDOW], + "hasNext": False, + "hasPrev": False, + "totalElements": 1, + "totalPages": 1, +} + + +# --------------------------------------------------------------------------- # +# Tool registration +# --------------------------------------------------------------------------- # + + +class TestMaintenanceWindowToolsRegistered: + """Every maintenance-window tool surfaces in the FastMCP registry.""" + + @pytest.mark.parametrize( + "name", + [ + "list_maintenance_windows", + "get_maintenance_window", + "create_maintenance_window", + "update_maintenance_window", + "cancel_maintenance_window", + ], + ) + def test_tool_registered( + self, registered_tools: RegisteredTools, name: str + ) -> None: + assert name in registered_tools, f"Missing tool: {name}" + assert registered_tools[name].description, ( + f"{name} must have a non-empty description (LLM docs)" + ) + + +# --------------------------------------------------------------------------- # +# HTTP wire contract — verifies path / method / body for each tool +# --------------------------------------------------------------------------- # + + +def _call_tool(tool_name: str, arguments: dict[str, Any]) -> Any: + return asyncio.run(mcp.call_tool(tool_name, arguments)) + + +def _stub_sdk_client() -> MagicMock: + """Return a minimal SDK-shaped stub for ``_http_client`` to walk. + + ``_http_client`` calls ``get_client(...).monitors._client`` to + reuse the SDK's configured ``httpx.Client``. The tests patch the + api_get/post/put/delete helpers separately and never hit a real + network, so any object with the right attribute chain works.""" + mock = MagicMock() + mock.monitors._client = MagicMock(name="httpx.Client") + return mock + + +@pytest.fixture(autouse=True) +def _stub_get_client(request: pytest.FixtureRequest) -> Any: + """Auto-patch ``get_client`` for tests that exercise the wire calls. + + The schema-hygiene and registration tests don't need this — they + introspect the ``inputSchema`` only — so the fixture is a no-op + when a test class doesn't ask for it. Tests opt in by living + under one of the ``HttpContract`` classes (which also patch the + api_* helpers); other classes get the real ``get_client`` (which + in turn raises ``DevhelmAuthError`` if no token is configured — + that's fine because they never call a tool).""" + classname = request.node.cls.__name__ if request.node.cls else "" + if "HttpContract" not in classname: + yield None + return + with patch( + "devhelm_mcp.tools.maintenance_windows.get_client", + return_value=_stub_sdk_client(), + ) as p: + yield p + + +class TestListMaintenanceWindowsHttpContract: + """``list_maintenance_windows`` GETs ``/api/v1/maintenance-windows``.""" + + def test_no_filters_omits_query_params(self) -> None: + captured: dict[str, Any] = {} + + def fake_api_get( + client: Any, path: str, params: dict[str, Any] | None = None + ) -> dict[str, Any]: + captured.setdefault("calls", []).append((path, params)) + return _SAMPLE_ENVELOPE_LIST + + with patch( + "devhelm_mcp.tools.maintenance_windows.api_get", + side_effect=fake_api_get, + ): + _call_tool("list_maintenance_windows", {}) + + assert captured["calls"], "expected at least one GET" + path, params = captured["calls"][0] + assert path == "/api/v1/maintenance-windows" + # The list endpoint accepts only ``monitorId`` and ``filter`` + # query params; when the LLM passes neither we send no params + # so the API can't reject on an unexpected key. + assert params is None or params == {} + + def test_monitor_id_and_status_flow_through_as_query_params(self) -> None: + captured: dict[str, Any] = {} + + def fake_api_get( + client: Any, path: str, params: dict[str, Any] | None = None + ) -> dict[str, Any]: + captured["params"] = dict(params or {}) + return _SAMPLE_ENVELOPE_LIST + + with patch( + "devhelm_mcp.tools.maintenance_windows.api_get", + side_effect=fake_api_get, + ): + _call_tool( + "list_maintenance_windows", + { + "monitor_id": "22222222-2222-2222-2222-222222222222", + "status": "active", + }, + ) + + assert captured["params"]["monitorId"] == ( + "22222222-2222-2222-2222-222222222222" + ) + # API contract: query key is ``filter``, value is ``active`` / + # ``upcoming`` (lowercase). Pinning here so a future docstring + # rewrite that nudges the LLM toward UPPERCASE values doesn't + # break the wire contract. + assert captured["params"]["filter"] == "active" + + +class TestGetMaintenanceWindowHttpContract: + def test_uses_correct_path_with_url_encoded_id(self) -> None: + captured: dict[str, Any] = {} + + def fake_api_get( + client: Any, path: str, params: dict[str, Any] | None = None + ) -> dict[str, Any]: + captured["path"] = path + return _SAMPLE_ENVELOPE_SINGLE + + with patch( + "devhelm_mcp.tools.maintenance_windows.api_get", + side_effect=fake_api_get, + ): + _call_tool( + "get_maintenance_window", + {"window_id": "11111111-1111-1111-1111-111111111111"}, + ) + + assert ( + captured["path"] + == "/api/v1/maintenance-windows/11111111-1111-1111-1111-111111111111" + ) + + +class TestCreateMaintenanceWindowHttpContract: + """``create_maintenance_window`` POSTs to ``/api/v1/maintenance-windows`` + with a Pydantic-validated ``CreateMaintenanceWindowRequest`` body. + + The body parameters use camelCase aliases (``startsAt``, ``endsAt``, + ``monitorId``, …) on the wire. Verifying them pins the OpenAPI + contract regardless of whether the SDK's internal Python field + names are snake_case or camelCase. + """ + + @staticmethod + def _wire_body(captured_body: Any) -> dict[str, Any]: + """Render a captured request body the way ``httpx`` would. + + ``api_post`` accepts a Pydantic model and the SDK's + ``_serialize_body`` calls ``model_dump(mode="json", + by_alias=True, exclude_none=True)`` to produce the wire + bytes. Replicate that here so the test asserts the exact + camelCase / RFC-3339 shape the API will receive. + """ + from pydantic import BaseModel + + assert isinstance(captured_body, BaseModel), ( + "create/update tools must hand a validated Pydantic model " + "to api_post, not a raw dict (raw dicts get rejected by " + "the SDK's _serialize_body)." + ) + return captured_body.model_dump(mode="json", by_alias=True, exclude_none=True) + + def test_body_serialised_with_camelcase_aliases(self) -> None: + captured: dict[str, Any] = {} + + def fake_api_post(client: Any, path: str, body: Any) -> dict[str, Any]: + captured["path"] = path + captured["body"] = body + return _SAMPLE_ENVELOPE_SINGLE + + with patch( + "devhelm_mcp.tools.maintenance_windows.api_post", + side_effect=fake_api_post, + ): + _call_tool( + "create_maintenance_window", + { + "body": { + "startsAt": "2026-05-15T14:00:00Z", + "endsAt": "2026-05-15T15:00:00Z", + "monitorId": "22222222-2222-2222-2222-222222222222", + "reason": "deploy v0.7.3", + "suppressAlerts": True, + }, + }, + ) + + assert captured["path"] == "/api/v1/maintenance-windows" + wire = self._wire_body(captured["body"]) + assert wire["startsAt"] == "2026-05-15T14:00:00Z" + assert wire["endsAt"] == "2026-05-15T15:00:00Z" + assert wire["monitorId"] == "22222222-2222-2222-2222-222222222222" + assert wire["reason"] == "deploy v0.7.3" + assert wire["suppressAlerts"] is True + + def test_org_wide_window_omits_monitor_id_when_null(self) -> None: + # ``monitorId=null`` in the input is the documented "org-wide" + # marker, but the SDK serialises with ``exclude_none=True`` so + # the wire body must drop the key entirely. The API treats + # absent and null identically, so this is purely a wire-shape + # invariant we want to keep stable. + captured: dict[str, Any] = {} + + def fake_api_post(client: Any, path: str, body: Any) -> dict[str, Any]: + captured["body"] = body + return _SAMPLE_ENVELOPE_SINGLE + + with patch( + "devhelm_mcp.tools.maintenance_windows.api_post", + side_effect=fake_api_post, + ): + _call_tool( + "create_maintenance_window", + { + "body": { + "startsAt": "2026-05-15T14:00:00Z", + "endsAt": "2026-05-15T15:00:00Z", + }, + }, + ) + + wire = self._wire_body(captured["body"]) + assert "monitorId" not in wire + + +class TestUpdateMaintenanceWindowHttpContract: + def test_puts_to_window_path_with_full_body(self) -> None: + from pydantic import BaseModel + + captured: dict[str, Any] = {} + + def fake_api_put(client: Any, path: str, body: Any) -> dict[str, Any]: + captured["path"] = path + captured["body"] = body + return _SAMPLE_ENVELOPE_SINGLE + + with patch( + "devhelm_mcp.tools.maintenance_windows.api_put", + side_effect=fake_api_put, + ): + _call_tool( + "update_maintenance_window", + { + "window_id": "11111111-1111-1111-1111-111111111111", + "body": { + "startsAt": "2026-05-15T14:00:00Z", + "endsAt": "2026-05-15T16:30:00Z", + "reason": "deploy still running", + }, + }, + ) + + assert ( + captured["path"] + == "/api/v1/maintenance-windows/11111111-1111-1111-1111-111111111111" + ) + assert isinstance(captured["body"], BaseModel) + wire = captured["body"].model_dump( + mode="json", by_alias=True, exclude_none=True + ) + assert wire["endsAt"] == "2026-05-15T16:30:00Z" + assert wire["reason"] == "deploy still running" + + +class TestCancelMaintenanceWindowHttpContract: + def test_deletes_at_window_path_and_returns_friendly_string(self) -> None: + captured: dict[str, Any] = {} + + def fake_api_delete(client: Any, path: str) -> None: + captured["path"] = path + + with patch( + "devhelm_mcp.tools.maintenance_windows.api_delete", + side_effect=fake_api_delete, + ): + result = _call_tool( + "cancel_maintenance_window", + {"window_id": "11111111-1111-1111-1111-111111111111"}, + ) + + assert ( + captured["path"] + == "/api/v1/maintenance-windows/11111111-1111-1111-1111-111111111111" + ) + # The tool returns a plain string for the LLM to echo back to + # the user. ``mcp.call_tool`` wraps that as a TextContent block. + rendered = result.content[0].text if result.content else "" + assert "cancelled" in rendered.lower() + + +# --------------------------------------------------------------------------- # +# Schema hygiene — input schema must not leak api_token or managedBy +# --------------------------------------------------------------------------- # + + +_MAINTENANCE_TOOLS = [ + "list_maintenance_windows", + "get_maintenance_window", + "create_maintenance_window", + "update_maintenance_window", + "cancel_maintenance_window", +] + + +def _body_schema_for(tools: RegisteredTools, name: str) -> dict[str, Any] | None: + """Resolve a tool's body sub-schema if it has one (create / update only).""" + params = tools[name].parameters + body = params.get("properties", {}).get("body") + if not isinstance(body, dict): + return None + if "$ref" in body: + ref_name = body["$ref"].rsplit("/", 1)[-1] + return params.get("$defs", {}).get(ref_name) # type: ignore[no-any-return] + return body + + +class TestMaintenanceWindowSchemaHygiene: + """``api_token`` is auto-resolved from header / env and must never + appear in the tool's JSON Schema (P2.Bug7 — leaks token into the + LLM's prompt context). ``managedBy`` is not part of the + maintenance-window API today, but pinning the assertion now means + a future SDK regen that bolts the field on can't silently expose + it to the LLM.""" + + @pytest.mark.parametrize("name", _MAINTENANCE_TOOLS) + def test_api_token_not_in_properties( + self, registered_tools: RegisteredTools, name: str + ) -> None: + properties = registered_tools[name].parameters.get("properties", {}) + assert "api_token" not in properties, ( + f"{name} leaks api_token into the LLM-facing input schema" + ) + + @pytest.mark.parametrize("name", _MAINTENANCE_TOOLS) + def test_api_token_not_required( + self, registered_tools: RegisteredTools, name: str + ) -> None: + required = registered_tools[name].parameters.get("required", []) + assert "api_token" not in required, ( + f"{name} requires api_token — must resolve from header / env" + ) + + @pytest.mark.parametrize( + "name", ["create_maintenance_window", "update_maintenance_window"] + ) + def test_body_schema_does_not_expose_managed_by( + self, registered_tools: RegisteredTools, name: str + ) -> None: + schema = _body_schema_for(registered_tools, name) + assert schema is not None, f"{name} has no body schema to inspect" + # Belt-and-suspenders: neither in properties nor in required. + properties = schema.get("properties", {}) if isinstance(schema, dict) else {} + required = schema.get("required", []) if isinstance(schema, dict) else [] + assert "managedBy" not in properties, ( + f"{name} body schema must not expose managedBy to the LLM" + ) + assert "managedBy" not in required, ( + f"{name} body schema must not require managedBy" + ) + + +# --------------------------------------------------------------------------- # +# Error surfacing — upstream failures must set isError=true +# --------------------------------------------------------------------------- # + + +class TestMaintenanceWindowErrorsBubbleUp: + """If the API rejects the request, the tool must surface the failure + via ``ToolError`` so FastMCP returns ``isError=True`` (P1.Bug3 + semantics — silent-success on 4xx is a known footgun for AI agents).""" + + def test_create_propagates_api_error(self) -> None: + async def go() -> Any: + from fastmcp import Client + + async with Client(mcp) as client: + return await client.call_tool( + "create_maintenance_window", + { + "body": { + "startsAt": "2026-05-15T14:00:00Z", + "endsAt": "2026-05-15T13:00:00Z", + }, + }, + raise_on_error=False, + ) + + # Patch the underlying httpx client lookup so we can inject the + # SDK error without mocking ``api_post`` (this exercises the + # full error-formatting path through the SDK's helpers). + mock_client = MagicMock() + mock_client.monitors._client = MagicMock() + + def boom(*args: Any, **kwargs: Any) -> None: + raise DevhelmApiError( + "endsAt must be after startsAt", + status=400, + code="VALIDATION_FAILED", + request_id="req_abc", + ) + + with ( + patch( + "devhelm_mcp.tools.maintenance_windows.get_client", + return_value=mock_client, + ), + patch("devhelm_mcp.tools.maintenance_windows.api_post", side_effect=boom), + ): + result = asyncio.run(go()) + + assert result.is_error is True + text = result.content[0].text + assert "ApiError (400 VALIDATION_FAILED)" in text + assert "request_id=req_abc" in text diff --git a/tests/test_tools.py b/tests/test_tools.py index 4e53a02..9149293 100644 --- a/tests/test_tools.py +++ b/tests/test_tools.py @@ -86,6 +86,11 @@ "get_current_deploy_lock", "release_deploy_lock", "force_release_deploy_lock", + "list_maintenance_windows", + "get_maintenance_window", + "create_maintenance_window", + "update_maintenance_window", + "cancel_maintenance_window", "get_status_overview", "list_status_pages", "get_status_page",