feat: stop tracking task & get active tracking task

* stop tracking task
* get active tracking task
* some enhancements

Author: reminia

.env.example

@@ -1,4 +1,4 @@
 LINEAR_API_KEY=your_linear_api_key_here
-TIMETRACKING_API_KEY=your_trackingtime_api_key_here 
 # can be empty
-LINEAR_TEAM=your_team
+LINEAR_TEAM=your_linear_team
+TRACKINGTIME_API_KEY=your_trackingtime_api_key_here 

…/task_tracker/api/timetracking_client.py …/task_tracker/api/trackingtime_client.pysrc/task_tracker/api/timetracking_client.py renamed to src/task_tracker/api/trackingtime_client.py src/task_tracker/api/timetracking_client.py renamed to src/task_tracker/api/trackingtime_client.py

@@ -1,66 +1,77 @@
-from typing import Optional, Union
+from typing import Optional
 import aiohttp
 from datetime import datetime, timezone
 from task_tracker.config import settings
 
 
-class TimeTrackingClient:
+class TrackingTimeClient:
     def __init__(self):
-        self.api_key = settings.TIMETRACKING_API_KEY
+        self.api_key = settings.TRACKINGTIME_API_KEY
         self.base_url = "https://app.trackingtime.co/api/v4"
         self.auth = aiohttp.BasicAuth.decode(f"Basic {self.api_key}")
         self.headers = {"Content-Type": "application/json"}
 
-    async def start_tracking(self, project: str, description: str) -> Union[dict, str]:
-        """Start tracking a task
+    async def start_tracking(self, project: str, description: str) -> dict:
+        """Start time tracking a task
 
         Args:
-            project: Name of the project to create the task in
-            description: Description/name of the task
+            project: name of the project to create the task in
+            description: description of the task
 
         Returns:
-            Union[dict, str]: Response from TrackingTime API - either JSON dict or raw text if not JSON
+            dict: json containing the tracking_task_id
         """
-        async with aiohttp.ClientSession(auth=self.auth) as session:
+        async with aiohttp.ClientSession(auth=self.auth, headers=self.headers) as session:
             current_time = datetime.now(
                 timezone.utc).strftime("%Y-%m-%d %H:%M:%S")
             async with session.post(
                 f"{self.base_url}/tasks/track",
                 params={
                     "date": current_time,
                     "task_name": description,
-                    "project_name": project
+                    "project_name": project,
+                    "return_task": "true"
                 }
             ) as response:
                 response.raise_for_status()
-                return await response.json()
+                full_response = await response.json()
+                return {"tracking_task_id": full_response["data"]["id"]}
 
-    async def stop_tracking(self, entry_id: str) -> dict:
-        """Stop time tracking for a task using TrackingTime API
+    async def stop_tracking(self, task_id: str) -> dict:
+        """Stop time tracking a task given its ID
 
         Args:
-            entry_id: The ID of the time entry to stop
+            task_id: The ID of the task to stop
 
         Returns:
-            dict: Response from TrackingTime API containing the completed time entry
+            dict: Response from TrackingTime API containing the task details
         """
         async with aiohttp.ClientSession(auth=self.auth, headers=self.headers) as session:
