Merge pull request #11 from WebexCommunity/claude/health-check-tool-KJCkn

Author: adamweeks

src/webex_bot_mcp/main.py

@@ -53,6 +53,8 @@
     update_webex_team, delete_webex_team,
     list_webex_team_memberships, add_webex_team_membership,
     delete_webex_team_membership,
+    # Diagnostic tools
+    webex_health_check,
 )
 
 # Import version and error handling from common
@@ -135,6 +137,9 @@ async def dispatch(self, request, call_next):
 mcp.tool()(add_webex_team_membership)
 mcp.tool()(delete_webex_team_membership)
 
+# Diagnostic tools
+mcp.tool()(webex_health_check)
+
 
 # ========== RESOURCES ==========
 # Resources provide contextual information that AI assistants can read
@@ -1117,7 +1122,7 @@ def server_version():
             "streamable-http", "stdio",
             "error-handling", "versioning"
         ],
-        "tools_count": 34,
+        "tools_count": 35,
         "resources_count": 11,
         "prompts_count": 8,
         "breaking_changes": {

src/webex_bot_mcp/tools/__init__.py

@@ -33,6 +33,10 @@
     delete_webex_team_membership,
 )
 
+from .diagnostics import (
+    webex_health_check,
+)
+
 __all__ = [
     # Room functions
     'list_webex_rooms', 'create_webex_room', 'update_webex_room',
@@ -60,4 +64,6 @@
     'update_webex_team', 'delete_webex_team',
     'list_webex_team_memberships', 'add_webex_team_membership',
     'delete_webex_team_membership',
+    # Diagnostic tools
+    'webex_health_check',
 ]

src/webex_bot_mcp/tools/diagnostics.py

@@ -0,0 +1,129 @@
+"""
+Diagnostic tools for Webex Bot MCP Server.
+"""
+import sys
+import time
+from typing import Any, Dict
+
+from .common import (
+    MCP_SERVER_VERSION,
+    WebexErrorCodes,
+    WebexTokenMissingError,
+    create_success_response,
+    get_webex_api,
+)
+
+
+def _check_token() -> Dict[str, Any]:
+    start = time.time()
+    try:
+        me = get_webex_api().people.me()
+        return {
+            "status": "ok",
+            "bot_id": me.id,
+            "bot_name": me.displayName,
+            "bot_email": me.emails[0] if me.emails else None,
+            "org_id": me.orgId,
+            "duration_ms": round((time.time() - start) * 1000),
+        }
+    except WebexTokenMissingError as e:
+        return {
+            "status": "error",
+            "error_code": WebexErrorCodes.UNAUTHORIZED,
+            "message": str(e),
+            "duration_ms": round((time.time() - start) * 1000),
+        }
+    except Exception as e:
+        err = str(e).lower()
+        if "unauthorized" in err or "invalid token" in err or "401" in err:
+            code = WebexErrorCodes.UNAUTHORIZED
+        elif "forbidden" in err or "403" in err:
+            code = WebexErrorCodes.FORBIDDEN
+        else:
+            code = WebexErrorCodes.WEBEX_API_ERROR
+        return {
+            "status": "error",
+            "error_code": code,
+            "message": str(e),
+            "duration_ms": round((time.time() - start) * 1000),
+        }
+
+
+def _check_rooms(max_rooms: int = 10) -> Dict[str, Any]:
+    start = time.time()
+    try:
+        rooms = list(get_webex_api().rooms.list(max=max_rooms))
+        room_types: Dict[str, int] = {}
+        for r in rooms:
+            rt = getattr(r, "type", "unknown")
+            room_types[rt] = room_types.get(rt, 0) + 1
+        return {
+            "status": "ok",
+            "total_visible": len(rooms),
+            "room_types": room_types,
+            "sample": [
+                {"id": r.id, "title": r.title, "type": getattr(r, "type", "unknown")}
+                for r in rooms[:3]
+            ],
+            "duration_ms": round((time.time() - start) * 1000),
+        }
+    except Exception as e:
+        return {
+            "status": "error",
+            "message": str(e),
+            "duration_ms": round((time.time() - start) * 1000),
+        }
+
+
+def webex_health_check(include_rooms: bool = True) -> Dict[str, Any]:
+    """
+    Validate Webex connectivity and return a structured diagnostic status.
+
+    Checks the bot token (identity + API reachability) and, optionally, lists
+    visible rooms to confirm room-access permissions. Designed for agents that
+    need to self-diagnose before starting a workflow.
+
+    Args:
+        include_rooms: When True (default), also verify room-list access.
+
+    Returns:
+        Standardized success response whose ``data`` contains:
+        - ``overall_status``: "healthy", "degraded", or "unhealthy"
+        - ``checks``: per-check result dicts keyed by check name
+        - ``duration_ms``: total wall-clock time for all checks
+        - ``server_version``: running MCP server version
+        - ``python_version``: Python interpreter version string
+
+        ``overall_status`` is "healthy" when all checks pass, "unhealthy" when
+        the token check fails, and "degraded" when the token is valid but a
+        secondary check (e.g. rooms) errors.
+    """
+    wall_start = time.time()
+
+    token_check = _check_token()
+    checks: Dict[str, Any] = {"token": token_check}
+
+    if include_rooms:
+        if token_check["status"] == "ok":
+            checks["rooms"] = _check_rooms()
+        else:
+            checks["rooms"] = {
+                "status": "skipped",
+                "message": "Token check failed; room check skipped.",
+            }
+
+    statuses = [c["status"] for c in checks.values()]
+    if any(s == "error" for s in statuses):
+        overall = "unhealthy" if token_check["status"] == "error" else "degraded"
+    else:
+        overall = "healthy"
+
+    return create_success_response(
+        data={
+            "overall_status": overall,
+            "checks": checks,
+            "duration_ms": round((time.time() - wall_start) * 1000),
+            "server_version": MCP_SERVER_VERSION,
+            "python_version": sys.version.split()[0],
+        }
+    )

