Things MCP Server

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 from . 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})