fix(webui): satisfy strict skill marketplace typing

nanobot/webui/skills_api.py

@@ -6,7 +6,7 @@ import json
 import shlex
 import tempfile
 from pathlib import Path
-from typing import Any
+from typing import Any, cast
 
 from nanobot.agent.skills import SkillsLoader
 from nanobot.config.loader import load_config, save_config
@@ -188,37 +188,42 @@ def _nanobot_metadata(metadata: dict[str, Any] | None) -> dict[str, Any]:
     raw = metadata.get("metadata")
     if isinstance(raw, str):
         try:
-            raw = json.loads(raw)
+            raw = cast(object, json.loads(raw))
         except (json.JSONDecodeError, TypeError):
             return {}
     if not isinstance(raw, dict):
         return {}
-    payload = raw.get("nanobot", raw.get("openclaw", {}))
+    metadata_payload = cast(dict[str, Any], raw)
-    return payload if isinstance(payload, dict) else {}
+    payload = metadata_payload.get("nanobot", metadata_payload.get("openclaw", {}))
+    return cast(dict[str, Any], payload) if isinstance(payload, dict) else {}
 
 
 def _install_options(metadata: dict[str, Any] | None) -> list[dict[str, str]]:
     """Return safe, copyable setup commands declared by a skill."""
-    install = _nanobot_metadata(metadata).get("install")
+    raw_install = _nanobot_metadata(metadata).get("install")
-    if not isinstance(install, list):
+    if not isinstance(raw_install, list):
         return []
+    install = cast(list[object], raw_install)
 
     options: list[dict[str, str]] = []
     for item in install:
         if not isinstance(item, dict):
             continue
-        kind = item.get("kind")
+        install_item = cast(dict[str, object], item)
-        package = item.get("formula") if kind == "brew" else item.get("package")
+        kind = install_item.get("kind")
+        if not isinstance(kind, str) or kind not in {"brew", "apt"}:
+            continue
+        package = (
+            install_item.get("formula") if kind == "brew" else install_item.get("package")
+        )
         if not isinstance(package, str) or not package.strip():
             continue
         if kind == "brew":
             command = f"brew install {shlex.quote(package.strip())}"
-        elif kind == "apt":
-            command = f"sudo apt-get install -y {shlex.quote(package.strip())}"
         else:
