mac-messages-mcp

Instructions

Fuzzy search for messages containing the search_term within the last N hours.
Returns messages that match the search term with a similarity score.

Args:
    search_term: The text to search for in messages.
    hours: How many hours back to search (default 24). Must be positive.
    threshold: Similarity threshold for matching (0.0 to 1.0, default 0.6). Lower is more lenient.

Implementation Reference

• mac_messages_mcp/server.py:220-249 (handler)

  MCP tool handler and registration for 'tool_fuzzy_search_messages'. Performs input validation on hours and threshold, logs the search, calls the core fuzzy_search_messages helper, and handles exceptions.

  @mcp.tool()
  def tool_fuzzy_search_messages(
      ctx: Context, search_term: str, hours: int = 24, threshold: float = 0.6
  ) -> str:
      """
      Fuzzy search for messages containing the search_term within the last N hours.
      Returns messages that match the search term with a similarity score.
  
      Args:
          search_term: The text to search for in messages.
          hours: How many hours back to search (default 24). Must be positive.
          threshold: Similarity threshold for matching (0.0 to 1.0, default 0.6). Lower is more lenient.
      """
      if not (0.0 <= threshold <= 1.0):
          return "Error: Threshold must be between 0.0 and 1.0."
      if hours <= 0:
          return "Error: Hours must be a positive integer."
  
      logger.info(
          f"Tool: Fuzzy searching messages for '{search_term}' in last {hours} hours with threshold {threshold}"
      )
      try:
          result = fuzzy_search_messages(
              search_term=search_term, hours=hours, threshold=threshold
          )
          return result
      except Exception as e:
          logger.error(f"Error in tool_fuzzy_search_messages: {e}", exc_info=True)
          return f"An unexpected error occurred during fuzzy message search: {str(e)}"

