{
"event": "PreToolUse",
"tool_name": "Write",
"tool_input": {
"file_path": "\/var\/www\/mcp-servers\/mcp-tasks\/tools\/task_tools\/crud_tools.py",
"content": "\"\"\"Task CRUD tools - Get, Create, Update, Delete.\"\"\"\nimport json\nimport time\nfrom datetime import datetime\n\nimport sys\nsys.path.insert(0, \"\/var\/www\/mcp-servers\/mcp-tasks\")\n\nfrom config import Config\nfrom domain.contracts import Task, TaskStatus, TaskType\nfrom .base import get_repo, get_task_logger, validate_type, log_tool_call\n\n\ndef register_crud_tools(mcp):\n \"\"\"Register task CRUD tools.\"\"\"\n\n repo = get_repo()\n logger = get_task_logger()\n\n @mcp.tool()\n def tasks_get(id: int) -> dict:\n \"\"\"\n Holt Details eines einzelnen Tasks inkl. Assignments und Results.\n\n Args:\n id: Task-ID\n\n Returns:\n Task mit allen Details oder Fehlermeldung\n \"\"\"\n start = time.time()\n request_str = json.dumps({\"id\": id})\n\n try:\n task = repo.find_by_id(id)\n\n if not task:\n log_tool_call(logger, \"tasks_get\", request_str, \"denied\", task_id=id, error_message=\"Task not found\")\n return {\"success\": False, \"error\": f\"Task {id} not found\"}\n\n assignments = repo.get_assignments(id)\n results = repo.get_results(id)\n\n duration = int((time.time() - start) * 1000)\n log_tool_call(logger, \"tasks_get\", request_str, \"success\", duration, task_id=id)\n\n return {\n \"success\": True,\n \"task\": task.to_dict(),\n \"assignments\": [a.to_dict() for a in assignments],\n \"results\": [r.to_dict() for r in results],\n }\n\n except Exception as e:\n duration = int((time.time() - start) * 1000)\n log_tool_call(logger, \"tasks_get\", request_str, \"error\", duration, task_id=id, error_message=str(e))\n return {\"success\": False, \"error\": str(e)}\n\n @mcp.tool()\n def tasks_create(\n title: str,\n description: str | None = None,\n type: str = \"ai_task\",\n parent_task_id: int | None = None,\n due_date: str | None = None,\n ) -> dict:\n \"\"\"\n Erstellt einen neuen Task.\n\n Args:\n title: Aufgabentitel (required)\n description: Ausführliche Beschreibung\n type: Task-Typ (human_task, ai_task, mixed)\n parent_task_id: ID des Parent-Tasks für Subtasks\n due_date: Fälligkeitsdatum (ISO 8601)\n\n Returns:\n Erstellter Task oder Fehlermeldung\n \"\"\"\n start = time.time()\n request_str = json.dumps({\"title\": title[:100], \"type\": type})\n\n try:\n if not title or not title.strip():\n log_tool_call(logger, \"tasks_create\", request_str, \"denied\", error_message=\"Title is required\")\n return {\"success\": False, \"error\": \"Title is required\"}\n\n valid, error = validate_type(type)\n if not valid:\n log_tool_call(logger, \"tasks_create\", request_str, \"denied\", error_message=error)\n return {\"success\": False, \"error\": error}\n\n task = Task(\n title=title.strip(),\n description=description,\n type=TaskType(type),\n status=TaskStatus.PENDING,\n created_by=\"mcp-tasks\",\n created_by_type=\"ai\",\n parent_task_id=parent_task_id,\n due_date=datetime.fromisoformat(due_date) if due_date else None,\n )\n\n task_id = repo.create(task)\n task.id = task_id\n\n duration = int((time.time() - start) * 1000)\n log_tool_call(logger, \"tasks_create\", request_str, \"success\", duration, task_id=task_id)\n\n return {\"success\": True, \"task\": task.to_dict(), \"message\": f\"Task #{task_id} created\"}\n\n except Exception as e:\n duration = int((time.time() - start) * 1000)\n log_tool_call(logger, \"tasks_create\", request_str, \"error\", duration, error_message=str(e))\n return {\"success\": False, \"error\": str(e)}\n\n @mcp.tool()\n def tasks_update(\n id: int,\n title: str | None = None,\n description: str | None = None,\n type: str | None = None,\n due_date: str | None = None,\n ) -> dict:\n \"\"\"\n Aktualisiert Task-Felder.\n\n Args:\n id: Task-ID (required)\n title: Neuer Titel\n description: Neue Beschreibung\n type: Neuer Typ\n due_date: Neues Datum (ISO 8601)\n\n Returns:\n Aktualisierter Task oder Fehlermeldung\n \"\"\"\n start = time.time()\n request_str = json.dumps({\"id\": id})\n\n try:\n task = repo.find_by_id(id)\n if not task:\n log_tool_call(logger, \"tasks_update\", request_str, \"denied\", task_id=id, error_message=\"Task not found\")\n return {\"success\": False, \"error\": f\"Task {id} not found\"}\n\n updates = {}\n if title is not None:\n updates[\"title\"] = title.strip()\n if description is not None:\n updates[\"description\"] = description\n if type is not None:\n valid, error = validate_type(type)\n if not valid:\n return {\"success\": False, \"error\": error}\n updates[\"type\"] = type\n if due_date is not None:\n updates[\"due_date\"] = datetime.fromisoformat(due_date)\n\n if not updates:\n return {\"success\": False, \"error\": \"No fields to update\"}\n\n repo.update(id, updates)\n updated_task = repo.find_by_id(id)\n\n duration = int((time.time() - start) * 1000)\n log_tool_call(logger, \"tasks_update\", request_str, \"success\", duration, task_id=id)\n\n return {\"success\": True, \"task\": updated_task.to_dict(), \"message\": f\"Task #{id} updated\"}\n\n except Exception as e:\n duration = int((time.time() - start) * 1000)\n log_tool_call(logger, \"tasks_update\", request_str, \"error\", duration, task_id=id, error_message=str(e))\n return {\"success\": False, \"error\": str(e)}\n\n @mcp.tool()\n def tasks_delete(id: int) -> dict:\n \"\"\"\n Löscht einen Task.\n\n Args:\n id: Task-ID (required)\n\n Returns:\n Bestätigung oder Fehlermeldung\n \"\"\"\n start = time.time()\n request_str = json.dumps({\"id\": id})\n\n try:\n task = repo.find_by_id(id)\n if not task:\n log_tool_call(logger, \"tasks_delete\", request_str, \"denied\", task_id=id, error_message=\"Task not found\")\n return {\"success\": False, \"error\": f\"Task {id} not found\"}\n\n deleted = repo.delete(id)\n\n duration = int((time.time() - start) * 1000)\n log_tool_call(logger, \"tasks_delete\", request_str, \"success\" if deleted else \"error\", duration, task_id=id)\n\n return {\n \"success\": deleted,\n \"message\": f\"Task #{id} deleted\" if deleted else f\"Failed to delete Task #{id}\",\n }\n\n except Exception as e:\n duration = int((time.time() - start) * 1000)\n log_tool_call(logger, \"tasks_delete\", request_str, \"error\", duration, task_id=id, error_message=str(e))\n return {\"success\": False, \"error\": str(e)}\n"
}
}
{
"tool_response": {
"type": "create",
"filePath": "\/var\/www\/mcp-servers\/mcp-tasks\/tools\/task_tools\/crud_tools.py",
"content": "\"\"\"Task CRUD tools - Get, Create, Update, Delete.\"\"\"\nimport json\nimport time\nfrom datetime import datetime\n\nimport sys\nsys.path.insert(0, \"\/var\/www\/mcp-servers\/mcp-tasks\")\n\nfrom config import Config\nfrom domain.contracts import Task, TaskStatus, TaskType\nfrom .base import get_repo, get_task_logger, validate_type, log_tool_call\n\n\ndef register_crud_tools(mcp):\n \"\"\"Register task CRUD tools.\"\"\"\n\n repo = get_repo()\n logger = get_task_logger()\n\n @mcp.tool()\n def tasks_get(id: int) -> dict:\n \"\"\"\n Holt Details eines einzelnen Tasks inkl. Assignments und Results.\n\n Args:\n id: Task-ID\n\n Returns:\n Task mit allen Details oder Fehlermeldung\n \"\"\"\n start = time.time()\n request_str = json.dumps({\"id\": id})\n\n try:\n task = repo.find_by_id(id)\n\n if not task:\n log_tool_call(logger, \"tasks_get\", request_str, \"denied\", task_id=id, error_message=\"Task not found\")\n return {\"success\": False, \"error\": f\"Task {id} not found\"}\n\n assignments = repo.get_assignments(id)\n results = repo.get_results(id)\n\n duration = int((time.time() - start) * 1000)\n log_tool_call(logger, \"tasks_get\", request_str, \"success\", duration, task_id=id)\n\n return {\n \"success\": True,\n \"task\": task.to_dict(),\n \"assignments\": [a.to_dict() for a in assignments],\n \"results\": [r.to_dict() for r in results],\n }\n\n except Exception as e:\n duration = int((time.time() - start) * 1000)\n log_tool_call(logger, \"tasks_get\", request_str, \"error\", duration, task_id=id, error_message=str(e))\n return {\"success\": False, \"error\": str(e)}\n\n @mcp.tool()\n def tasks_create(\n title: str,\n description: str | None = None,\n type: str = \"ai_task\",\n parent_task_id: int | None = None,\n due_date: str | None = None,\n ) -> dict:\n \"\"\"\n Erstellt einen neuen Task.\n\n Args:\n title: Aufgabentitel (required)\n description: Ausführliche Beschreibung\n type: Task-Typ (human_task, ai_task, mixed)\n parent_task_id: ID des Parent-Tasks für Subtasks\n due_date: Fälligkeitsdatum (ISO 8601)\n\n Returns:\n Erstellter Task oder Fehlermeldung\n \"\"\"\n start = time.time()\n request_str = json.dumps({\"title\": title[:100], \"type\": type})\n\n try:\n if not title or not title.strip():\n log_tool_call(logger, \"tasks_create\", request_str, \"denied\", error_message=\"Title is required\")\n return {\"success\": False, \"error\": \"Title is required\"}\n\n valid, error = validate_type(type)\n if not valid:\n log_tool_call(logger, \"tasks_create\", request_str, \"denied\", error_message=error)\n return {\"success\": False, \"error\": error}\n\n task = Task(\n title=title.strip(),\n description=description,\n type=TaskType(type),\n status=TaskStatus.PENDING,\n created_by=\"mcp-tasks\",\n created_by_type=\"ai\",\n parent_task_id=parent_task_id,\n due_date=datetime.fromisoformat(due_date) if due_date else None,\n )\n\n task_id = repo.create(task)\n task.id = task_id\n\n duration = int((time.time() - start) * 1000)\n log_tool_call(logger, \"tasks_create\", request_str, \"success\", duration, task_id=task_id)\n\n return {\"success\": True, \"task\": task.to_dict(), \"message\": f\"Task #{task_id} created\"}\n\n except Exception as e:\n duration = int((time.time() - start) * 1000)\n log_tool_call(logger, \"tasks_create\", request_str, \"error\", duration, error_message=str(e))\n return {\"success\": False, \"error\": str(e)}\n\n @mcp.tool()\n def tasks_update(\n id: int,\n title: str | None = None,\n description: str | None = None,\n type: str | None = None,\n due_date: str | None = None,\n ) -> dict:\n \"\"\"\n Aktualisiert Task-Felder.\n\n Args:\n id: Task-ID (required)\n title: Neuer Titel\n description: Neue Beschreibung\n type: Neuer Typ\n due_date: Neues Datum (ISO 8601)\n\n Returns:\n Aktualisierter Task oder Fehlermeldung\n \"\"\"\n start = time.time()\n request_str = json.dumps({\"id\": id})\n\n try:\n task = repo.find_by_id(id)\n if not task:\n log_tool_call(logger, \"tasks_update\", request_str, \"denied\", task_id=id, error_message=\"Task not found\")\n return {\"success\": False, \"error\": f\"Task {id} not found\"}\n\n updates = {}\n if title is not None:\n updates[\"title\"] = title.strip()\n if description is not None:\n updates[\"description\"] = description\n if type is not None:\n valid, error = validate_type(type)\n if not valid:\n return {\"success\": False, \"error\": error}\n updates[\"type\"] = type\n if due_date is not None:\n updates[\"due_date\"] = datetime.fromisoformat(due_date)\n\n if not updates:\n return {\"success\": False, \"error\": \"No fields to update\"}\n\n repo.update(id, updates)\n updated_task = repo.find_by_id(id)\n\n duration = int((time.time() - start) * 1000)\n log_tool_call(logger, \"tasks_update\", request_str, \"success\", duration, task_id=id)\n\n return {\"success\": True, \"task\": updated_task.to_dict(), \"message\": f\"Task #{id} updated\"}\n\n except Exception as e:\n duration = int((time.time() - start) * 1000)\n log_tool_call(logger, \"tasks_update\", request_str, \"error\", duration, task_id=id, error_message=str(e))\n return {\"success\": False, \"error\": str(e)}\n\n @mcp.tool()\n def tasks_delete(id: int) -> dict:\n \"\"\"\n Löscht einen Task.\n\n Args:\n id: Task-ID (required)\n\n Returns:\n Bestätigung oder Fehlermeldung\n \"\"\"\n start = time.time()\n request_str = json.dumps({\"id\": id})\n\n try:\n task = repo.find_by_id(id)\n if not task:\n log_tool_call(logger, \"tasks_delete\", request_str, \"denied\", task_id=id, error_message=\"Task not found\")\n return {\"success\": False, \"error\": f\"Task {id} not found\"}\n\n deleted = repo.delete(id)\n\n duration = int((time.time() - start) * 1000)\n log_tool_call(logger, \"tasks_delete\", request_str, \"success\" if deleted else \"error\", duration, task_id=id)\n\n return {\n \"success\": deleted,\n \"message\": f\"Task #{id} deleted\" if deleted else f\"Failed to delete Task #{id}\",\n }\n\n except Exception as e:\n duration = int((time.time() - start) * 1000)\n log_tool_call(logger, \"tasks_delete\", request_str, \"error\", duration, task_id=id, error_message=str(e))\n return {\"success\": False, \"error\": str(e)}\n",
"structuredPatch": [],
"originalFile": null
}
}