Things MCP Server

src
things_mcp
import urllib.parse
import webbrowser
import things
import subprocess
import platform
import random
import time
import logging
import json
from typing import Optional, Dict, Any, Union, Callable
from utils import app_state, circuit_breaker, dead_letter_queue, rate_limiter, is_things_running

logger = logging.getLogger(__name__)
        
def launch_things() -> bool:
    """Launch Things app if not already running.
    
    Returns:
        bool: True if successful, False otherwise
    """
    try:
        if is_things_running():
            return True
            
        result = subprocess.run(
            ['open', '-a', 'Things3'],
            capture_output=True,
            text=True,
            check=False
        )
        
        # Give Things time to launch
        time.sleep(2)
        
        return is_things_running()
    except Exception as e:
        logger.error(f"Error launching Things: {str(e)}")
        return False

def execute_url(url: str) -> bool:
    """Execute a Things URL by opening it in the default browser.
    Returns True if successful, False otherwise.
    """
    # Ensure any + signs in the URL are replaced with %20
    url = url.replace("+", "%20")
    
    # Log the URL for debugging
    logger.debug(f"Executing URL: {url}")
    
    # Apply rate limiting
    rate_limiter.wait_if_needed()
    
    # Check if circuit breaker allows the operation
    if not circuit_breaker.allow_operation():
        logger.warning("Circuit breaker is open, blocking operation")
        return False
    
    try:
        # Check if Things is running, attempt to launch if not
        if not is_things_running():
            logger.info("Things is not running, attempting to launch")
            if not launch_things():
                logger.error("Failed to launch Things")
                circuit_breaker.record_failure()
                return False
        
        # Execute the URL
        result = webbrowser.open(url)
        
        if not result:
            circuit_breaker.record_failure()
            logger.error(f"Failed to open URL: {url}")
            return False
        
        # Add a small delay to allow Things time to process the command
        # Add jitter to prevent thundering herd problem
        delay = 0.5 + random.uniform(0, 0.2)  # 0.5-0.7 seconds
        time.sleep(delay)
        
        circuit_breaker.record_success()
        return True
    except Exception as e:
        logger.error(f"Failed to execute URL: {url}, Error: {str(e)}")
        circuit_breaker.record_failure()
        return False


def execute_xcallback_url(action: str, params: Dict[str, Any]) -> bool:
    """Execute a Things X-Callback-URL.
    
    Args:
        action: The X-Callback action to perform
        params: Parameters for the action
        
    Returns:
        bool: True if successful, False otherwise
    """
    # The correct format for Things URLs (no 'x-callback-url/' prefix)
    base_url = "things:///"
    
    # Add callback parameters, but only if we need them
    # For now, avoid using callbacks since we don't have a handler for them
    callback_params = params.copy()
    
    # Don't add callback URLs - this avoids the "no application set to open URL" error
    # If we need callbacks later, we'd need to register a URL handler for our app
    
    # Construct URL - action is part of the path (not a separate query parameter)
    url = f"{base_url}{action}?{urllib.parse.urlencode(callback_params)}"
    
    # Log the URL for debugging
    logger.debug(f"Executing URL: {url}")
    
    return execute_url(url)

def construct_url(command: str, params: Dict[str, Any]) -> str:
    """Construct a Things URL from command and parameters."""
    # Pre-process all string parameters to replace any + signs with spaces
    cleaned_params = {}
    for key, value in params.items():
        if isinstance(value, str):
            # Replace any + signs with spaces in the original input
            cleaned_params[key] = value.replace("+", " ")
        else:
            cleaned_params[key] = value
    
    # Use the cleaned params from now on
    params = cleaned_params
    
    # Start with base URL
    url = f"things:///{command}"
    
    # Get authentication token if needed - applies to all commands to ensure reliability
    try:
        # Import here to avoid circular imports
        import config
        
        # Get token from config system
        token = config.get_things_auth_token()
        
        if token:
            # Add token to all params for consistent behavior
            params['auth-token'] = token
            logger.debug(f"Auth token from config used for {command} operation")
        else:
            logger.warning(f"No Things auth token found in config. URL may not work without a token.")
            # Note: We continue without a token, which may cause the operation to fail
    except Exception as e:
        logger.error(f"Error getting auth token: {str(e)}")
        # Continue without token - the operation may fail
    
    # Disable JSON API for now as it's causing formatting issues
    # JSON API is currently experimental and unreliable
    # We'll use the standard URL scheme instead which is more reliable
    use_json_api = False
    
    if False and command in ['add'] and use_json_api:
        # This code is disabled but kept for reference
        logger.info("JSON API is currently disabled due to formatting issues")
        pass
    
    # Standard URL scheme encoding
    if params:
        encoded_params = []
        for key, value in params.items():
            if value is None:
                continue
            # Handle boolean values
            if isinstance(value, bool):
                value = str(value).lower()
            # Handle lists (for tags, checklist items etc)
            elif isinstance(value, list) and key == 'tags':
                # Important: Tags are sensitive to formatting in Things URL scheme
                # Based on testing, using a simple comma-separated list without spaces works best
                encoded_tags = []
                for tag in value:
                    # Ensure tag is properly encoded as string
                    tag_str = str(tag).strip()
                    if tag_str:  # Only add non-empty tags
                        encoded_tags.append(tag_str)
                
                # Only include non-empty tag lists
                if encoded_tags:
                    # Join with commas - Things expects comma-separated tags without spaces between commas
                    # Use a simple comma with no spacing for maximum compatibility
                    value = ','.join(encoded_tags)
                else:
                    # If no valid tags, don't include this parameter
                    continue
            # Handle other lists
            elif isinstance(value, list):
                value = ','.join(str(v) for v in value)
            
            # Ensure proper encoding of the value - use quote_plus to handle spaces correctly
            # Then replace + with %20 to ensure Things handles spaces correctly
            encoded_value = urllib.parse.quote(str(value), safe='')
            # Replace + with %20 for better compatibility with Things
            encoded_value = encoded_value.replace('+', '%20')
            encoded_params.append(f"{key}={encoded_value}")
        
        url += "?" + "&".join(encoded_params)
    
    return url