-            continue
+            command = f"sudo apt-get install -y {shlex.quote(package.strip())}"
-        option_id = item.get("id")
+        option_id = install_item.get("id")
-        label = item.get("label")
+        label = install_item.get("label")
         options.append(
             {
                 "id": option_id if isinstance(option_id, str) else kind,

nanobot/webui/skills_marketplace.py

@@ -12,7 +12,7 @@ import tempfile
 import time
 import zipfile
 from pathlib import Path, PurePosixPath
-from typing import Any
+from typing import Any, cast
 from urllib.parse import quote, urlparse
 
 import httpx
@@ -56,6 +56,12 @@ _weekly_cache: dict[tuple[str, str], list[int]] = {}
 _weekly_cache_expires_at = 0.0
 
 
+def _response_json_object(response: httpx.Response) -> dict[str, Any] | None:
+    """Narrow an untyped HTTP JSON response at the external-data boundary."""
+    payload = cast(object, response.json())
+    return cast(dict[str, Any], payload) if isinstance(payload, dict) else None
+
+
 class SkillsMarketplaceError(Exception):
     """A safe error that can be returned to the WebUI."""
 
@@ -117,7 +123,7 @@ async def _trending_skills_sh_skills(
         async with _skills_client() as client:
             response = await client.get(_TRENDING_URL)
             response.raise_for_status()
-            payload = response.json()
+            payload = _response_json_object(response) or {}
     except (httpx.HTTPError, ValueError) as exc:
         raise SkillsMarketplaceError(
             "skills.sh trending skills are temporarily unavailable",
@@ -125,16 +131,17 @@ async def _trending_skills_sh_skills(
         ) from exc
 
     installed = _installed_skill_names(workspace_path)
-    rows = payload.get("skills", []) if isinstance(payload, dict) else []
+    rows = payload.get("skills", [])
     skills: list[dict[str, Any]] = []
     seen_sources: set[str] = set()
     for rank, row in enumerate(rows, start=1):
         if not isinstance(row, dict):
             continue
-        source = row.get("source")
+        row_payload = cast(dict[str, Any], row)
+        source = row_payload.get("source")
         if not isinstance(source, str) or source in seen_sources:
             continue
-        skill = _marketplace_skill(row, installed, rank=rank)
+        skill = _marketplace_skill(row_payload, installed, rank=rank)
         if skill is None:
             continue
         seen_sources.add(source)
@@ -207,7 +214,7 @@ async def _search_skills_sh_skills(
                 params={"q": normalized, "limit": min(max(limit, 1), 50)},
             )
             response.raise_for_status()
-            payload = response.json()
+            payload = _response_json_object(response) or {}
     except (httpx.HTTPError, ValueError) as exc:
         raise SkillsMarketplaceError(
             "skills.sh search is temporarily unavailable",
@@ -215,12 +222,12 @@ async def _search_skills_sh_skills(
         ) from exc
 
     installed = _installed_skill_names(workspace_path)
-    rows = payload.get("skills", []) if isinstance(payload, dict) else []
+    rows = payload.get("skills", [])
-    skills = []
+    skills: list[dict[str, Any]] = []
     for row in rows:
         if not isinstance(row, dict):
             continue
-        skill = _marketplace_skill(row, installed)
+        skill = _marketplace_skill(cast(dict[str, Any], row), installed)
         if skill is not None:
             skills.append(skill)
 
@@ -245,7 +252,7 @@ async def _search_skillhub_skills(
                 params={"q": normalized, "limit": min(max(limit, 1), 50)},
             )
             response.raise_for_status()
-            payload = response.json()
+            payload = _response_json_object(response) or {}
     except (httpx.HTTPError, ValueError) as exc:
         raise SkillsMarketplaceError(
             "SkillHub search is temporarily unavailable",
@@ -253,12 +260,12 @@ async def _search_skillhub_skills(
         ) from exc
 
     installed = _installed_skill_names(workspace_path)
-    rows = payload.get("results", []) if isinstance(payload, dict) else []
+    rows = payload.get("results", [])
     skills = [
         skill
         for row in rows
         if isinstance(row, dict)
-        if (skill := _skillhub_skill(row, installed)) is not None
+        if (skill := _skillhub_skill(cast(dict[str, Any], row), installed)) is not None
     ]
     return {
         "query": normalized,
@@ -277,7 +284,7 @@ async def _trending_skillhub_skills(
         async with _skillhub_client() as client:
             response = await client.get(_SKILLHUB_TRENDING_URL)
             response.raise_for_status()
-            payload = response.json()
+            payload = _response_json_object(response) or {}
     except (httpx.HTTPError, ValueError) as exc:
         raise SkillsMarketplaceError(
             "SkillHub trending skills are temporarily unavailable",
@@ -285,12 +292,12 @@ async def _trending_skillhub_skills(
         ) from exc
 
     installed = _installed_skill_names(workspace_path)
-    rows = payload.get("skills", []) if isinstance(payload, dict) else []
+    rows = payload.get("skills", [])
     skills: list[dict[str, Any]] = []
     for rank, row in enumerate(rows, start=1):
         if not isinstance(row, dict):
             continue
-        skill = _skillhub_skill(row, installed, rank=rank)
+        skill = _skillhub_skill(cast(dict[str, Any], row), installed, rank=rank)
         if skill is not None:
             skills.append(skill)
         if len(skills) >= min(max(limit, 1), 20):
@@ -538,9 +545,10 @@ async def _skillhub_latest_version(client: httpx.AsyncClient, skill_id: str) ->
         f"{_SKILLHUB_API_BASE_URL}/api/v1/skills/{quote(skill_id, safe='')}"
     )
     response.raise_for_status()
-    payload = response.json()
+    payload = _response_json_object(response) or {}
-    latest = payload.get("latestVersion", {}) if isinstance(payload, dict) else {}
+    raw_latest = payload.get("latestVersion", {})
-    version = latest.get("version") if isinstance(latest, dict) else None
+    latest = cast(dict[str, Any], raw_latest) if isinstance(raw_latest, dict) else {}
+    version = latest.get("version")
     if not isinstance(version, str) or _VERSION_RE.fullmatch(version) is None:
         raise SkillsMarketplaceError(
             "SkillHub did not return a valid skill version",
@@ -559,8 +567,8 @@ async def _skillhub_signature(
         f"{quote(skill_id, safe='')}/versions/{quote(version, safe='')}/signature"
     )
     response.raise_for_status()
-    payload = response.json()
+    payload = _response_json_object(response)
-    if not isinstance(payload, dict):
+    if payload is None:
         raise SkillsMarketplaceError(
             "SkillHub returned an invalid package fingerprint",
             status=502,
@@ -786,7 +794,8 @@ def _skillhub_skill(
         display_name = skill_id
 
     namespace = row.get("namespace")
-    handle = namespace.get("handle") if isinstance(namespace, dict) else None
+    namespace_payload = cast(dict[str, Any], namespace) if isinstance(namespace, dict) else {}
+    handle = namespace_payload.get("handle")
     if not isinstance(handle, str) or not handle.strip():
         owner = row.get("owner_name") or row.get("ownerName")
         handle = owner if isinstance(owner, str) and owner.strip() else "community"
@@ -795,11 +804,11 @@ def _skillhub_skill(
     installs = row.get("installs")
     downloads = row.get("downloads")
     publisher = row.get("publisher")
-    verified = bool(isinstance(publisher, dict) and publisher.get("verified") is True)
+    publisher_payload = cast(dict[str, Any], publisher) if isinstance(publisher, dict) else {}
+    verified = publisher_payload.get("verified") is True
     labels = row.get("labels")
-    requires_api_key = bool(
+    labels_payload = cast(dict[str, Any], labels) if isinstance(labels, dict) else {}
-        isinstance(labels, dict) and str(labels.get("requires_api_key", "")).lower() == "true"
+    requires_api_key = str(labels_payload.get("requires_api_key", "")).lower() == "true"
-    )
     version = row.get("version")
     if not isinstance(version, str) or _VERSION_RE.fullmatch(version) is None:
         version = ""
@@ -846,21 +855,22 @@ async def _load_weekly_installs(
             continue
         try:
             response.raise_for_status()
-            payload = response.json()
+            payload = _response_json_object(response) or {}
         except (httpx.HTTPError, ValueError):
             continue
         successful = True
-        rows = payload.get("skills", []) if isinstance(payload, dict) else []
+        rows = payload.get("skills", [])
         for row in rows:
             if not isinstance(row, dict):
                 continue
-            source = row.get("source")
+            row_payload = cast(dict[str, Any], row)
-            skill_id = row.get("skillId")
+            source = row_payload.get("source")
-            values = row.get("weeklyInstalls")
+            skill_id = row_payload.get("skillId")
+            values = row_payload.get("weeklyInstalls")
             if isinstance(source, str) and isinstance(skill_id, str) and isinstance(values, list):
                 clean = [
                     value
-                    for value in values
+                    for value in cast(list[object], values)
                     if isinstance(value, int) and not isinstance(value, bool) and value >= 0
                 ]
                 if len(clean) >= 2:
@@ -875,7 +885,7 @@ async def _load_weekly_installs(
 def _valid_skill_refs(skill_ids: list[str]) -> list[tuple[str, str]]:
     refs: list[tuple[str, str]] = []
     for value in skill_ids[:20]:
-        if not isinstance(value, str) or "/" not in value:
+        if "/" not in value:
             continue
         source, skill_id = value.rsplit("/", 1)
         ref = (source, skill_id)

nanobot/webui/ws_http.py

@@ -197,7 +197,9 @@ class GatewayHTTPHandler:
         self.ingress = ingress
         self.workspaces = workspaces
         self.skills_workspace_path = skills_workspace_path
-        self.disabled_skills = disabled_skills if disabled_skills is not None else set()
+        self.disabled_skills: set[str] = (
+            disabled_skills if disabled_skills is not None else set()
+        )
         self.skill_state_action = skill_state_action
         self._skill_install_lock = asyncio.Lock()
         self.cron_service = cron_service