fix(tools): harden OpenWebUI tool workflows

dev/verify_mcp_catalog.py

@@ -0,0 +1,88 @@
+#!/usr/bin/env python3
+"""Verify a Streamable HTTP MCP catalogue without invoking any tool."""
+
+from __future__ import annotations
+
+import argparse
+import asyncio
+import json
+
+from mcp import ClientSession
+from mcp.client.streamable_http import streamablehttp_client
+
+
+def invalid_patterns(value, path: str = "schema") -> list[str]:
+    issues = []
+    if isinstance(value, dict):
+        pattern = value.get("pattern")
+        if isinstance(pattern, str) and not (
+            pattern.startswith("^") and pattern.endswith("$")
+        ):
+            issues.append(f"{path}.pattern={pattern!r}")
+        for key, item in value.items():
+            issues.extend(invalid_patterns(item, f"{path}.{key}"))
+    elif isinstance(value, list):
+        for index, item in enumerate(value):
+            issues.extend(invalid_patterns(item, f"{path}[{index}]"))
+    return issues
+
+
+async def verify(
+    url: str,
+    required: set[str],
+    forbidden: set[str],
+    max_tools: int,
+    max_schema_chars: int,
+    timeout: float,
+) -> None:
+    async with asyncio.timeout(timeout):
+        async with streamablehttp_client(url) as (read_stream, write_stream, _):
+            async with ClientSession(read_stream, write_stream) as session:
+                await session.initialize()
+                response = await session.list_tools()
+    names = {tool.name for tool in response.tools}
+    schemas = [tool.inputSchema for tool in response.tools]
+    schema_chars = len(json.dumps(schemas, ensure_ascii=False, separators=(",", ":")))
+    pattern_issues = [
+        issue
+        for index, schema in enumerate(schemas)
+        for issue in invalid_patterns(schema, f"tools[{index}]")
+    ]
+    missing = required - names
+    exposed = forbidden & names
+    if missing or exposed or len(names) > max_tools or schema_chars > max_schema_chars or pattern_issues:
+        raise SystemExit(
+            f"MCP catalogue mismatch: missing={sorted(missing)}, "
+            f"forbidden={sorted(exposed)}, count={len(names)}/{max_tools}, "
+            f"schema_chars={schema_chars}/{max_schema_chars}, "
+            f"invalid_patterns={pattern_issues}, actual={sorted(names)}"
+        )
+    print(
+        f"MCP_CATALOG_OK count={len(names)} schema_chars={schema_chars} tools="
+        + ",".join(sorted(names))
+    )
+
+
+def main() -> None:
+    parser = argparse.ArgumentParser()
+    parser.add_argument("url")
+    parser.add_argument("--require", action="append", default=[])
+    parser.add_argument("--forbid", action="append", default=[])
+    parser.add_argument("--max-tools", type=int, default=64)
+    parser.add_argument("--max-schema-chars", type=int, default=100_000)
+    parser.add_argument("--timeout", type=float, default=10.0)
+    args = parser.parse_args()
+    asyncio.run(
+        verify(
+            args.url,
+            set(args.require),
+            set(args.forbid),
+            args.max_tools,
+            args.max_schema_chars,
+            args.timeout,
+        )
+    )
+
+
+if __name__ == "__main__":
+    main()