{
"event": "PreToolUse",
"tool_name": "Write",
"tool_input": {
"file_path": "\/var\/www\/mcp-servers\/mcp_db\/tools\/stats_tool.py",
"content": "\"\"\"Stats Tool - Logging-Statistiken\"\"\"\n\nfrom constants import MAX_ROWS, PERCENT_HALF\n\nfrom config import Config\nfrom infrastructure.db_connection import DatabaseConnection\n\n\ndef register_stats_tool(mcp):\n \"\"\"Registriert db_stats Tool\"\"\"\n\n @mcp.tool()\n def db_stats(limit: int = PERCENT_HALF) -> dict:\n \"\"\"\n Zeigt letzte Queries aus mcp_log.\n\n Args:\n limit: Anzahl der Eintraege (1-100)\n\n Returns:\n Dict mit logs (List[dict])\n \"\"\"\n limit = min(max(limit, 1), MAX_ROWS)\n\n try:\n with DatabaseConnection.get_connection(Config.LOG_DB_NAME) as conn:\n # buffered=True verhindert \"Unread result found\" Fehler\n cursor = conn.cursor(dictionary=True, buffered=True)\n cursor.execute(\n \"\"\"SELECT id, timestamp, client_name, request, status,\n duration_ms, error_message\n FROM mcp_log\n ORDER BY timestamp DESC\n LIMIT %s\"\"\",\n (limit,),\n )\n logs = cursor.fetchall()\n cursor.close()\n\n # Convert datetime objects to strings for JSON serialization\n for log in logs:\n if log.get(\"timestamp\"):\n log[\"timestamp\"] = log[\"timestamp\"].isoformat()\n\n return {\"logs\": logs, \"count\": len(logs)}\n\n except Exception as e:\n return {\"error\": str(e), \"logs\": []}\n"
}
}
{
"tool_response": {
"type": "update",
"filePath": "\/var\/www\/mcp-servers\/mcp_db\/tools\/stats_tool.py",
"content": "\"\"\"Stats Tool - Logging-Statistiken\"\"\"\n\nfrom constants import MAX_ROWS, PERCENT_HALF\n\nfrom config import Config\nfrom infrastructure.db_connection import DatabaseConnection\n\n\ndef register_stats_tool(mcp):\n \"\"\"Registriert db_stats Tool\"\"\"\n\n @mcp.tool()\n def db_stats(limit: int = PERCENT_HALF) -> dict:\n \"\"\"\n Zeigt letzte Queries aus mcp_log.\n\n Args:\n limit: Anzahl der Eintraege (1-100)\n\n Returns:\n Dict mit logs (List[dict])\n \"\"\"\n limit = min(max(limit, 1), MAX_ROWS)\n\n try:\n with DatabaseConnection.get_connection(Config.LOG_DB_NAME) as conn:\n # buffered=True verhindert \"Unread result found\" Fehler\n cursor = conn.cursor(dictionary=True, buffered=True)\n cursor.execute(\n \"\"\"SELECT id, timestamp, client_name, request, status,\n duration_ms, error_message\n FROM mcp_log\n ORDER BY timestamp DESC\n LIMIT %s\"\"\",\n (limit,),\n )\n logs = cursor.fetchall()\n cursor.close()\n\n # Convert datetime objects to strings for JSON serialization\n for log in logs:\n if log.get(\"timestamp\"):\n log[\"timestamp\"] = log[\"timestamp\"].isoformat()\n\n return {\"logs\": logs, \"count\": len(logs)}\n\n except Exception as e:\n return {\"error\": str(e), \"logs\": []}\n",
"structuredPatch": [
{
"oldStart": 1,
"oldLines": 9,
"newStart": 1,
"newLines": 5,
"lines": [
" \"\"\"Stats Tool - Logging-Statistiken\"\"\"",
"-import sys",
" ",
"-sys.path.insert(0, \"\/var\/www\/mcp-servers\/mcp_db\")",
"-sys.path.insert(0, \"\/var\/www\/mcp-servers\/shared\")",
"-",
" from constants import MAX_ROWS, PERCENT_HALF",
" ",
" from config import Config"
]
},
{
"oldStart": 19,
"oldLines": 7,
"newStart": 15,
"newLines": 7,
"lines": [
" Zeigt letzte Queries aus mcp_log.",
" ",
" Args:",
"- limit: Anzahl der Einträge (1-100)",
"+ limit: Anzahl der Eintraege (1-100)",
" ",
" Returns:",
" Dict mit logs (List[dict])"
]
}
],
"originalFile": "\"\"\"Stats Tool - Logging-Statistiken\"\"\"\nimport sys\n\nsys.path.insert(0, \"\/var\/www\/mcp-servers\/mcp_db\")\nsys.path.insert(0, \"\/var\/www\/mcp-servers\/shared\")\n\nfrom constants import MAX_ROWS, PERCENT_HALF\n\nfrom config import Config\nfrom infrastructure.db_connection import DatabaseConnection\n\n\ndef register_stats_tool(mcp):\n \"\"\"Registriert db_stats Tool\"\"\"\n\n @mcp.tool()\n def db_stats(limit: int = PERCENT_HALF) -> dict:\n \"\"\"\n Zeigt letzte Queries aus mcp_log.\n\n Args:\n limit: Anzahl der Einträge (1-100)\n\n Returns:\n Dict mit logs (List[dict])\n \"\"\"\n limit = min(max(limit, 1), MAX_ROWS)\n\n try:\n with DatabaseConnection.get_connection(Config.LOG_DB_NAME) as conn:\n # buffered=True verhindert \"Unread result found\" Fehler\n cursor = conn.cursor(dictionary=True, buffered=True)\n cursor.execute(\n \"\"\"SELECT id, timestamp, client_name, request, status,\n duration_ms, error_message\n FROM mcp_log\n ORDER BY timestamp DESC\n LIMIT %s\"\"\",\n (limit,),\n )\n logs = cursor.fetchall()\n cursor.close()\n\n # Convert datetime objects to strings for JSON serialization\n for log in logs:\n if log.get(\"timestamp\"):\n log[\"timestamp\"] = log[\"timestamp\"].isoformat()\n\n return {\"logs\": logs, \"count\": len(logs)}\n\n except Exception as e:\n return {\"error\": str(e), \"logs\": []}\n"
}
}