50 lines
1.8 KiB
Python
50 lines
1.8 KiB
Python
#!/usr/bin/env python3
|
|
"""Verify MCPHub server endpoints without invoking write operations."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import asyncio
|
|
import json
|
|
import pathlib
|
|
|
|
import httpx
|
|
from mcp import ClientSession
|
|
from mcp.client.streamable_http import streamable_http_client
|
|
|
|
|
|
async def verify(base_url: str, servers: list[str], token: str) -> None:
|
|
results: dict[str, object] = {}
|
|
for server in servers:
|
|
url = f"{base_url.rstrip('/')}/mcp/{server}"
|
|
try:
|
|
async with httpx.AsyncClient(headers={"Authorization": f"Bearer {token}"}) as client:
|
|
transport = streamable_http_client(url, http_client=client)
|
|
async with transport as (reader, writer, _):
|
|
async with ClientSession(reader, writer) as session:
|
|
await session.initialize()
|
|
tools = await session.list_tools()
|
|
results[server] = {
|
|
"ok": True,
|
|
"tool_count": len(tools.tools),
|
|
"tools": [tool.name for tool in tools.tools],
|
|
}
|
|
except Exception as exc:
|
|
results[server] = {"ok": False, "error": str(exc)[:300]}
|
|
print(json.dumps(results, ensure_ascii=False, indent=2))
|
|
if not all(item.get("ok") for item in results.values()):
|
|
raise SystemExit(1)
|
|
|
|
|
|
def main() -> None:
|
|
parser = argparse.ArgumentParser()
|
|
parser.add_argument("base_url")
|
|
parser.add_argument("--token-file", required=True, type=pathlib.Path)
|
|
parser.add_argument("servers", nargs="+")
|
|
args = parser.parse_args()
|
|
asyncio.run(verify(args.base_url, args.servers, args.token_file.read_text(encoding="utf-8").strip()))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|