mac-messages-mcp

Instructions

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()