tests/test_diagnostics.py

@@ -0,0 +1,211 @@
+"""
+Unit tests for webex_health_check tool.
+
+The webexpythonsdk and python-dotenv stubs are set up before any project
+module is imported so no real credentials or SDK installation are needed.
+"""
+import os
+import sys
+import types
+import unittest
+from unittest.mock import MagicMock, patch
+
+# ── Provide fake env var and SDK before any tools module is imported ──────────
+os.environ.setdefault("WEBEX_ACCESS_TOKEN", "test-token")
+
+_sdk_mod = types.ModuleType("webexpythonsdk")
+_sdk_mod.WebexAPI = MagicMock(return_value=MagicMock())
+sys.modules.setdefault("webexpythonsdk", _sdk_mod)
+
+if "dotenv" not in sys.modules:
+    _dotenv_mod = types.ModuleType("dotenv")
+    _dotenv_mod.load_dotenv = lambda *a, **kw: None
+    sys.modules["dotenv"] = _dotenv_mod
+
+from webex_bot_mcp.tools.diagnostics import webex_health_check  # noqa: E402
+from webex_bot_mcp.tools.common import WebexTokenMissingError    # noqa: E402
+import webex_bot_mcp.tools.diagnostics as _diag_mod             # noqa: E402
+
+
+# ── helpers ───────────────────────────────────────────────────────────────────
+
+def _fake_me(**overrides):
+    me = MagicMock()
+    me.id = "bot-id-123"
+    me.displayName = "Test Bot"
+    me.emails = ["testbot@example.com"]
+    me.orgId = "org-456"
+    for k, v in overrides.items():
+        setattr(me, k, v)
+    return me
+
+
+def _fake_room(room_id="R1", title="Room One", room_type="group"):
+    r = MagicMock()
+    r.id = room_id
+    r.title = title
+    r.type = room_type
+    return r
+
+
+class TestWebexHealthCheckHealthy(unittest.TestCase):
+    """All checks pass → overall_status healthy."""
+
+    def setUp(self):
+        self.mock_api = MagicMock()
+        self.mock_api.people.me.return_value = _fake_me()
+        self.mock_api.rooms.list.return_value = iter([
+            _fake_room("R1", "Alpha", "group"),
+            _fake_room("R2", "Beta", "direct"),
+        ])
+
+    def test_overall_status_healthy(self):
+        with patch.object(_diag_mod, "get_webex_api", return_value=self.mock_api):
+            result = webex_health_check()
+        self.assertTrue(result["success"])
+        self.assertEqual(result["data"]["overall_status"], "healthy")
+
+    def test_token_check_fields(self):
+        with patch.object(_diag_mod, "get_webex_api", return_value=self.mock_api):
+            result = webex_health_check()
+        token = result["data"]["checks"]["token"]
+        self.assertEqual(token["status"], "ok")
+        self.assertEqual(token["bot_id"], "bot-id-123")
+        self.assertEqual(token["bot_name"], "Test Bot")
+        self.assertEqual(token["bot_email"], "testbot@example.com")
+        self.assertIn("duration_ms", token)
+
+    def test_rooms_check_fields(self):
+        with patch.object(_diag_mod, "get_webex_api", return_value=self.mock_api):
+            result = webex_health_check()
+        rooms = result["data"]["checks"]["rooms"]
+        self.assertEqual(rooms["status"], "ok")
+        self.assertEqual(rooms["total_visible"], 2)
+        self.assertEqual(rooms["room_types"], {"group": 1, "direct": 1})
+        self.assertEqual(len(rooms["sample"]), 2)
+
+    def test_top_level_metadata(self):
+        with patch.object(_diag_mod, "get_webex_api", return_value=self.mock_api):
+            result = webex_health_check()
+        data = result["data"]
+        self.assertIn("duration_ms", data)
+        self.assertIn("server_version", data)
+        self.assertIn("python_version", data)
+        self.assertIn("timestamp", result)
+        self.assertIn("server_version", result)
+
+
+class TestWebexHealthCheckNoRooms(unittest.TestCase):
+    """include_rooms=False skips room check."""
+
+    def setUp(self):
+        self.mock_api = MagicMock()
+        self.mock_api.people.me.return_value = _fake_me()
+
+    def test_rooms_check_absent(self):
+        with patch.object(_diag_mod, "get_webex_api", return_value=self.mock_api):
+            result = webex_health_check(include_rooms=False)
+        self.assertNotIn("rooms", result["data"]["checks"])
+
+    def test_still_healthy(self):
+        with patch.object(_diag_mod, "get_webex_api", return_value=self.mock_api):
+            result = webex_health_check(include_rooms=False)
+        self.assertEqual(result["data"]["overall_status"], "healthy")
+
+    def test_rooms_list_not_called(self):
+        with patch.object(_diag_mod, "get_webex_api", return_value=self.mock_api):
+            webex_health_check(include_rooms=False)
+        self.mock_api.rooms.list.assert_not_called()
+
+
+class TestWebexHealthCheckTokenMissing(unittest.TestCase):
+    """WebexTokenMissingError → unhealthy, room check skipped."""
+
+    def _raise_missing(self):
+        raise WebexTokenMissingError("No token available.")
+
+    def test_overall_status_unhealthy(self):
+        with patch.object(_diag_mod, "get_webex_api", side_effect=self._raise_missing):
+            result = webex_health_check()
+        self.assertTrue(result["success"])
+        self.assertEqual(result["data"]["overall_status"], "unhealthy")
+
+    def test_token_check_error(self):
+        with patch.object(_diag_mod, "get_webex_api", side_effect=self._raise_missing):
+            result = webex_health_check()
+        token = result["data"]["checks"]["token"]
+        self.assertEqual(token["status"], "error")
+        self.assertEqual(token["error_code"], "E401")
+
+    def test_rooms_skipped(self):
+        with patch.object(_diag_mod, "get_webex_api", side_effect=self._raise_missing):
+            result = webex_health_check()
+        rooms = result["data"]["checks"]["rooms"]
+        self.assertEqual(rooms["status"], "skipped")
+
+
+class TestWebexHealthCheckTokenAuthError(unittest.TestCase):
+    """HTTP 401 from people.me → unhealthy."""
+
+    def setUp(self):
+        self.mock_api = MagicMock()
+        self.mock_api.people.me.side_effect = Exception("401 Unauthorized")
+
+    def test_overall_status_unhealthy(self):
+        with patch.object(_diag_mod, "get_webex_api", return_value=self.mock_api):
+            result = webex_health_check()
+        self.assertEqual(result["data"]["overall_status"], "unhealthy")
+
+    def test_token_error_code_e401(self):
+        with patch.object(_diag_mod, "get_webex_api", return_value=self.mock_api):
+            result = webex_health_check()
+        self.assertEqual(result["data"]["checks"]["token"]["error_code"], "E401")
+
+
+class TestWebexHealthCheckRoomError(unittest.TestCase):
+    """Token OK but rooms.list fails → degraded."""
+
+    def setUp(self):
+        self.mock_api = MagicMock()
+        self.mock_api.people.me.return_value = _fake_me()
+        self.mock_api.rooms.list.side_effect = Exception("network error")
+
+    def test_overall_status_degraded(self):
+        with patch.object(_diag_mod, "get_webex_api", return_value=self.mock_api):
+            result = webex_health_check()
+        self.assertEqual(result["data"]["overall_status"], "degraded")
+
+    def test_token_still_ok(self):
+        with patch.object(_diag_mod, "get_webex_api", return_value=self.mock_api):
+            result = webex_health_check()
+        self.assertEqual(result["data"]["checks"]["token"]["status"], "ok")
+
+    def test_rooms_status_error(self):
+        with patch.object(_diag_mod, "get_webex_api", return_value=self.mock_api):
+            result = webex_health_check()
+        self.assertEqual(result["data"]["checks"]["rooms"]["status"], "error")
+
+
+class TestWebexHealthCheckSampleCap(unittest.TestCase):
+    """Sample rooms list is capped at 3 entries."""
+
+    def setUp(self):
+        self.mock_api = MagicMock()
+        self.mock_api.people.me.return_value = _fake_me()
+        self.mock_api.rooms.list.return_value = iter([
+            _fake_room(f"R{i}", f"Room {i}") for i in range(8)
+        ])
+
+    def test_sample_capped_at_three(self):
+        with patch.object(_diag_mod, "get_webex_api", return_value=self.mock_api):
+            result = webex_health_check()
+        self.assertEqual(len(result["data"]["checks"]["rooms"]["sample"]), 3)
+
+    def test_total_visible_is_full_count(self):
+        with patch.object(_diag_mod, "get_webex_api", return_value=self.mock_api):
+            result = webex_health_check()
+        self.assertEqual(result["data"]["checks"]["rooms"]["total_visible"], 8)
+
+
+if __name__ == "__main__":
+    unittest.main()