• mac_messages_mcp/messages.py:802-948 (helper)

  Core implementation of fuzzy message search. Queries recent messages from chat.db (last N hours, limited to 500), extracts text from text or attributedBody fields, cleans text, uses thefuzz.fuzz.WRatio for similarity scoring against search_term above threshold, formats results with timestamps, scores, sender, and group chat info.

  def fuzzy_search_messages(
      search_term: str,
      hours: int = 24,
      threshold: float = 0.6,  # Default threshold adjusted for thefuzz
  ) -> str:
      """
      Fuzzy search for messages containing the search_term within the last N hours.
  
      Args:
          search_term: The string to search for in message content.
          hours: Number of hours to look back (default: 24).
          threshold: Minimum similarity score (0.0-1.0) to consider a match (default: 0.6 for WRatio).
                     A lower threshold allows for more lenient matching.
  
      Returns:
          Formatted string with matching messages and their scores, or an error/no results message.
      """
      # Input validation
      if not search_term or not search_term.strip():
          return "Error: Search term cannot be empty."
      
      if hours < 0:
          return "Error: Hours cannot be negative. Please provide a positive number."
      
      # Prevent integer overflow - limit to reasonable maximum (10 years)
      MAX_HOURS = 10 * 365 * 24  # 87,600 hours
      if hours > MAX_HOURS:
          return f"Error: Hours value too large. Maximum allowed is {MAX_HOURS} hours (10 years)."
      
      if not (0.0 <= threshold <= 1.0):
          return "Error: Threshold must be between 0.0 and 1.0."
      
      # Calculate the timestamp for X hours ago
      current_time = datetime.now(timezone.utc)
      hours_ago_dt = current_time - timedelta(hours=hours)
      apple_epoch = datetime(2001, 1, 1, tzinfo=timezone.utc)
      seconds_since_apple_epoch = (hours_ago_dt - apple_epoch).total_seconds()
      
      # Convert to nanoseconds (Apple's format)
      nanoseconds_since_apple_epoch = int(seconds_since_apple_epoch * 1_000_000_000)
      timestamp_str = str(nanoseconds_since_apple_epoch)
  
      # Build the SQL query to get all messages in the time window
      # Limiting to 500 messages to avoid performance issues with very large message histories.
      query = """
      SELECT
          m.ROWID,
          m.date,
          m.text,
          m.attributedBody,
          m.is_from_me,
          m.handle_id,
          m.cache_roomnames
      FROM
          message m
      WHERE
          CAST(m.date AS TEXT) > ?
      ORDER BY m.date DESC
      LIMIT 500
      """
      params = (timestamp_str,)
      raw_messages = query_messages_db(query, params)
  
      if not raw_messages:
          return f"No messages found in the last {hours} hours to search."
      if "error" in raw_messages[0]:
          return f"Error accessing messages: {raw_messages[0]['error']}"
  
      message_candidates = []
      for msg_dict in raw_messages:
          body = msg_dict.get("text") or extract_body_from_attributed(
              msg_dict.get("attributedBody")
          )
          if body and body.strip():
              message_candidates.append((body, msg_dict))
  
      if not message_candidates:
          return f"No message content found to search in the last {hours} hours."
  
      # --- New fuzzy matching logic using thefuzz ---
      cleaned_search_term = clean_name(search_term).lower()
      # thefuzz scores are 0-100. Scale the input threshold (0.0-1.0).
      scaled_threshold = threshold * 100
  
      matched_messages_with_scores = []
      for original_message_text, msg_dict_value in message_candidates:
          # We use the original_message_text for matching, which might contain HTML entities etc.
          # clean_name will handle basic cleaning like emoji removal.
          cleaned_candidate_text = clean_name(original_message_text).lower()
  
          # Using WRatio for a good balance of matching strategies.
          score_from_thefuzz = fuzz.WRatio(cleaned_search_term, cleaned_candidate_text)
  
          if score_from_thefuzz >= scaled_threshold:
              # Store score as 0.0-1.0 for consistency with how threshold is defined
              matched_messages_with_scores.append(
                  (original_message_text, msg_dict_value, score_from_thefuzz / 100.0)
              )
      matched_messages_with_scores.sort(
          key=lambda x: x[2], reverse=True
      )  # Sort by score desc
  
      if not matched_messages_with_scores:
          return f"No messages found matching '{search_term}' with a threshold of {threshold} in the last {hours} hours."
  
      chat_mapping = get_chat_mapping()
      formatted_results = []
      for _matched_text, msg_dict, score in matched_messages_with_scores:
          original_body = (
              msg_dict.get("text")
              or extract_body_from_attributed(msg_dict.get("attributedBody"))
              or "[No displayable content]"
          )
  
          apple_offset = (
              978307200  # Seconds between Unix epoch and Apple epoch (2001-01-01)
          )
          msg_timestamp_ns = int(msg_dict["date"])
          # Ensure timestamp is in seconds for fromtimestamp
          msg_timestamp_s = (
              msg_timestamp_ns / 1_000_000_000
              if len(str(msg_timestamp_ns)) > 10
              else msg_timestamp_ns
          )
          date_val = datetime.fromtimestamp(
              msg_timestamp_s + apple_offset, tz=timezone.utc
          )
          date_str = date_val.astimezone().strftime("%Y-%m-%d %H:%M:%S")
  
          direction = (
              "You" if msg_dict["is_from_me"] else get_contact_name(msg_dict["handle_id"])
          )
          group_chat_name = (
              chat_mapping.get(msg_dict.get("cache_roomnames"))
              if msg_dict.get("cache_roomnames")
              else None
          )
          message_prefix = f"[{date_str}] (Score: {score:.2f})" + (
              f" [{group_chat_name}]" if group_chat_name else ""
          )
          formatted_results.append(f"{message_prefix} {direction}: {original_body}")
  
      return (
          f"Found {len(matched_messages_with_scores)} messages matching '{search_term}':\n"
          + "\n".join(formatted_results)
      )