{
"event": "PreToolUse",
"tool_name": "Write",
"tool_input": {
"file_path": "\/var\/www\/mcp-servers\/mcp_contracts\/tools\/contract_tools_components\/contract_reporter.py",
"content": "\"\"\"Contract Reporting and Logging Module.\"\"\"\n\nimport json\nimport time\nfrom datetime import datetime\nfrom typing import Any, Callable, Optional\n\nfrom config import Config\nfrom shared.domain import LogEntry\nfrom shared.infrastructure import get_logger\n\nfrom .constants import (\n ERROR_MESSAGE_MAX_LENGTH,\n LOG_STATUS_DENIED,\n LOG_STATUS_ERROR,\n LOG_STATUS_SUCCESS,\n MS_PER_SECOND,\n)\n\n\nclass ContractReporter:\n \"\"\"Handles logging and response formatting for contract operations.\"\"\"\n\n def __init__(self):\n \"\"\"Initialize reporter with logger.\"\"\"\n self.logger = get_logger(\"mcp-contracts\", Config)\n\n def log_operation(\n self,\n tool_name: str,\n request_data: dict[str, Any],\n status: str,\n duration_ms: int,\n error_message: Optional[str] = None,\n ) -> None:\n \"\"\"\n Log a contract operation.\n\n Args:\n tool_name: Name of the tool\/operation\n request_data: Request parameters as dict\n status: Operation status (success, error, denied)\n duration_ms: Duration in milliseconds\n error_message: Optional error message\n \"\"\"\n request_str = json.dumps(request_data)\n error_truncated = None\n if error_message:\n error_truncated = error_message[:ERROR_MESSAGE_MAX_LENGTH]\n\n log_entry = LogEntry(\n timestamp=datetime.now(),\n client_name=\"mcp-contracts\",\n tool_name=tool_name,\n request=request_str,\n status=status,\n duration_ms=duration_ms,\n error_message=error_truncated,\n )\n self.logger.log(log_entry)\n\n def execute_with_logging(\n self,\n tool_name: str,\n request_data: dict[str, Any],\n operation: Callable[[], dict[str, Any]],\n ) -> dict[str, Any]:\n \"\"\"\n Execute an operation with automatic logging.\n\n Args:\n tool_name: Name of the tool\/operation\n request_data: Request parameters as dict\n operation: Callable that executes the operation\n\n Returns:\n Operation result dictionary\n \"\"\"\n start = time.time()\n\n try:\n result = operation()\n duration = int((time.time() - start) * MS_PER_SECOND)\n\n # Log success\n self.log_operation(\n tool_name=tool_name,\n request_data=request_data,\n status=LOG_STATUS_SUCCESS,\n duration_ms=duration,\n )\n\n return result\n\n except Exception as e:\n duration = int((time.time() - start) * MS_PER_SECOND)\n\n # Log error\n self.log_operation(\n tool_name=tool_name,\n request_data=request_data,\n status=LOG_STATUS_ERROR,\n duration_ms=duration,\n error_message=str(e),\n )\n\n return {\"success\": False, \"error\": str(e)}\n\n def log_not_found(\n self,\n tool_name: str,\n request_data: dict[str, Any],\n message: str = \"Contract not found\",\n ) -> None:\n \"\"\"\n Log a not-found error.\n\n Args:\n tool_name: Name of the tool\/operation\n request_data: Request parameters as dict\n message: Error message\n \"\"\"\n self.log_operation(\n tool_name=tool_name,\n request_data=request_data,\n status=LOG_STATUS_DENIED,\n duration_ms=0,\n error_message=message,\n )\n\n @staticmethod\n def format_success_response(data: dict[str, Any]) -> dict[str, Any]:\n \"\"\"\n Format a successful response.\n\n Args:\n data: Response data\n\n Returns:\n Formatted response with success flag\n \"\"\"\n return {\"success\": True, **data}\n\n @staticmethod\n def format_error_response(error: str) -> dict[str, Any]:\n \"\"\"\n Format an error response.\n\n Args:\n error: Error message\n\n Returns:\n Formatted error response\n \"\"\"\n return {\"success\": False, \"error\": error}\n\n @staticmethod\n def format_list_response(\n items: list[Any],\n total: int,\n limit: int,\n compact: bool = True,\n ) -> dict[str, Any]:\n \"\"\"\n Format a list response.\n\n Args:\n items: List of items (with to_dict or to_dict_compact methods)\n total: Total count\n limit: Limit used\n compact: Whether to use compact format\n\n Returns:\n Formatted list response\n \"\"\"\n formatted_items = []\n for item in items:\n if hasattr(item, \"to_dict_compact\") and compact:\n formatted_items.append(item.to_dict_compact())\n elif hasattr(item, \"to_dict\"):\n formatted_items.append(item.to_dict())\n else:\n formatted_items.append(item)\n\n return ContractReporter.format_success_response(\n {\n \"contracts\": formatted_items,\n \"total\": total,\n \"limit\": limit,\n }\n )\n"
}
}
{
"tool_response": {
"type": "update",
"filePath": "\/var\/www\/mcp-servers\/mcp_contracts\/tools\/contract_tools_components\/contract_reporter.py",
"content": "\"\"\"Contract Reporting and Logging Module.\"\"\"\n\nimport json\nimport time\nfrom datetime import datetime\nfrom typing import Any, Callable, Optional\n\nfrom config import Config\nfrom shared.domain import LogEntry\nfrom shared.infrastructure import get_logger\n\nfrom .constants import (\n ERROR_MESSAGE_MAX_LENGTH,\n LOG_STATUS_DENIED,\n LOG_STATUS_ERROR,\n LOG_STATUS_SUCCESS,\n MS_PER_SECOND,\n)\n\n\nclass ContractReporter:\n \"\"\"Handles logging and response formatting for contract operations.\"\"\"\n\n def __init__(self):\n \"\"\"Initialize reporter with logger.\"\"\"\n self.logger = get_logger(\"mcp-contracts\", Config)\n\n def log_operation(\n self,\n tool_name: str,\n request_data: dict[str, Any],\n status: str,\n duration_ms: int,\n error_message: Optional[str] = None,\n ) -> None:\n \"\"\"\n Log a contract operation.\n\n Args:\n tool_name: Name of the tool\/operation\n request_data: Request parameters as dict\n status: Operation status (success, error, denied)\n duration_ms: Duration in milliseconds\n error_message: Optional error message\n \"\"\"\n request_str = json.dumps(request_data)\n error_truncated = None\n if error_message:\n error_truncated = error_message[:ERROR_MESSAGE_MAX_LENGTH]\n\n log_entry = LogEntry(\n timestamp=datetime.now(),\n client_name=\"mcp-contracts\",\n tool_name=tool_name,\n request=request_str,\n status=status,\n duration_ms=duration_ms,\n error_message=error_truncated,\n )\n self.logger.log(log_entry)\n\n def execute_with_logging(\n self,\n tool_name: str,\n request_data: dict[str, Any],\n operation: Callable[[], dict[str, Any]],\n ) -> dict[str, Any]:\n \"\"\"\n Execute an operation with automatic logging.\n\n Args:\n tool_name: Name of the tool\/operation\n request_data: Request parameters as dict\n operation: Callable that executes the operation\n\n Returns:\n Operation result dictionary\n \"\"\"\n start = time.time()\n\n try:\n result = operation()\n duration = int((time.time() - start) * MS_PER_SECOND)\n\n # Log success\n self.log_operation(\n tool_name=tool_name,\n request_data=request_data,\n status=LOG_STATUS_SUCCESS,\n duration_ms=duration,\n )\n\n return result\n\n except Exception as e:\n duration = int((time.time() - start) * MS_PER_SECOND)\n\n # Log error\n self.log_operation(\n tool_name=tool_name,\n request_data=request_data,\n status=LOG_STATUS_ERROR,\n duration_ms=duration,\n error_message=str(e),\n )\n\n return {\"success\": False, \"error\": str(e)}\n\n def log_not_found(\n self,\n tool_name: str,\n request_data: dict[str, Any],\n message: str = \"Contract not found\",\n ) -> None:\n \"\"\"\n Log a not-found error.\n\n Args:\n tool_name: Name of the tool\/operation\n request_data: Request parameters as dict\n message: Error message\n \"\"\"\n self.log_operation(\n tool_name=tool_name,\n request_data=request_data,\n status=LOG_STATUS_DENIED,\n duration_ms=0,\n error_message=message,\n )\n\n @staticmethod\n def format_success_response(data: dict[str, Any]) -> dict[str, Any]:\n \"\"\"\n Format a successful response.\n\n Args:\n data: Response data\n\n Returns:\n Formatted response with success flag\n \"\"\"\n return {\"success\": True, **data}\n\n @staticmethod\n def format_error_response(error: str) -> dict[str, Any]:\n \"\"\"\n Format an error response.\n\n Args:\n error: Error message\n\n Returns:\n Formatted error response\n \"\"\"\n return {\"success\": False, \"error\": error}\n\n @staticmethod\n def format_list_response(\n items: list[Any],\n total: int,\n limit: int,\n compact: bool = True,\n ) -> dict[str, Any]:\n \"\"\"\n Format a list response.\n\n Args:\n items: List of items (with to_dict or to_dict_compact methods)\n total: Total count\n limit: Limit used\n compact: Whether to use compact format\n\n Returns:\n Formatted list response\n \"\"\"\n formatted_items = []\n for item in items:\n if hasattr(item, \"to_dict_compact\") and compact:\n formatted_items.append(item.to_dict_compact())\n elif hasattr(item, \"to_dict\"):\n formatted_items.append(item.to_dict())\n else:\n formatted_items.append(item)\n\n return ContractReporter.format_success_response(\n {\n \"contracts\": formatted_items,\n \"total\": total,\n \"limit\": limit,\n }\n )\n",
"structuredPatch": [
{
"oldStart": 1,
"oldLines": 26,
"newStart": 1,
"newLines": 29,
"lines": [
"-\"\"\"Contract Reporting and Logging Module\"\"\"",
"-import time",
"+\"\"\"Contract Reporting and Logging Module.\"\"\"",
"+",
" import json",
"-from typing import Optional, Any, Callable",
"+import time",
"+from datetime import datetime",
"+from typing import Any, Callable, Optional",
" ",
"-from domain.contracts import LogEntry",
"-from infrastructure.protokoll_logger import get_logger",
"+from config import Config",
"+from shared.domain import LogEntry",
"+from shared.infrastructure import get_logger",
" ",
" from .constants import (",
" ERROR_MESSAGE_MAX_LENGTH,",
"- LOG_STATUS_SUCCESS,",
"- LOG_STATUS_ERROR,",
" LOG_STATUS_DENIED,",
"- MS_PER_SECOND",
"+ LOG_STATUS_ERROR,",
"+ LOG_STATUS_SUCCESS,",
"+ MS_PER_SECOND,",
" )",
" ",
" ",
" class ContractReporter:",
"- \"\"\"Handles logging and response formatting for contract operations\"\"\"",
"+ \"\"\"Handles logging and response formatting for contract operations.\"\"\"",
" ",
" def __init__(self):",
"- \"\"\"Initialize reporter with logger\"\"\"",
"- self.logger = get_logger()",
"+ \"\"\"Initialize reporter with logger.\"\"\"",
"+ self.logger = get_logger(\"mcp-contracts\", Config)",
" ",
" def log_operation(",
" self,"
]
},
{
"oldStart": 28,
"oldLines": 7,
"newStart": 31,
"newLines": 7,
"lines": [
" request_data: dict[str, Any],",
" status: str,",
" duration_ms: int,",
"- error_message: Optional[str] = None",
"+ error_message: Optional[str] = None,",
" ) -> None:",
" \"\"\"",
" Log a contract operation."
]
},
{
"oldStart": 46,
"oldLines": 6,
"newStart": 49,
"newLines": 8,
"lines": [
" error_truncated = error_message[:ERROR_MESSAGE_MAX_LENGTH]",
" ",
" log_entry = LogEntry(",
"+ timestamp=datetime.now(),",
"+ client_name=\"mcp-contracts\",",
" tool_name=tool_name,",
" request=request_str,",
" status=status,"
]
},
{
"oldStart": 58,
"oldLines": 7,
"newStart": 63,
"newLines": 7,
"lines": [
" self,",
" tool_name: str,",
" request_data: dict[str, Any],",
"- operation: Callable[[], dict[str, Any]]",
"+ operation: Callable[[], dict[str, Any]],",
" ) -> dict[str, Any]:",
" \"\"\"",
" Execute an operation with automatic logging."
]
},
{
"oldStart": 82,
"oldLines": 7,
"newStart": 87,
"newLines": 7,
"lines": [
" tool_name=tool_name,",
" request_data=request_data,",
" status=LOG_STATUS_SUCCESS,",
"- duration_ms=duration",
"+ duration_ms=duration,",
" )",
" ",
" return result"
]
},
{
"oldStart": 96,
"oldLines": 7,
"newStart": 101,
"newLines": 7,
"lines": [
" request_data=request_data,",
" status=LOG_STATUS_ERROR,",
" duration_ms=duration,",
"- error_message=str(e)",
"+ error_message=str(e),",
" )",
" ",
" return {\"success\": False, \"error\": str(e)}"
]
},
{
"oldStart": 105,
"oldLines": 7,
"newStart": 110,
"newLines": 7,
"lines": [
" self,",
" tool_name: str,",
" request_data: dict[str, Any],",
"- message: str = \"Contract not found\"",
"+ message: str = \"Contract not found\",",
" ) -> None:",
" \"\"\"",
" Log a not-found error."
]
},
{
"oldStart": 120,
"oldLines": 7,
"newStart": 125,
"newLines": 7,
"lines": [
" request_data=request_data,",
" status=LOG_STATUS_DENIED,",
" duration_ms=0,",
"- error_message=message",
"+ error_message=message,",
" )",
" ",
" @staticmethod"
]
},
{
"oldStart": 154,
"oldLines": 7,
"newStart": 159,
"newLines": 7,
"lines": [
" items: list[Any],",
" total: int,",
" limit: int,",
"- compact: bool = True",
"+ compact: bool = True,",
" ) -> dict[str, Any]:",
" \"\"\"",
" Format a list response."
]
},
{
"oldStart": 170,
"oldLines": 15,
"newStart": 175,
"newLines": 17,
"lines": [
" \"\"\"",
" formatted_items = []",
" for item in items:",
"- if hasattr(item, 'to_dict_compact') and compact:",
"+ if hasattr(item, \"to_dict_compact\") and compact:",
" formatted_items.append(item.to_dict_compact())",
"- elif hasattr(item, 'to_dict'):",
"+ elif hasattr(item, \"to_dict\"):",
" formatted_items.append(item.to_dict())",
" else:",
" formatted_items.append(item)",
" ",
"- return ContractReporter.format_success_response({",
"- \"contracts\": formatted_items,",
"- \"total\": total,",
"- \"limit\": limit,",
"- })",
"+ return ContractReporter.format_success_response(",
"+ {",
"+ \"contracts\": formatted_items,",
"+ \"total\": total,",
"+ \"limit\": limit,",
"+ }",
"+ )"
]
}
],
"originalFile": "\"\"\"Contract Reporting and Logging Module\"\"\"\nimport time\nimport json\nfrom typing import Optional, Any, Callable\n\nfrom domain.contracts import LogEntry\nfrom infrastructure.protokoll_logger import get_logger\n\nfrom .constants import (\n ERROR_MESSAGE_MAX_LENGTH,\n LOG_STATUS_SUCCESS,\n LOG_STATUS_ERROR,\n LOG_STATUS_DENIED,\n MS_PER_SECOND\n)\n\n\nclass ContractReporter:\n \"\"\"Handles logging and response formatting for contract operations\"\"\"\n\n def __init__(self):\n \"\"\"Initialize reporter with logger\"\"\"\n self.logger = get_logger()\n\n def log_operation(\n self,\n tool_name: str,\n request_data: dict[str, Any],\n status: str,\n duration_ms: int,\n error_message: Optional[str] = None\n ) -> None:\n \"\"\"\n Log a contract operation.\n\n Args:\n tool_name: Name of the tool\/operation\n request_data: Request parameters as dict\n status: Operation status (success, error, denied)\n duration_ms: Duration in milliseconds\n error_message: Optional error message\n \"\"\"\n request_str = json.dumps(request_data)\n error_truncated = None\n if error_message:\n error_truncated = error_message[:ERROR_MESSAGE_MAX_LENGTH]\n\n log_entry = LogEntry(\n tool_name=tool_name,\n request=request_str,\n status=status,\n duration_ms=duration_ms,\n error_message=error_truncated,\n )\n self.logger.log(log_entry)\n\n def execute_with_logging(\n self,\n tool_name: str,\n request_data: dict[str, Any],\n operation: Callable[[], dict[str, Any]]\n ) -> dict[str, Any]:\n \"\"\"\n Execute an operation with automatic logging.\n\n Args:\n tool_name: Name of the tool\/operation\n request_data: Request parameters as dict\n operation: Callable that executes the operation\n\n Returns:\n Operation result dictionary\n \"\"\"\n start = time.time()\n\n try:\n result = operation()\n duration = int((time.time() - start) * MS_PER_SECOND)\n\n # Log success\n self.log_operation(\n tool_name=tool_name,\n request_data=request_data,\n status=LOG_STATUS_SUCCESS,\n duration_ms=duration\n )\n\n return result\n\n except Exception as e:\n duration = int((time.time() - start) * MS_PER_SECOND)\n\n # Log error\n self.log_operation(\n tool_name=tool_name,\n request_data=request_data,\n status=LOG_STATUS_ERROR,\n duration_ms=duration,\n error_message=str(e)\n )\n\n return {\"success\": False, \"error\": str(e)}\n\n def log_not_found(\n self,\n tool_name: str,\n request_data: dict[str, Any],\n message: str = \"Contract not found\"\n ) -> None:\n \"\"\"\n Log a not-found error.\n\n Args:\n tool_name: Name of the tool\/operation\n request_data: Request parameters as dict\n message: Error message\n \"\"\"\n self.log_operation(\n tool_name=tool_name,\n request_data=request_data,\n status=LOG_STATUS_DENIED,\n duration_ms=0,\n error_message=message\n )\n\n @staticmethod\n def format_success_response(data: dict[str, Any]) -> dict[str, Any]:\n \"\"\"\n Format a successful response.\n\n Args:\n data: Response data\n\n Returns:\n Formatted response with success flag\n \"\"\"\n return {\"success\": True, **data}\n\n @staticmethod\n def format_error_response(error: str) -> dict[str, Any]:\n \"\"\"\n Format an error response.\n\n Args:\n error: Error message\n\n Returns:\n Formatted error response\n \"\"\"\n return {\"success\": False, \"error\": error}\n\n @staticmethod\n def format_list_response(\n items: list[Any],\n total: int,\n limit: int,\n compact: bool = True\n ) -> dict[str, Any]:\n \"\"\"\n Format a list response.\n\n Args:\n items: List of items (with to_dict or to_dict_compact methods)\n total: Total count\n limit: Limit used\n compact: Whether to use compact format\n\n Returns:\n Formatted list response\n \"\"\"\n formatted_items = []\n for item in items:\n if hasattr(item, 'to_dict_compact') and compact:\n formatted_items.append(item.to_dict_compact())\n elif hasattr(item, 'to_dict'):\n formatted_items.append(item.to_dict())\n else:\n formatted_items.append(item)\n\n return ContractReporter.format_success_response({\n \"contracts\": formatted_items,\n \"total\": total,\n \"limit\": limit,\n })\n"
}
}