mac-messages-mcp

Instructions

Get recent messages from the Messages app.

Args:
    hours: Number of hours to look back (default: 24)
    contact: Filter by contact name, phone number, or email (optional)
            Use "contact:N" to select a specific contact from previous matches

Implementation Reference

• mac_messages_mcp/server.py:36-55 (handler)

  The tool handler function decorated with @mcp.tool(), which registers and implements the tool_get_recent_messages. It validates inputs, calls the core get_recent_messages helper, handles exceptions, and returns formatted recent messages or error.

  @mcp.tool()
  def tool_get_recent_messages(ctx: Context, hours: int = 24, contact: str = None) -> str:
      """
      Get recent messages from the Messages app.
      
      Args:
          hours: Number of hours to look back (default: 24)
          contact: Filter by contact name, phone number, or email (optional)
                  Use "contact:N" to select a specific contact from previous matches
      """
      logger.info(f"Getting recent messages: hours={hours}, contact={contact}")
      try:
          # Handle contacts that are passed as numbers
          if contact is not None:
              contact = str(contact)
          result = get_recent_messages(hours=hours, contact=contact)
          return result
      except Exception as e:
          logger.error(f"Error in get_recent_messages: {str(e)}")
          return f"Error getting messages: {str(e)}"

• mac_messages_mcp/messages.py:583-797 (helper)

  Core helper function containing the exact implementation logic for retrieving, filtering, processing, and formatting recent messages from the macOS Messages database (chat.db). Handles contact lookup, fuzzy matching, SQL queries, Apple timestamp conversion, attributedBody parsing, and output formatting.

  def get_recent_messages(hours: int = 24, contact: Optional[str] = None) -> str:
      """
      Get recent messages from the Messages app using attributedBody for content.
      
      Args:
          hours: Number of hours to look back (default: 24)
          contact: Filter by contact name, phone number, or email (optional)
                  Use "contact:N" to select a specific contact from previous matches
      
      Returns:
          Formatted string with recent messages
      """
      # Input validation
      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)."
      
      handle_id = None
      
      # If contact is specified, try to resolve it
      if contact:
          # Convert to string to ensure phone numbers work properly
          contact = str(contact).strip()
          
          # Handle contact selection format (contact:N)
          if contact.lower().startswith("contact:"):
              try:
                  # Extract the number after the colon
                  contact_parts = contact.split(":", 1)
                  if len(contact_parts) < 2 or not contact_parts[1].strip():
                      return "Error: Invalid contact selection format. Use 'contact:N' where N is a positive number."
                  
                  # Get the selected index (1-based)
                  try:
                      index = int(contact_parts[1].strip()) - 1
                  except ValueError:
                      return "Error: Contact selection must be a number. Use 'contact:N' where N is a positive number."
                  
                  # Validate index is not negative
                  if index < 0:
                      return "Error: Contact selection must be a positive number (starting from 1)."
                  
                  # Get the most recent contact matches from global cache
                  if not hasattr(get_recent_messages, "recent_matches") or not get_recent_messages.recent_matches:
                      return "No recent contact matches available. Please search for a contact first."
                  
                  if index >= len(get_recent_messages.recent_matches):
                      return f"Invalid selection. Please choose a number between 1 and {len(get_recent_messages.recent_matches)}."
                  
                  # Get the selected contact's phone number
                  contact = get_recent_messages.recent_matches[index]['phone']
              except Exception as e:
                  return f"Error processing contact selection: {str(e)}"
          
          # Check if contact might be a name rather than a phone number or email
          if not all(c.isdigit() or c in '+- ()@.' for c in contact):
              # Try fuzzy matching
              matches = find_contact_by_name(contact)
              
              if not matches:
                  return f"No contacts found matching '{contact}'."
              
              if len(matches) == 1:
                  # Single match, use its phone number
                  contact = matches[0]['phone']
              else:
                  # Store the matches for later selection
                  get_recent_messages.recent_matches = matches
                  
                  # Multiple matches, return them all
                  contact_list = "\n".join([f"{i+1}. {c['name']} ({c['phone']})" for i, c in enumerate(matches[:10])])
                  return f"Multiple contacts found matching '{contact}'. Please specify which one using 'contact:N' where N is the number:\n{contact_list}"
          
          # At this point, contact should be a phone number or email
          # Try to find handle_id with improved phone number matching
          if '@' in contact:
              # This is an email
              query = "SELECT ROWID FROM handle WHERE id = ?"
              results = query_messages_db(query, (contact,))
              if results and not "error" in results[0] and len(results) > 0:
                  handle_id = results[0]["ROWID"]
          else:
              # This is a phone number - try various formats
              handle_id = find_handle_by_phone(contact)
              
          if not handle_id:
              # Try a direct search in message table to see if any messages exist
              normalized = normalize_phone_number(contact)
              query = """
              SELECT COUNT(*) as count 
              FROM message m
              JOIN handle h ON m.handle_id = h.ROWID
              WHERE h.id LIKE ?
              """
              results = query_messages_db(query, (f"%{normalized}%",))
              
              if results and not "error" in results[0] and results[0].get("count", 0) == 0:
                  # No messages found but the query was valid
                  return f"No message history found with '{contact}'."
              else:
                  # Could not find the handle at all
                  return f"Could not find any messages with contact '{contact}'. Verify the phone number or email is correct."
      
      # Calculate the timestamp for X hours ago
      current_time = datetime.now(timezone.utc)
      hours_ago = current_time - timedelta(hours=hours)
      
      # Convert to Apple's timestamp format (nanoseconds since 2001-01-01)
      # Apple's Core Data uses nanoseconds, not seconds
      apple_epoch = datetime(2001, 1, 1, tzinfo=timezone.utc)
      seconds_since_apple_epoch = (hours_ago - apple_epoch).total_seconds()
      
      # Convert to nanoseconds (Apple's format)
      nanoseconds_since_apple_epoch = int(seconds_since_apple_epoch * 1_000_000_000)
      
      # Make sure we're using a string representation for the timestamp
      # to avoid integer overflow issues when binding to SQLite
      timestamp_str = str(nanoseconds_since_apple_epoch)
      
      # Build the SQL query - use attributedBody field and text
      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) > ? 
      """
      
      params = (timestamp_str,)
      
      # Add contact filter if handle_id was found
      if handle_id:
          query += "AND m.handle_id = ? "
          params = (timestamp_str, handle_id)
      
      query += "ORDER BY m.date DESC LIMIT 100"
      
      # Execute the query
      messages = query_messages_db(query, params)
      
      # Format the results
      if not messages:
          return "No messages found in the specified time period."
      
      if "error" in messages[0]:
          return f"Error accessing messages: {messages[0]['error']}"
      
      # Get chat mapping for group chat names
      chat_mapping = get_chat_mapping()
      
      formatted_messages = []
      for msg in messages:
          # Get the message content from text or attributedBody
          if msg.get('text'):
              body = msg['text']
          elif msg.get('attributedBody'):
              body = extract_body_from_attributed(msg['attributedBody'])
              if not body:
                  # Skip messages with no content
                  continue
          else:
              # Skip empty messages
              continue
          
          # Convert Apple timestamp to readable date
          try:
              # Convert Apple timestamp to datetime
              date_string = '2001-01-01'
              mod_date = datetime.strptime(date_string, '%Y-%m-%d')
              unix_timestamp = int(mod_date.timestamp()) * 1000000000
              
              # Handle both nanosecond and second format timestamps
              msg_timestamp = int(msg["date"])
              if len(str(msg_timestamp)) > 10:  # It's in nanoseconds
                  new_date = int((msg_timestamp + unix_timestamp) / 1000000000)
              else:  # It's already in seconds
                  new_date = mod_date.timestamp() + msg_timestamp
                  
              date_str = datetime.fromtimestamp(new_date).strftime("%Y-%m-%d %H:%M:%S")
          except (ValueError, TypeError, OverflowError) as e:
              # If conversion fails, use a placeholder
              date_str = "Unknown date"
              print(f"Date conversion error: {e} for timestamp {msg['date']}")
          
          direction = "You" if msg["is_from_me"] else get_contact_name(msg["handle_id"])
          
          # Check if this is a group chat
          group_chat_name = None
          if msg.get('cache_roomnames'):
              group_chat_name = chat_mapping.get(msg['cache_roomnames'])
          
          message_prefix = f"[{date_str}]"
          if group_chat_name:
              message_prefix += f" [{group_chat_name}]"
          
          formatted_messages.append(
              f"{message_prefix} {direction}: {body}"
          )
      
      if not formatted_messages:
          return "No messages found in the specified time period."
          
      return "\n".join(formatted_messages)

• mac_messages_mcp/server.py:36-36 (registration)

  MCP tool registration decorator applied to the tool_get_recent_messages handler function.

  @mcp.tool()