140 lines
5.2 KiB
Python
140 lines
5.2 KiB
Python
#!/usr/bin/env python3
|
|
"""Test a RUNNING BookStack MCP server end to end.
|
|
|
|
Run it with the venv the installer created, so no extra install is needed:
|
|
|
|
/opt/bookstack-mcp/venv/bin/python test_live.py \
|
|
--url http://127.0.0.1:8080/mcp --token YOUR_STATIC_TOKEN
|
|
|
|
Add --search "some term" to also run a real query against your wiki.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import asyncio
|
|
import sys
|
|
|
|
GREEN, RED, YELLOW, DIM, RESET = "\033[32m", "\033[31m", "\033[33m", "\033[2m", "\033[0m"
|
|
|
|
|
|
def ok(msg: str) -> None:
|
|
print(f"{GREEN} PASS{RESET} {msg}")
|
|
|
|
|
|
def fail(msg: str, hint: str = "") -> None:
|
|
print(f"{RED} FAIL{RESET} {msg}")
|
|
if hint:
|
|
print(f"{DIM} {hint}{RESET}")
|
|
|
|
|
|
def info(msg: str) -> None:
|
|
print(f"{DIM} {msg}{RESET}")
|
|
|
|
|
|
async def main() -> int:
|
|
parser = argparse.ArgumentParser()
|
|
parser.add_argument("--url", default="http://127.0.0.1:8080/mcp")
|
|
parser.add_argument("--token", help="A value from MCP_STATIC_TOKENS")
|
|
parser.add_argument("--search", help="Optional: run a real search query")
|
|
args = parser.parse_args()
|
|
|
|
from fastmcp import Client
|
|
from fastmcp.client.auth import BearerAuth
|
|
|
|
print(f"\nTesting {args.url}\n")
|
|
|
|
auth = BearerAuth(args.token) if args.token else None
|
|
failures = 0
|
|
|
|
# 1. Connect and list tools
|
|
try:
|
|
async with Client(args.url, auth=auth) as client:
|
|
tools = sorted(t.name for t in await client.list_tools())
|
|
ok(f"connected, {len(tools)} tools available")
|
|
info(", ".join(tools))
|
|
|
|
writable = [t for t in tools if t.startswith(("create_", "update_", "delete_"))]
|
|
if writable:
|
|
print(f"{YELLOW} NOTE{RESET} write tools are exposed: {', '.join(writable)}")
|
|
info("set BOOKSTACK_READ_ONLY=true if you didn't intend this")
|
|
|
|
# 2. Reach BookStack itself
|
|
try:
|
|
result = await client.call_tool("list_books", {"count": 5})
|
|
books = result.data.get("books", [])
|
|
ok(f"BookStack API reachable, {result.data.get('total', len(books))} books visible")
|
|
for b in books[:5]:
|
|
info(f"[{b['id']}] {b['name']}")
|
|
if not books:
|
|
print(f"{YELLOW} NOTE{RESET} no books visible to the token's user")
|
|
info("check that user's role permissions in BookStack")
|
|
except Exception as exc:
|
|
failures += 1
|
|
fail("could not list books", str(exc)[:200])
|
|
|
|
# 3. Optional real search
|
|
if args.search:
|
|
try:
|
|
result = await client.call_tool(
|
|
"search_content", {"query": args.search, "count": 5}
|
|
)
|
|
hits = result.data.get("results", [])
|
|
ok(f"search for {args.search!r} returned {len(hits)} result(s)")
|
|
for h in hits:
|
|
info(f"{h['type']} [{h['id']}] {h['name']}")
|
|
except Exception as exc:
|
|
failures += 1
|
|
fail("search failed", str(exc)[:200])
|
|
|
|
except Exception as exc:
|
|
text = str(exc)
|
|
failures += 1
|
|
fail("could not connect", text[:200])
|
|
if "401" in text or "Unauthorized" in text:
|
|
info("Token rejected. Check it matches a value in MCP_STATIC_TOKENS,")
|
|
info("or omit --token if MCP_AUTH_MODE=none.")
|
|
elif "Connection" in text or "refused" in text:
|
|
info("Nothing listening. Check: systemctl status bookstack-mcp")
|
|
return 1
|
|
|
|
# 4. OAuth discovery, only meaningful for public URLs
|
|
if args.url.startswith("https://"):
|
|
import httpx
|
|
|
|
base = args.url.rsplit("/", 1)[0]
|
|
path = args.url.rsplit("/", 1)[1]
|
|
for name, endpoint in [
|
|
("protected resource metadata", f"{base}/.well-known/oauth-protected-resource/{path}"),
|
|
("authorization server metadata", f"{base}/.well-known/oauth-authorization-server"),
|
|
]:
|
|
try:
|
|
async with httpx.AsyncClient(timeout=15, follow_redirects=True) as http:
|
|
response = await http.get(endpoint)
|
|
ctype = response.headers.get("content-type", "")
|
|
if response.status_code == 200 and "json" in ctype:
|
|
ok(f"{name} served")
|
|
elif "html" in ctype:
|
|
failures += 1
|
|
fail(f"{name} returned HTML, not JSON")
|
|
info("Your reverse proxy is serving a login page here.")
|
|
info("Add an Authelia access_control bypass rule for this host.")
|
|
else:
|
|
failures += 1
|
|
fail(f"{name} returned HTTP {response.status_code}")
|
|
info("Proxy the whole hostname, not just the /mcp path.")
|
|
except Exception as exc:
|
|
failures += 1
|
|
fail(f"{name} unreachable", str(exc)[:150])
|
|
|
|
print()
|
|
if failures:
|
|
print(f"{RED}{failures} check(s) failed.{RESET}\n")
|
|
return 1
|
|
print(f"{GREEN}All checks passed.{RESET}\n")
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(asyncio.run(main()))
|