OneSignal MCP Server

import os import json import re import requests import logging from typing import List, Dict, Any, Optional, Union from mcp.server.fastmcp import FastMCP, Context from dotenv import load_dotenv # Server information __version__ = "2.1.0" # Configure logging logging.basicConfig( level=logging.INFO, # Default level, will be overridden by env var if set format='%(asctime)s - %(name)s - %(levelname)s - %(message)s', handlers=[ logging.StreamHandler() ] ) logger = logging.getLogger("onesignal-mcp") # Load environment variables from .env file load_dotenv() logger.info("Environment variables loaded") # Get log level from environment, default to INFO, and ensure it's uppercase log_level_str = os.getenv("LOG_LEVEL", "INFO").upper() valid_log_levels = ["DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"] if log_level_str not in valid_log_levels: logger.warning(f"Invalid LOG_LEVEL '{log_level_str}' found in environment. Using INFO instead.") log_level_str = "INFO" # Apply the validated log level logger.setLevel(log_level_str) # Initialize the MCP server, passing the validated log level mcp = FastMCP("onesignal-server", log_level=log_level_str) logger.info(f"OneSignal MCP server initialized with log level: {log_level_str}") # OneSignal API configuration # v2 API keys use the new base URL without /api/v1 ONESIGNAL_API_URL = "https://api.onesignal.com" ONESIGNAL_ORG_API_KEY = os.getenv("ONESIGNAL_ORG_API_KEY", "") # Class to manage app configurations class AppConfig: def __init__(self, app_id: str, api_key: str, name: str = None): self.app_id = app_id self.api_key = api_key self.name = name or app_id def __str__(self): return f"{self.name} ({self.app_id})" # Dictionary to store app configurations app_configs: Dict[str, AppConfig] = {} # Load app configurations from environment variables # Mandible app configuration mandible_app_id = os.getenv("ONESIGNAL_MANDIBLE_APP_ID", "") or os.getenv("ONESIGNAL_APP_ID", "") mandible_api_key = os.getenv("ONESIGNAL_MANDIBLE_API_KEY", "") or os.getenv("ONESIGNAL_API_KEY", "") if mandible_app_id and mandible_api_key: app_configs["mandible"] = AppConfig(mandible_app_id, mandible_api_key, "Mandible") current_app_key = "mandible" logger.info(f"Mandible app configured with ID: {mandible_app_id}") # Weird Brains app configuration weirdbrains_app_id = os.getenv("ONESIGNAL_WEIRDBRAINS_APP_ID", "") weirdbrains_api_key = os.getenv("ONESIGNAL_WEIRDBRAINS_API_KEY", "") if weirdbrains_app_id and weirdbrains_api_key: app_configs["weirdbrains"] = AppConfig(weirdbrains_app_id, weirdbrains_api_key, "Weird Brains") if not current_app_key: current_app_key = "weirdbrains" logger.info(f"Weird Brains app configured with ID: {weirdbrains_app_id}") # Fallback for default app configuration if not app_configs: default_app_id = os.getenv("ONESIGNAL_APP_ID", "") default_api_key = os.getenv("ONESIGNAL_API_KEY", "") if default_app_id and default_api_key: app_configs["default"] = AppConfig(default_app_id, default_api_key, "Default App") current_app_key = "default" logger.info(f"Default app configured with ID: {default_app_id}") else: current_app_key = None logger.warning("No app configurations found. Use add_app to add an app configuration.") # Function to add a new app configuration def add_app_config(key: str, app_id: str, api_key: str, name: str = None) -> None: """Add a new app configuration to the available apps. Args: key: Unique identifier for this app configuration app_id: OneSignal App ID api_key: OneSignal REST API Key name: Display name for the app (optional) """ app_configs[key] = AppConfig(app_id, api_key, name or key) logger.info(f"Added app configuration '{key}' with ID: {app_id}") # Function to switch the current app def set_current_app(app_key: str) -> bool: """Set the current app to use for API requests. Args: app_key: The key of the app configuration to use Returns: True if successful, False if the app key doesn't exist """ global current_app_key if app_key in app_configs: current_app_key = app_key logger.info(f"Switched to app '{app_key}'") return True logger.error(f"Failed to switch app: '{app_key}' not found") return False # Function to get the current app configuration def get_current_app() -> Optional[AppConfig]: """Get the current app configuration. Returns: The current AppConfig or None if no app is set """ if current_app_key and current_app_key in app_configs: return app_configs[current_app_key] logger.warning("No current app is set. Use switch_app(key) to select an app.") return None # Helper function to determine whether to use Organization API Key def requires_org_api_key(endpoint: str) -> bool: """Determine if an endpoint requires the Organization API Key instead of a REST API Key. Args: endpoint: The API endpoint path Returns: True if the endpoint requires Organization API Key, False otherwise """ # Organization-level endpoints that require Organization API Key # Note: "apps" alone requires org key, but "apps/{app_id}/segments" etc. use app key import re # Exact matches that require org key if endpoint == "apps": return True # apps/{app_id} for viewing/updating a single app requires org key # But apps/{app_id}/segments, apps/{app_id}/users, etc. use app key if re.match(r"^apps/[^/]+$", endpoint): return True # apps/{app_id}/auth/tokens for API key management requires org key if re.match(r"^apps/[^/]+/auth/tokens", endpoint): return True # Export endpoints require org key if endpoint == "notifications/csv_export" or endpoint.startswith("notifications/csv_export/"): return True return False # Helper function for OneSignal API requests async def make_onesignal_request( endpoint: str, method: str = "GET", data: Dict[str, Any] = None, params: Dict[str, Any] = None, use_org_key: bool = None, app_key: str = None ) -> Dict[str, Any]: """Make a request to the OneSignal API with proper authentication. Args: endpoint: API endpoint path method: HTTP method (GET, POST, PUT, DELETE) data: Request body for POST/PUT requests params: Query parameters for GET requests use_org_key: Whether to use the organization API key instead of the REST API key If None, will be automatically determined based on the endpoint app_key: The key of the app configuration to use (uses current app if None) Returns: API response as dictionary """ headers = { "Content-Type": "application/json", "Accept": "application/json", } # If use_org_key is not explicitly specified, determine it based on the endpoint if use_org_key is None: use_org_key = requires_org_api_key(endpoint) # Determine which app configuration to use app_config = None if not use_org_key: if app_key and app_key in app_configs: app_config = app_configs[app_key] elif current_app_key and current_app_key in app_configs: app_config = app_configs[current_app_key] if not app_config: error_msg = "No app configuration available. Use set_current_app or specify app_key." logger.error(error_msg) return {"error": error_msg} # Check if it's a v2 API key if app_config.api_key.startswith("os_v2_"): headers["Authorization"] = f"Key {app_config.api_key}" else: headers["Authorization"] = f"Basic {app_config.api_key}" else: if not ONESIGNAL_ORG_API_KEY: error_msg = "Organization API Key not configured. Set the ONESIGNAL_ORG_API_KEY environment variable." logger.error(error_msg) return {"error": error_msg} # Check if it's a v2 API key if ONESIGNAL_ORG_API_KEY.startswith("os_v2_"): headers["Authorization"] = f"Key {ONESIGNAL_ORG_API_KEY}" else: headers["Authorization"] = f"Basic {ONESIGNAL_ORG_API_KEY}" url = f"{ONESIGNAL_API_URL}/{endpoint}" # If using app-specific endpoint and not using org key, add app_id to params if not already present if not use_org_key and app_config: if params is None: params = {} if "app_id" not in params and not endpoint.startswith("apps/"): params["app_id"] = app_config.app_id # For POST/PUT requests, add app_id to data if not already present if data is not None and method in ["POST", "PUT"] and "app_id" not in data and not endpoint.startswith("apps/"): data["app_id"] = app_config.app_id try: logger.debug(f"Making {method} request to {url}") logger.debug(f"Using {'Organization API Key' if use_org_key else 'App REST API Key'}") logger.debug(f"Authorization header type: {headers['Authorization'].split(' ')[0]}") if method == "GET": response = requests.get(url, headers=headers, params=params, timeout=30) elif method == "POST": response = requests.post(url, headers=headers, params=params, json=data, timeout=30) elif method == "PUT": response = requests.put(url, headers=headers, params=params, json=data, timeout=30) elif method == "DELETE": response = requests.delete(url, headers=headers, params=params, timeout=30) elif method == "PATCH": response = requests.patch(url, headers=headers, params=params, json=data, timeout=30) else: error_msg = f"Unsupported HTTP method: {method}" logger.error(error_msg) return {"error": error_msg} response.raise_for_status() return response.json() if response.text else {} except requests.exceptions.RequestException as e: error_message = f"Error: {str(e)}" try: if hasattr(e, 'response') and e.response is not None: error_data = e.response.json() if isinstance(error_data, dict): error_message = f"Error: {error_data.get('errors', [e.response.reason])[0]}" except Exception: pass logger.error(f"API request failed: {error_message}") return {"error": error_message} except Exception as e: error_message = f"Unexpected error: {str(e)}" logger.exception(error_message) return {"error": error_message} # Resource for OneSignal configuration information @mcp.resource("onesignal://config") def get_onesignal_config() -> str: """Get information about the OneSignal configuration""" current_app = get_current_app() app_list = "\n".join([f"- {key}: {app}" for key, app in app_configs.items()]) return f""" OneSignal Server Configuration: Version: {__version__} API URL: {ONESIGNAL_API_URL} Organization API Key Status: {'Configured' if ONESIGNAL_ORG_API_KEY else 'Not configured'} Available Apps: {app_list or "No apps configured"} Current App: {current_app.name if current_app else 'None'} This MCP server provides tools for: - Viewing and managing messages (push notifications, emails, SMS) - Managing users and subscriptions - Viewing and managing segments - Creating and managing templates - Viewing app information - Managing multiple OneSignal applications Make sure you have set the appropriate environment variables in your .env file. """ # === App Management Tools === @mcp.tool() async def list_apps() -> str: """List all configured OneSignal apps in this server.""" if not app_configs: return "No apps configured. Use add_app to add a new app configuration." current_app = get_current_app() result = ["Configured OneSignal Apps:"] for key, app in app_configs.items(): current_marker = " (current)" if current_app and key == current_app_key else "" result.append(f"- {key}: {app.name} (App ID: {app.app_id}){current_marker}") return "\n".join(result) @mcp.tool() async def add_app(key: str, app_id: str, api_key: str, name: str = None) -> str: """Add a new OneSignal app configuration locally. Args: key: Unique identifier for this app configuration app_id: OneSignal App ID api_key: OneSignal REST API Key name: Display name for the app (optional) """ if not key or not app_id or not api_key: return "Error: All parameters (key, app_id, api_key) are required." if key in app_configs: return f"Error: App key '{key}' already exists. Use a different key or update_app to modify it." add_app_config(key, app_id, api_key, name) # If this is the first app, set it as current global current_app_key if len(app_configs) == 1: current_app_key = key return f"Successfully added app '{key}' with name '{name or key}'." @mcp.tool() async def update_local_app_config(key: str, app_id: str = None, api_key: str = None, name: str = None) -> str: """Update an existing local OneSignal app configuration. Args: key: The key of the app configuration to update locally app_id: New OneSignal App ID (optional) api_key: New OneSignal REST API Key (optional) name: New display name for the app (optional) """ if key not in app_configs: return f"Error: App key '{key}' not found." app = app_configs[key] updated = [] if app_id: app.app_id = app_id updated.append("App ID") if api_key: app.api_key = api_key updated.append("API Key") if name: app.name = name updated.append("Name") if not updated: return "No changes were made. Specify at least one parameter to update." logger.info(f"Updated app '{key}': {', '.join(updated)}") return f"Successfully updated app '{key}': {', '.join(updated)}." @mcp.tool() async def remove_app(key: str) -> str: """Remove a local OneSignal app configuration. Args: key: The key of the app configuration to remove locally """ if key not in app_configs: return f"Error: App key '{key}' not found." global current_app_key if current_app_key == key: if len(app_configs) > 1: # Set current to another app other_keys = [k for k in app_configs.keys() if k != key] current_app_key = other_keys[0] logger.info(f"Current app changed to '{current_app_key}' after removing '{key}'") else: current_app_key = None logger.warning("No current app set after removing the only app configuration") del app_configs[key] logger.info(f"Removed app configuration '{key}'") return f"Successfully removed app '{key}'." @mcp.tool() async def switch_app(key: str) -> str: """Switch the current app to use for API requests. Args: key: The key of the app configuration to use """ if key not in app_configs: return f"Error: App key '{key}' not found. Available apps: {', '.join(app_configs.keys()) or 'None'}" global current_app_key current_app_key = key app = app_configs[key] return f"Switched to app '{key}' ({app.name})." # === Message Management Tools === @mcp.tool() async def send_push_notification(title: str, message: str, segments: List[str] = None, external_ids: List[str] = None, data: Dict[str, Any] = None) -> Dict[str, Any]: """Send a new push notification through OneSignal. Args: title: Notification title. message: Notification message content. segments: List of segments to include (e.g., ["Subscribed Users"]). external_ids: List of external user IDs to target. data: Additional data to include with the notification (optional). """ app_config = get_current_app() if not app_config: return {"error": "No app currently selected. Use switch_app to select an app."} if not segments and not external_ids: segments = ["Subscribed Users"] # Default if no target specified notification_data = { "app_id": app_config.app_id, "contents": {"en": message}, "headings": {"en": title}, "target_channel": "push" } if segments: notification_data["included_segments"] = segments if external_ids: # Assuming make_onesignal_request handles converting list to JSON notification_data["include_external_user_ids"] = external_ids if data: notification_data["data"] = data # This endpoint uses app-specific REST API Key result = await make_onesignal_request("notifications", method="POST", data=notification_data, use_org_key=False) return result @mcp.tool() async def view_messages(limit: int = 20, offset: int = 0, kind: int = None) -> Dict[str, Any]: """View recent messages sent through OneSignal. Args: limit: Maximum number of messages to return (default: 20, max: 50) offset: Result offset for pagination (default: 0) kind: Filter by message type (0=Dashboard, 1=API, 3=Automated) (optional) """ app_config = get_current_app() if not app_config: return {"error": "No app currently selected. Use switch_app to select an app."} params = {"limit": min(limit, 50), "offset": offset} if kind is not None: params["kind"] = kind # This endpoint uses app-specific REST API Key result = await make_onesignal_request("notifications", method="GET", params=params, use_org_key=False) # Return the raw JSON result for flexibility return result @mcp.tool() async def view_message_details(message_id: str) -> Dict[str, Any]: """Get detailed information about a specific message. Args: message_id: The ID of the message to retrieve details for """ app_config = get_current_app() if not app_config: return {"error": "No app currently selected. Use switch_app to select an app."} # This endpoint uses app-specific REST API Key result = await make_onesignal_request(f"notifications/{message_id}", method="GET", use_org_key=False) # Return the raw JSON result return result @mcp.tool() async def view_message_history(message_id: str, event: str) -> Dict[str, Any]: """View the history / recipients of a message based on events. Args: message_id: The ID of the message. event: The event type to track (e.g., 'sent', 'clicked'). """ app_config = get_current_app() if not app_config: return {"error": "No app currently selected. Use switch_app to select an app."} data = { "app_id": app_config.app_id, "events": event, "email": get_current_app().name + "-history@example.com" # Requires an email to send the CSV report } # Endpoint uses REST API Key result = await make_onesignal_request(f"notifications/{message_id}/history", method="POST", data=data, use_org_key=False) return result @mcp.tool() async def cancel_message(message_id: str) -> Dict[str, Any]: """Cancel a scheduled message that hasn't been delivered yet. Args: message_id: The ID of the message to cancel """ app_config = get_current_app() if not app_config: return {"error": "No app currently selected. Use switch_app to select an app."} # This endpoint uses app-specific REST API Key result = await make_onesignal_request(f"notifications/{message_id}", method="DELETE", use_org_key=False) return result # === Segment Management Tools === @mcp.tool() async def view_segments() -> str: """List all segments available in your OneSignal app.""" app_config = get_current_app() if not app_config: return "No app currently selected. Use switch_app to select an app." # This endpoint requires app_id in the URL path endpoint = f"apps/{app_config.app_id}/segments" result = await make_onesignal_request(endpoint, method="GET", use_org_key=False) # Check if result is a dictionary with an error if isinstance(result, dict) and "error" in result: return f"Error retrieving segments: {result['error']}" # Handle different response formats if isinstance(result, dict): # Some endpoints return segments in a wrapper object segments = result.get("segments", []) elif isinstance(result, list): # Direct list of segments segments = result else: return f"Unexpected response format: {type(result)}" if not segments: return "No segments found." output = "Segments:\n\n" for segment in segments: if isinstance(segment, dict): output += f"ID: {segment.get('id')}\n" output += f"Name: {segment.get('name')}\n" output += f"Created: {segment.get('created_at')}\n" output += f"Updated: {segment.get('updated_at')}\n" output += f"Active: {segment.get('is_active', False)}\n" output += f"Read Only: {segment.get('read_only', False)}\n\n" return output @mcp.tool() async def create_segment(name: str, filters: str) -> str: """Create a new segment in your OneSignal app. Args: name: Name of the segment filters: JSON string representing the filters for this segment (e.g., '[{"field":"tag","key":"level","relation":"=","value":"10"}]') """ try: parsed_filters = json.loads(filters) except json.JSONDecodeError: return "Error: The filters parameter must be a valid JSON string." data = { "name": name, "filters": parsed_filters } endpoint = f"apps/{get_current_app().app_id}/segments" result = await make_onesignal_request(endpoint, method="POST", data=data) if "error" in result: return f"Error creating segment: {result['error']}" return f"Segment '{name}' created successfully with ID: {result.get('id')}" @mcp.tool() async def delete_segment(segment_id: str) -> str: """Delete a segment from your OneSignal app. Args: segment_id: ID of the segment to delete """ endpoint = f"apps/{get_current_app().app_id}/segments/{segment_id}" result = await make_onesignal_request(endpoint, method="DELETE") if "error" in result: return f"Error deleting segment: {result['error']}" return f"Segment '{segment_id}' deleted successfully" # === Template Management Tools === @mcp.tool() async def view_templates() -> str: """List all templates available in your OneSignal app.""" app_config = get_current_app() if not app_config: return "No app currently selected. Use switch_app to select an app." # Templates endpoint uses query param for app_id params = {"app_id": app_config.app_id} result = await make_onesignal_request("templates", method="GET", params=params, use_org_key=False) if "error" in result: return f"Error retrieving templates: {result['error']}" templates = result.get("templates", []) if not templates: return "No templates found." output = "Templates:\n\n" for template in templates: output += f"ID: {template.get('id')}\n" output += f"Name: {template.get('name')}\n" output += f"Created: {template.get('created_at')}\n" output += f"Updated: {template.get('updated_at')}\n\n" return output @mcp.tool() async def view_template_details(template_id: str) -> str: """Get detailed information about a specific template. Args: template_id: The ID of the template to retrieve details for """ params = {"app_id": get_current_app().app_id} result = await make_onesignal_request(f"templates/{template_id}", method="GET", params=params) if "error" in result: return f"Error fetching template details: {result['error']}" # Format the template details in a readable way heading = result.get("headings", {}).get("en", "No heading") if isinstance(result.get("headings"), dict) else "No heading" content = result.get("contents", {}).get("en", "No content") if isinstance(result.get("contents"), dict) else "No content" details = [ f"ID: {result.get('id')}", f"Name: {result.get('name')}", f"Title: {heading}", f"Message: {content}", f"Platform: {result.get('platform')}", f"Created: {result.get('created_at')}" ] return "\n".join(details) @mcp.tool() async def create_template(name: str, title: str, message: str) -> str: """Create a new template in your OneSignal app. Args: name: Name of the template title: Title/heading of the template message: Content/message of the template """ app_config = get_current_app() if not app_config: return "No app currently selected. Use switch_app to select an app." data = { "app_id": app_config.app_id, "name": name, "headings": {"en": title}, "contents": {"en": message} } # Templates endpoint uses app_id in the body result = await make_onesignal_request("templates", method="POST", data=data) if "error" in result: return f"Error creating template: {result['error']}" return f"Template '{name}' created successfully with ID: {result.get('id')}" # === App Information Tools === @mcp.tool() async def view_app_details() -> str: """Get detailed information about the configured OneSignal app.""" app_config = get_current_app() if not app_config: return "No app currently selected. Use switch_app to select an app." # This endpoint requires the app_id in the URL and Organization API Key result = await make_onesignal_request(f"apps/{app_config.app_id}", method="GET", use_org_key=True) if "error" in result: return f"Error retrieving app details: {result['error']}" output = f"ID: {result.get('id')}\n" output += f"Name: {result.get('name')}\n" output += f"Created: {result.get('created_at')}\n" output += f"Updated: {result.get('updated_at')}\n" output += f"GCM: {'Configured' if result.get('gcm_key') else 'Not Configured'}\n" output += f"APNS: {'Configured' if result.get('apns_env') else 'Not Configured'}\n" output += f"Chrome: {'Configured' if result.get('chrome_web_key') else 'Not Configured'}\n" output += f"Safari: {'Configured' if result.get('safari_site_origin') else 'Not Configured'}\n" output += f"Email: {'Configured' if result.get('email_marketing') else 'Not Configured'}\n" output += f"SMS: {'Configured' if result.get('sms_marketing') else 'Not Configured'}\n" return output @mcp.tool() async def view_apps() -> str: """List all OneSignal applications for the organization (requires Organization API Key).""" result = await make_onesignal_request("apps", method="GET", use_org_key=True) if "error" in result: if "401" in str(result["error"]) or "403" in str(result["error"]): return ("Error: Your Organization API Key is either not configured or doesn't have permission to view all apps. " "Make sure you've set the ONESIGNAL_ORG_API_KEY environment variable with a valid Organization API Key. " "Organization API Keys can be found in your OneSignal dashboard under Organizations > Keys & IDs.") return f"Error fetching applications: {result['error']}" if not result: return "No applications found." apps_info = [] for app in result: apps_info.append( f"ID: {app.get('id')}\n" f"Name: {app.get('name')}\n" f"GCM: {'Configured' if app.get('gcm_key') else 'Not Configured'}\n" f"APNS: {'Configured' if app.get('apns_env') else 'Not Configured'}\n" f"Created: {app.get('created_at')}" ) return "Applications:\n\n" + "\n\n".join(apps_info) # === Organization-level Tools === @mcp.tool() async def create_app(name: str, site_name: str = None) -> str: """Create a new OneSignal application at the organization level (requires Organization API Key). Args: name: Name of the new application site_name: Optional name of the website for the application """ data = { "name": name } if site_name: data["site_name"] = site_name result = await make_onesignal_request("apps", method="POST", data=data, use_org_key=True) if "error" in result: if "401" in str(result["error"]) or "403" in str(result["error"]): return ("Error: Your Organization API Key is either not configured or doesn't have permission to create apps. " "Make sure you've set the ONESIGNAL_ORG_API_KEY environment variable with a valid Organization API Key.") return f"Error creating application: {result['error']}" return f"Application '{name}' created successfully with ID: {result.get('id')}" @mcp.tool() async def update_app(app_id: str, name: str = None, site_name: str = None) -> str: """Update an existing OneSignal application at the organization level (requires Organization API Key). Args: app_id: ID of the app to update name: New name for the application (optional) site_name: New site name for the application (optional) """ data = {} if name: data["name"] = name if site_name: data["site_name"] = site_name if not data: return "Error: No update parameters provided. Specify at least one parameter to update." result = await make_onesignal_request(f"apps/{app_id}", method="PUT", data=data, use_org_key=True) if "error" in result: if "401" in str(result["error"]) or "403" in str(result["error"]): return ("Error: Your Organization API Key is either not configured or doesn't have permission to update apps. " "Make sure you've set the ONESIGNAL_ORG_API_KEY environment variable with a valid Organization API Key.") return f"Error updating application: {result['error']}" return f"Application '{app_id}' updated successfully" @mcp.tool() async def view_app_api_keys(app_id: str) -> str: """View API keys for a specific OneSignal app (requires Organization API Key). Args: app_id: The ID of the app to retrieve API keys for """ result = await make_onesignal_request(f"apps/{app_id}/auth/tokens", use_org_key=True) if "error" in result: if "401" in str(result["error"]) or "403" in str(result["error"]): return ("Error: Your Organization API Key is either not configured or doesn't have permission to view API keys. " "Make sure you've set the ONESIGNAL_ORG_API_KEY environment variable with a valid Organization API Key.") return f"Error fetching API keys: {result['error']}" if not result.get("tokens", []): return f"No API keys found for app ID: {app_id}" keys_info = [] for key in result.get("tokens", []): keys_info.append( f"ID: {key.get('id')}\n" f"Name: {key.get('name')}\n" f"Created: {key.get('created_at')}\n" f"Updated: {key.get('updated_at')}\n" f"IP Allowlist Mode: {key.get('ip_allowlist_mode', 'disabled')}" ) return f"API Keys for App {app_id}:\n\n" + "\n\n".join(keys_info) @mcp.tool() async def create_app_api_key(app_id: str, name: str) -> str: """Create a new API key for a specific OneSignal app (requires Organization API Key). Args: app_id: The ID of the app to create an API key for name: Name for the new API key """ data = { "name": name } result = await make_onesignal_request(f"apps/{app_id}/auth/tokens", method="POST", data=data, use_org_key=True) if "error" in result: if "401" in str(result["error"]) or "403" in str(result["error"]): return ("Error: Your Organization API Key is either not configured or doesn't have permission to create API keys. " "Make sure you've set the ONESIGNAL_ORG_API_KEY environment variable with a valid Organization API Key.") return f"Error creating API key: {result['error']}" # Format the API key details for display key_details = ( f"API Key '{name}' created successfully!\n\n" f"Key ID: {result.get('id')}\n" f"Token: {result.get('token')}\n\n" f"IMPORTANT: Save this token now! You won't be able to see the full token again." ) return key_details # === User Management Tools === @mcp.tool() async def create_user(name: str = None, email: str = None, external_id: str = None, tags: Dict[str, str] = None) -> Dict[str, Any]: """Create a new user in OneSignal. Args: name: User's name (optional) email: User's email address (optional) external_id: External user ID for identification (required for most use cases) tags: Additional user tags/properties (optional) """ app_config = get_current_app() if not app_config: return {"error": "No app currently selected. Use switch_app to select an app."} # Build the proper User Model structure data = {} # Identity section - external_id is the main identifier if external_id: data["identity"] = {"external_id": external_id} # Properties section - includes tags and other user properties properties = {} if tags: properties["tags"] = tags if properties: data["properties"] = properties # Subscriptions section - add email subscription if provided if email: data["subscriptions"] = [{ "type": "Email", "token": email, "enabled": True }] if not data.get("identity") and not data.get("subscriptions"): return {"error": "Must provide at least an external_id or email to create a user"} # Use the correct endpoint: apps/{app_id}/users result = await make_onesignal_request(f"apps/{app_config.app_id}/users", method="POST", data=data) return result @mcp.tool() async def view_user(user_id: str) -> Dict[str, Any]: """Get detailed information about a specific user. Args: user_id: The user identifier. Can be a onesignal_id or external_id. Will try onesignal_id first, then external_id. """ app_config = get_current_app() if not app_config: return {"error": "No app currently selected. Use switch_app to select an app."} # Try onesignal_id first (UUID format check) import re uuid_pattern = r'^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$' if re.match(uuid_pattern, user_id.lower()): alias_label = "onesignal_id" else: alias_label = "external_id" # Use the correct endpoint: apps/{app_id}/users/by/{alias_label}/{alias_id} result = await make_onesignal_request( f"apps/{app_config.app_id}/users/by/{alias_label}/{user_id}", method="GET" ) return result @mcp.tool() async def update_user(user_id: str, name: str = None, email: str = None, tags: Dict[str, str] = None) -> Dict[str, Any]: """Update an existing user's information. Args: user_id: The user identifier (onesignal_id or external_id) name: New name for the user (optional) email: New email address (optional) tags: New or updated tags/properties (optional) """ app_config = get_current_app() if not app_config: return {"error": "No app currently selected. Use switch_app to select an app."} # Build properties update properties = {} if tags: properties["tags"] = tags if not properties and not email: return {"error": "No update parameters provided. Specify tags or email."} data = {} if properties: data["properties"] = properties # Determine alias type import re uuid_pattern = r'^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$' if re.match(uuid_pattern, user_id.lower()): alias_label = "onesignal_id" else: alias_label = "external_id" # Use the correct endpoint: apps/{app_id}/users/by/{alias_label}/{alias_id} result = await make_onesignal_request( f"apps/{app_config.app_id}/users/by/{alias_label}/{user_id}", method="PATCH", data=data ) return result @mcp.tool() async def delete_user(user_id: str) -> Dict[str, Any]: """Delete a user and all their subscriptions. Args: user_id: The user identifier (onesignal_id or external_id) """ app_config = get_current_app() if not app_config: return {"error": "No app currently selected. Use switch_app to select an app."} # Determine alias type import re uuid_pattern = r'^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$' if re.match(uuid_pattern, user_id.lower()): alias_label = "onesignal_id" else: alias_label = "external_id" # Use the correct endpoint: apps/{app_id}/users/by/{alias_label}/{alias_id} result = await make_onesignal_request( f"apps/{app_config.app_id}/users/by/{alias_label}/{user_id}", method="DELETE" ) return result @mcp.tool() async def view_user_identity(user_id: str) -> Dict[str, Any]: """Get user identity information (aliases). Args: user_id: The user identifier (onesignal_id or external_id) """ app_config = get_current_app() if not app_config: return {"error": "No app currently selected. Use switch_app to select an app."} # Determine alias type import re uuid_pattern = r'^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$' if re.match(uuid_pattern, user_id.lower()): alias_label = "onesignal_id" else: alias_label = "external_id" # Use the correct endpoint: apps/{app_id}/users/by/{alias_label}/{alias_id}/identity result = await make_onesignal_request( f"apps/{app_config.app_id}/users/by/{alias_label}/{user_id}/identity", method="GET" ) return result @mcp.tool() async def create_or_update_alias(user_id: str, alias_label: str, alias_id: str) -> Dict[str, Any]: """Create or update a user alias. Args: user_id: The OneSignal User ID (onesignal_id or external_id to look up the user) alias_label: The type/label of the alias to create (e.g., "phone_number", "custom_id") alias_id: The alias identifier value """ app_config = get_current_app() if not app_config: return {"error": "No app currently selected. Use switch_app to select an app."} data = { "identity": { alias_label: alias_id } } # Determine lookup alias type import re uuid_pattern = r'^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$' if re.match(uuid_pattern, user_id.lower()): lookup_alias = "onesignal_id" else: lookup_alias = "external_id" # Use the correct endpoint: apps/{app_id}/users/by/{alias_label}/{alias_id}/identity result = await make_onesignal_request( f"apps/{app_config.app_id}/users/by/{lookup_alias}/{user_id}/identity", method="PATCH", data=data ) return result @mcp.tool() async def delete_alias(user_id: str, alias_label: str) -> Dict[str, Any]: """Delete a user alias. Args: user_id: The OneSignal User ID (onesignal_id or external_id to look up the user) alias_label: The type/label of the alias to delete """ app_config = get_current_app() if not app_config: return {"error": "No app currently selected. Use switch_app to select an app."} # Determine lookup alias type import re uuid_pattern = r'^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$' if re.match(uuid_pattern, user_id.lower()): lookup_alias = "onesignal_id" else: lookup_alias = "external_id" # Use the correct endpoint: apps/{app_id}/users/by/{alias_label}/{alias_id}/identity/{alias_to_delete} result = await make_onesignal_request( f"apps/{app_config.app_id}/users/by/{lookup_alias}/{user_id}/identity/{alias_label}", method="DELETE" ) return result # === Subscription Management Tools === @mcp.tool() async def create_subscription(user_id: str, subscription_type: str, identifier: str) -> Dict[str, Any]: """Create a new subscription for a user. Args: user_id: The user identifier (onesignal_id or external_id) subscription_type: Type of subscription (Email, SMS, iOSPush, AndroidPush, etc.) identifier: Email address, phone number (E.164), or push token """ app_config = get_current_app() if not app_config: return {"error": "No app currently selected. Use switch_app to select an app."} data = { "subscription": { "type": subscription_type, "token": identifier, "enabled": True } } # Determine lookup alias type import re uuid_pattern = r'^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$' if re.match(uuid_pattern, user_id.lower()): alias_label = "onesignal_id" else: alias_label = "external_id" # Use the correct endpoint: apps/{app_id}/users/by/{alias_label}/{alias_id}/subscriptions result = await make_onesignal_request( f"apps/{app_config.app_id}/users/by/{alias_label}/{user_id}/subscriptions", method="POST", data=data ) return result @mcp.tool() async def update_subscription(user_id: str, subscription_id: str, enabled: bool = None) -> Dict[str, Any]: """Update a user's subscription. Args: user_id: The user identifier (onesignal_id or external_id) - used for context subscription_id: The ID of the subscription to update enabled: Whether the subscription should be enabled or disabled (optional) """ app_config = get_current_app() if not app_config: return {"error": "No app currently selected. Use switch_app to select an app."} data = {} if enabled is not None: data["enabled"] = enabled # Set notification_types based on enabled status data["notification_types"] = 1 if enabled else -31 if not data: return {"error": "No update parameters provided. Specify enabled status."} # Subscription updates use: apps/{app_id}/subscriptions/{subscription_id} result = await make_onesignal_request( f"apps/{app_config.app_id}/subscriptions/{subscription_id}", method="PATCH", data=data ) return result @mcp.tool() async def delete_subscription(user_id: str, subscription_id: str) -> Dict[str, Any]: """Delete a user's subscription. Args: user_id: The user identifier (onesignal_id or external_id) - used for context subscription_id: The ID of the subscription to delete """ app_config = get_current_app() if not app_config: return {"error": "No app currently selected. Use switch_app to select an app."} # Subscription deletes use: apps/{app_id}/subscriptions/{subscription_id} result = await make_onesignal_request( f"apps/{app_config.app_id}/subscriptions/{subscription_id}", method="DELETE" ) return result @mcp.tool() async def transfer_subscription(user_id: str, subscription_id: str, new_user_id: str) -> Dict[str, Any]: """Transfer a subscription from one user to another. Args: user_id: The current user identifier (for context) subscription_id: The ID of the subscription to transfer new_user_id: The user identifier (onesignal_id or external_id) to transfer the subscription to """ app_config = get_current_app() if not app_config: return {"error": "No app currently selected. Use switch_app to select an app."} # Determine the new user's identity for the transfer import re uuid_pattern = r'^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$' # Build the identity object for the new user if re.match(uuid_pattern, new_user_id.lower()): identity = {"onesignal_id": new_user_id} else: identity = {"external_id": new_user_id} data = {"identity": identity} # Transfer endpoint: apps/{app_id}/subscriptions/{subscription_id}/owner result = await make_onesignal_request( f"apps/{app_config.app_id}/subscriptions/{subscription_id}/owner", method="PATCH", data=data ) return result @mcp.tool() async def unsubscribe_email(token: str) -> Dict[str, Any]: """Unsubscribe an email subscription using an unsubscribe token. Args: token: The unsubscribe token from the email """ app_config = get_current_app() if not app_config: return {"error": "No app currently selected. Use switch_app to select an app."} data = { "token": token } result = await make_onesignal_request("email/unsubscribe", method="POST", data=data) return result # === NEW: Email & SMS Messaging Tools === @mcp.tool() async def send_email(subject: str, body: str, email_body: str = None, include_emails: List[str] = None, segments: List[str] = None, external_ids: List[str] = None, template_id: str = None) -> Dict[str, Any]: """Send an email through OneSignal. Args: subject: Email subject line body: Plain text email content email_body: HTML email content (optional) include_emails: List of specific email addresses to target segments: List of segments to include external_ids: List of external user IDs to target template_id: Email template ID to use """ app_config = get_current_app() if not app_config: return {"error": "No app currently selected. Use switch_app to select an app."} email_data = { "app_id": app_config.app_id, "email_subject": subject, "email_body": email_body or body, "target_channel": "email" } # Set targeting if include_emails: email_data["include_emails"] = include_emails elif external_ids: email_data["include_external_user_ids"] = external_ids elif segments: email_data["included_segments"] = segments else: email_data["included_segments"] = ["Subscribed Users"] if template_id: email_data["template_id"] = template_id result = await make_onesignal_request("notifications", method="POST", data=email_data) return result @mcp.tool() async def send_sms(message: str, phone_numbers: List[str] = None, segments: List[str] = None, external_ids: List[str] = None, media_url: str = None) -> Dict[str, Any]: """Send an SMS/MMS through OneSignal. Args: message: SMS message content phone_numbers: List of phone numbers in E.164 format segments: List of segments to include external_ids: List of external user IDs to target media_url: URL for MMS media attachment """ app_config = get_current_app() if not app_config: return {"error": "No app currently selected. Use switch_app to select an app."} sms_data = { "app_id": app_config.app_id, "contents": {"en": message}, "target_channel": "sms" } if phone_numbers: sms_data["include_phone_numbers"] = phone_numbers elif external_ids: sms_data["include_external_user_ids"] = external_ids elif segments: sms_data["included_segments"] = segments else: return {"error": "SMS requires phone_numbers, external_ids, or segments"} if media_url: sms_data["mms_media_url"] = media_url result = await make_onesignal_request("notifications", method="POST", data=sms_data) return result @mcp.tool() async def send_transactional_message(channel: str, content: Dict[str, str], recipients: Dict[str, Any], template_id: str = None, custom_data: Dict[str, Any] = None) -> Dict[str, Any]: """Send a transactional message (immediate delivery, no scheduling). Args: channel: Channel to send on ("push", "email", "sms") content: Message content (format depends on channel) recipients: Targeting information (include_external_user_ids, include_emails, etc.) template_id: Template ID to use custom_data: Custom data to include """ app_config = get_current_app() if not app_config: return {"error": "No app currently selected. Use switch_app to select an app."} message_data = { "app_id": app_config.app_id, "target_channel": channel, "is_transactional": True } # Set content based on channel if channel == "email": message_data["email_subject"] = content.get("subject", "") message_data["email_body"] = content.get("body", "") else: message_data["contents"] = content # Set recipients message_data.update(recipients) if template_id: message_data["template_id"] = template_id if custom_data: message_data["data"] = custom_data result = await make_onesignal_request("notifications", method="POST", data=message_data) return result # === NEW: Enhanced Template Management === @mcp.tool() async def update_template(template_id: str, name: str = None, title: str = None, message: str = None) -> Dict[str, Any]: """Update an existing template. Args: template_id: ID of the template to update name: New name for the template title: New title/heading for the template message: New content/message for the template """ app_config = get_current_app() if not app_config: return {"error": "No app currently selected. Use switch_app to select an app."} data = {} if name: data["name"] = name if title: data["headings"] = {"en": title} if message: data["contents"] = {"en": message} if not data: return {"error": "No update parameters provided"} params = {"app_id": app_config.app_id} result = await make_onesignal_request(f"templates/{template_id}", method="PATCH", data=data, params=params) return result @mcp.tool() async def delete_template(template_id: str) -> Dict[str, Any]: """Delete a template from your OneSignal app. Args: template_id: ID of the template to delete """ app_config = get_current_app() if not app_config: return {"error": "No app currently selected. Use switch_app to select an app."} params = {"app_id": app_config.app_id} result = await make_onesignal_request(f"templates/{template_id}", method="DELETE", params=params) if "error" not in result: return {"success": f"Template '{template_id}' deleted successfully"} return result @mcp.tool() async def copy_template_to_app(template_id: str, target_app_id: str, new_name: str = None) -> Dict[str, Any]: """Copy a template to another OneSignal app. Args: template_id: ID of the template to copy target_app_id: ID of the app to copy the template to new_name: Optional new name for the copied template """ data = {"app_id": target_app_id} if new_name: data["name"] = new_name result = await make_onesignal_request(f"templates/{template_id}/copy", method="POST", data=data) return result # === NEW: Live Activities (iOS) === @mcp.tool() async def start_live_activity(activity_id: str, push_token: str, subscription_id: str, activity_attributes: Dict[str, Any], content_state: Dict[str, Any]) -> Dict[str, Any]: """Start a new iOS Live Activity. Args: activity_id: Unique identifier for the activity push_token: Push token for the Live Activity subscription_id: Subscription ID for the user activity_attributes: Static attributes for the activity content_state: Initial dynamic content state """ data = { "activity_id": activity_id, "push_token": push_token, "subscription_id": subscription_id, "activity_attributes": activity_attributes, "content_state": content_state } result = await make_onesignal_request(f"live_activities/{activity_id}/start", method="POST", data=data) return result @mcp.tool() async def update_live_activity(activity_id: str, name: str, event: str, content_state: Dict[str, Any], dismissal_date: int = None, priority: int = None, sound: str = None) -> Dict[str, Any]: """Update an existing iOS Live Activity. Args: activity_id: ID of the activity to update name: Name identifier for the update event: Event type ("update" or "end") content_state: Updated dynamic content state dismissal_date: Unix timestamp for automatic dismissal priority: Notification priority (5-10) sound: Sound file name for the update """ data = { "name": name, "event": event, "content_state": content_state } if dismissal_date: data["dismissal_date"] = dismissal_date if priority: data["priority"] = priority if sound: data["sound"] = sound result = await make_onesignal_request(f"live_activities/{activity_id}/update", method="POST", data=data) return result @mcp.tool() async def end_live_activity(activity_id: str, subscription_id: str, dismissal_date: int = None, priority: int = None) -> Dict[str, Any]: """End an iOS Live Activity. Args: activity_id: ID of the activity to end subscription_id: Subscription ID associated with the activity dismissal_date: Unix timestamp for dismissal priority: Notification priority (5-10) """ data = { "subscription_id": subscription_id, "event": "end" } if dismissal_date: data["dismissal_date"] = dismissal_date if priority: data["priority"] = priority result = await make_onesignal_request(f"live_activities/{activity_id}/end", method="POST", data=data) return result # === NEW: Analytics & Outcomes === @mcp.tool() async def view_outcomes(outcome_names: List[str], outcome_time_range: str = None, outcome_platforms: List[str] = None, outcome_attribution: str = None) -> Dict[str, Any]: """View outcomes data for your OneSignal app. Args: outcome_names: List of outcome names to fetch data for outcome_time_range: Time range for data (e.g., "1d", "1mo") outcome_platforms: Filter by platforms (e.g., ["ios", "android"]) outcome_attribution: Attribution model ("direct" or "influenced") """ app_config = get_current_app() if not app_config: return {"error": "No app currently selected. Use switch_app to select an app."} params = {"outcome_names": outcome_names} if outcome_time_range: params["outcome_time_range"] = outcome_time_range if outcome_platforms: params["outcome_platforms"] = outcome_platforms if outcome_attribution: params["outcome_attribution"] = outcome_attribution result = await make_onesignal_request(f"apps/{app_config.app_id}/outcomes", method="GET", params=params) return result # === NEW: Export Functions === @mcp.tool() async def export_messages_csv(start_date: str = None, end_date: str = None, event_types: List[str] = None) -> Dict[str, Any]: """Export messages/notifications data to CSV. Args: start_date: Start date for export (ISO 8601 format) end_date: End date for export (ISO 8601 format) event_types: List of event types to export """ data = {} if start_date: data["start_date"] = start_date if end_date: data["end_date"] = end_date if event_types: data["event_types"] = event_types result = await make_onesignal_request("notifications/csv_export", method="POST", data=data, use_org_key=True) return result # === NEW: API Key Management === @mcp.tool() async def delete_app_api_key(app_id: str, key_id: str) -> Dict[str, Any]: """Delete an API key from a specific OneSignal app. Args: app_id: The ID of the app key_id: The ID of the API key to delete """ result = await make_onesignal_request(f"apps/{app_id}/auth/tokens/{key_id}", method="DELETE", use_org_key=True) if "error" not in result: return {"success": f"API key '{key_id}' deleted successfully"} return result @mcp.tool() async def update_app_api_key(app_id: str, key_id: str, name: str = None, scopes: List[str] = None) -> Dict[str, Any]: """Update an API key for a specific OneSignal app. Args: app_id: The ID of the app key_id: The ID of the API key to update name: New name for the API key scopes: New list of permission scopes """ data = {} if name: data["name"] = name if scopes: data["scopes"] = scopes if not data: return {"error": "No update parameters provided"} result = await make_onesignal_request(f"apps/{app_id}/auth/tokens/{key_id}", method="PATCH", data=data, use_org_key=True) return result @mcp.tool() async def rotate_app_api_key(app_id: str, key_id: str) -> Dict[str, Any]: """Rotate an API key (generate new token while keeping permissions). Args: app_id: The ID of the app key_id: The ID of the API key to rotate """ result = await make_onesignal_request(f"apps/{app_id}/auth/tokens/{key_id}/rotate", method="POST", use_org_key=True) if "error" not in result: return { "success": f"API key rotated successfully", "new_token": result.get("token"), "warning": "Save the new token now! You won't be able to see it again." } return result # === Custom Events API === @mcp.tool() async def create_custom_events(events: List[Dict[str, Any]]) -> Dict[str, Any]: """Create custom events to track user actions in OneSignal. Custom events can represent any action users take in your application, such as completing a purchase, viewing content, or achieving milestones. Used for Journey automation and user behavior tracking. Args: events: List of event objects. Each event should have: - name (str): Event name (e.g., "purchase", "signup") - external_id (str): User's external ID (recommended) - onesignal_id (str): User's OneSignal ID (alternative to external_id) - timestamp (str): ISO 8601 timestamp (optional, defaults to now) - properties (dict): Additional event properties (optional) Example: events = [ {"name": "purchase", "external_id": "user123", "properties": {"amount": 99.99}}, {"name": "level_complete", "external_id": "user456", "properties": {"level": 5}} ] """ app_config = get_current_app() if not app_config: return {"error": "No app currently selected. Use switch_app to select an app."} data = {"events": events} result = await make_onesignal_request( f"apps/{app_config.app_id}/custom_events", method="POST", data=data ) if "error" not in result: return {"success": f"Successfully created {len(events)} custom event(s)"} return result # === Export Audience Activity API === @mcp.tool() async def export_audience_activity(message_id: str) -> Dict[str, Any]: """Export a CSV report of audience-level delivery and engagement data for a message. This includes sent, delivered, clicked, failed, and unsubscribed events across Push, Email, and SMS channels. Args: message_id: The ID of the message to export activity for Returns: A dict with csv_file_url to download the report (valid for 3 days). Note: Large exports may take several minutes. Poll the URL until available. """ app_config = get_current_app() if not app_config: return {"error": "No app currently selected. Use switch_app to select an app."} params = {"app_id": app_config.app_id} result = await make_onesignal_request( f"notifications/{message_id}/export_events", method="POST", params=params ) if "error" not in result and "csv_file_url" in result: return { "success": True, "csv_file_url": result["csv_file_url"], "note": "File may take a few minutes to generate. Poll the URL until available. Valid for 3 days." } return result # === Enhanced Push Notification with Advanced Options === @mcp.tool() async def send_advanced_push_notification( title: str, message: str, # Targeting options (use one) segments: List[str] = None, external_ids: List[str] = None, filters: List[Dict[str, Any]] = None, # Scheduling send_after: str = None, delayed_option: str = None, delivery_time_of_day: str = None, # Advanced options data: Dict[str, Any] = None, url: str = None, web_url: str = None, app_url: str = None, # Images big_picture: str = None, ios_attachments: Dict[str, str] = None, chrome_web_image: str = None, # Buttons buttons: List[Dict[str, str]] = None, web_buttons: List[Dict[str, str]] = None, # Delivery options ttl: int = None, priority: int = None, throttle_rate_per_minute: int = None, # iOS specific ios_badge_type: str = None, ios_badge_count: int = None, ios_sound: str = None, # Android specific android_channel_id: str = None, android_accent_color: str = None, android_led_color: str = None, # Idempotency idempotency_key: str = None, # Name for tracking name: str = None ) -> Dict[str, Any]: """Send a push notification with advanced options. Args: title: Notification title message: Notification message content Targeting (use one): - segments: List of segments (e.g., ["Subscribed Users"]) - external_ids: List of external user IDs - filters: List of filter objects for dynamic targeting Example: [{"field": "tag", "key": "level", "relation": ">", "value": "10"}] Scheduling: - send_after: ISO 8601 datetime or Unix timestamp for scheduled delivery - delayed_option: "timezone" for per-user timezone delivery, "last-active" for user activity - delivery_time_of_day: Time of day for timezone delivery (e.g., "9:00AM") Content: - data: Custom data payload (JSON) - url: URL to open on click - web_url: URL for web push only - app_url: Deep link URL for mobile Images: - big_picture: URL for large image (Android/Web) - ios_attachments: Dict of attachment ID to URL for iOS - chrome_web_image: URL for Chrome web push image Buttons: - buttons: Mobile action buttons [{"id": "btn1", "text": "Click Me", "icon": "icon.png"}] - web_buttons: Web action buttons [{"id": "btn1", "text": "Click", "url": "https://..."}] Delivery: - ttl: Time to live in seconds (default: 3 days) - priority: 1-10, with 10 being high priority - throttle_rate_per_minute: Messages per minute (Pro+ plans) iOS: - ios_badge_type: "Increase", "SetTo", or "None" - ios_badge_count: Badge number - ios_sound: Sound file name Android: - android_channel_id: Notification channel ID - android_accent_color: ARGB hex color (e.g., "FF00FF00") - android_led_color: LED color ARGB hex Other: - idempotency_key: Unique key to prevent duplicate sends (valid 24h) - name: Internal name for tracking in dashboard """ app_config = get_current_app() if not app_config: return {"error": "No app currently selected. Use switch_app to select an app."} notification_data = { "app_id": app_config.app_id, "contents": {"en": message}, "headings": {"en": title}, "target_channel": "push" } # Targeting if filters: notification_data["filters"] = filters elif external_ids: notification_data["include_aliases"] = {"external_id": external_ids} notification_data["target_channel"] = "push" elif segments: notification_data["included_segments"] = segments else: notification_data["included_segments"] = ["Subscribed Users"] # Scheduling if send_after: notification_data["send_after"] = send_after if delayed_option: notification_data["delayed_option"] = delayed_option if delivery_time_of_day: notification_data["delivery_time_of_day"] = delivery_time_of_day # Content if data: notification_data["data"] = data if url: notification_data["url"] = url if web_url: notification_data["web_url"] = web_url if app_url: notification_data["app_url"] = app_url # Images if big_picture: notification_data["big_picture"] = big_picture if ios_attachments: notification_data["ios_attachments"] = ios_attachments if chrome_web_image: notification_data["chrome_web_image"] = chrome_web_image # Buttons if buttons: notification_data["buttons"] = buttons if web_buttons: notification_data["web_buttons"] = web_buttons # Delivery options if ttl is not None: notification_data["ttl"] = ttl if priority is not None: notification_data["priority"] = priority if throttle_rate_per_minute is not None: notification_data["throttle_rate_per_minute"] = throttle_rate_per_minute # iOS specific if ios_badge_type: notification_data["ios_badgeType"] = ios_badge_type if ios_badge_count is not None: notification_data["ios_badgeCount"] = ios_badge_count if ios_sound: notification_data["ios_sound"] = ios_sound # Android specific if android_channel_id: notification_data["android_channel_id"] = android_channel_id if android_accent_color: notification_data["android_accent_color"] = android_accent_color if android_led_color: notification_data["android_led_color"] = android_led_color # Name if name: notification_data["name"] = name # Headers for idempotency headers = {} if idempotency_key: notification_data["idempotency_key"] = idempotency_key result = await make_onesignal_request("notifications", method="POST", data=notification_data) return result # === Enhanced Email with Advanced Options === @mcp.tool() async def send_advanced_email( subject: str, body: str, # Targeting (use one) include_emails: List[str] = None, external_ids: List[str] = None, segments: List[str] = None, filters: List[Dict[str, Any]] = None, # Content email_body: str = None, email_preheader: str = None, template_id: str = None, # Sender info email_from_name: str = None, email_from_address: str = None, email_reply_to_address: str = None, # Scheduling send_after: str = None, # Tracking disable_email_click_tracking: bool = None, # Other name: str = None, idempotency_key: str = None ) -> Dict[str, Any]: """Send an email with advanced options. Args: subject: Email subject line body: Plain text email content (fallback) Targeting (use one): - include_emails: List of specific email addresses - external_ids: List of external user IDs - segments: List of segments - filters: Dynamic targeting filters Content: - email_body: HTML email body (overrides body) - email_preheader: Preview text shown in inbox - template_id: Use a predefined email template Sender: - email_from_name: Sender display name - email_from_address: Sender email address - email_reply_to_address: Reply-to address Scheduling: - send_after: ISO 8601 datetime for scheduled send Tracking: - disable_email_click_tracking: Disable click tracking Other: - name: Internal name for tracking - idempotency_key: Prevent duplicate sends """ app_config = get_current_app() if not app_config: return {"error": "No app currently selected. Use switch_app to select an app."} email_data = { "app_id": app_config.app_id, "email_subject": subject, "email_body": email_body or f"<html><body>{body}</body></html>", "target_channel": "email" } # Targeting if filters: email_data["filters"] = filters elif include_emails: email_data["include_email_tokens"] = include_emails elif external_ids: email_data["include_aliases"] = {"external_id": external_ids} elif segments: email_data["included_segments"] = segments else: email_data["included_segments"] = ["Subscribed Users"] # Content if email_preheader: email_data["email_preheader"] = email_preheader if template_id: email_data["template_id"] = template_id # Sender info if email_from_name: email_data["email_from_name"] = email_from_name if email_from_address: email_data["email_from_address"] = email_from_address if email_reply_to_address: email_data["email_reply_to_address"] = email_reply_to_address # Scheduling if send_after: email_data["send_after"] = send_after # Tracking if disable_email_click_tracking is not None: email_data["disable_email_click_tracking"] = disable_email_click_tracking # Other if name: email_data["name"] = name if idempotency_key: email_data["idempotency_key"] = idempotency_key result = await make_onesignal_request("notifications", method="POST", data=email_data) return result # === Enhanced SMS with Advanced Options === @mcp.tool() async def send_advanced_sms( message: str, # Targeting (use one) phone_numbers: List[str] = None, external_ids: List[str] = None, segments: List[str] = None, filters: List[Dict[str, Any]] = None, # MMS media_url: str = None, # Scheduling send_after: str = None, # Sender sms_from: str = None, # Other name: str = None, idempotency_key: str = None ) -> Dict[str, Any]: """Send an SMS/MMS with advanced options. Args: message: SMS message content Targeting (use one): - phone_numbers: List of phone numbers in E.164 format - external_ids: List of external user IDs - segments: List of segments - filters: Dynamic targeting filters MMS: - media_url: URL for MMS media attachment (image/video) Scheduling: - send_after: ISO 8601 datetime for scheduled send Sender: - sms_from: Sender phone number (must be configured in OneSignal) Other: - name: Internal name for tracking - idempotency_key: Prevent duplicate sends """ app_config = get_current_app() if not app_config: return {"error": "No app currently selected. Use switch_app to select an app."} sms_data = { "app_id": app_config.app_id, "contents": {"en": message}, "target_channel": "sms" } # Targeting if filters: sms_data["filters"] = filters elif phone_numbers: sms_data["include_phone_numbers"] = phone_numbers elif external_ids: sms_data["include_aliases"] = {"external_id": external_ids} elif segments: sms_data["included_segments"] = segments else: return {"error": "Must specify phone_numbers, external_ids, segments, or filters"} # MMS if media_url: sms_data["sms_media_urls"] = [media_url] # Scheduling if send_after: sms_data["send_after"] = send_after # Sender if sms_from: sms_data["sms_from"] = sms_from # Other if name: sms_data["name"] = name if idempotency_key: sms_data["idempotency_key"] = idempotency_key result = await make_onesignal_request("notifications", method="POST", data=sms_data) return result # === Start Live Activity (Enhanced) === @mcp.tool() async def start_live_activity_advanced( activity_type: str, activity_id: str, event_attributes: Dict[str, Any], event_updates: Dict[str, Any], name: str, title: str, message: str, # Targeting (use one) external_ids: List[str] = None, subscription_ids: List[str] = None, segments: List[str] = None, filters: List[Dict[str, Any]] = None, # Options stale_date: int = None, priority: int = None, ios_relevance_score: float = None, idempotency_key: str = None ) -> Dict[str, Any]: """Start a Live Activity on iOS devices with full control. Args: activity_type: The Live Activity type defined in your app (e.g., "DeliveryAttributes") activity_id: Unique ID to track this Live Activity (use UUID) event_attributes: Static data to initialize the Live Activity event_updates: Dynamic content for the Live Activity (ContentState) name: Internal name for tracking title: Push notification title message: Push notification message Targeting (use one): - external_ids: List of user external IDs - subscription_ids: List of subscription IDs - segments: List of segments - filters: Dynamic targeting filters Options: - stale_date: Unix timestamp when activity becomes stale - priority: 5 (normal) or 10 (high) - ios_relevance_score: 0-1, higher shows in Dynamic Island - idempotency_key: Prevent duplicate sends """ app_config = get_current_app() if not app_config: return {"error": "No app currently selected. Use switch_app to select an app."} data = { "event": "start", "activity_id": activity_id, "event_attributes": event_attributes, "event_updates": event_updates, "name": name, "contents": {"en": message}, "headings": {"en": title} } # Targeting if external_ids: data["include_aliases"] = {"external_id": external_ids} elif subscription_ids: data["include_subscription_ids"] = subscription_ids elif segments: data["included_segments"] = segments elif filters: data["filters"] = filters else: return {"error": "Must specify targeting: external_ids, subscription_ids, segments, or filters"} # Options if stale_date is not None: data["stale_date"] = stale_date if priority is not None: data["priority"] = priority if ios_relevance_score is not None: data["ios_relevance_score"] = ios_relevance_score if idempotency_key: data["idempotency_key"] = idempotency_key result = await make_onesignal_request( f"apps/{app_config.app_id}/activities/activity/{activity_type}", method="POST", data=data ) return result # Run the server if __name__ == "__main__": # Run the server mcp.run()