Fix status-cache poisoning, add HTTP timeouts, coerce custom status names

- _status_choices no longer caches failures: a transient /ticket_form_fields
  error used to cache [] forever, silently degrading unresolved=True to an
  unfiltered listing (including Resolved/Closed) until container restart.
  unresolved=True and list_ticket_statuses now return an explicit error when
  the status list can't be fetched instead of degrading silently.
- All API calls (server and skill helper) now use a 30s timeout with a clean
  error, so a hung connection can't block a tool call indefinitely.
- _coerce_status falls back to the instance's live status list, so custom
  names like "Working" coerce to their id instead of 400ing at the API.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

mcp-server/server.py

@@ -75,7 +75,7 @@ def _request(method: str, path: str, params: dict[str, Any] | None = None,
     data = json.dumps(body).encode() if body is not None else None
     req = urllib.request.Request(url, data=data, method=method.upper(), headers=headers)
     try:
-        with urllib.request.urlopen(req) as resp:
+        with urllib.request.urlopen(req, timeout=30) as resp:
             raw = resp.read().decode()
             result: dict[str, Any] = {"ok": True, "status": resp.status,
                                       "data": json.loads(raw) if raw else {}}
@@ -98,6 +98,8 @@ def _request(method: str, path: str, params: dict[str, Any] | None = None,
                 "url": url, "method": method.upper()}
     except urllib.error.URLError as e:
         return {"ok": False, "error": f"Network error: {e.reason}", "url": url}
+    except TimeoutError:
+        return {"ok": False, "error": "Request timed out after 30s", "url": url}
 
 
 def _out(result: dict[str, Any]) -> str:
@@ -109,6 +111,11 @@ def _coerce_status(v: Any) -> Any:
         for k, name in TICKET_STATUS.items():
             if name.lower() == v.lower():
                 return k
+        # Not a default name — try the instance's full status list, which
+        # includes custom statuses like "Working" or "Waiting Vendor Support".
+        for c in _status_choices():
+            if str(c.get("value", "")).lower() == v.lower():
+                return c["id"]
     return v
 
 
@@ -128,17 +135,24 @@ _status_cache: dict[str, Any] = {"choices": None}
 
 def _status_choices() -> list[dict[str, Any]]:
     """Fetch (and cache) the instance's ticket status choices via the status
-    field on /ticket_form_fields."""
+    field on /ticket_form_fields.
-    if _status_cache["choices"] is None:
+
-        res = _request("GET", "/ticket_form_fields")
+    Only a successful, non-empty result is cached: a transient API failure must
-        choices: list[dict[str, Any]] = []
+    not permanently disable status-aware features (unresolved=True would
-        if res.get("ok"):
+    otherwise silently degrade to an unfiltered listing until restart)."""
-            for f in res["data"].get("ticket_fields", []):
+    if _status_cache["choices"] is not None:
-                if f.get("name") == "status" or f.get("field_type") == "default_status":
+        return _status_cache["choices"]
-                    choices = f.get("choices", [])
+    res = _request("GET", "/ticket_form_fields")
-                    break
+    if not res.get("ok"):
+        return []
+    choices: list[dict[str, Any]] = []
+    for f in res["data"].get("ticket_fields", []):
+        if f.get("name") == "status" or f.get("field_type") == "default_status":
+            choices = f.get("choices", [])
+            break
+    if choices:
         _status_cache["choices"] = choices
-    return _status_cache["choices"]
+    return choices
 
 
 def _is_terminal(choice: dict[str, Any]) -> bool:
@@ -182,9 +196,16 @@ def list_tickets(query: str | None = None, unresolved: bool = False,
     """
     if unresolved:
         ids = _open_status_ids()
-        if ids:
+        if not ids:
-            clause = "(" + " OR ".join(f"status:{i}" for i in ids) + ")"
+            # Never degrade silently to an unfiltered listing (which would
-            query = f"({query}) AND {clause}" if query else clause
+            # include Resolved/Closed tickets) — surface the failure instead.
+            return _out({"ok": False, "error":
+                         "unresolved=True needs this instance's status list, but "
+                         "fetching /ticket_form_fields failed. Retry, or build an "
+                         "explicit (status:.. OR ..) query using "
+                         "list_ticket_statuses()."})
+        clause = "(" + " OR ".join(f"status:{i}" for i in ids) + ")"
+        query = f"({query}) AND {clause}" if query else clause
     if query:
         params = {"query": f'"{query}"', "workspace_id": workspace_id,
                   "page": page, "per_page": min(per_page, 100)}
@@ -200,8 +221,12 @@ def list_ticket_statuses() -> str:
     """List every ticket status defined in this Freshservice instance (defaults +
     custom), each flagged `terminal` (Resolved/Closed) or open. Use this to build
     status filters correctly, or just call list_tickets(unresolved=True)."""
+    choices = _status_choices()
+    if not choices:
+        return _out({"ok": False, "error":
+                     "Could not fetch ticket statuses from /ticket_form_fields."})
     statuses = [{"id": c.get("id"), "label": c.get("value"),
-                 "terminal": _is_terminal(c)} for c in _status_choices()]
+                 "terminal": _is_terminal(c)} for c in choices]
     return _out({"ok": True, "statuses": statuses,
                  "open_ids": _open_status_ids()})
 

scripts/fs.py

@@ -93,7 +93,7 @@ def request(method: str, path: str, params: dict[str, str], data: str | None) ->
     req = urllib.request.Request(url, data=body, method=method.upper(), headers=headers)
 
     try:
-        with urllib.request.urlopen(req) as resp:
+        with urllib.request.urlopen(req, timeout=30) as resp:
             raw = resp.read().decode()
             out = {"ok": True, "status": resp.status,
                    "data": json.loads(raw) if raw else {}}
@@ -116,6 +116,8 @@ def request(method: str, path: str, params: dict[str, str], data: str | None) ->
                 "method": method.upper(), "url": url}
     except urllib.error.URLError as e:
         return {"ok": False, "error": f"Network error: {e.reason}", "url": url}
+    except TimeoutError:
+        return {"ok": False, "error": "Request timed out after 30s", "url": url}
 
 
 def main() -> None: