{
"event": "PostToolUse",
"tool_name": "Read",
"tool_response": {
"type": "text",
"file": {
"filePath": "\/var\/www\/mcp-servers\/mcp_tasks\/tools\/task_tools\/base.py",
"content": "\"\"\"Base utilities for task tools.\"\"\"\nimport json\nimport time\nfrom typing import Any, Callable\nfrom functools import wraps\n\nimport sys\nsys.path.insert(0, \"\/var\/www\/mcp-servers\/mcp_tasks\")\n\nfrom config import Config\nfrom domain.contracts import LogEntry\nfrom infrastructure.task_repository import TaskRepository\nfrom infrastructure.protokoll_logger import get_logger\n\n\ndef get_repo():\n \"\"\"Get TaskRepository singleton.\"\"\"\n return TaskRepository()\n\n\ndef get_task_logger():\n \"\"\"Get protokoll logger singleton.\"\"\"\n return get_logger()\n\n\ndef validate_status(status: str) -> tuple[bool, str | None]:\n \"\"\"Validate status value.\"\"\"\n if status and status not in Config.VALID_STATUSES:\n return False, f\"Invalid status: {status}\"\n return True, None\n\n\ndef validate_type(task_type: str) -> tuple[bool, str | None]:\n \"\"\"Validate task type value.\"\"\"\n if task_type and task_type not in Config.VALID_TYPES:\n return False, f\"Invalid type: {task_type}\"\n return True, None\n\n\ndef validate_executor_type(executor_type: str) -> tuple[bool, str | None]:\n \"\"\"Validate executor type value.\"\"\"\n if executor_type not in Config.VALID_EXECUTOR_TYPES:\n return False, f\"Invalid executor_type: {executor_type}\"\n return True, None\n\n\ndef log_tool_call(\n logger,\n tool_name: str,\n request_str: str,\n status: str,\n duration_ms: int = 0,\n task_id: int | None = None,\n error_message: str | None = None,\n):\n \"\"\"Log a tool call to protokoll.\"\"\"\n logger.log(LogEntry(\n tool_name=tool_name,\n request=request_str,\n status=status,\n duration_ms=duration_ms,\n task_id=task_id,\n error_message=error_message[:200] if error_message else None,\n ))\n",
"numLines": 65,
"startLine": 1,
"totalLines": 65
}
}
}