Productive Simple MCP

Instructions

Implementation Reference

• tools.py:98-136 (handler)

  The core handler function that executes the get_task tool logic: fetches task data from the Productive API client, sanitizes the response with filter_response, ensures time tracking fields have default values, and handles API and unexpected errors.

  async def get_task(ctx: Context, task_id: int) -> ToolResult:
      """Fetch a single task by internal ID.
  
      Developer notes:
      - Wraps client.get_task(task_id).
      - Applies utils.filter_response to sanitize output.
      - Ensures time tracking fields are always present (initial_estimate, worked_time, billable_time, remaining_time).
      - Raises ProductiveAPIError on failure.
      """
      try:
          await ctx.info(f"Fetching task with ID: {task_id}")
          result = await client.get_task(task_id)
          await ctx.info("Successfully retrieved task")
          
          filtered = filter_response(result)
          
          # Ensure time tracking fields are always present at the top level
          if "data" in filtered and "attributes" in filtered["data"]:
              attributes = filtered["data"]["attributes"]
              
              # Set default values for time tracking fields if missing
              time_fields = {
                  "initial_estimate": 0,
                  "worked_time": 0,
                  "billable_time": 0,
                  "remaining_time": 0
              }
              
              for field, default_value in time_fields.items():
                  if field not in attributes or attributes[field] is None:
                      attributes[field] = default_value
          
          return filtered
          
      except ProductiveAPIError as e:
          await _handle_productive_api_error(ctx, e, f"task {task_id}")
      except Exception as e:
          await ctx.error(f"Unexpected error fetching task: {str(e)}")
          raise e

• server.py:250-269 (registration)

  MCP tool registration using @mcp.tool decorator. Delegates execution to tools.get_task and provides input schema via Annotated Field for task_id, along with comprehensive docstring describing usage and output.

  @mcp.tool
  async def get_task(
      ctx: Context,
      task_id: Annotated[
          int, Field(description="The unique Productive task identifier (internal ID)")
      ],
  ) -> Dict[str, Any]:
      """Get detailed task information by its internal ID.
  
      Use this when you have the internal task ID (e.g., 14677418).
      For looking up tasks by their project-specific number (e.g., #960), use get_project_task instead.
  
      Returns task details including:
      - Task title, description, and status (open/closed)
      - Due date, start date, and creation/update timestamps
      - Time tracking: initial estimate, remaining time, billable time, and worked time (in minutes)
      - Todo counts: total and open
      """
      return await tools.get_task(ctx=ctx, task_id=task_id)

• server.py:253-256 (schema)

  Input schema definition for the tool using Pydantic Annotated and Field, specifying task_id as required integer with description.

      task_id: Annotated[
          int, Field(description="The unique Productive task identifier (internal ID)")
      ],
  ) -> Dict[str, Any]:

• productive_client.py:89-91 (helper)

  Helper method in ProductiveClient that performs the actual HTTP GET request to the Productive API /tasks/{task_id} endpoint, including workflow_status in the response.

  async def get_task(self, task_id: int) -> Dict[str, Any]:
      """Get task by ID with workflow_status always included"""
      return await self._request("GET", f"/tasks/{str(task_id)}", params={"include": "workflow_status"})