# Directory Structure
```
├── .env.example
├── .github
│ └── workflows
│ └── ci.yml
├── .gitignore
├── check_loaded_key.py
├── CONTRIBUTING.md
├── debug_api_key.py
├── examples
│ ├── send_invite_email.py
│ └── send_notification.py
├── implementation_examples.md
├── LICENSE
├── missing_endpoints_analysis.md
├── onesignal_refactored
│ ├── __init__.py
│ ├── api_client.py
│ ├── config.py
│ ├── server.py
│ └── tools
│ ├── __init__.py
│ ├── analytics.py
│ ├── live_activities.py
│ ├── messages.py
│ └── templates.py
├── onesignal_refactoring_summary.md
├── onesignal_server.py
├── onesignal_tools_list.md
├── README.md
├── requirements.txt
├── setup.py
├── test_api_key_validity.py
├── test_auth_fix.py
├── test_onesignal_mcp.py
├── test_segments_debug.py
└── tests
├── __init__.py
└── test_onesignal_server.py
```
# Files
--------------------------------------------------------------------------------
/.gitignore:
--------------------------------------------------------------------------------
```
# Environment variables
.env
# Python
__pycache__/
*.py[cod]
*$py.class
*.so
.Python
env/
build/
develop-eggs/
dist/
downloads/
eggs/
.eggs/
lib/
lib64/
parts/
sdist/
var/
*.egg-info/
.installed.cfg
*.egg
# Virtual Environment
venv/
ENV/
# IDE files
.idea/
.vscode/
*.swp
*.swo
```
--------------------------------------------------------------------------------
/.env.example:
--------------------------------------------------------------------------------
```
# App-specific credentials
# Mandible app
ONESIGNAL_MANDIBLE_APP_ID=your_mandible_app_id_here
ONESIGNAL_MANDIBLE_API_KEY=your_mandible_api_key_here
# Weird Brains app
ONESIGNAL_WEIRDBRAINS_APP_ID=your_weirdbrains_app_id_here
ONESIGNAL_WEIRDBRAINS_API_KEY=your_weirdbrains_api_key_here
# Default app credentials (will be used if app-specific credentials not found)
ONESIGNAL_APP_ID=your_default_app_id_here
ONESIGNAL_API_KEY=your_default_api_key_here
# Organization API key (used for organization-level operations)
ONESIGNAL_ORG_API_KEY=your_organization_api_key_here
```
--------------------------------------------------------------------------------
/README.md:
--------------------------------------------------------------------------------
```markdown
# OneSignal MCP Server
A comprehensive Model Context Protocol (MCP) server for interacting with the OneSignal API. This server provides a complete interface for managing push notifications, emails, SMS, users, devices, segments, templates, analytics, and more through OneSignal's REST API.
[](https://opensource.org/licenses/MIT)
[](https://github.com/weirdbrains/onesignal-mcp)
[](https://github.com/weirdbrains/onesignal-mcp)
## Overview
This MCP server provides comprehensive access to the [OneSignal REST API](https://documentation.onesignal.com/reference/rest-api-overview), offering **57 tools** that cover all major OneSignal operations:
### 🚀 Key Features
- **Multi-channel Messaging**: Send push notifications, emails, SMS, and transactional messages
- **User & Device Management**: Complete CRUD operations for users, devices, and subscriptions
- **Advanced Segmentation**: Create and manage user segments with complex filters
- **Template System**: Create, update, and manage message templates
- **iOS Live Activities**: Full support for iOS Live Activities
- **Analytics & Export**: View outcomes data and export to CSV
- **Multi-App Support**: Manage multiple OneSignal applications seamlessly
- **API Key Management**: Create, update, rotate, and delete API keys
- **Organization-level Operations**: Manage apps across your entire organization
## Requirements
- Python 3.7 or higher
- `python-dotenv` package
- `requests` package
- `mcp` package
- OneSignal account with API credentials
## Installation
### Option 1: Clone from GitHub
```bash
# Clone the repository
git clone https://github.com/weirdbrains/onesignal-mcp.git
cd onesignal-mcp
# Install dependencies
pip install -r requirements.txt
```
### Option 2: Install as a Package (Coming Soon)
```bash
pip install onesignal-mcp
```
## Configuration
1. Create a `.env` file in the root directory with your OneSignal credentials:
```
# Default app credentials (optional, you can also add apps via the API)
ONESIGNAL_APP_ID=your_app_id_here
ONESIGNAL_API_KEY=your_rest_api_key_here
# Organization API key (for org-level operations)
ONESIGNAL_ORG_API_KEY=your_organization_api_key_here
# Optional: Multiple app configurations
ONESIGNAL_MANDIBLE_APP_ID=mandible_app_id
ONESIGNAL_MANDIBLE_API_KEY=mandible_api_key
ONESIGNAL_WEIRDBRAINS_APP_ID=weirdbrains_app_id
ONESIGNAL_WEIRDBRAINS_API_KEY=weirdbrains_api_key
# Logging level (DEBUG, INFO, WARNING, ERROR, CRITICAL)
LOG_LEVEL=INFO
```
2. Find your OneSignal credentials:
- **App ID**: Settings > Keys & IDs > OneSignal App ID
- **REST API Key**: Settings > Keys & IDs > REST API Key
- **Organization API Key**: Organization Settings > API Keys
## Usage
### Running the Server
```bash
python onesignal_server.py
```
The server will start and register itself with the MCP system, making all 57 tools available for use.
## Complete Tools Reference (57 Tools)
### 📱 App Management (5 tools)
- `list_apps` - List all configured OneSignal apps
- `add_app` - Add a new OneSignal app configuration locally
- `update_local_app_config` - Update an existing local app configuration
- `remove_app` - Remove a local OneSignal app configuration
- `switch_app` - Switch the current app to use for API requests
### 📨 Messaging (8 tools)
- `send_push_notification` - Send a push notification
- `send_email` - Send an email through OneSignal
- `send_sms` - Send an SMS/MMS through OneSignal
- `send_transactional_message` - Send immediate delivery messages
- `view_messages` - View recent messages sent
- `view_message_details` - Get detailed information about a message
- `view_message_history` - View message history/recipients
- `cancel_message` - Cancel a scheduled message
### 📱 Devices/Players (6 tools)
- `view_devices` - View devices subscribed to your app
- `view_device_details` - Get detailed information about a device
- `add_player` - Add a new player/device
- `edit_player` - Edit an existing player/device
- `delete_player` - Delete a player/device record
- `edit_tags_with_external_user_id` - Bulk edit tags by external ID
### 🎯 Segments (3 tools)
- `view_segments` - List all segments
- `create_segment` - Create a new segment
- `delete_segment` - Delete a segment
### 📄 Templates (6 tools)
- `view_templates` - List all templates
- `view_template_details` - Get template details
- `create_template` - Create a new template
- `update_template` - Update an existing template
- `delete_template` - Delete a template
- `copy_template_to_app` - Copy template to another app
### 🏢 Apps (6 tools)
- `view_app_details` - Get details about configured app
- `view_apps` - List all organization apps
- `create_app` - Create a new OneSignal application
- `update_app` - Update an existing application
- `view_app_api_keys` - View API keys for an app
- `create_app_api_key` - Create a new API key
### 🔑 API Key Management (3 tools)
- `delete_app_api_key` - Delete an API key
- `update_app_api_key` - Update an API key
- `rotate_app_api_key` - Rotate an API key
### 👤 Users (6 tools)
- `create_user` - Create a new user
- `view_user` - View user details
- `update_user` - Update user information
- `delete_user` - Delete a user
- `view_user_identity` - Get user identity information
- `view_user_identity_by_subscription` - Get identity by subscription
### 🏷️ Aliases (3 tools)
- `create_or_update_alias` - Create or update user alias
- `delete_alias` - Delete a user alias
- `create_alias_by_subscription` - Create alias by subscription ID
### 📬 Subscriptions (5 tools)
- `create_subscription` - Create a new subscription
- `update_subscription` - Update a subscription
- `delete_subscription` - Delete a subscription
- `transfer_subscription` - Transfer subscription between users
- `unsubscribe_email` - Unsubscribe using email token
### 🎯 Live Activities (3 tools)
- `start_live_activity` - Start iOS Live Activity
- `update_live_activity` - Update iOS Live Activity
- `end_live_activity` - End iOS Live Activity
### 📊 Analytics & Export (3 tools)
- `view_outcomes` - View outcomes/conversion data
- `export_players_csv` - Export player data to CSV
- `export_messages_csv` - Export messages to CSV
## Usage Examples
### Multi-Channel Messaging
```python
# Send a push notification
await send_push_notification(
title="Hello World",
message="This is a test notification",
segments=["Subscribed Users"]
)
# Send an email
await send_email(
subject="Welcome!",
body="Thank you for joining us",
email_body="<html><body><h1>Welcome!</h1></body></html>",
include_emails=["[email protected]"]
)
# Send an SMS
await send_sms(
message="Your verification code is 12345",
phone_numbers=["+15551234567"]
)
# Send a transactional message
await send_transactional_message(
channel="email",
content={"subject": "Order Confirmation", "body": "Your order has been confirmed"},
recipients={"include_external_user_ids": ["user123"]}
)
```
### User and Device Management
```python
# Create a user
user = await create_user(
name="John Doe",
email="[email protected]",
external_id="user123",
tags={"plan": "premium", "joined": "2024-01-01"}
)
# Add a device
device = await add_player(
device_type=1, # Android
identifier="device_token_here",
language="en",
tags={"app_version": "1.0.0"}
)
# Update user tags across all devices
await edit_tags_with_external_user_id(
external_user_id="user123",
tags={"last_active": "2024-01-15", "purchases": "5"}
)
```
### iOS Live Activities
```python
# Start a Live Activity
await start_live_activity(
activity_id="delivery_123",
push_token="live_activity_push_token",
subscription_id="user_subscription_id",
activity_attributes={"order_number": "12345"},
content_state={"status": "preparing", "eta": "15 mins"}
)
# Update the Live Activity
await update_live_activity(
activity_id="delivery_123",
name="delivery_update",
event="update",
content_state={"status": "on_the_way", "eta": "5 mins"}
)
```
### Analytics and Export
```python
# View conversion outcomes
outcomes = await view_outcomes(
outcome_names=["purchase", "session_duration"],
outcome_time_range="7d",
outcome_platforms=["ios", "android"]
)
# Export player data
export = await export_players_csv(
start_date="2024-01-01T00:00:00Z",
end_date="2024-01-31T23:59:59Z",
segment_names=["Active Users"]
)
```
## Testing
The server includes a comprehensive test suite. To run tests:
```bash
# Run the test script
python test_onesignal_mcp.py
# Or use unittest
python -m unittest discover tests
```
## Error Handling
The server provides consistent error handling:
- All errors are returned in a standardized format
- Detailed error messages help identify issues
- Automatic retry logic for transient failures
- Proper authentication error messages
## Rate Limiting
OneSignal enforces rate limits on API requests:
- Standard limit: 10 requests per second
- Bulk operations: May have lower limits
- The server includes guidance on handling rate limits
## Contributing
We welcome contributions! Please see [CONTRIBUTING.md](CONTRIBUTING.md) for guidelines.
## License
This project is licensed under the MIT License - see the [LICENSE](LICENSE) file for details.
## Acknowledgements
- [OneSignal](https://onesignal.com/) for their excellent notification service
- The MCP community for the Model Context Protocol
- All contributors to this project
```
--------------------------------------------------------------------------------
/CONTRIBUTING.md:
--------------------------------------------------------------------------------
```markdown
# Contributing to OneSignal MCP Server
Thank you for considering contributing to the OneSignal MCP Server! This document provides guidelines and instructions for contributing to this project.
## Code of Conduct
Please be respectful and considerate of others when contributing to this project. We aim to foster an inclusive and welcoming community.
## How to Contribute
### Reporting Bugs
If you find a bug, please create an issue with the following information:
- A clear, descriptive title
- Steps to reproduce the bug
- Expected behavior
- Actual behavior
- Any relevant logs or error messages
- Your environment (OS, Python version, etc.)
### Suggesting Features
If you have an idea for a new feature, please create an issue with:
- A clear, descriptive title
- A detailed description of the feature
- Any relevant examples or use cases
- Why this feature would be beneficial
### Pull Requests
1. Fork the repository
2. Create a new branch for your changes
3. Make your changes
4. Run tests to ensure your changes don't break existing functionality
5. Submit a pull request
### Development Setup
1. Clone the repository
2. Install dependencies:
```
pip install -r requirements.txt
```
3. Create a `.env` file with your OneSignal credentials (see `.env.example`)
## Coding Standards
- Follow PEP 8 style guidelines
- Write docstrings for all functions, classes, and modules
- Include type hints where appropriate
- Write tests for new functionality
## Testing
Before submitting a pull request, please ensure that all tests pass. You can run tests with:
```
# TODO: Add testing instructions once tests are implemented
```
## Documentation
Please update documentation when making changes:
- Update docstrings for modified functions
- Update the README.md if necessary
- Add examples for new functionality
## License
By contributing to this project, you agree that your contributions will be licensed under the project's [MIT License](LICENSE).
```
--------------------------------------------------------------------------------
/tests/__init__.py:
--------------------------------------------------------------------------------
```python
```
--------------------------------------------------------------------------------
/requirements.txt:
--------------------------------------------------------------------------------
```
python-dotenv>=1.0.0
requests>=2.31.0
```
--------------------------------------------------------------------------------
/onesignal_refactored/tools/__init__.py:
--------------------------------------------------------------------------------
```python
"""OneSignal MCP Server Tools - API endpoint implementations."""
from . import messages
from . import templates
from . import live_activities
from . import analytics
__all__ = [
"messages",
"templates",
"live_activities",
"analytics"
]
```
--------------------------------------------------------------------------------
/onesignal_refactored/__init__.py:
--------------------------------------------------------------------------------
```python
"""OneSignal MCP Server - Refactored Implementation."""
from .config import app_manager, AppConfig
from .api_client import api_client, OneSignalAPIError
from .server import mcp, __version__
__all__ = [
"app_manager",
"AppConfig",
"api_client",
"OneSignalAPIError",
"mcp",
"__version__"
]
```
--------------------------------------------------------------------------------
/setup.py:
--------------------------------------------------------------------------------
```python
from setuptools import setup, find_packages
with open("README.md", "r", encoding="utf-8") as fh:
long_description = fh.read()
setup(
name="onesignal-mcp",
version="1.0.0",
author="Weirdbrains",
author_email="[email protected]",
description="A Model Context Protocol (MCP) server for interacting with the OneSignal API",
long_description=long_description,
long_description_content_type="text/markdown",
url="https://github.com/weirdbrains/onesignal-mcp",
packages=find_packages(),
classifiers=[
"Programming Language :: Python :: 3",
"License :: OSI Approved :: MIT License",
"Operating System :: OS Independent",
],
python_requires=">=3.7",
install_requires=[
"python-dotenv>=1.0.0",
"requests>=2.31.0",
],
include_package_data=True,
)
```
--------------------------------------------------------------------------------
/examples/send_notification.py:
--------------------------------------------------------------------------------
```python
#!/usr/bin/env python
"""
Example script demonstrating how to send a notification using the OneSignal MCP server.
"""
import asyncio
import sys
import os
# Add the parent directory to the path so we can import the server module
sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), '..')))
# Import the server module
from onesignal_server import send_notification
async def main():
"""Send a test notification to all subscribed users."""
print("Sending a test notification...")
result = await send_notification(
title="Hello from OneSignal MCP",
message="This is a test notification sent from the example script.",
segment="Subscribed Users",
data={"custom_key": "custom_value"}
)
if "error" in result:
print(f"Error: {result['error']}")
else:
print(f"Success! Notification ID: {result.get('id')}")
print(f"Recipients: {result.get('recipients')}")
if __name__ == "__main__":
asyncio.run(main())
```
--------------------------------------------------------------------------------
/.github/workflows/ci.yml:
--------------------------------------------------------------------------------
```yaml
name: CI
on:
push:
branches: [ main ]
pull_request:
branches: [ main ]
jobs:
lint:
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v3
- name: Set up Python
uses: actions/setup-python@v4
with:
python-version: '3.9'
- name: Install dependencies
run: |
python -m pip install --upgrade pip
pip install flake8 black isort
if [ -f requirements.txt ]; then pip install -r requirements.txt; fi
- name: Lint with flake8
run: |
flake8 . --count --select=E9,F63,F7,F82 --show-source --statistics
- name: Check formatting with black
run: |
black --check .
- name: Check imports with isort
run: |
isort --check-only --profile black .
test:
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v3
- name: Set up Python
uses: actions/setup-python@v4
with:
python-version: '3.9'
- name: Install dependencies
run: |
python -m pip install --upgrade pip
pip install pytest pytest-cov
if [ -f requirements.txt ]; then pip install -r requirements.txt; fi
- name: Test with pytest
run: |
# Uncomment when tests are added
# pytest --cov=. --cov-report=xml
echo "Tests will be added in a future update"
```
--------------------------------------------------------------------------------
/check_loaded_key.py:
--------------------------------------------------------------------------------
```python
#!/usr/bin/env python3
"""Check what API key is currently loaded from .env"""
import os
from dotenv import load_dotenv
# Load environment variables
load_dotenv()
# Get the current API key
api_key = os.getenv("ONESIGNAL_MANDIBLE_API_KEY")
print("Currently loaded Mandible API key:")
print(f"Length: {len(api_key) if api_key else 'None'}")
print(f"Prefix: {api_key[:20] if api_key else 'None'}...")
print(f"Suffix: ...{api_key[-10:] if api_key else 'None'}")
# Also check if we can read the .env file directly
print("\nReading .env file directly:")
try:
with open('.env', 'r') as f:
for line in f:
if 'ONESIGNAL_MANDIBLE_API_KEY' in line and not line.strip().startswith('#'):
key_from_file = line.split('=', 1)[1].strip().strip('"').strip("'")
print(f"Length: {len(key_from_file)}")
print(f"Prefix: {key_from_file[:20]}...")
print(f"Suffix: ...{key_from_file[-10:]}")
if api_key != key_from_file:
print("\n⚠️ WARNING: The loaded key differs from what's in .env!")
print(" You need to restart the MCP server to load the new key.")
else:
print("\n✅ The loaded key matches what's in .env")
break
except Exception as e:
print(f"Error reading .env file: {e}")
```
--------------------------------------------------------------------------------
/test_segments_debug.py:
--------------------------------------------------------------------------------
```python
#!/usr/bin/env python3
"""Debug script to test the segments endpoint with detailed output"""
import os
import requests
from dotenv import load_dotenv
# Load environment variables
load_dotenv()
# Get credentials
app_id = os.getenv("ONESIGNAL_MANDIBLE_APP_ID")
api_key = os.getenv("ONESIGNAL_MANDIBLE_API_KEY")
print(f"App ID: {app_id}")
print(f"API Key type: {'v2' if api_key.startswith('os_v2_') else 'v1'}")
print(f"API Key prefix: {api_key[:15]}...")
# Try different authentication methods
url = f"https://api.onesignal.com/apps/{app_id}/segments"
print(f"\nTesting URL: {url}")
# Test 1: Using 'Key' authorization for v2 API key
headers1 = {
"Authorization": f"Key {api_key}",
"Accept": "application/json",
"Content-Type": "application/json"
}
print("\n1. Testing with 'Key' authorization header...")
try:
response = requests.get(url, headers=headers1)
print(f"Status: {response.status_code}")
print(f"Response: {response.text[:200]}")
except Exception as e:
print(f"Error: {e}")
# Test 2: Using 'Basic' authorization
headers2 = {
"Authorization": f"Basic {api_key}",
"Accept": "application/json",
"Content-Type": "application/json"
}
print("\n2. Testing with 'Basic' authorization header...")
try:
response = requests.get(url, headers=headers2)
print(f"Status: {response.status_code}")
print(f"Response: {response.text[:200]}")
except Exception as e:
print(f"Error: {e}")
# Test 3: Without app_id in URL path (using query param)
url2 = "https://api.onesignal.com/segments"
params = {"app_id": app_id}
print(f"\n3. Testing with app_id as query param: {url2}")
print(f"Params: {params}")
try:
response = requests.get(url2, headers=headers1, params=params)
print(f"Status: {response.status_code}")
print(f"Response: {response.text[:200]}")
except Exception as e:
print(f"Error: {e}")
```
--------------------------------------------------------------------------------
/examples/send_invite_email.py:
--------------------------------------------------------------------------------
```python
#!/usr/bin/env python
"""
Example script demonstrating how to send invitation emails using the OneSignal MCP server.
This showcases the functionality that replaces SendGrid's invitation system.
"""
import asyncio
import sys
import os
# Add the parent directory to the path so we can import the server module
sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), '..')))
# Import the server module
from onesignal_server import send_invite_email, send_bulk_invites
async def send_single_invite():
"""Send a single invitation email."""
print("Sending a single invitation email...")
result = await send_invite_email(
email="[email protected]",
first_name="John",
invite_url="https://yourapp.com/invite/abc123",
inviter_name="Jane Smith",
app_name="Your Amazing App",
expiry_days=7
)
if "error" in result:
print(f"Error: {result['error']}")
else:
print(f"Success! Email sent to [email protected]")
print(f"Details: {result}")
async def send_multiple_invites():
"""Send multiple invitation emails at once."""
print("\nSending multiple invitation emails...")
invites = [
{
"email": "[email protected]",
"first_name": "User",
"invite_url": "https://yourapp.com/invite/user1",
"inviter_name": "Team Admin"
},
{
"email": "[email protected]",
"first_name": "Another",
"invite_url": "https://yourapp.com/invite/user2",
"inviter_name": "Team Admin"
}
]
results = await send_bulk_invites(
invites=invites,
app_name="Your Amazing App",
expiry_days=7
)
print(f"Sent {len(results)} invitation emails")
for i, result in enumerate(results):
if "error" in result:
print(f"Error sending to {invites[i]['email']}: {result['error']}")
else:
print(f"Successfully sent to {invites[i]['email']}")
async def main():
"""Run both examples."""
await send_single_invite()
await send_multiple_invites()
if __name__ == "__main__":
asyncio.run(main())
```
--------------------------------------------------------------------------------
/missing_endpoints_analysis.md:
--------------------------------------------------------------------------------
```markdown
# OneSignal MCP Server - Missing Endpoints Analysis
## High Priority Missing Endpoints
### 1. Messaging Endpoints
- **Email-specific endpoint** (`/notifications` with email channel)
- Currently only generic push notifications are supported
- **SMS-specific endpoint** (`/notifications` with SMS channel)
- No dedicated SMS sending functionality
- **Transactional Messages** (`/notifications` with specific flags)
- Critical for automated/triggered messages
### 2. Live Activities (iOS)
- **Start Live Activity** (`/live_activities/{activity_id}/start`)
- **Update Live Activity** (`/live_activities/{activity_id}/update`)
- **End Live Activity** (`/live_activities/{activity_id}/end`)
### 3. Template Management
- **Update Template** (`PATCH /templates/{template_id}`)
- **Delete Template** (`DELETE /templates/{template_id}`)
- **Copy Template to Another App** (`POST /templates/{template_id}/copy`)
### 4. API Key Management
- **Delete API Key** (`DELETE /apps/{app_id}/auth/tokens/{token_id}`)
- **Update API Key** (`PATCH /apps/{app_id}/auth/tokens/{token_id}`)
- **Rotate API Key** (`POST /apps/{app_id}/auth/tokens/{token_id}/rotate`)
### 5. Analytics/Outcomes
- **View Outcomes** (`GET /apps/{app_id}/outcomes`)
- Essential for tracking conversion metrics
### 6. Export Functionality
- **Export Subscriptions CSV** (`POST /players/csv_export`)
- **Export Audience Activity CSV** (`POST /notifications/csv_export`)
### 7. Player/Device Management (Legacy but still used)
- **Add a Player** (`POST /players`)
- **Edit Player** (`PUT /players/{player_id}`)
- **Edit Tags with External User ID** (`PUT /users/{external_user_id}`)
- **Delete Player Record** (`DELETE /players/{player_id}`)
### 8. Additional User/Subscription Endpoints
- **View User Identity by Subscription** (`GET /apps/{app_id}/subscriptions/{subscription_id}/identity`)
- **Create Alias by Subscription** (`PATCH /apps/{app_id}/subscriptions/{subscription_id}/identity`)
## Medium Priority Missing Features
### 1. In-App Messages
- No endpoints for managing in-app messages
### 2. Webhooks Management
- No endpoints for configuring webhooks
### 3. Journey/Automation APIs
- No support for automated messaging journeys
## Implementation Priority
1. **Email and SMS endpoints** - Critical for multi-channel messaging
2. **Transactional Messages** - Essential for automated notifications
3. **Template management completeness** - Update and delete operations
4. **Export functionality** - Important for data management
5. **Analytics/Outcomes** - Necessary for measuring effectiveness
```
--------------------------------------------------------------------------------
/test_auth_fix.py:
--------------------------------------------------------------------------------
```python
#!/usr/bin/env python3
"""
Test script to verify the authentication fix for OneSignal MCP server
"""
import asyncio
import json
import logging
import os
from onesignal_server import (
view_segments,
view_templates,
get_current_app,
app_configs,
make_onesignal_request,
logger
)
# Enable DEBUG logging
logging.basicConfig(level=logging.DEBUG)
logger.setLevel(logging.DEBUG)
async def test_direct_api_call():
"""Test direct API call to debug the issue"""
print("\nTesting direct API call...")
current_app = get_current_app()
# Test direct segments call
endpoint = f"apps/{current_app.app_id}/segments"
print(f"Endpoint: {endpoint}")
print(f"App ID: {current_app.app_id}")
print(f"API Key length: {len(current_app.api_key)}")
result = await make_onesignal_request(endpoint, method="GET", use_org_key=False)
print(f"Direct API result: {json.dumps(result, indent=2)}")
async def test_endpoints():
"""Test the fixed endpoints"""
print("Testing OneSignal Authentication Fix")
print("=" * 50)
# Check current app configuration
current_app = get_current_app()
if not current_app:
print("❌ No app configured. Please ensure your .env file has:")
print(" ONESIGNAL_APP_ID=your_app_id_here")
print(" ONESIGNAL_API_KEY=your_rest_api_key_here")
return
print(f"✅ Current app: {current_app.name} (ID: {current_app.app_id})")
print(f" API Key: {'*' * 20}{current_app.api_key[-10:]}")
print()
# Test segments endpoint
print("Testing view_segments()...")
try:
segments_result = await view_segments()
if "Error retrieving segments:" in segments_result:
print(f"❌ Segments test failed: {segments_result}")
else:
print("✅ Segments endpoint is working!")
print(f" Result preview: {segments_result[:200]}...")
except Exception as e:
print(f"❌ Segments test error: {str(e)}")
print()
# Test templates endpoint
print("Testing view_templates()...")
try:
templates_result = await view_templates()
if "Error retrieving templates:" in templates_result:
print(f"❌ Templates test failed: {templates_result}")
else:
print("✅ Templates endpoint is working!")
print(f" Result preview: {templates_result[:200]}...")
except Exception as e:
print(f"❌ Templates test error: {str(e)}")
# Test direct API call for debugging
await test_direct_api_call()
print()
print("Test completed!")
if __name__ == "__main__":
asyncio.run(test_endpoints())
```
--------------------------------------------------------------------------------
/debug_api_key.py:
--------------------------------------------------------------------------------
```python
#!/usr/bin/env python3
"""
Debug script to check API key format and test authentication
"""
import os
import requests
from dotenv import load_dotenv
# Load environment variables
load_dotenv()
# Get the API credentials
app_id = os.getenv("ONESIGNAL_MANDIBLE_APP_ID", "")
api_key = os.getenv("ONESIGNAL_MANDIBLE_API_KEY", "")
org_key = os.getenv("ONESIGNAL_ORG_API_KEY", "")
print("API Key Debugging")
print("=" * 50)
print(f"App ID: {app_id}")
print(f"REST API Key length: {len(api_key)}")
print(f"REST API Key prefix: {api_key[:10] if api_key else 'None'}")
print(f"Org API Key length: {len(org_key)}")
print(f"Org API Key prefix: {org_key[:10] if org_key else 'None'}")
# Check for common issues
if api_key:
if api_key.startswith('"') or api_key.endswith('"'):
print("⚠️ WARNING: API key contains quotes")
if ' ' in api_key:
print("⚠️ WARNING: API key contains spaces")
if '\n' in api_key or '\r' in api_key:
print("⚠️ WARNING: API key contains newlines")
if api_key.startswith('Basic '):
print("⚠️ WARNING: API key already contains 'Basic ' prefix")
# Test segments endpoint with REST API key
print("\n1. Testing segments endpoint with REST API key...")
url = f"https://api.onesignal.com/apps/{app_id}/segments"
headers = {
"Authorization": f"Basic {api_key}",
"Content-Type": "application/json",
"Accept": "application/json"
}
print(f"URL: {url}")
try:
response = requests.get(url, headers=headers, timeout=10)
print(f"Response status: {response.status_code}")
print(f"Response: {response.text[:500] if response.text else 'Empty response'}")
if response.status_code == 200:
print("✅ Segments endpoint working with REST API key!")
else:
print("❌ Segments endpoint failed with REST API key")
except Exception as e:
print(f"❌ Request failed: {str(e)}")
# Test app details endpoint with Org API key
print("\n2. Testing app details endpoint with Org API key...")
url = f"https://api.onesignal.com/apps/{app_id}"
headers = {
"Authorization": f"Basic {org_key}",
"Content-Type": "application/json",
"Accept": "application/json"
}
print(f"URL: {url}")
try:
response = requests.get(url, headers=headers, timeout=10)
print(f"Response status: {response.status_code}")
if response.status_code == 200:
print("✅ App details endpoint working with Org API key!")
else:
print(f"❌ App details endpoint failed: {response.text[:200]}")
except Exception as e:
print(f"❌ Request failed: {str(e)}")
# Test templates endpoint
print("\n3. Testing templates endpoint with REST API key...")
url = f"https://api.onesignal.com/apps/{app_id}/templates"
headers = {
"Authorization": f"Basic {api_key}",
"Content-Type": "application/json",
"Accept": "application/json"
}
print(f"URL: {url}")
try:
response = requests.get(url, headers=headers, timeout=10)
print(f"Response status: {response.status_code}")
if response.status_code == 200:
print("✅ Templates endpoint working with REST API key!")
else:
print(f"❌ Templates endpoint failed: {response.text[:200]}")
except Exception as e:
print(f"❌ Request failed: {str(e)}")
```
--------------------------------------------------------------------------------
/onesignal_refactored/tools/live_activities.py:
--------------------------------------------------------------------------------
```python
"""Live Activities management tools for OneSignal MCP server."""
from typing import Dict, Any, Optional
from ..api_client import api_client
async def start_live_activity(
activity_id: str,
push_token: str,
subscription_id: str,
activity_attributes: Dict[str, Any],
content_state: Dict[str, Any],
**kwargs
) -> Dict[str, Any]:
"""
Start a new Live Activity for iOS.
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
**kwargs: Additional parameters
"""
data = {
"activity_id": activity_id,
"push_token": push_token,
"subscription_id": subscription_id,
"activity_attributes": activity_attributes,
"content_state": content_state
}
data.update(kwargs)
return await api_client.request(
f"live_activities/{activity_id}/start",
method="POST",
data=data
)
async def update_live_activity(
activity_id: str,
name: str,
event: str,
content_state: Dict[str, Any],
dismissal_date: Optional[int] = None,
priority: Optional[int] = None,
sound: Optional[str] = None,
**kwargs
) -> Dict[str, Any]:
"""
Update an existing 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
**kwargs: Additional parameters
"""
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
data.update(kwargs)
return await api_client.request(
f"live_activities/{activity_id}/update",
method="POST",
data=data
)
async def end_live_activity(
activity_id: str,
subscription_id: str,
dismissal_date: Optional[int] = None,
priority: Optional[int] = None,
**kwargs
) -> Dict[str, Any]:
"""
End a 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)
**kwargs: Additional parameters
"""
data = {
"subscription_id": subscription_id,
"event": "end"
}
if dismissal_date:
data["dismissal_date"] = dismissal_date
if priority:
data["priority"] = priority
data.update(kwargs)
return await api_client.request(
f"live_activities/{activity_id}/end",
method="POST",
data=data
)
async def get_live_activity_status(
activity_id: str,
subscription_id: str
) -> Dict[str, Any]:
"""
Get the status of a Live Activity.
Args:
activity_id: ID of the activity
subscription_id: Subscription ID associated with the activity
"""
params = {"subscription_id": subscription_id}
return await api_client.request(
f"live_activities/{activity_id}/status",
method="GET",
params=params
)
```
--------------------------------------------------------------------------------
/onesignal_refactored/tools/analytics.py:
--------------------------------------------------------------------------------
```python
"""Analytics and outcomes tools for OneSignal MCP server."""
from typing import Dict, Any, Optional, List
from ..api_client import api_client
from ..config import app_manager
async def view_outcomes(
outcome_names: List[str],
outcome_time_range: Optional[str] = None,
outcome_platforms: Optional[List[str]] = None,
outcome_attribution: Optional[str] = None,
**kwargs
) -> 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")
**kwargs: Additional parameters
"""
app_config = app_manager.get_current_app()
if not app_config:
raise ValueError("No app currently selected")
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
params.update(kwargs)
return await api_client.request(
f"apps/{app_config.app_id}/outcomes",
method="GET",
params=params
)
async def export_players_csv(
start_date: Optional[str] = None,
end_date: Optional[str] = None,
segment_names: Optional[List[str]] = None,
**kwargs
) -> Dict[str, Any]:
"""
Export player/subscription data to CSV.
Args:
start_date: Start date for export (ISO 8601 format)
end_date: End date for export (ISO 8601 format)
segment_names: List of segment names to export
**kwargs: Additional export parameters
"""
data = {}
if start_date:
data["start_date"] = start_date
if end_date:
data["end_date"] = end_date
if segment_names:
data["segment_names"] = segment_names
data.update(kwargs)
return await api_client.request(
"players/csv_export",
method="POST",
data=data,
use_org_key=True
)
async def export_audience_activity_csv(
start_date: Optional[str] = None,
end_date: Optional[str] = None,
event_types: Optional[List[str]] = None,
**kwargs
) -> Dict[str, Any]:
"""
Export audience activity events 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
**kwargs: Additional export parameters
"""
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
data.update(kwargs)
return await api_client.request(
"notifications/csv_export",
method="POST",
data=data,
use_org_key=True
)
def format_outcomes_response(outcomes: Dict[str, Any]) -> str:
"""Format outcomes response for display."""
if not outcomes or "outcomes" not in outcomes:
return "No outcomes data available."
output = "Outcomes Report:\n\n"
for outcome in outcomes.get("outcomes", []):
output += f"Outcome: {outcome.get('id')}\n"
output += f"Total Count: {outcome.get('aggregation', {}).get('count', 0)}\n"
output += f"Total Value: {outcome.get('aggregation', {}).get('sum', 0)}\n"
# Platform breakdown
platforms = outcome.get('platforms', {})
if platforms:
output += "Platform Breakdown:\n"
for platform, data in platforms.items():
output += f" {platform}: Count={data.get('count', 0)}, Value={data.get('sum', 0)}\n"
output += "\n"
return output
```
--------------------------------------------------------------------------------
/onesignal_tools_list.md:
--------------------------------------------------------------------------------
```markdown
# Complete OneSignal MCP Tools List
The OneSignal MCP server now includes **57 tools** covering all major OneSignal API functionality:
## App Management (5 tools)
1. **list_apps** - List all configured OneSignal apps
2. **add_app** - Add a new OneSignal app configuration locally
3. **update_local_app_config** - Update an existing local app configuration
4. **remove_app** - Remove a local OneSignal app configuration
5. **switch_app** - Switch the current app to use for API requests
## Messaging (8 tools)
6. **send_push_notification** - Send a push notification
7. **send_email** *(NEW)* - Send an email through OneSignal
8. **send_sms** *(NEW)* - Send an SMS/MMS through OneSignal
9. **send_transactional_message** *(NEW)* - Send immediate delivery messages
10. **view_messages** - View recent messages sent
11. **view_message_details** - Get detailed information about a message
12. **view_message_history** - View message history/recipients
13. **cancel_message** - Cancel a scheduled message
## Devices/Players (6 tools)
14. **view_devices** - View devices subscribed to your app
15. **view_device_details** - Get detailed information about a device
16. **add_player** *(NEW)* - Add a new player/device
17. **edit_player** *(NEW)* - Edit an existing player/device
18. **delete_player** *(NEW)* - Delete a player/device record
19. **edit_tags_with_external_user_id** *(NEW)* - Bulk edit tags by external ID
## Segments (3 tools)
20. **view_segments** - List all segments
21. **create_segment** - Create a new segment
22. **delete_segment** - Delete a segment
## Templates (6 tools)
23. **view_templates** - List all templates
24. **view_template_details** - Get template details
25. **create_template** - Create a new template
26. **update_template** *(NEW)* - Update an existing template
27. **delete_template** *(NEW)* - Delete a template
28. **copy_template_to_app** *(NEW)* - Copy template to another app
## Apps (6 tools)
29. **view_app_details** - Get details about configured app
30. **view_apps** - List all organization apps
31. **create_app** - Create a new OneSignal application
32. **update_app** - Update an existing application
33. **view_app_api_keys** - View API keys for an app
34. **create_app_api_key** - Create a new API key
## API Key Management (3 tools) *(NEW)*
35. **delete_app_api_key** - Delete an API key
36. **update_app_api_key** - Update an API key
37. **rotate_app_api_key** - Rotate an API key
## Users (6 tools)
38. **create_user** - Create a new user
39. **view_user** - View user details
40. **update_user** - Update user information
41. **delete_user** - Delete a user
42. **view_user_identity** - Get user identity information
43. **view_user_identity_by_subscription** *(NEW)* - Get identity by subscription
## Aliases (3 tools)
44. **create_or_update_alias** - Create or update user alias
45. **delete_alias** - Delete a user alias
46. **create_alias_by_subscription** *(NEW)* - Create alias by subscription ID
## Subscriptions (5 tools)
47. **create_subscription** - Create a new subscription
48. **update_subscription** - Update a subscription
49. **delete_subscription** - Delete a subscription
50. **transfer_subscription** - Transfer subscription between users
51. **unsubscribe_email** - Unsubscribe using email token
## Live Activities (3 tools) *(NEW)*
52. **start_live_activity** - Start iOS Live Activity
53. **update_live_activity** - Update iOS Live Activity
54. **end_live_activity** - End iOS Live Activity
## Analytics & Export (3 tools) *(NEW)*
55. **view_outcomes** - View outcomes/conversion data
56. **export_players_csv** - Export player data to CSV
57. **export_messages_csv** - Export messages to CSV
## Summary by Category
- **App Management**: 5 tools
- **Messaging**: 8 tools (3 new)
- **Devices/Players**: 6 tools (4 new)
- **Segments**: 3 tools
- **Templates**: 6 tools (3 new)
- **Apps**: 6 tools
- **API Keys**: 3 tools (all new)
- **Users**: 6 tools (1 new)
- **Aliases**: 3 tools (1 new)
- **Subscriptions**: 5 tools
- **Live Activities**: 3 tools (all new)
- **Analytics**: 3 tools (all new)
**Total**: 57 tools (21 newly added)
```
--------------------------------------------------------------------------------
/onesignal_refactored/tools/templates.py:
--------------------------------------------------------------------------------
```python
"""Template management tools for OneSignal MCP server."""
from typing import Dict, Any, Optional
from ..api_client import api_client
from ..config import app_manager
async def create_template(
name: str,
title: str,
message: str,
**kwargs
) -> Dict[str, Any]:
"""
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
**kwargs: Additional template parameters
"""
app_config = app_manager.get_current_app()
if not app_config:
raise ValueError("No app currently selected")
data = {
"app_id": app_config.app_id,
"name": name,
"headings": {"en": title},
"contents": {"en": message}
}
data.update(kwargs)
return await api_client.request("templates", method="POST", data=data)
async def update_template(
template_id: str,
name: Optional[str] = None,
title: Optional[str] = None,
message: Optional[str] = None,
**kwargs
) -> 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
**kwargs: Additional template parameters
"""
data = {}
if name:
data["name"] = name
if title:
data["headings"] = {"en": title}
if message:
data["contents"] = {"en": message}
data.update(kwargs)
if not data:
raise ValueError("No update parameters provided")
return await api_client.request(
f"templates/{template_id}",
method="PATCH",
data=data
)
async def view_templates() -> Dict[str, Any]:
"""List all templates available in your OneSignal app."""
return await api_client.request("templates", method="GET")
async def view_template_details(template_id: str) -> Dict[str, Any]:
"""
Get detailed information about a specific template.
Args:
template_id: The ID of the template to retrieve
"""
app_config = app_manager.get_current_app()
if not app_config:
raise ValueError("No app currently selected")
params = {"app_id": app_config.app_id}
return await api_client.request(
f"templates/{template_id}",
method="GET",
params=params
)
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
"""
return await api_client.request(
f"templates/{template_id}",
method="DELETE"
)
async def copy_template_to_app(
template_id: str,
target_app_id: str,
new_name: Optional[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
return await api_client.request(
f"templates/{template_id}/copy",
method="POST",
data=data
)
def format_template_list(templates_response: Dict[str, Any]) -> str:
"""Format template list response for display."""
templates = templates_response.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
def format_template_details(template: Dict[str, Any]) -> str:
"""Format template details for display."""
heading = template.get("headings", {}).get("en", "No heading")
content = template.get("contents", {}).get("en", "No content")
details = [
f"ID: {template.get('id')}",
f"Name: {template.get('name')}",
f"Title: {heading}",
f"Message: {content}",
f"Platform: {template.get('platform')}",
f"Created: {template.get('created_at')}"
]
return "\n".join(details)
```
--------------------------------------------------------------------------------
/test_api_key_validity.py:
--------------------------------------------------------------------------------
```python
#!/usr/bin/env python3
"""Test script to verify OneSignal API key validity"""
import os
import requests
from dotenv import load_dotenv
# Load environment variables
load_dotenv()
# Get credentials
app_id = os.getenv("ONESIGNAL_MANDIBLE_APP_ID")
api_key = os.getenv("ONESIGNAL_MANDIBLE_API_KEY")
print("OneSignal API Key Validation Test")
print("=" * 50)
if not app_id or not api_key:
print("❌ ERROR: Missing credentials!")
print(" Make sure your .env file contains:")
print(" ONESIGNAL_MANDIBLE_APP_ID=your_app_id")
print(" ONESIGNAL_MANDIBLE_API_KEY=your_api_key")
exit(1)
print(f"App ID: {app_id}")
print(f"API Key length: {len(api_key)}")
print(f"API Key prefix: {api_key[:15]}...")
# Determine API key type
is_v2_key = api_key.startswith("os_v2_")
print(f"API Key type: {'v2' if is_v2_key else 'v1'}")
# Set up headers based on key type
headers = {
"Accept": "application/json",
"Content-Type": "application/json"
}
if is_v2_key:
headers["Authorization"] = f"Key {api_key}"
print("Using Authorization: Key <api_key>")
else:
headers["Authorization"] = f"Basic {api_key}"
print("Using Authorization: Basic <api_key>")
print("\n" + "=" * 50)
print("Testing API Key validity...")
print("=" * 50)
# Test 1: Get app details (requires valid API key)
print("\n1. Testing app details endpoint (requires valid app-specific API key)...")
url = f"https://api.onesignal.com/api/v1/apps/{app_id}"
params = {} # No params needed for this endpoint
try:
response = requests.get(url, headers=headers, params=params)
print(f" URL: {url}")
print(f" Status: {response.status_code}")
if response.status_code == 200:
print(" ✅ SUCCESS: API key is valid!")
data = response.json()
print(f" App Name: {data.get('name', 'N/A')}")
print(f" Created: {data.get('created_at', 'N/A')}")
elif response.status_code == 401:
print(" ❌ FAILED: Authentication error - API key is invalid or doesn't have permission")
print(f" Response: {response.text}")
elif response.status_code == 403:
print(" ❌ FAILED: Forbidden - API key doesn't have permission for this app")
print(f" Response: {response.text}")
else:
print(f" ❌ FAILED: Unexpected status code")
print(f" Response: {response.text}")
except Exception as e:
print(f" ❌ ERROR: {e}")
# Test 2: List notifications (requires valid API key with proper permissions)
print("\n2. Testing notifications endpoint...")
url = "https://api.onesignal.com/api/v1/notifications"
params = {"app_id": app_id, "limit": 1}
try:
response = requests.get(url, headers=headers, params=params)
print(f" URL: {url}")
print(f" Params: {params}")
print(f" Status: {response.status_code}")
if response.status_code == 200:
print(" ✅ SUCCESS: Can list notifications")
data = response.json()
print(f" Total notifications: {data.get('total_count', 0)}")
else:
print(f" ❌ FAILED: Status {response.status_code}")
print(f" Response: {response.text[:200]}...")
except Exception as e:
print(f" ❌ ERROR: {e}")
# Test 3: Check segments endpoint (the one causing issues)
print("\n3. Testing segments endpoint (the problematic one)...")
url = f"https://api.onesignal.com/api/v1/apps/{app_id}/segments"
try:
response = requests.get(url, headers=headers)
print(f" URL: {url}")
print(f" Status: {response.status_code}")
if response.status_code == 200:
print(" ✅ SUCCESS: Can access segments")
data = response.json()
print(f" Segments found: {len(data) if isinstance(data, list) else 'N/A'}")
else:
print(f" ❌ FAILED: Status {response.status_code}")
print(f" Response: {response.text}")
except Exception as e:
print(f" ❌ ERROR: {e}")
print("\n" + "=" * 50)
print("Summary:")
print("=" * 50)
if api_key.startswith("os_v2_"):
print("You're using a v2 API key (starts with 'os_v2_')")
print("Make sure this key has the necessary permissions in OneSignal dashboard:")
print("- View App Details")
print("- View Notifications")
print("- View Segments")
print("\nTo check/update permissions:")
print("1. Go to OneSignal Dashboard")
print("2. Navigate to Settings > Keys & IDs")
print("3. Find your REST API Key")
print("4. Check the permissions assigned to it")
else:
print("You're using a v1 API key")
print("Consider upgrading to a v2 API key for better security and permissions control")
```
--------------------------------------------------------------------------------
/onesignal_refactored/config.py:
--------------------------------------------------------------------------------
```python
"""Configuration management for OneSignal MCP server."""
import os
import logging
from typing import Dict, Optional
from dataclasses import dataclass
from dotenv import load_dotenv
# Load environment variables
load_dotenv()
# Configure logger
logger = logging.getLogger("onesignal-mcp.config")
# API Configuration
ONESIGNAL_API_URL = "https://api.onesignal.com/api/v1"
ONESIGNAL_ORG_API_KEY = os.getenv("ONESIGNAL_ORG_API_KEY", "")
@dataclass
class AppConfig:
"""Configuration for a OneSignal application."""
app_id: str
api_key: str
name: str
def __str__(self):
return f"{self.name} ({self.app_id})"
class AppManager:
"""Manages OneSignal app configurations."""
def __init__(self):
self.app_configs: Dict[str, AppConfig] = {}
self.current_app_key: Optional[str] = None
self._load_from_environment()
def _load_from_environment(self):
"""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:
self.add_app("mandible", mandible_app_id, mandible_api_key, "Mandible")
self.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:
self.add_app("weirdbrains", weirdbrains_app_id, weirdbrains_api_key, "Weird Brains")
if not self.current_app_key:
self.current_app_key = "weirdbrains"
logger.info(f"Weird Brains app configured with ID: {weirdbrains_app_id}")
# Fallback for default app configuration
if not self.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:
self.add_app("default", default_app_id, default_api_key, "Default App")
self.current_app_key = "default"
logger.info(f"Default app configured with ID: {default_app_id}")
else:
logger.warning("No app configurations found. Use add_app to add an app configuration.")
def add_app(self, key: str, app_id: str, api_key: str, name: Optional[str] = None) -> None:
"""Add a new app configuration."""
self.app_configs[key] = AppConfig(app_id, api_key, name or key)
logger.info(f"Added app configuration '{key}' with ID: {app_id}")
def update_app(self, key: str, app_id: Optional[str] = None,
api_key: Optional[str] = None, name: Optional[str] = None) -> bool:
"""Update an existing app configuration."""
if key not in self.app_configs:
return False
app = self.app_configs[key]
if app_id:
app.app_id = app_id
if api_key:
app.api_key = api_key
if name:
app.name = name
logger.info(f"Updated app configuration '{key}'")
return True
def remove_app(self, key: str) -> bool:
"""Remove an app configuration."""
if key not in self.app_configs:
return False
if self.current_app_key == key:
# Switch to another app if available
other_keys = [k for k in self.app_configs.keys() if k != key]
self.current_app_key = other_keys[0] if other_keys else None
del self.app_configs[key]
logger.info(f"Removed app configuration '{key}'")
return True
def set_current_app(self, key: str) -> bool:
"""Set the current app to use for API requests."""
if key in self.app_configs:
self.current_app_key = key
logger.info(f"Switched to app '{key}'")
return True
return False
def get_current_app(self) -> Optional[AppConfig]:
"""Get the current app configuration."""
if self.current_app_key and self.current_app_key in self.app_configs:
return self.app_configs[self.current_app_key]
return None
def get_app(self, key: str) -> Optional[AppConfig]:
"""Get a specific app configuration."""
return self.app_configs.get(key)
def list_apps(self) -> Dict[str, AppConfig]:
"""Get all app configurations."""
return self.app_configs.copy()
# Global app manager instance
app_manager = AppManager()
def requires_org_api_key(endpoint: str) -> bool:
"""Determine if an endpoint requires the Organization API Key."""
org_level_endpoints = [
"apps", # Managing apps
"players/csv_export", # Export users
"notifications/csv_export" # Export notifications
]
return any(endpoint == ep or endpoint.startswith(f"{ep}/") for ep in org_level_endpoints)
```
--------------------------------------------------------------------------------
/onesignal_refactoring_summary.md:
--------------------------------------------------------------------------------
```markdown
# OneSignal MCP Server Refactoring Summary
## Overview
This document summarizes the analysis of the OneSignal MCP server implementation against the official OneSignal REST API documentation and provides a comprehensive refactoring plan.
## 1. Missing API Endpoints Analysis
### High Priority Missing Endpoints
#### Messaging
- **Email-specific endpoint** - Send emails with HTML content and templates
- **SMS-specific endpoint** - Send SMS/MMS messages
- **Transactional Messages** - Immediate delivery messages without scheduling
#### Live Activities (iOS)
- **Start Live Activity** - Initialize iOS Live Activities
- **Update Live Activity** - Update running Live Activities
- **End Live Activity** - Terminate Live Activities
#### Template Management
- **Update Template** - Modify existing templates
- **Delete Template** - Remove templates
- **Copy Template** - Duplicate templates across apps
#### API Key Management
- **Delete API Key** - Remove API keys
- **Update API Key** - Modify API key permissions
- **Rotate API Key** - Generate new key while maintaining permissions
#### Analytics & Export
- **View Outcomes** - Track conversion metrics
- **Export Subscriptions CSV** - Export user data
- **Export Audience Activity CSV** - Export event data
#### Player/Device Management (Legacy)
- **Add Player** - Register new devices
- **Edit Player** - Update device information
- **Edit Tags with External User ID** - Bulk tag updates
- **Delete Player Record** - Remove device records
## 2. Refactored Architecture
### New Module Structure
```
onesignal_refactored/
├── __init__.py
├── config.py # App configuration management
├── api_client.py # API request handling
├── tools/
│ ├── __init__.py
│ ├── messages.py # All messaging endpoints
│ ├── templates.py # Template management
│ ├── live_activities.py # iOS Live Activities
│ ├── analytics.py # Outcomes and exports
│ ├── users.py # User management
│ ├── devices.py # Device/player management
│ ├── segments.py # Segment management
│ ├── apps.py # App management
│ └── subscriptions.py # Subscription management
└── server.py # Main MCP server entry point
```
### Key Improvements
#### 1. Centralized Configuration (`config.py`)
- `AppConfig` dataclass for app configurations
- `AppManager` class for managing multiple apps
- Environment variable loading
- Automatic organization API key detection
#### 2. Unified API Client (`api_client.py`)
- `OneSignalAPIClient` class with centralized request handling
- Automatic authentication method selection
- Consistent error handling with custom exceptions
- Request/response logging
#### 3. Modular Tool Organization
Each module focuses on a specific API domain:
- Better code organization
- Easier maintenance
- Clear separation of concerns
- Reduced code duplication
### 3. New Features Implementation
#### Email Sending
```python
async def send_email(
subject: str,
body: str,
email_body: Optional[str] = None,
segments: Optional[List[str]] = None,
include_emails: Optional[List[str]] = None,
external_ids: Optional[List[str]] = None,
template_id: Optional[str] = None
) -> Dict[str, Any]
```
#### SMS Sending
```python
async def send_sms(
message: str,
phone_numbers: Optional[List[str]] = None,
segments: Optional[List[str]] = None,
external_ids: Optional[List[str]] = None,
media_url: Optional[str] = None
) -> Dict[str, Any]
```
#### Transactional Messages
```python
async def send_transactional_message(
channel: str,
content: Dict[str, str],
recipients: Dict[str, Any],
template_id: Optional[str] = None,
custom_data: Optional[Dict[str, Any]] = None
) -> Dict[str, Any]
```
#### Live Activities
```python
async def start_live_activity(...)
async def update_live_activity(...)
async def end_live_activity(...)
```
#### Analytics & Outcomes
```python
async def view_outcomes(...)
async def export_players_csv(...)
async def export_audience_activity_csv(...)
```
## 4. Migration Guide
### Step 1: Create New Directory Structure
```bash
mkdir -p onesignal_refactored/tools
```
### Step 2: Implement Core Modules
1. Copy `config.py` for app configuration
2. Copy `api_client.py` for API requests
3. Implement tool modules based on provided templates
### Step 3: Update MCP Server Registration
```python
# In server.py
from mcp.server.fastmcp import FastMCP
from .tools import messages, templates, live_activities, analytics
mcp = FastMCP("onesignal-server")
# Register all tools
@mcp.tool()
async def send_email(...):
return await messages.send_email(...)
# Continue for all tools...
```
### Step 4: Test Implementation
1. Test each new endpoint individually
2. Verify error handling
3. Check authentication switching
4. Validate response formatting
## 5. Benefits of Refactoring
1. **Better Maintainability** - Modular structure makes updates easier
2. **Reduced Duplication** - Shared API client eliminates repeated code
3. **Enhanced Error Handling** - Consistent error messages and logging
4. **Feature Completeness** - Support for all major OneSignal features
5. **Improved Testing** - Easier to unit test individual modules
6. **Better Documentation** - Clear module boundaries and responsibilities
## 6. Future Enhancements
1. **Async/Await Optimization** - Better concurrency handling
2. **Response Caching** - Cache frequently accessed data
3. **Batch Operations** - Support bulk operations where applicable
4. **Webhook Support** - Add webhook configuration endpoints
5. **In-App Messaging** - Support for in-app message management
6. **Rate Limiting** - Implement client-side rate limiting
7. **Retry Logic** - Automatic retry for failed requests
## Implementation Priority
1. **Phase 1** - Core refactoring (config, api_client)
2. **Phase 2** - Missing messaging endpoints (email, SMS, transactional)
3. **Phase 3** - Template and Live Activity completion
4. **Phase 4** - Analytics and export functionality
5. **Phase 5** - Legacy player/device endpoints
This refactoring will transform the OneSignal MCP server into a comprehensive, maintainable solution that supports all major OneSignal API features.
```
--------------------------------------------------------------------------------
/onesignal_refactored/api_client.py:
--------------------------------------------------------------------------------
```python
"""API client for OneSignal REST API requests."""
import logging
import requests
from typing import Dict, Any, Optional
from .config import (
ONESIGNAL_API_URL,
ONESIGNAL_ORG_API_KEY,
app_manager,
requires_org_api_key
)
logger = logging.getLogger("onesignal-mcp.api_client")
class OneSignalAPIError(Exception):
"""Custom exception for OneSignal API errors."""
pass
class OneSignalAPIClient:
"""Client for making requests to the OneSignal API."""
def __init__(self):
self.api_url = ONESIGNAL_API_URL
self.timeout = 30
async def request(
self,
endpoint: str,
method: str = "GET",
data: Optional[Dict[str, Any]] = None,
params: Optional[Dict[str, Any]] = None,
use_org_key: Optional[bool] = None,
app_key: Optional[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, PATCH)
data: Request body for POST/PUT/PATCH requests
params: Query parameters for GET requests
use_org_key: Whether to use the organization API key
app_key: The key of the app configuration to use
Returns:
API response as dictionary
Raises:
OneSignalAPIError: If the API request fails
"""
headers = {
"Content-Type": "application/json",
"Accept": "application/json",
}
# Determine authentication method
if use_org_key is None:
use_org_key = requires_org_api_key(endpoint)
# Set authentication header
if use_org_key:
if not ONESIGNAL_ORG_API_KEY:
raise OneSignalAPIError(
"Organization API Key not configured. "
"Set the ONESIGNAL_ORG_API_KEY environment variable."
)
headers["Authorization"] = f"Basic {ONESIGNAL_ORG_API_KEY}"
else:
# Get app configuration
app_config = None
if app_key:
app_config = app_manager.get_app(app_key)
else:
app_config = app_manager.get_current_app()
if not app_config:
raise OneSignalAPIError(
"No app configuration available. "
"Use set_current_app or specify app_key."
)
headers["Authorization"] = f"Basic {app_config.api_key}"
# Add app_id to params/data if needed
if params is None:
params = {}
if "app_id" not in params and not endpoint.startswith("apps/"):
params["app_id"] = app_config.app_id
if data is not None and method in ["POST", "PUT", "PATCH"]:
if "app_id" not in data and not endpoint.startswith("apps/"):
data["app_id"] = app_config.app_id
url = f"{self.api_url}/{endpoint}"
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'}")
response = self._make_request(method, url, headers, params, data)
response.raise_for_status()
return response.json() if response.text else {}
except requests.exceptions.HTTPError as e:
error_message = self._extract_error_message(e)
logger.error(f"API request failed: {error_message}")
raise OneSignalAPIError(error_message) from e
except requests.exceptions.RequestException as e:
error_message = f"Request failed: {str(e)}"
logger.error(error_message)
raise OneSignalAPIError(error_message) from e
except Exception as e:
error_message = f"Unexpected error: {str(e)}"
logger.exception(error_message)
raise OneSignalAPIError(error_message) from e
def _make_request(
self,
method: str,
url: str,
headers: Dict[str, str],
params: Optional[Dict[str, Any]],
data: Optional[Dict[str, Any]]
) -> requests.Response:
"""Make the actual HTTP request."""
method = method.upper()
if method == "GET":
return requests.get(url, headers=headers, params=params, timeout=self.timeout)
elif method == "POST":
return requests.post(url, headers=headers, json=data, timeout=self.timeout)
elif method == "PUT":
return requests.put(url, headers=headers, json=data, timeout=self.timeout)
elif method == "DELETE":
return requests.delete(url, headers=headers, timeout=self.timeout)
elif method == "PATCH":
return requests.patch(url, headers=headers, json=data, timeout=self.timeout)
else:
raise ValueError(f"Unsupported HTTP method: {method}")
def _extract_error_message(self, error: requests.exceptions.HTTPError) -> str:
"""Extract a meaningful error message from the HTTP error."""
try:
if hasattr(error, 'response') and error.response is not None:
error_data = error.response.json()
if isinstance(error_data, dict):
# Try different error message formats
if 'errors' in error_data:
errors = error_data['errors']
if isinstance(errors, list) and errors:
return f"Error: {errors[0]}"
elif isinstance(errors, str):
return f"Error: {errors}"
elif 'error' in error_data:
return f"Error: {error_data['error']}"
elif 'message' in error_data:
return f"Error: {error_data['message']}"
return f"Error: {error.response.reason} (Status: {error.response.status_code})"
except Exception:
pass
return f"Error: {str(error)}"
# Global API client instance
api_client = OneSignalAPIClient()
```
--------------------------------------------------------------------------------
/tests/test_onesignal_server.py:
--------------------------------------------------------------------------------
```python
import unittest
from unittest.mock import patch, MagicMock
import os
import sys
import json
import asyncio
# Add the parent directory to sys.path to import the server module
sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), '..')))
# Import the server module
import onesignal_server
class TestOneSignalServer(unittest.TestCase):
"""Test cases for the OneSignal MCP server."""
def setUp(self):
"""Set up test environment before each test."""
# Mock environment variables
self.env_patcher = patch.dict('os.environ', {
'ONESIGNAL_APP_ID': 'test-app-id',
'ONESIGNAL_API_KEY': 'test-api-key',
'ONESIGNAL_ORG_API_KEY': 'test-org-api-key'
})
self.env_patcher.start()
# Reset app configurations for each test
onesignal_server.app_configs = {}
onesignal_server.current_app_key = None
# Initialize with test app
onesignal_server.add_app_config('test', 'test-app-id', 'test-api-key', 'Test App')
onesignal_server.current_app_key = 'test'
def tearDown(self):
"""Clean up after each test."""
self.env_patcher.stop()
def test_app_config(self):
"""Test AppConfig class."""
app = onesignal_server.AppConfig('app-id', 'api-key', 'App Name')
self.assertEqual(app.app_id, 'app-id')
self.assertEqual(app.api_key, 'api-key')
self.assertEqual(app.name, 'App Name')
self.assertEqual(str(app), 'App Name (app-id)')
def test_add_app_config(self):
"""Test adding app configurations."""
onesignal_server.add_app_config('new-app', 'new-app-id', 'new-api-key', 'New App')
self.assertIn('new-app', onesignal_server.app_configs)
self.assertEqual(onesignal_server.app_configs['new-app'].app_id, 'new-app-id')
self.assertEqual(onesignal_server.app_configs['new-app'].api_key, 'new-api-key')
self.assertEqual(onesignal_server.app_configs['new-app'].name, 'New App')
def test_set_current_app(self):
"""Test setting the current app."""
# Add a second app
onesignal_server.add_app_config('second', 'second-app-id', 'second-api-key')
# Test switching to an existing app
result = onesignal_server.set_current_app('second')
self.assertTrue(result)
self.assertEqual(onesignal_server.current_app_key, 'second')
# Test switching to a non-existent app
result = onesignal_server.set_current_app('non-existent')
self.assertFalse(result)
self.assertEqual(onesignal_server.current_app_key, 'second') # Should not change
def test_get_current_app(self):
"""Test getting the current app configuration."""
current_app = onesignal_server.get_current_app()
self.assertIsNotNone(current_app)
self.assertEqual(current_app.app_id, 'test-app-id')
self.assertEqual(current_app.api_key, 'test-api-key')
# Test with no current app
onesignal_server.current_app_key = None
current_app = onesignal_server.get_current_app()
self.assertIsNone(current_app)
@patch('requests.get')
def test_make_onesignal_request_get(self, mock_get):
"""Test making a GET request to the OneSignal API."""
# Mock the response
mock_response = MagicMock()
mock_response.json.return_value = {'success': True}
mock_response.text = json.dumps({'success': True})
mock_get.return_value = mock_response
# Make the request and run it through the event loop
loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)
try:
result = loop.run_until_complete(
onesignal_server.make_onesignal_request('notifications', 'GET', params={'limit': 10})
)
# Check that the request was made correctly
mock_get.assert_called_once()
args, kwargs = mock_get.call_args
self.assertEqual(kwargs['headers']['Authorization'], 'Key test-api-key')
self.assertEqual(kwargs['params']['app_id'], 'test-app-id')
self.assertEqual(kwargs['params']['limit'], 10)
# Check the result
self.assertEqual(result, {'success': True})
finally:
loop.close()
@patch('requests.post')
def test_make_onesignal_request_post(self, mock_post):
"""Test making a POST request to the OneSignal API."""
# Mock the response
mock_response = MagicMock()
mock_response.json.return_value = {'id': 'notification-id'}
mock_response.text = json.dumps({'id': 'notification-id'})
mock_post.return_value = mock_response
# Make the request
data = {'contents': {'en': 'Test message'}}
loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)
try:
result = loop.run_until_complete(
onesignal_server.make_onesignal_request('notifications', 'POST', data=data)
)
# Check that the request was made correctly
mock_post.assert_called_once()
args, kwargs = mock_post.call_args
self.assertEqual(kwargs['headers']['Authorization'], 'Key test-api-key')
self.assertEqual(kwargs['json']['app_id'], 'test-app-id')
self.assertEqual(kwargs['json']['contents']['en'], 'Test message')
# Check the result
self.assertEqual(result, {'id': 'notification-id'})
finally:
loop.close()
@patch('requests.get')
@patch('onesignal_server.ONESIGNAL_ORG_API_KEY', 'test-org-api-key')
def test_make_onesignal_request_with_org_key(self, mock_get):
"""Test making a request with the organization API key."""
# Mock the response
mock_response = MagicMock()
mock_response.json.return_value = {'apps': []}
mock_response.text = json.dumps({'apps': []})
mock_get.return_value = mock_response
# Make the request
loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)
try:
result = loop.run_until_complete(
onesignal_server.make_onesignal_request('apps', 'GET', use_org_key=True)
)
# Check that the request was made correctly
mock_get.assert_called_once()
args, kwargs = mock_get.call_args
self.assertEqual(kwargs['headers']['Authorization'], 'Key test-org-api-key')
# Check the result
self.assertEqual(result, {'apps': []})
finally:
loop.close()
@patch('requests.get')
def test_make_onesignal_request_error_handling(self, mock_get):
"""Test error handling in make_onesignal_request."""
# Mock a request exception
mock_get.side_effect = Exception('Test error')
# Make the request
loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)
try:
result = loop.run_until_complete(
onesignal_server.make_onesignal_request('notifications')
)
# Check the result
self.assertIn('error', result)
self.assertEqual(result['error'], 'Unexpected error: Test error')
finally:
loop.close()
if __name__ == '__main__':
unittest.main()
```
--------------------------------------------------------------------------------
/onesignal_refactored/tools/messages.py:
--------------------------------------------------------------------------------
```python
"""Message management tools for OneSignal MCP server."""
import json
from typing import List, Dict, Any, Optional
from ..api_client import api_client, OneSignalAPIError
from ..config import app_manager
async def send_push_notification(
title: str,
message: str,
segments: Optional[List[str]] = None,
include_player_ids: Optional[List[str]] = None,
external_ids: Optional[List[str]] = None,
data: Optional[Dict[str, Any]] = None,
**kwargs
) -> Dict[str, Any]:
"""
Send a push notification through OneSignal.
Args:
title: Notification title
message: Notification message content
segments: List of segments to include
include_player_ids: List of specific player IDs to target
external_ids: List of external user IDs to target
data: Additional data to include with the notification
**kwargs: Additional notification parameters
"""
notification_data = {
"contents": {"en": message},
"headings": {"en": title},
"target_channel": "push"
}
# Set targeting
if not any([segments, include_player_ids, external_ids]):
segments = ["Subscribed Users"]
if segments:
notification_data["included_segments"] = segments
if include_player_ids:
notification_data["include_player_ids"] = include_player_ids
if external_ids:
notification_data["include_external_user_ids"] = external_ids
if data:
notification_data["data"] = data
# Add any additional parameters
notification_data.update(kwargs)
return await api_client.request("notifications", method="POST", data=notification_data)
async def send_email(
subject: str,
body: str,
email_body: Optional[str] = None,
segments: Optional[List[str]] = None,
include_emails: Optional[List[str]] = None,
external_ids: Optional[List[str]] = None,
template_id: Optional[str] = None,
**kwargs
) -> Dict[str, Any]:
"""
Send an email through OneSignal.
Args:
subject: Email subject line
body: Plain text email content
email_body: HTML email content (optional)
segments: List of segments to include
include_emails: List of specific email addresses to target
external_ids: List of external user IDs to target
template_id: Email template ID to use
**kwargs: Additional email parameters
"""
email_data = {
"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
# Add any additional parameters
email_data.update(kwargs)
return await api_client.request("notifications", method="POST", data=email_data)
async def send_sms(
message: str,
phone_numbers: Optional[List[str]] = None,
segments: Optional[List[str]] = None,
external_ids: Optional[List[str]] = None,
media_url: Optional[str] = None,
**kwargs
) -> Dict[str, Any]:
"""
Send an SMS 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
**kwargs: Additional SMS parameters
"""
sms_data = {
"contents": {"en": message},
"target_channel": "sms"
}
# Set targeting
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:
raise OneSignalAPIError(
"SMS requires phone_numbers, external_ids, or segments to be specified"
)
if media_url:
sms_data["mms_media_url"] = media_url
# Add any additional parameters
sms_data.update(kwargs)
return await api_client.request("notifications", method="POST", data=sms_data)
async def send_transactional_message(
channel: str,
content: Dict[str, str],
recipients: Dict[str, Any],
template_id: Optional[str] = None,
custom_data: Optional[Dict[str, Any]] = None,
**kwargs
) -> 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
template_id: Template ID to use
custom_data: Custom data to include
**kwargs: Additional parameters
"""
message_data = {
"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
# Add any additional parameters
message_data.update(kwargs)
return await api_client.request("notifications", method="POST", data=message_data)
async def view_messages(
limit: int = 20,
offset: int = 0,
kind: Optional[int] = None
) -> Dict[str, Any]:
"""
View recent messages sent through OneSignal.
Args:
limit: Maximum number of messages to return (max: 50)
offset: Result offset for pagination
kind: Filter by message type (0=Dashboard, 1=API, 3=Automated)
"""
params = {"limit": min(limit, 50), "offset": offset}
if kind is not None:
params["kind"] = kind
return await api_client.request("notifications", method="GET", params=params)
async def view_message_details(message_id: str) -> Dict[str, Any]:
"""Get detailed information about a specific message."""
return await api_client.request(f"notifications/{message_id}", method="GET")
async def cancel_message(message_id: str) -> Dict[str, Any]:
"""Cancel a scheduled message that hasn't been delivered yet."""
return await api_client.request(f"notifications/{message_id}", method="DELETE")
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 = app_manager.get_current_app()
if not app_config:
raise OneSignalAPIError("No app currently selected")
data = {
"app_id": app_config.app_id,
"events": event,
"email": f"{app_config.name}[email protected]"
}
return await api_client.request(
f"notifications/{message_id}/history",
method="POST",
data=data
)
async def export_messages_csv(
start_date: Optional[str] = None,
end_date: Optional[str] = None,
**kwargs
) -> Dict[str, Any]:
"""
Export messages to CSV.
Args:
start_date: Start date for export (ISO 8601 format)
end_date: End date for export (ISO 8601 format)
**kwargs: Additional export parameters
"""
data = {}
if start_date:
data["start_date"] = start_date
if end_date:
data["end_date"] = end_date
data.update(kwargs)
return await api_client.request(
"notifications/csv_export",
method="POST",
data=data,
use_org_key=True
)
```
--------------------------------------------------------------------------------
/test_onesignal_mcp.py:
--------------------------------------------------------------------------------
```python
#!/usr/bin/env python3
"""
Test script for OneSignal MCP Server
This script tests all the available functions in the OneSignal MCP server.
Usage:
1. Set up your environment variables in .env file:
- ONESIGNAL_APP_ID
- ONESIGNAL_API_KEY
- ONESIGNAL_ORG_API_KEY
2. Run the OneSignal MCP server:
python onesignal_server.py
3. In another terminal, run this test script:
python test_onesignal_mcp.py
"""
import asyncio
import json
import sys
from datetime import datetime
from typing import Dict, Any, List
# Test configuration
TEST_CONFIG = {
"test_email": "[email protected]",
"test_phone": "+15551234567", # E.164 format
"test_external_id": "test_user_123",
"test_player_id": None, # Will be populated during tests
"test_template_id": None, # Will be populated during tests
"test_user_id": None, # Will be populated during tests
"test_subscription_id": None, # Will be populated during tests
}
# Test results tracking
test_results = {
"passed": 0,
"failed": 0,
"skipped": 0,
"errors": []
}
def print_test_header(test_name: str):
"""Print a formatted test header."""
print(f"\n{'='*60}")
print(f"Testing: {test_name}")
print(f"{'='*60}")
def print_result(success: bool, message: str, result: Any = None):
"""Print test result with formatting."""
if success:
print(f"✅ {message}")
test_results["passed"] += 1
else:
print(f"❌ {message}")
test_results["failed"] += 1
if result:
test_results["errors"].append({
"test": message,
"error": result
})
if result and isinstance(result, dict):
print(f" Response: {json.dumps(result, indent=2)}")
async def test_app_management():
"""Test app management functions."""
print_test_header("App Management")
# Test listing apps
print("\n1. Testing list_apps...")
# Simulate function call - in real MCP, this would be via the MCP protocol
# For testing, you'll need to call these through the MCP client
print(" ⚠️ App management tests require MCP client implementation")
test_results["skipped"] += 1
async def test_messaging():
"""Test messaging functions."""
print_test_header("Messaging Functions")
tests = [
{
"name": "Send Push Notification",
"function": "send_push_notification",
"params": {
"title": "Test Push",
"message": "This is a test push notification",
"segments": ["Subscribed Users"]
}
},
{
"name": "Send Email",
"function": "send_email",
"params": {
"subject": "Test Email",
"body": "This is a test email",
"include_emails": [TEST_CONFIG["test_email"]]
}
},
{
"name": "Send SMS",
"function": "send_sms",
"params": {
"message": "Test SMS message",
"phone_numbers": [TEST_CONFIG["test_phone"]]
}
},
{
"name": "Send Transactional Message",
"function": "send_transactional_message",
"params": {
"channel": "push",
"content": {"en": "Transactional test"},
"recipients": {"include_external_user_ids": [TEST_CONFIG["test_external_id"]]}
}
}
]
for test in tests:
print(f"\n{test['name']}...")
print(f" Function: {test['function']}")
print(f" Params: {json.dumps(test['params'], indent=6)}")
print(" ⚠️ Requires MCP client to execute")
test_results["skipped"] += 1
async def test_templates():
"""Test template management functions."""
print_test_header("Template Management")
# Create template test
print("\n1. Create Template")
create_params = {
"name": f"Test Template {datetime.now().strftime('%Y%m%d_%H%M%S')}",
"title": "Test Template Title",
"message": "Test template message content"
}
print(f" Params: {json.dumps(create_params, indent=6)}")
# Update template test
print("\n2. Update Template")
if TEST_CONFIG["test_template_id"]:
update_params = {
"template_id": TEST_CONFIG["test_template_id"],
"name": "Updated Test Template",
"title": "Updated Title"
}
print(f" Params: {json.dumps(update_params, indent=6)}")
else:
print(" ⚠️ Skipped: No template ID available")
# Delete template test
print("\n3. Delete Template")
print(" ⚠️ Skipped: Preserving test template")
test_results["skipped"] += 3
async def test_live_activities():
"""Test iOS Live Activities functions."""
print_test_header("iOS Live Activities")
activity_tests = [
{
"name": "Start Live Activity",
"params": {
"activity_id": "test_activity_123",
"push_token": "test_push_token",
"subscription_id": "test_subscription",
"activity_attributes": {"event": "Test Event"},
"content_state": {"status": "active"}
}
},
{
"name": "Update Live Activity",
"params": {
"activity_id": "test_activity_123",
"name": "test_update",
"event": "update",
"content_state": {"status": "updated"}
}
},
{
"name": "End Live Activity",
"params": {
"activity_id": "test_activity_123",
"subscription_id": "test_subscription"
}
}
]
for test in activity_tests:
print(f"\n{test['name']}...")
print(f" Params: {json.dumps(test['params'], indent=6)}")
print(" ⚠️ Requires iOS app with Live Activities support")
test_results["skipped"] += 1
async def test_analytics():
"""Test analytics and export functions."""
print_test_header("Analytics & Export")
# View outcomes
print("\n1. View Outcomes")
outcomes_params = {
"outcome_names": ["session_duration", "purchase"],
"outcome_time_range": "1d",
"outcome_platforms": ["ios", "android"]
}
print(f" Params: {json.dumps(outcomes_params, indent=6)}")
# Export functions
export_tests = [
{
"name": "Export Players CSV",
"params": {
"start_date": "2024-01-01T00:00:00Z",
"end_date": "2024-12-31T23:59:59Z"
}
},
{
"name": "Export Messages CSV",
"params": {
"start_date": "2024-01-01T00:00:00Z",
"end_date": "2024-12-31T23:59:59Z"
}
}
]
for test in export_tests:
print(f"\n{test['name']}...")
print(f" Params: {json.dumps(test['params'], indent=6)}")
print(" ⚠️ Requires Organization API Key")
test_results["skipped"] += 1
async def test_user_management():
"""Test user management functions."""
print_test_header("User Management")
# Create user
print("\n1. Create User")
create_user_params = {
"name": "Test User",
"email": TEST_CONFIG["test_email"],
"external_id": TEST_CONFIG["test_external_id"],
"tags": {"test": "true", "created": datetime.now().isoformat()}
}
print(f" Params: {json.dumps(create_user_params, indent=6)}")
# Other user operations
user_tests = [
"View User",
"Update User",
"View User Identity",
"Create/Update Alias",
"Delete Alias"
]
for test in user_tests:
print(f"\n{test}")
print(f" ⚠️ Requires valid user_id from create_user")
test_results["skipped"] += 1
async def test_player_management():
"""Test player/device management functions."""
print_test_header("Player/Device Management")
# Add player
print("\n1. Add Player")
add_player_params = {
"device_type": 1, # Android
"identifier": "test_device_token",
"language": "en",
"tags": {"test_device": "true"}
}
print(f" Params: {json.dumps(add_player_params, indent=6)}")
# Other player operations
player_tests = [
"Edit Player",
"Delete Player",
"Edit Tags with External User ID"
]
for test in player_tests:
print(f"\n{test}")
print(f" ⚠️ Requires valid player_id")
test_results["skipped"] += 1
async def test_subscription_management():
"""Test subscription management functions."""
print_test_header("Subscription Management")
subscription_tests = [
{
"name": "Create Subscription",
"params": {
"user_id": "test_user_id",
"subscription_type": "email",
"identifier": TEST_CONFIG["test_email"]
}
},
{
"name": "Update Subscription",
"params": {
"user_id": "test_user_id",
"subscription_id": "test_subscription_id",
"enabled": False
}
},
{
"name": "Transfer Subscription",
"params": {
"user_id": "test_user_id",
"subscription_id": "test_subscription_id",
"new_user_id": "new_test_user_id"
}
},
{
"name": "Delete Subscription",
"params": {
"user_id": "test_user_id",
"subscription_id": "test_subscription_id"
}
}
]
for test in subscription_tests:
print(f"\n{test['name']}...")
print(f" Params: {json.dumps(test['params'], indent=6)}")
print(" ⚠️ Requires valid user_id and subscription_id")
test_results["skipped"] += 1
async def test_api_key_management():
"""Test API key management functions."""
print_test_header("API Key Management")
print("\n⚠️ API Key management requires Organization API Key")
print(" and should be tested carefully to avoid breaking access")
api_key_tests = [
"View App API Keys",
"Create App API Key",
"Update App API Key",
"Rotate App API Key",
"Delete App API Key"
]
for test in api_key_tests:
print(f"\n{test}")
print(" ⚠️ Skipped for safety")
test_results["skipped"] += 1
def print_summary():
"""Print test summary."""
print(f"\n\n{'='*60}")
print("TEST SUMMARY")
print(f"{'='*60}")
print(f"✅ Passed: {test_results['passed']}")
print(f"❌ Failed: {test_results['failed']}")
print(f"⚠️ Skipped: {test_results['skipped']}")
if test_results["errors"]:
print(f"\n\nERRORS:")
for error in test_results["errors"]:
print(f"\n- {error['test']}")
print(f" Error: {error['error']}")
async def main():
"""Run all tests."""
print("OneSignal MCP Server Test Suite")
print("================================")
print("\nNOTE: This test script shows the structure of all available functions.")
print("To actually execute these tests, you need to:")
print("1. Run the OneSignal MCP server")
print("2. Use an MCP client to connect and call the functions")
print("3. Have valid OneSignal API credentials in your .env file")
# Run test categories
await test_app_management()
await test_messaging()
await test_templates()
await test_live_activities()
await test_analytics()
await test_user_management()
await test_player_management()
await test_subscription_management()
await test_api_key_management()
# Print summary
print_summary()
if __name__ == "__main__":
asyncio.run(main())
```
--------------------------------------------------------------------------------
/implementation_examples.md:
--------------------------------------------------------------------------------
```markdown
# Implementation Examples for Missing OneSignal API Endpoints
This document provides concrete examples of how to add the missing API endpoints to the current `onesignal_server.py` file without requiring a full refactor.
## 1. Email Sending Endpoint
Add this function after the existing `send_push_notification` function:
```python
@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
```
## 2. SMS Sending Endpoint
```python
@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
```
## 3. Template Management Completions
### Update Template
```python
@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
```
### Delete Template
```python
@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
```
## 4. Live Activities
### Start Live Activity
```python
@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
```
### Update Live Activity
```python
@mcp.tool()
async def update_live_activity(activity_id: str, name: str, event: str,
content_state: Dict[str, Any],
dismissal_date: int = 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
"""
data = {
"name": name,
"event": event,
"content_state": content_state
}
if dismissal_date:
data["dismissal_date"] = dismissal_date
result = await make_onesignal_request(f"live_activities/{activity_id}/update",
method="POST", data=data)
return result
```
## 5. Analytics & Outcomes
### View Outcomes
```python
@mcp.tool()
async def view_outcomes(outcome_names: List[str],
outcome_time_range: str = None,
outcome_platforms: List[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"])
"""
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
result = await make_onesignal_request(f"apps/{app_config.app_id}/outcomes",
method="GET", params=params)
return result
```
## 6. Export Functions
### Export Players CSV
```python
@mcp.tool()
async def export_players_csv(start_date: str = None, end_date: str = None,
segment_names: List[str] = None) -> Dict[str, Any]:
"""Export player/subscription data to CSV.
Args:
start_date: Start date for export (ISO 8601 format)
end_date: End date for export (ISO 8601 format)
segment_names: List of segment names to export
"""
data = {}
if start_date:
data["start_date"] = start_date
if end_date:
data["end_date"] = end_date
if segment_names:
data["segment_names"] = segment_names
result = await make_onesignal_request("players/csv_export",
method="POST", data=data, use_org_key=True)
return result
```
## 7. API Key Management
### Delete API Key
```python
@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
```
### Update API Key
```python
@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
```
## 8. Player/Device Management
### Add a Player
```python
@mcp.tool()
async def add_player(device_type: int, identifier: str = None,
language: str = None, tags: Dict[str, str] = None) -> Dict[str, Any]:
"""Add a new player/device to OneSignal.
Args:
device_type: Device type (0=iOS, 1=Android, etc.)
identifier: Push token or player ID
language: Language code (e.g., "en")
tags: Dictionary of tags to assign
"""
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,
"device_type": device_type
}
if identifier:
data["identifier"] = identifier
if language:
data["language"] = language
if tags:
data["tags"] = tags
result = await make_onesignal_request("players", method="POST", data=data)
return result
```
### Edit Player
```python
@mcp.tool()
async def edit_player(player_id: str, tags: Dict[str, str] = None,
external_user_id: str = None, language: str = None) -> Dict[str, Any]:
"""Edit an existing player/device.
Args:
player_id: The player ID to edit
tags: Dictionary of tags to update
external_user_id: External user ID to assign
language: Language code to update
"""
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}
if tags:
data["tags"] = tags
if external_user_id:
data["external_user_id"] = external_user_id
if language:
data["language"] = language
if len(data) == 1: # Only app_id
return {"error": "No update parameters provided"}
result = await make_onesignal_request(f"players/{player_id}",
method="PUT", data=data)
return result
```
## Implementation Notes
1. **Error Handling**: All functions should return consistent error formats
2. **Authentication**: Use `use_org_key=True` for organization-level endpoints
3. **Validation**: Add input validation where necessary
4. **Documentation**: Include clear docstrings with parameter descriptions
5. **Testing**: Test each endpoint with valid OneSignal credentials
These implementations can be added directly to the existing `onesignal_server.py` file to provide immediate support for the missing API endpoints.
```
--------------------------------------------------------------------------------
/onesignal_refactored/server.py:
--------------------------------------------------------------------------------
```python
"""OneSignal MCP Server - Refactored implementation."""
import os
import logging
from mcp.server.fastmcp import FastMCP, Context
from .config import app_manager
from .api_client import OneSignalAPIError
from .tools import (
messages,
templates,
live_activities,
analytics,
)
# Version information
__version__ = "2.0.0"
# Configure logging
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
handlers=[logging.StreamHandler()]
)
logger = logging.getLogger("onesignal-mcp")
# Get log level from environment
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}'. Using INFO instead.")
log_level_str = "INFO"
logger.setLevel(log_level_str)
# Initialize MCP server
mcp = FastMCP("onesignal-server", settings={"log_level": log_level_str})
logger.info(f"OneSignal MCP server v{__version__} initialized")
# === Configuration Resource ===
@mcp.resource("onesignal://config")
def get_onesignal_config() -> str:
"""Get information about the OneSignal configuration."""
current_app = app_manager.get_current_app()
app_list = "\n".join([
f"- {key}: {app}" for key, app in app_manager.list_apps().items()
])
return f"""
OneSignal Server Configuration:
Version: {__version__}
Current App: {current_app.name if current_app else 'None'}
Available Apps:
{app_list or "No apps configured"}
This refactored MCP server provides comprehensive tools for:
- Multi-channel messaging (Push, Email, SMS)
- Transactional messages
- Template management
- Live Activities (iOS)
- Analytics and outcomes
- User and subscription management
- App configuration management
- Data export functionality
"""
# === App Management Tools ===
@mcp.tool()
async def list_apps() -> str:
"""List all configured OneSignal apps."""
apps = app_manager.list_apps()
if not apps:
return "No apps configured. Use add_app to add a new app configuration."
current_app = app_manager.get_current_app()
result = ["Configured OneSignal Apps:"]
for key, app in apps.items():
current_marker = " (current)" if current_app and key == app_manager.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."""
if not all([key, app_id, api_key]):
return "Error: All parameters (key, app_id, api_key) are required."
if key in app_manager.list_apps():
return f"Error: App key '{key}' already exists."
app_manager.add_app(key, app_id, api_key, name)
if len(app_manager.list_apps()) == 1:
app_manager.set_current_app(key)
return f"Successfully added app '{key}' with name '{name or key}'."
@mcp.tool()
async def switch_app(key: str) -> str:
"""Switch the current app to use for API requests."""
if app_manager.set_current_app(key):
app = app_manager.get_app(key)
return f"Switched to app '{key}' ({app.name})."
else:
available = ", ".join(app_manager.list_apps().keys()) or "None"
return f"Error: App key '{key}' not found. Available apps: {available}"
# === Messaging Tools ===
@mcp.tool()
async def send_push_notification(
title: str,
message: str,
segments: list = None,
include_player_ids: list = None,
external_ids: list = None,
data: dict = None
) -> dict:
"""Send a push notification through OneSignal."""
try:
return await messages.send_push_notification(
title=title,
message=message,
segments=segments,
include_player_ids=include_player_ids,
external_ids=external_ids,
data=data
)
except OneSignalAPIError as e:
return {"error": str(e)}
@mcp.tool()
async def send_email(
subject: str,
body: str,
include_emails: list = None,
segments: list = None,
external_ids: list = None,
template_id: str = None
) -> dict:
"""Send an email through OneSignal."""
try:
return await messages.send_email(
subject=subject,
body=body,
include_emails=include_emails,
segments=segments,
external_ids=external_ids,
template_id=template_id
)
except OneSignalAPIError as e:
return {"error": str(e)}
@mcp.tool()
async def send_sms(
message: str,
phone_numbers: list = None,
segments: list = None,
external_ids: list = None,
media_url: str = None
) -> dict:
"""Send an SMS/MMS through OneSignal."""
try:
return await messages.send_sms(
message=message,
phone_numbers=phone_numbers,
segments=segments,
external_ids=external_ids,
media_url=media_url
)
except OneSignalAPIError as e:
return {"error": str(e)}
@mcp.tool()
async def send_transactional_message(
channel: str,
content: dict,
recipients: dict,
template_id: str = None,
custom_data: dict = None
) -> dict:
"""Send a transactional message (immediate delivery)."""
try:
return await messages.send_transactional_message(
channel=channel,
content=content,
recipients=recipients,
template_id=template_id,
custom_data=custom_data
)
except OneSignalAPIError as e:
return {"error": str(e)}
@mcp.tool()
async def view_messages(limit: int = 20, offset: int = 0, kind: int = None) -> dict:
"""View recent messages sent through OneSignal."""
try:
return await messages.view_messages(limit=limit, offset=offset, kind=kind)
except OneSignalAPIError as e:
return {"error": str(e)}
@mcp.tool()
async def view_message_details(message_id: str) -> dict:
"""Get detailed information about a specific message."""
try:
return await messages.view_message_details(message_id)
except OneSignalAPIError as e:
return {"error": str(e)}
@mcp.tool()
async def cancel_message(message_id: str) -> dict:
"""Cancel a scheduled message."""
try:
return await messages.cancel_message(message_id)
except OneSignalAPIError as e:
return {"error": str(e)}
@mcp.tool()
async def export_messages_csv(start_date: str = None, end_date: str = None) -> dict:
"""Export messages to CSV (requires Organization API Key)."""
try:
return await messages.export_messages_csv(
start_date=start_date,
end_date=end_date
)
except OneSignalAPIError as e:
return {"error": str(e)}
# === Template Tools ===
@mcp.tool()
async def create_template(name: str, title: str, message: str) -> dict:
"""Create a new template."""
try:
result = await templates.create_template(name=name, title=title, message=message)
return {"success": f"Template '{name}' created with ID: {result.get('id')}"}
except Exception as e:
return {"error": str(e)}
@mcp.tool()
async def update_template(
template_id: str,
name: str = None,
title: str = None,
message: str = None
) -> dict:
"""Update an existing template."""
try:
await templates.update_template(
template_id=template_id,
name=name,
title=title,
message=message
)
return {"success": f"Template '{template_id}' updated successfully"}
except Exception as e:
return {"error": str(e)}
@mcp.tool()
async def view_templates() -> str:
"""List all templates."""
try:
result = await templates.view_templates()
return templates.format_template_list(result)
except Exception as e:
return f"Error: {str(e)}"
@mcp.tool()
async def view_template_details(template_id: str) -> str:
"""Get template details."""
try:
result = await templates.view_template_details(template_id)
return templates.format_template_details(result)
except Exception as e:
return f"Error: {str(e)}"
@mcp.tool()
async def delete_template(template_id: str) -> dict:
"""Delete a template."""
try:
await templates.delete_template(template_id)
return {"success": f"Template '{template_id}' deleted successfully"}
except Exception as e:
return {"error": str(e)}
@mcp.tool()
async def copy_template_to_app(
template_id: str,
target_app_id: str,
new_name: str = None
) -> dict:
"""Copy a template to another app."""
try:
result = await templates.copy_template_to_app(
template_id=template_id,
target_app_id=target_app_id,
new_name=new_name
)
return {"success": f"Template copied successfully. New ID: {result.get('id')}"}
except Exception as e:
return {"error": str(e)}
# === Live Activities Tools ===
@mcp.tool()
async def start_live_activity(
activity_id: str,
push_token: str,
subscription_id: str,
activity_attributes: dict,
content_state: dict
) -> dict:
"""Start a new iOS Live Activity."""
try:
return await live_activities.start_live_activity(
activity_id=activity_id,
push_token=push_token,
subscription_id=subscription_id,
activity_attributes=activity_attributes,
content_state=content_state
)
except OneSignalAPIError as e:
return {"error": str(e)}
@mcp.tool()
async def update_live_activity(
activity_id: str,
name: str,
event: str,
content_state: dict,
dismissal_date: int = None,
priority: int = None,
sound: str = None
) -> dict:
"""Update an iOS Live Activity."""
try:
return await live_activities.update_live_activity(
activity_id=activity_id,
name=name,
event=event,
content_state=content_state,
dismissal_date=dismissal_date,
priority=priority,
sound=sound
)
except OneSignalAPIError as e:
return {"error": str(e)}
@mcp.tool()
async def end_live_activity(
activity_id: str,
subscription_id: str,
dismissal_date: int = None,
priority: int = None
) -> dict:
"""End an iOS Live Activity."""
try:
return await live_activities.end_live_activity(
activity_id=activity_id,
subscription_id=subscription_id,
dismissal_date=dismissal_date,
priority=priority
)
except OneSignalAPIError as e:
return {"error": str(e)}
# === Analytics Tools ===
@mcp.tool()
async def view_outcomes(
outcome_names: list,
outcome_time_range: str = None,
outcome_platforms: list = None,
outcome_attribution: str = None
) -> str:
"""View outcomes data for your app."""
try:
result = await analytics.view_outcomes(
outcome_names=outcome_names,
outcome_time_range=outcome_time_range,
outcome_platforms=outcome_platforms,
outcome_attribution=outcome_attribution
)
return analytics.format_outcomes_response(result)
except Exception as e:
return f"Error: {str(e)}"
@mcp.tool()
async def export_players_csv(
start_date: str = None,
end_date: str = None,
segment_names: list = None
) -> dict:
"""Export player data to CSV (requires Organization API Key)."""
try:
return await analytics.export_players_csv(
start_date=start_date,
end_date=end_date,
segment_names=segment_names
)
except OneSignalAPIError as e:
return {"error": str(e)}
@mcp.tool()
async def export_audience_activity_csv(
start_date: str = None,
end_date: str = None,
event_types: list = None
) -> dict:
"""Export audience activity to CSV (requires Organization API Key)."""
try:
return await analytics.export_audience_activity_csv(
start_date=start_date,
end_date=end_date,
event_types=event_types
)
except OneSignalAPIError as e:
return {"error": str(e)}
# Run the server
if __name__ == "__main__":
mcp.run()
```
--------------------------------------------------------------------------------
/onesignal_server.py:
--------------------------------------------------------------------------------
```python
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 + "[email protected]" # 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()
```