{
"event": "PreToolUse",
"tool_name": "Write",
"tool_input": {
"file_path": "\/var\/www\/mcp-servers\/shared\/infrastructure\/protokoll_logger.py",
"content": "\"\"\"\nInfrastructure Logger for mcp_log table.\n\nImports LogEntry from Domain - does NOT define it here.\n\"\"\"\n\nimport logging\nimport sys\nfrom typing import TYPE_CHECKING\n\nfrom shared.domain.log_entry import LogEntry\nfrom shared.infrastructure.simple_db_connection import SimpleDbConnection\n\nif TYPE_CHECKING:\n from shared.config_base import LogDatabaseConfig\n\n\nclass ProtokollLogger:\n \"\"\"Fail-Safe Logger for mcp_log table.\"\"\"\n\n def __init__(self, client_name: str, config: \"LogDatabaseConfig\") -> None:\n self.client_name = client_name\n self.config = config\n self._logger = logging.getLogger(f\"mcp.{client_name}\")\n\n def log(self, entry: LogEntry) -> None:\n \"\"\"Write log entry. Errors only go to stderr.\"\"\"\n try:\n with SimpleDbConnection.get_log_connection(self.config) as conn:\n with conn.cursor() as cursor:\n request_str = self._format_request(entry)\n cursor.execute(\n \"\"\"INSERT INTO mcp_log\n (timestamp, client_name, request, status, duration_ms, error_message)\n VALUES (%s, %s, %s, %s, %s, %s)\"\"\",\n (\n entry.timestamp,\n self.client_name,\n request_str[:500],\n entry.status,\n entry.duration_ms,\n entry.error_message[:500] if entry.error_message else None,\n ),\n )\n except Exception as e:\n print(f\"CRITICAL: {self.client_name} log failed: {e}\", file=sys.stderr)\n\n def _format_request(self, entry: LogEntry) -> str:\n parts = []\n if entry.tool_name:\n parts.append(f\"[{entry.tool_name}]\")\n if entry.context_id:\n parts.append(f\"id={entry.context_id}\")\n parts.append(entry.request[:400] if entry.request else \"\")\n return \" \".join(parts)\n\n\n# === Registry with Test Support ===\n\n_logger_instances: dict[str, ProtokollLogger] = {}\n\n\ndef get_logger(client_name: str, config: \"LogDatabaseConfig\") -> ProtokollLogger:\n \"\"\"Singleton-Factory for Logger.\"\"\"\n if client_name not in _logger_instances:\n _logger_instances[client_name] = ProtokollLogger(client_name, config)\n return _logger_instances[client_name]\n\n\ndef clear_logger_registry() -> None:\n \"\"\"\n Reset the logger registry.\n\n FOR TESTS ONLY! Enables isolated test execution.\n \"\"\"\n _logger_instances.clear()\n"
}
}
{
"tool_response": {
"type": "create",
"filePath": "\/var\/www\/mcp-servers\/shared\/infrastructure\/protokoll_logger.py",
"content": "\"\"\"\nInfrastructure Logger for mcp_log table.\n\nImports LogEntry from Domain - does NOT define it here.\n\"\"\"\n\nimport logging\nimport sys\nfrom typing import TYPE_CHECKING\n\nfrom shared.domain.log_entry import LogEntry\nfrom shared.infrastructure.simple_db_connection import SimpleDbConnection\n\nif TYPE_CHECKING:\n from shared.config_base import LogDatabaseConfig\n\n\nclass ProtokollLogger:\n \"\"\"Fail-Safe Logger for mcp_log table.\"\"\"\n\n def __init__(self, client_name: str, config: \"LogDatabaseConfig\") -> None:\n self.client_name = client_name\n self.config = config\n self._logger = logging.getLogger(f\"mcp.{client_name}\")\n\n def log(self, entry: LogEntry) -> None:\n \"\"\"Write log entry. Errors only go to stderr.\"\"\"\n try:\n with SimpleDbConnection.get_log_connection(self.config) as conn:\n with conn.cursor() as cursor:\n request_str = self._format_request(entry)\n cursor.execute(\n \"\"\"INSERT INTO mcp_log\n (timestamp, client_name, request, status, duration_ms, error_message)\n VALUES (%s, %s, %s, %s, %s, %s)\"\"\",\n (\n entry.timestamp,\n self.client_name,\n request_str[:500],\n entry.status,\n entry.duration_ms,\n entry.error_message[:500] if entry.error_message else None,\n ),\n )\n except Exception as e:\n print(f\"CRITICAL: {self.client_name} log failed: {e}\", file=sys.stderr)\n\n def _format_request(self, entry: LogEntry) -> str:\n parts = []\n if entry.tool_name:\n parts.append(f\"[{entry.tool_name}]\")\n if entry.context_id:\n parts.append(f\"id={entry.context_id}\")\n parts.append(entry.request[:400] if entry.request else \"\")\n return \" \".join(parts)\n\n\n# === Registry with Test Support ===\n\n_logger_instances: dict[str, ProtokollLogger] = {}\n\n\ndef get_logger(client_name: str, config: \"LogDatabaseConfig\") -> ProtokollLogger:\n \"\"\"Singleton-Factory for Logger.\"\"\"\n if client_name not in _logger_instances:\n _logger_instances[client_name] = ProtokollLogger(client_name, config)\n return _logger_instances[client_name]\n\n\ndef clear_logger_registry() -> None:\n \"\"\"\n Reset the logger registry.\n\n FOR TESTS ONLY! Enables isolated test execution.\n \"\"\"\n _logger_instances.clear()\n",
"structuredPatch": [],
"originalFile": null
}
}