+            current_time = datetime.now(
+                timezone.utc).strftime("%Y-%m-%d %H:%M:%S")
             async with session.post(
-                f"{self.base_url}/time_entries/{entry_id}/stop",
-                json={
-                    "end_date": datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M:%S")
+                f"{self.base_url}/tasks/stop",
+                params={
+                    "date": current_time,
+                    "task_id": task_id,
+                    "return_task": "true"
                 }
             ) as response:
                 response.raise_for_status()
                 return await response.json()
 
-    async def get_active_tracking(self) -> Optional[dict]:
-        """Get currently active tracking session if any using TrackingTime API
+    async def get_tracking_task(self, filter: str = "TRACKING") -> Optional[dict]:
+        """Fetch the tracking task with the given filter, defaults to "TRACKING"
 
         Returns:
             Optional[dict]: Active time entry if exists, None otherwise
         """
         async with aiohttp.ClientSession(auth=self.auth, headers=self.headers) as session:
-            async with session.get(f"{self.base_url}/time_entries/current") as response:
+            async with session.get(
+                f"{self.base_url}/tasks",
+                params={
+                    "filter": filter
+                }
+            ) as response:
                 response.raise_for_status()
                 return await response.json()

src/task_tracker/config.py

@@ -4,7 +4,7 @@ class Settings(BaseSettings):
     LINEAR_API_KEY: str
     LINEAR_TEAM: str | None = None
 
-    TIMETRACKING_API_KEY: str
+    TRACKINGTIME_API_KEY: str
 
     class Config:
         env_file = ".env"

src/task_tracker/server.py

@@ -3,24 +3,24 @@
 import logging
 from typing import Any, Dict, List
 
-from mcp.server.models import InitializationOptions
 from mcp.server import NotificationOptions, Server
+from mcp.server.models import InitializationOptions
 from mcp.server.stdio import stdio_server
 import mcp.types as types
 
 from task_tracker.api.linear_client import LinearClient
-from task_tracker.api.timetracking_client import TimeTrackingClient
-
-linear_client = asyncio.run(LinearClient.create())
-timetracking_client = TimeTrackingClient()
-server = Server("task-tracker")
+from task_tracker.api.trackingtime_client import TrackingTimeClient
 
 logging.basicConfig(
     level=logging.INFO,
     format='%(asctime)s - %(levelname)s - %(message)s'
 )
 logger = logging.getLogger("task-tracker")
-logger.info("task-tracker server started")
+logger.info("starting task-tracker server")
+
+linear_client = asyncio.run(LinearClient.create())
+trackingtime_client = TrackingTimeClient()
+server = Server("task-tracker")
 
 
 @server.list_tools()
@@ -42,88 +42,91 @@ async def handle_list_tools() -> List[types.Tool]:
             }
         ),
         types.Tool(
-            name="log_time",
-            description="Log time for a task",
+            name="set_current_team",
+            description="Set the current Linear team by name",
             inputSchema={
                 "type": "object",
                 "properties": {
-                    "task_id": {"type": "string"},
-                    "duration": {"type": "integer"},
-                    "description": {"type": "string"}
+                    "team_name": {"type": "string"}
                 },
-                "required": ["task_id", "duration", "description"]
+                "required": ["team_name"]
             }
         ),
         types.Tool(
-            name="start_tracking",
-            description="Start time tracking for a task",
+            name="get_my_tasks",
+            description=(
+                "Get the Linear tasks assigned to me. "
+                "Support task status: backlog, unstarted, started, completed, canceled, triage. "
+                "Default is unstarted."
+            ),
             inputSchema={
                 "type": "object",
                 "properties": {
-                    "project": {
-                        "type": "string",
-                        "description": "task project"
-                    },
-                    "description": {
-                        "type": "string",
-                        "description": "task name"
+                    "status": {
+                        "type": "array",
+                        "items": {"type": "string"},
+                        "optional": True,
+                        "description": "List of task status to filter by"
                     }
                 },
-                "required": ["project", "description"]
+                "required": []
             }
         ),
         types.Tool(
-            name="stop_tracking",
-            description="Stop current time tracking",
+            name="search_tasks",
+            description="Search Linear tasks by title or identifier",
             inputSchema={
                 "type": "object",
                 "properties": {
-                    "entry_id": {"type": "string"}
+                    "search_term": {
+                        "type": "string",
+                        "description": "Text to search for in task titles or identifiers"
+                    }
                 },
-                "required": ["entry_id"]
+                "required": ["search_term"]
             }
         ),
         types.Tool(
-            name="set_current_team",
-            description="Set the current team by name",
+            name="start_tracking",
+            description="Start time tracking for a task",
             inputSchema={
                 "type": "object",
                 "properties": {
-                    "team_name": {"type": "string"}
+                    "project": {
+                        "type": "string",
+                        "description": "task project"
+                    },
+                    "description": {
+                        "type": "string",
+                        "description": "task name"
+                    }
                 },
-                "required": ["team_name"]
+                "required": ["project", "description"]
             }
         ),
         types.Tool(
-            name="get_my_tasks",
-            description="Get tasks assigned to me. Support task status: backlog, unstarted, started, completed, canceled, triage. Default is unstarted.",
+            name="stop_tracking",
+            description="Stop current time tracking",
             inputSchema={
                 "type": "object",
                 "properties": {
-                    "status": {
-                        "type": "array",
-                        "items": {"type": "string"},
-                        "optional": True,
-                        "description": "List of task status to filter by"
+                    "task_id": {
+                        "type": "string",
+                        "description": "task id"
                     }
                 },
-                "required": []
+                "required": ["task_id"]
             }
         ),
         types.Tool(
-            name="search_tasks",
-            description="Search for tasks by title or identifier",
+            name="get_active_tracking",
+            description="Get the currently active tracking task",
             inputSchema={
                 "type": "object",
-                "properties": {
-                    "search_term": {
-                        "type": "string",
-                        "description": "Text to search for in task titles or identifiers"
-                    }
-                },
-                "required": ["search_term"]
+                "properties": {},
+                "required": []
             }
-        )
+        ),
     ]
 
 
@@ -147,37 +150,6 @@ async def handle_call_tool(name: str, arguments: Dict[str, Any] | None) -> List[
                     json.dumps(result, indent=2)}"
             )]
 
-        elif name == "log_time":
-            result = await timetracking_client.log_time(
-                task_id=arguments["task_id"],
-                duration=arguments["duration"],
-                description=arguments["description"]
-            )
-            return [types.TextContent(
-                type="text",
-                text=f"Time logged successfully: {
-                    json.dumps(result, indent=2)}"
-            )]
-
-        elif name == "start_tracking":
-            result = await timetracking_client.start_tracking(
-                project=arguments["project"],
-                description=arguments["description"]
-            )
-            return [types.TextContent(
-                type="text",
-                text=f"Time tracking started: {json.dumps(result, indent=2)}"
-            )]
-
-        elif name == "stop_tracking":
-            result = await timetracking_client.stop_tracking(
-                entry_id=arguments["entry_id"]
-            )
-            return [types.TextContent(
-                type="text",
-                text=f"Time tracking stopped: {json.dumps(result, indent=2)}"
-            )]
-
         elif name == "set_current_team":
             team_name = arguments["team_name"]
             await linear_client.set_current_team(team_name)
@@ -215,6 +187,38 @@ async def handle_call_tool(name: str, arguments: Dict[str, Any] | None) -> List[
                     json.dumps(tasks, indent=2)}"
             )]
 
+        elif name == "start_tracking":
+            result = await trackingtime_client.start_tracking(
+                project=arguments["project"],
+                description=arguments["description"]
+            )
+            return [types.TextContent(
+                type="text",
+                text=f"Time tracking started: {json.dumps(result, indent=2)}"
+            )]
+
+        elif name == "stop_tracking":
+            result = await trackingtime_client.stop_tracking(
+                task_id=arguments["task_id"]
+            )
+            return [types.TextContent(
+                type="text",
+                text=f"Time tracking stopped: {json.dumps(result, indent=2)}"
+            )]
+
+        elif name == "get_active_tracking":
+            result = await trackingtime_client.get_tracking_task(filter="TRACKING")
+            if not result:
+                return [types.TextContent(
+                    type="text",
+                    text="No active time tracking task found"
+                )]
+            return [types.TextContent(
+                type="text",
+                text=f"Current tracking task:\n{
+                    json.dumps(result, indent=2)}"
+            )]
+
         raise ValueError(f"Unknown tool: {name}")
 
     except Exception as e: