import os
import json
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", settings={"log_level": log_level_str})
logger.info(f"OneSignal MCP server initialized with log level: {log_level_str}")
# OneSignal API configuration
ONESIGNAL_API_URL = "https://api.onesignal.com/api/v1"
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
org_level_endpoints = [
"apps", # Managing apps
"notifications/csv_export" # Export notifications
]
# Check if endpoint starts with or matches any org-level endpoint
for org_endpoint in org_level_endpoints:
if endpoint == org_endpoint or endpoint.startswith(f"{org_endpoint}/"):
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, json=data, timeout=30)
elif method == "PUT":
response = requests.put(url, headers=headers, json=data, timeout=30)
elif method == "DELETE":
response = requests.delete(url, headers=headers, timeout=30)
elif method == "PATCH":
response = requests.patch(url, headers=headers, 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."
# This endpoint requires app_id in the URL path
endpoint = f"apps/{app_config.app_id}/templates"
result = await make_onesignal_request(endpoint, method="GET", 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 = {
"name": name,
"headings": {"en": title},
"contents": {"en": message}
}
# This endpoint requires app_id in the URL path
endpoint = f"apps/{app_config.app_id}/templates"
result = await make_onesignal_request(endpoint, 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 (optional)
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."}
data = {}
if name:
data["name"] = name
if email:
data["email"] = email
if external_id:
data["external_user_id"] = external_id
if tags:
data["tags"] = tags
result = await make_onesignal_request("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 OneSignal User ID 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."}
result = await make_onesignal_request(f"users/{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 OneSignal User ID to update
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."}
data = {}
if name:
data["name"] = name
if email:
data["email"] = email
if tags:
data["tags"] = tags
if not data:
return {"error": "No update parameters provided"}
result = await make_onesignal_request(f"users/{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 OneSignal User ID to delete
"""
app_config = get_current_app()
if not app_config:
return {"error": "No app currently selected. Use switch_app to select an app."}
result = await make_onesignal_request(f"users/{user_id}", method="DELETE")
return result
@mcp.tool()
async def view_user_identity(user_id: str) -> Dict[str, Any]:
"""Get user identity information.
Args:
user_id: The OneSignal User ID to retrieve identity for
"""
app_config = get_current_app()
if not app_config:
return {"error": "No app currently selected. Use switch_app to select an app."}
result = await make_onesignal_request(f"users/{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
alias_label: The type/label of the alias (e.g., "email", "phone", "external")
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 = {
"alias": {
alias_label: alias_id
}
}
result = await make_onesignal_request(f"users/{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
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."}
result = await make_onesignal_request(f"users/{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 OneSignal User ID
subscription_type: Type of subscription ("email", "sms", "push")
identifier: Email address or phone number for the subscription
"""
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,
"identifier": identifier
}
}
result = await make_onesignal_request(f"users/{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 OneSignal User ID
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
result = await make_onesignal_request(f"users/{user_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 OneSignal User ID
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."}
result = await make_onesignal_request(f"users/{user_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 OneSignal User ID
subscription_id: The ID of the subscription to transfer
new_user_id: The OneSignal User 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."}
data = {
"new_user_id": new_user_id
}
result = await make_onesignal_request(f"users/{user_id}/subscriptions/{subscription_id}/transfer", 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
"""
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"}
result = await make_onesignal_request(f"templates/{template_id}",
method="PATCH", data=data)
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
"""
result = await make_onesignal_request(f"templates/{template_id}",
method="DELETE")
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
# Run the server
if __name__ == "__main__":
# Run the server
mcp.run()