def should_use_json_api() -> bool:
    """Determine if the JSON API should be used based on Things version."""
    from utils import detect_things_version
    
    version = detect_things_version()
    if not version:
        # Default to using JSON API if version can't be determined
        return True
    
    try:
        # Parse version string (e.g., '3.15.4')
        major, minor, _ = map(int, version.split('.'))
        
        # JSON API is available in Things 3.4+
        return major > 3 or (major == 3 and minor >= 4)
    except Exception:
        # Default to standard URL scheme if version parsing fails
        return False

def add_todo(title: str, notes: Optional[str] = None, when: Optional[str] = None,
             deadline: Optional[str] = None, tags: Optional[list[str]] = None,
             checklist_items: Optional[list[str]] = None, list_id: Optional[str] = None,
             list_title: Optional[str] = None, heading: Optional[str] = None,
             completed: Optional[bool] = None) -> str:
    """Construct URL to add a new todo."""
    params = {
        'title': title,
        'notes': notes,
        'when': when,
        'deadline': deadline,
        'checklist-items': '\n'.join(checklist_items) if checklist_items else None,
        'list-id': list_id,
        'list': list_title,
        'heading': heading,
        'completed': completed
    }
    
    # Handle tags separately since they need to be comma-separated
    if tags:
        params['tags'] = ','.join(tags)
    return construct_url('add', {k: v for k, v in params.items() if v is not None})

def add_project(title: str, notes: Optional[str] = None, when: Optional[str] = None,
                deadline: Optional[str] = None, tags: Optional[list[str]] = None,
                area_id: Optional[str] = None, area_title: Optional[str] = None,
                todos: Optional[list[str]] = None) -> str:
    """Construct URL to add a new project."""
    params = {
        'title': title,
        'notes': notes,
        'when': when,
        'deadline': deadline,
        'area-id': area_id,
        'area': area_title,
        # Change todos to be newline separated
        'to-dos': '\n'.join(todos) if todos else None
    }
    
    # Handle tags separately since they need to be comma-separated
    if tags:
        params['tags'] = ','.join(tags)
        
    return construct_url('add-project', {k: v for k, v in params.items() if v is not None})

def update_todo(id: str, title: Optional[str] = None, notes: Optional[str] = None,
                when: Optional[str] = None, deadline: Optional[str] = None,
                tags: Optional[Union[list[str], str]] = None, add_tags: Optional[Union[list[str], str]] = None, 
                checklist_items: Optional[list[str]] = None,
                completed: Optional[bool] = None, canceled: Optional[bool] = None) -> str:
    """Construct URL to update an existing todo."""
    params = {
        'id': id,
        'title': title,
        'notes': notes,
        'when': when,
        'deadline': deadline,
        'tags': tags,
        'add-tags': add_tags,  # Support for adding tags without replacing existing ones
        'checklist-items': '\n'.join(checklist_items) if checklist_items else None,
        'completed': completed,
        'canceled': canceled
    }
    return construct_url('update', {k: v for k, v in params.items() if v is not None})

def update_project(id: str, title: Optional[str] = None, notes: Optional[str] = None,
                   when: Optional[str] = None, deadline: Optional[str] = None,
                   tags: Optional[list[str]] = None, completed: Optional[bool] = None,
                   canceled: Optional[bool] = None) -> str:
    """Construct URL to update an existing project."""
    params = {
        'id': id,
        'title': title,
        'notes': notes,
        'when': when,
        'deadline': deadline,
        'tags': tags,
        'completed': completed,
        'canceled': canceled
    }
    return construct_url('update-project', {k: v for k, v in params.items() if v is not None})

def show(id: str, query: Optional[str] = None, filter_tags: Optional[list[str]] = None) -> str:
    """Construct URL to show a specific item or list."""
    params = {
        'id': id,
        'query': query,
        'filter': filter_tags
    }
    return construct_url('show', {k: v for k, v in params.items() if v is not None})

def search(query: str) -> str:
    """Construct URL to perform a search."""
    return construct_url('search', {'query': query})