Consolidate portable MCPs in Unraid MCPHub

platform/mcphub/probe-hub.py

@@ -0,0 +1,47 @@
+#!/usr/bin/env python3
+"""Invoke one explicitly chosen MCPHub tool and print a bounded result."""
+
+from __future__ import annotations
+
+import argparse
+import asyncio
+import base64
+import json
+import pathlib
+
+import httpx
+from mcp import ClientSession
+from mcp.client.streamable_http import streamable_http_client
+
+
+async def probe(url: str, token: str, tool: str, arguments: dict[str, object]) -> None:
+    async with httpx.AsyncClient(headers={"Authorization": f"Bearer {token}"}) as client:
+        async with streamable_http_client(url, http_client=client) as (reader, writer, _):
+            async with ClientSession(reader, writer) as session:
+                await session.initialize()
+                result = await session.call_tool(tool, arguments)
+                text = "\n".join(getattr(item, "text", "") for item in result.content)
+                print(text[:2000])
+                if result.isError:
+                    raise SystemExit(1)
+
+
+def main() -> None:
+    parser = argparse.ArgumentParser()
+    parser.add_argument("url")
+    parser.add_argument("tool")
+    parser.add_argument("arguments_b64", help="Base64-encoded JSON object")
+    parser.add_argument("--token-file", required=True, type=pathlib.Path)
+    args = parser.parse_args()
+    asyncio.run(
+        probe(
+            args.url,
+            args.token_file.read_text(encoding="utf-8").strip(),
+            args.tool,
+            json.loads(base64.b64decode(args.arguments_b64).decode("utf-8")),
+        )
+    )
+
+
+if __name__ == "__main__":
+    main()