Spaces:
Running
Running
File size: 3,532 Bytes
d497a35 69f247d d497a35 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 |
#!/usr/bin/env python3
"""Exercise all MCP tools exposed by the hosted server."""
from __future__ import annotations
import argparse
import asyncio
import json
import os
import sys
import uuid
from typing import Any, Dict
from fastmcp.client import Client, StreamableHttpTransport
from fastmcp.client.client import ToolError
def build_parser() -> argparse.ArgumentParser:
parser = argparse.ArgumentParser(description="Test MCP HTTP tools end-to-end")
parser.add_argument(
"--url",
default=os.environ.get("MCP_URL", "https://YOUR_USERNAME-document-mcp.hf.space/mcp"),
help="Hosted MCP endpoint base URL",
)
parser.add_argument(
"--token",
default=os.environ.get("MCP_TOKEN"),
help="Bearer token for authentication (or set MCP_TOKEN env variable)",
)
parser.add_argument(
"--note",
default=f"mcp-test-{uuid.uuid4().hex}.md",
help="Temporary note path to create and delete during the test",
)
return parser
async def exercise_tools(url: str, token: str, note_path: str) -> Dict[str, Any]:
transport = StreamableHttpTransport(url=url, headers={"Authorization": f"Bearer {token}"})
async with Client(transport, name="multi-tenant-audit") as client:
results: Dict[str, Any] = {}
tools = await client.list_tools()
results["list_tools"] = [tool.name for tool in tools]
results["list_notes_before"] = await client.call_tool("list_notes", {})
# Read the welcome note if it exists
try:
results["read_note"] = await client.call_tool(
"read_note", {"path": "Welcome.md"}
)
except ToolError as exc:
results["read_note_error"] = str(exc)
# Create a temporary note
body = "# MCP Test\n\nThis note was created by the MCP HTTP audit script."
results["write_note"] = await client.call_tool(
"write_note",
{
"path": note_path,
"body": body,
"title": "MCP Test",
"metadata": {"tags": ["mcp", "audit"]},
},
)
# Search for the word "MCP"
results["search_notes"] = await client.call_tool(
"search_notes", {"query": "MCP", "limit": 10}
)
# Fetch backlinks for the welcome note (if present)
try:
results["get_backlinks"] = await client.call_tool(
"get_backlinks", {"path": "Welcome.md"}
)
except ToolError as exc:
results["get_backlinks_error"] = str(exc)
# Fetch tags
results["get_tags"] = await client.call_tool("get_tags", {})
# Delete the temporary note and show list afterwards
results["delete_note"] = await client.call_tool(
"delete_note", {"path": note_path}
)
results["list_notes_after"] = await client.call_tool("list_notes", {})
return results
def main() -> None:
parser = build_parser()
args = parser.parse_args()
if not args.token:
parser.error("Bearer token must be provided via --token or MCP_TOKEN env variable")
try:
results = asyncio.run(exercise_tools(args.url, args.token, args.note))
except Exception as exc: # pragma: no cover
print(f"Error exercising MCP tools: {exc}", file=sys.stderr)
raise SystemExit(1) from exc
json.dump(results, sys.stdout, indent=2, sort_keys=True)
print()
if __name__ == "__main__":
main()
|