{
"event": "PreToolUse",
"tool_name": "Write",
"tool_input": {
"file_path": "\/var\/www\/mcp-servers\/mcp_tasks\/tools\/task_tools\/base.py",
"content": "\"\"\"Base utilities for task tools.\"\"\"\n\nimport json\nfrom datetime import datetime\nfrom typing import Any\n\nfrom config import Config\nfrom infrastructure.task_repository import TaskRepository\nfrom shared.domain import LogEntry\nfrom shared.infrastructure 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(\"mcp-tasks\", Config)\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(\n LogEntry(\n timestamp=datetime.now(),\n client_name=\"mcp-tasks\",\n tool_name=tool_name,\n request=request_str,\n status=status,\n duration_ms=duration_ms,\n context_id=task_id, # task_id maps to context_id in shared LogEntry\n error_message=error_message[:200] if error_message else None,\n )\n )\n"
}
}
{
"tool_response": {
"type": "update",
"filePath": "\/var\/www\/mcp-servers\/mcp_tasks\/tools\/task_tools\/base.py",
"content": "\"\"\"Base utilities for task tools.\"\"\"\n\nimport json\nfrom datetime import datetime\nfrom typing import Any\n\nfrom config import Config\nfrom infrastructure.task_repository import TaskRepository\nfrom shared.domain import LogEntry\nfrom shared.infrastructure 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(\"mcp-tasks\", Config)\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(\n LogEntry(\n timestamp=datetime.now(),\n client_name=\"mcp-tasks\",\n tool_name=tool_name,\n request=request_str,\n status=status,\n duration_ms=duration_ms,\n context_id=task_id, # task_id maps to context_id in shared LogEntry\n error_message=error_message[:200] if error_message else None,\n )\n )\n",
"structuredPatch": [
{
"oldStart": 1,
"oldLines": 16,
"newStart": 1,
"newLines": 13,
"lines": [
" \"\"\"Base utilities for task tools.\"\"\"",
"+",
" import json",
"-import time",
"-from typing import Any, Callable",
"-from functools import wraps",
"+from datetime import datetime",
"+from typing import Any",
" ",
"-import sys",
"-sys.path.insert(0, \"\/var\/www\/mcp-servers\/mcp_tasks\")",
"-",
" from config import Config",
"-from domain.contracts import LogEntry",
" from infrastructure.task_repository import TaskRepository",
"-from infrastructure.protokoll_logger import get_logger",
"+from shared.domain import LogEntry",
"+from shared.infrastructure import get_logger",
" ",
" ",
" def get_repo():"
]
},
{
"oldStart": 20,
"oldLines": 7,
"newStart": 17,
"newLines": 7,
"lines": [
" ",
" def get_task_logger():",
" \"\"\"Get protokoll logger singleton.\"\"\"",
"- return get_logger()",
"+ return get_logger(\"mcp-tasks\", Config)",
" ",
" ",
" def validate_status(status: str) -> tuple[bool, str | None]:"
]
},
{
"oldStart": 54,
"oldLines": 11,
"newStart": 51,
"newLines": 15,
"lines": [
" error_message: str | None = None,",
" ):",
" \"\"\"Log a tool call to protokoll.\"\"\"",
"- logger.log(LogEntry(",
"- tool_name=tool_name,",
"- request=request_str,",
"- status=status,",
"- duration_ms=duration_ms,",
"- task_id=task_id,",
"- error_message=error_message[:200] if error_message else None,",
"- ))",
"+ logger.log(",
"+ LogEntry(",
"+ timestamp=datetime.now(),",
"+ client_name=\"mcp-tasks\",",
"+ tool_name=tool_name,",
"+ request=request_str,",
"+ status=status,",
"+ duration_ms=duration_ms,",
"+ context_id=task_id, # task_id maps to context_id in shared LogEntry",
"+ error_message=error_message[:200] if error_message else None,",
"+ )",
"+ )"
]
}
],
"originalFile": "\"\"\"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"
}
}