add fastmcp test script

Author: fayedraza

mcp_debugger.py

@@ -0,0 +1,51 @@
+from fastmcp import FastMCP
+import sqlite3
+import json
+import os
+
+from pathlib import Path
+
+# Create the MCP server
+mcp = FastMCP("MCP Database Debugger")
+
+# Use absolute path relative to this script
+DB_PATH = Path(__file__).parent / "mcp.db"
+
+def get_db_connection():
+    if not os.path.exists(DB_PATH):
+        raise FileNotFoundError(f"Database not found at {DB_PATH}")
+    conn = sqlite3.connect(DB_PATH)
+    conn.row_factory = sqlite3.Row
+    return conn
+
+@mcp.tool()
+def get_database_stats() -> dict:
+    """Get counts of events and alerts in the database."""
+    conn = get_db_connection()
+    try:
+        cursor = conn.cursor()
+        total_events = cursor.execute("SELECT count(*) FROM mcp_auth_events").fetchone()[0]
+        total_alerts = cursor.execute("SELECT count(*) FROM mcp_alerts").fetchone()[0]
+        open_alerts = cursor.execute("SELECT count(*) FROM mcp_alerts WHERE status='open'").fetchone()[0]
+        return {
+            "total_events": total_events,
+            "total_alerts": total_alerts,
+            "open_alerts": open_alerts
+        }
+    finally:
+        conn.close()
+
+@mcp.tool()
+def get_recent_events(limit: int = 5) -> str:
+    """Get the most recent authentication events."""
+    conn = get_db_connection()
+    try:
+        cursor = conn.cursor()
+        rows = cursor.execute("SELECT * FROM mcp_auth_events ORDER BY timestamp DESC LIMIT ?", (limit,)).fetchall()
+        # Convert rows to dicts and handle datetime serialization via str default
+        return json.dumps([dict(row) for row in rows], default=str, indent=2)
+    finally:
+        conn.close()
+
+if __name__ == "__main__":
+    mcp.run(show_banner=False)

test_client.py

@@ -0,0 +1,43 @@
+from fastmcp import Client
+from mcp_debugger import mcp
+import asyncio
+import json
+
+async def main():
+    # client = Client(mcp) # This works according to overload signatures!
+
+    # Using 'with' context manager for automatic connection/cleanup
+    # Note: Client might be async. Let's assume standard usage pattern.
+    # FastMCP client docs usually imply sync usage for simple scripts or async.
+    # Given the __init__ signature didn't show async, but `run_async` did...
+    # The overload `Client(transport: FastMCP)` returns a Client.
+
+    print("Connecting to MCP Server...")
+    async with Client(mcp) as client:
+        print("\n--- Listing Tools ---")
+        tools = await client.list_tools()
+        for tool in tools:
+            print(f"- {tool.name}: {tool.description}")
+
+        print("\n--- Calling get_database_stats ---")
+        result = await client.call_tool("get_database_stats")
+        # FastMCP CallToolResult usually has 'content' or is iterable?
+        # Let's print the raw result to see what we got
+        print(f"Result type: {type(result)}")
+        print(f"Result content: {result}")
+
+        # If it has content, usually it's a list of TextContent or ImageContent path
+        if hasattr(result, 'content'):
+             for content in result.content:
+                 if hasattr(content, 'text'):
+                     print(content.text)
+
+        print(f"\n--- Calling get_recent_events (limit=2) ---")
+        result_events = await client.call_tool("get_recent_events", arguments={"limit": 2})
+        if hasattr(result_events, 'content'):
+             for content in result_events.content:
+                 if hasattr(content, 'text'):
+                     print(content.text)
+
+if __name__ == "__main__":
+    asyncio.run(main())