wf-mcp reorg big 1 the Folders have appeared

src/wf_mcp/shared/__init__.py

@@ -0,0 +1,25 @@
+from .errors import error_payload, root_exception
+from .names import (
+    ADMIN_NAMESPACE,
+    LdaNamespace,
+    ProxyToolName,
+    is_admin_tool_name,
+    namespaced_tool_name,
+    parse_namespaced_tool_name,
+)
+from .pagination import clamp_limit, make_cursor, paginate_items, parse_cursor
+
+__all__ = [
+    "ADMIN_NAMESPACE",
+    "LdaNamespace",
+    "ProxyToolName",
+    "clamp_limit",
+    "error_payload",
+    "is_admin_tool_name",
+    "make_cursor",
+    "namespaced_tool_name",
+    "paginate_items",
+    "parse_cursor",
+    "parse_namespaced_tool_name",
+    "root_exception",
+]

src/wf_mcp/shared/errors.py

@@ -0,0 +1,20 @@
+from __future__ import annotations
+
+
+def root_exception(exc: BaseException) -> BaseException:
+    current: BaseException = exc
+    while isinstance(current, ExceptionGroup) and current.exceptions:
+        nested = current.exceptions[0]
+        if isinstance(nested, BaseException):
+            current = nested
+            continue
+        break
+    return current
+
+
+def error_payload(exc: BaseException) -> dict[str, str]:
+    root = root_exception(exc)
+    return {
+        "error_type": type(root).__name__,
+        "error": str(root),
+    }

src/wf_mcp/shared/names.py

@@ -0,0 +1,50 @@
+from __future__ import annotations
+
+from dataclasses import dataclass
+
+from fastmcp.server.transforms import Namespace
+
+ADMIN_NAMESPACE = "wf.mcp"
+
+
+@dataclass(frozen=True, slots=True)
+class ProxyToolName:
+    proxy_name: str
+    connection_id: str
+    local_name: str
+
+
+def namespaced_tool_name(connection_id: str, local_name: str) -> str:
+    return f"{connection_id}_{local_name}"
+
+
+def parse_namespaced_tool_name(
+    proxy_name: str,
+    connection_ids: set[str],
+) -> ProxyToolName | None:
+    matches = [
+        connection_id
+        for connection_id in connection_ids
+        if proxy_name.startswith(f"{connection_id}_")
+    ]
+    if not matches:
+        return None
+    connection_id = max(matches, key=len)
+    local_name = proxy_name[len(connection_id) + 1 :]
+    if not local_name:
+        return None
+    return ProxyToolName(
+        proxy_name=proxy_name,
+        connection_id=connection_id,
+        local_name=local_name,
+    )
+
+
+def is_admin_tool_name(proxy_name: str) -> bool:
+    return proxy_name.startswith(f"{ADMIN_NAMESPACE}_")
+
+
+class LdaNamespace(Namespace):
+    def __init__(self, prefix: str) -> None:
+        super().__init__(prefix)
+        self._name_prefix = f"{prefix}."  # some good stuff

src/wf_mcp/shared/pagination.py

@@ -0,0 +1,44 @@
+from __future__ import annotations
+
+import base64
+import json
+from typing import TypeVar
+
+T = TypeVar("T")
+
+
+def parse_cursor(cursor: str | None) -> int:
+    if cursor is None:
+        return 0
+    try:
+        payload = json.loads(base64.urlsafe_b64decode(cursor.encode()).decode())
+    except Exception as exc:
+        raise ValueError("invalid cursor") from exc
+    start = payload.get("start")
+    if not isinstance(start, int) or start < 0:
+        raise ValueError("invalid cursor")
+    return start
+
+
+def make_cursor(start: int) -> str:
+    payload = json.dumps({"start": start}, separators=(",", ":")).encode()
+    return base64.urlsafe_b64encode(payload).decode()
+
+
+def clamp_limit(limit: int, *, default: int = 50, maximum: int = 200) -> int:
+    if limit <= 0:
+        return default
+    return min(limit, maximum)
+
+
+def paginate_items(
+    items: list[T],
+    *,
+    cursor: str | None,
+    limit: int,
+) -> tuple[list[T], str | None]:
+    page_limit = clamp_limit(limit)
+    start = parse_cursor(cursor)
+    end = start + page_limit
+    next_cursor = make_cursor(end) if end < len(items) else None
+    return items[start:end], next_cursor