mac-messages-mcp

Instructions

Input Schema

Input Schema (JSON Schema)

Implementation Reference

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

  The MCP tool handler function for tool_get_recent_messages, decorated with @mcp.tool() for registration. Delegates to the core get_recent_messages helper.
  @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 implementing the logic to query Messages database, resolve contacts/handles, extract message bodies, format timestamps and output recent messages.
  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)