backend dev: create plan

common/as_type.py

@@ -0,0 +1,187 @@
+import json
+import re
+from typing import Any
+
+from mcp_types import TextContent
+
+
+def _as_dict(result: Any) -> dict:
+    """Normalize MCP call_tool output to a dict."""
+    if result is None:
+        raise ValueError("MCP returned None")
+
+    # Direct dict
+    if isinstance(result, dict):
+        return result
+
+    if not isinstance(result, list):
+        raise ValueError(f"Unexpected MCP response shape: {type(result)}")
+
+    if not result:
+        raise ValueError("MCP returned an empty list")
+
+    # Find the first text block (dict or TextContent object)
+    text: str | None = None
+    for block in result:
+        if isinstance(block, TextContent):
+            text = block.text
+            break
+        if isinstance(block, dict) and block.get("type") == "text":
+            text = block.get("text")
+            break
+        # Last-resort: the block itself is structured data
+        if isinstance(block, dict) and "id" in block:
+            return block
+
+    if text is None:
+        raise ValueError(f"No text block found in {repr(result)[:300]}")
+
+    try:
+        parsed = json.loads(text)
+    except json.JSONDecodeError as e:
+        raise ValueError(f"Text block is not JSON: {text[:300]}") from e
+
+    if isinstance(parsed, dict):
+        return parsed
+    if isinstance(parsed, list) and parsed and isinstance(parsed[0], dict):
+        return parsed[0]
+    raise ValueError(f"Parsed JSON is neither dict nor list[dict]: {type(parsed)}")
+
+
+def _as_list(result: Any) -> list[dict]:
+    if result is None:
+        return []
+    if isinstance(result, list) and result and isinstance(result[0], TextContent):
+        text = result[0].text
+    elif isinstance(result, list) and result and isinstance(result[0], dict) and "type" in result[0]:
+        text = result[0].get("text", "")
+    elif isinstance(result, list):
+        return result  # already a list of dicts
+    else:
+        raise ValueError(f"Unexpected MCP response shape: {type(result)}")
+
+    parsed = json.loads(text)
+    if isinstance(parsed, list):
+        return parsed
+    if isinstance(parsed, dict):
+        for key in ("items", "comments", "issues", "results"):
+            if isinstance(parsed.get(key), list):
+                return parsed[key]
+        return [parsed]
+    raise ValueError(f"Parsed JSON is not a list: {type(parsed)}")
+
+def _first(d: dict, *keys, default=None):
+    for k in keys:
+        if k in d and d[k] is not None:
+            return d[k]
+    return default
+
+def _extract_base64(result: Any) -> str | None:
+    """Pull a base64 image payload out of an MCP call_tool result.
+
+    Handles all the shapes a YouTrack MCP server might return:
+      - ImageContent block (has .data and .mimeType)
+      - TextContent block with raw base64 or a JSON wrapper
+      - dict with {"data": "..."} or {"content": "..."}
+      - plain string of base64
+    """
+    if result is None:
+        return None
+
+    # Plain string
+    if isinstance(result, str):
+        return result if _looks_like_base64(result) else None
+
+    # Unwrap CallToolResult-like objects
+    content = getattr(result, "content", None)
+    if content is not None:
+        return _extract_base64(content)
+
+    # List of content blocks
+    if isinstance(result, list):
+        for block in result:
+            got = _extract_base64(block)
+            if got:
+                return got
+        return None
+
+    # Dict wrapper
+    if isinstance(result, dict):
+        # Image content: {"type": "image", "data": "...", "mimeType": "..."}
+        if result.get("type") == "image" and result.get("data"):
+            return result["data"]
+
+        # Direct data field
+        for key in ("data", "content", "base64", "image"):
+            if isinstance(result.get(key), str):
+                val = result[key]
+                if _looks_like_base64(val):
+                    return val
+
+        # Text wrapper whose text is base64 or JSON containing base64
+        text = result.get("text")
+        if isinstance(text, str):
+            try:
+                parsed = json.loads(text)
+                return _extract_base64(parsed)
+            except json.JSONDecodeError:
+                return text if _looks_like_base64(text) else None
+
+        return None
+
+    # SDK TextContent / ImageContent objects (attribute access)
+    if isinstance(result, TextContent):
+        text = result.text
+        try:
+            return _extract_base64(json.loads(text))
+        except json.JSONDecodeError:
+            return text if _looks_like_base64(text) else None
+
+    # ImageContent has .data and .mime_type (or .mimeType)
+    if hasattr(result, "data") and hasattr(result, "type") and getattr(result, "type") == "image":
+        return getattr(result, "data")
+
+    return None
+
+
+B64_RE = re.compile(r"^[A-Za-z0-9+/=\s]+$")
+def _looks_like_base64(s: str) -> bool:
+    """Very cheap check — base64 has no whitespace-heavy content and uses a limited alphabet."""
+    s = s.strip()
+    if len(s) < 32:
+        return False
+    return bool(B64_RE.match(s))
+
+async def _fetch_image_base64(self, issue_id: str, attachment_id: str) -> str:
+    result = await self.mcp.call_tool(
+        "issue_attachment_download",
+        {"issueId": issue_id, "attachmentId": attachment_id, "downloadToFile": False},
+    )
+    # Result contains base64 or a URL you can fetch
+    return result  # base64 string
+
+def _extract_text(result: Any) -> str:
+    """Pull a single string out of an MCP result, whichever shape it is."""
+    content = getattr(result, "content", None) or result
+    if isinstance(content, list) and content:
+        block = content[0]
+        if isinstance(block, dict):
+            return block.get("text") or json.dumps(block)
+    if isinstance(content, dict):
+        return content.get("text") or json.dumps(content)
+    return str(content)
+
+
+NOT_FOUND_HINTS = (
+    "not found",
+    "does not exist",
+    "doesn't exist",
+    "no such",
+    "unknown issue",
+    "404",
+)
+
+
+def _looks_like_not_found(text: str) -> bool:
+    lowered = text.lower()
+    return any(h in lowered for h in NOT_FOUND_HINTS)