This is page 2 of 2. Use http://codebase.md/weirdbrains/onesignal-mcp?lines=true&page={x} to view the full context.
# 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
--------------------------------------------------------------------------------
/onesignal_server.py:
--------------------------------------------------------------------------------
```python
1 | import os
2 | import json
3 | import requests
4 | import logging
5 | from typing import List, Dict, Any, Optional, Union
6 | from mcp.server.fastmcp import FastMCP, Context
7 | from dotenv import load_dotenv
8 |
9 | # Server information
10 | __version__ = "2.1.0"
11 |
12 | # Configure logging
13 | logging.basicConfig(
14 | level=logging.INFO, # Default level, will be overridden by env var if set
15 | format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
16 | handlers=[
17 | logging.StreamHandler()
18 | ]
19 | )
20 | logger = logging.getLogger("onesignal-mcp")
21 |
22 | # Load environment variables from .env file
23 | load_dotenv()
24 | logger.info("Environment variables loaded")
25 |
26 | # Get log level from environment, default to INFO, and ensure it's uppercase
27 | log_level_str = os.getenv("LOG_LEVEL", "INFO").upper()
28 | valid_log_levels = ["DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"]
29 | if log_level_str not in valid_log_levels:
30 | logger.warning(f"Invalid LOG_LEVEL '{log_level_str}' found in environment. Using INFO instead.")
31 | log_level_str = "INFO"
32 |
33 | # Apply the validated log level
34 | logger.setLevel(log_level_str)
35 |
36 | # Initialize the MCP server, passing the validated log level
37 | mcp = FastMCP("onesignal-server", settings={"log_level": log_level_str})
38 | logger.info(f"OneSignal MCP server initialized with log level: {log_level_str}")
39 |
40 | # OneSignal API configuration
41 | ONESIGNAL_API_URL = "https://api.onesignal.com/api/v1"
42 | ONESIGNAL_ORG_API_KEY = os.getenv("ONESIGNAL_ORG_API_KEY", "")
43 |
44 | # Class to manage app configurations
45 | class AppConfig:
46 | def __init__(self, app_id: str, api_key: str, name: str = None):
47 | self.app_id = app_id
48 | self.api_key = api_key
49 | self.name = name or app_id
50 |
51 | def __str__(self):
52 | return f"{self.name} ({self.app_id})"
53 |
54 | # Dictionary to store app configurations
55 | app_configs: Dict[str, AppConfig] = {}
56 |
57 | # Load app configurations from environment variables
58 | # Mandible app configuration
59 | mandible_app_id = os.getenv("ONESIGNAL_MANDIBLE_APP_ID", "") or os.getenv("ONESIGNAL_APP_ID", "")
60 | mandible_api_key = os.getenv("ONESIGNAL_MANDIBLE_API_KEY", "") or os.getenv("ONESIGNAL_API_KEY", "")
61 | if mandible_app_id and mandible_api_key:
62 | app_configs["mandible"] = AppConfig(mandible_app_id, mandible_api_key, "Mandible")
63 | current_app_key = "mandible"
64 | logger.info(f"Mandible app configured with ID: {mandible_app_id}")
65 |
66 | # Weird Brains app configuration
67 | weirdbrains_app_id = os.getenv("ONESIGNAL_WEIRDBRAINS_APP_ID", "")
68 | weirdbrains_api_key = os.getenv("ONESIGNAL_WEIRDBRAINS_API_KEY", "")
69 | if weirdbrains_app_id and weirdbrains_api_key:
70 | app_configs["weirdbrains"] = AppConfig(weirdbrains_app_id, weirdbrains_api_key, "Weird Brains")
71 | if not current_app_key:
72 | current_app_key = "weirdbrains"
73 | logger.info(f"Weird Brains app configured with ID: {weirdbrains_app_id}")
74 |
75 | # Fallback for default app configuration
76 | if not app_configs:
77 | default_app_id = os.getenv("ONESIGNAL_APP_ID", "")
78 | default_api_key = os.getenv("ONESIGNAL_API_KEY", "")
79 | if default_app_id and default_api_key:
80 | app_configs["default"] = AppConfig(default_app_id, default_api_key, "Default App")
81 | current_app_key = "default"
82 | logger.info(f"Default app configured with ID: {default_app_id}")
83 | else:
84 | current_app_key = None
85 | logger.warning("No app configurations found. Use add_app to add an app configuration.")
86 |
87 | # Function to add a new app configuration
88 | def add_app_config(key: str, app_id: str, api_key: str, name: str = None) -> None:
89 | """Add a new app configuration to the available apps.
90 |
91 | Args:
92 | key: Unique identifier for this app configuration
93 | app_id: OneSignal App ID
94 | api_key: OneSignal REST API Key
95 | name: Display name for the app (optional)
96 | """
97 | app_configs[key] = AppConfig(app_id, api_key, name or key)
98 | logger.info(f"Added app configuration '{key}' with ID: {app_id}")
99 |
100 | # Function to switch the current app
101 | def set_current_app(app_key: str) -> bool:
102 | """Set the current app to use for API requests.
103 |
104 | Args:
105 | app_key: The key of the app configuration to use
106 |
107 | Returns:
108 | True if successful, False if the app key doesn't exist
109 | """
110 | global current_app_key
111 | if app_key in app_configs:
112 | current_app_key = app_key
113 | logger.info(f"Switched to app '{app_key}'")
114 | return True
115 | logger.error(f"Failed to switch app: '{app_key}' not found")
116 | return False
117 |
118 | # Function to get the current app configuration
119 | def get_current_app() -> Optional[AppConfig]:
120 | """Get the current app configuration.
121 |
122 | Returns:
123 | The current AppConfig or None if no app is set
124 | """
125 | if current_app_key and current_app_key in app_configs:
126 | return app_configs[current_app_key]
127 | logger.warning("No current app is set. Use switch_app(key) to select an app.")
128 | return None
129 |
130 | # Helper function to determine whether to use Organization API Key
131 | def requires_org_api_key(endpoint: str) -> bool:
132 | """Determine if an endpoint requires the Organization API Key instead of a REST API Key.
133 |
134 | Args:
135 | endpoint: The API endpoint path
136 |
137 | Returns:
138 | True if the endpoint requires Organization API Key, False otherwise
139 | """
140 | # Organization-level endpoints that require Organization API Key
141 | org_level_endpoints = [
142 | "apps", # Managing apps
143 | "notifications/csv_export" # Export notifications
144 | ]
145 |
146 | # Check if endpoint starts with or matches any org-level endpoint
147 | for org_endpoint in org_level_endpoints:
148 | if endpoint == org_endpoint or endpoint.startswith(f"{org_endpoint}/"):
149 | return True
150 |
151 | return False
152 |
153 | # Helper function for OneSignal API requests
154 | async def make_onesignal_request(
155 | endpoint: str,
156 | method: str = "GET",
157 | data: Dict[str, Any] = None,
158 | params: Dict[str, Any] = None,
159 | use_org_key: bool = None,
160 | app_key: str = None
161 | ) -> Dict[str, Any]:
162 | """Make a request to the OneSignal API with proper authentication.
163 |
164 | Args:
165 | endpoint: API endpoint path
166 | method: HTTP method (GET, POST, PUT, DELETE)
167 | data: Request body for POST/PUT requests
168 | params: Query parameters for GET requests
169 | use_org_key: Whether to use the organization API key instead of the REST API key
170 | If None, will be automatically determined based on the endpoint
171 | app_key: The key of the app configuration to use (uses current app if None)
172 |
173 | Returns:
174 | API response as dictionary
175 | """
176 | headers = {
177 | "Content-Type": "application/json",
178 | "Accept": "application/json",
179 | }
180 |
181 | # If use_org_key is not explicitly specified, determine it based on the endpoint
182 | if use_org_key is None:
183 | use_org_key = requires_org_api_key(endpoint)
184 |
185 | # Determine which app configuration to use
186 | app_config = None
187 | if not use_org_key:
188 | if app_key and app_key in app_configs:
189 | app_config = app_configs[app_key]
190 | elif current_app_key and current_app_key in app_configs:
191 | app_config = app_configs[current_app_key]
192 |
193 | if not app_config:
194 | error_msg = "No app configuration available. Use set_current_app or specify app_key."
195 | logger.error(error_msg)
196 | return {"error": error_msg}
197 |
198 | # Check if it's a v2 API key
199 | if app_config.api_key.startswith("os_v2_"):
200 | headers["Authorization"] = f"Key {app_config.api_key}"
201 | else:
202 | headers["Authorization"] = f"Basic {app_config.api_key}"
203 | else:
204 | if not ONESIGNAL_ORG_API_KEY:
205 | error_msg = "Organization API Key not configured. Set the ONESIGNAL_ORG_API_KEY environment variable."
206 | logger.error(error_msg)
207 | return {"error": error_msg}
208 | # Check if it's a v2 API key
209 | if ONESIGNAL_ORG_API_KEY.startswith("os_v2_"):
210 | headers["Authorization"] = f"Key {ONESIGNAL_ORG_API_KEY}"
211 | else:
212 | headers["Authorization"] = f"Basic {ONESIGNAL_ORG_API_KEY}"
213 |
214 | url = f"{ONESIGNAL_API_URL}/{endpoint}"
215 |
216 | # If using app-specific endpoint and not using org key, add app_id to params if not already present
217 | if not use_org_key and app_config:
218 | if params is None:
219 | params = {}
220 | if "app_id" not in params and not endpoint.startswith("apps/"):
221 | params["app_id"] = app_config.app_id
222 |
223 | # For POST/PUT requests, add app_id to data if not already present
224 | if data is not None and method in ["POST", "PUT"] and "app_id" not in data and not endpoint.startswith("apps/"):
225 | data["app_id"] = app_config.app_id
226 |
227 | try:
228 | logger.debug(f"Making {method} request to {url}")
229 | logger.debug(f"Using {'Organization API Key' if use_org_key else 'App REST API Key'}")
230 | logger.debug(f"Authorization header type: {headers['Authorization'].split(' ')[0]}")
231 | if method == "GET":
232 | response = requests.get(url, headers=headers, params=params, timeout=30)
233 | elif method == "POST":
234 | response = requests.post(url, headers=headers, json=data, timeout=30)
235 | elif method == "PUT":
236 | response = requests.put(url, headers=headers, json=data, timeout=30)
237 | elif method == "DELETE":
238 | response = requests.delete(url, headers=headers, timeout=30)
239 | elif method == "PATCH":
240 | response = requests.patch(url, headers=headers, json=data, timeout=30)
241 | else:
242 | error_msg = f"Unsupported HTTP method: {method}"
243 | logger.error(error_msg)
244 | return {"error": error_msg}
245 |
246 | response.raise_for_status()
247 | return response.json() if response.text else {}
248 | except requests.exceptions.RequestException as e:
249 | error_message = f"Error: {str(e)}"
250 | try:
251 | if hasattr(e, 'response') and e.response is not None:
252 | error_data = e.response.json()
253 | if isinstance(error_data, dict):
254 | error_message = f"Error: {error_data.get('errors', [e.response.reason])[0]}"
255 | except Exception:
256 | pass
257 | logger.error(f"API request failed: {error_message}")
258 | return {"error": error_message}
259 | except Exception as e:
260 | error_message = f"Unexpected error: {str(e)}"
261 | logger.exception(error_message)
262 | return {"error": error_message}
263 |
264 | # Resource for OneSignal configuration information
265 | @mcp.resource("onesignal://config")
266 | def get_onesignal_config() -> str:
267 | """Get information about the OneSignal configuration"""
268 | current_app = get_current_app()
269 |
270 | app_list = "\n".join([f"- {key}: {app}" for key, app in app_configs.items()])
271 |
272 | return f"""
273 | OneSignal Server Configuration:
274 | Version: {__version__}
275 | API URL: {ONESIGNAL_API_URL}
276 | Organization API Key Status: {'Configured' if ONESIGNAL_ORG_API_KEY else 'Not configured'}
277 |
278 | Available Apps:
279 | {app_list or "No apps configured"}
280 |
281 | Current App: {current_app.name if current_app else 'None'}
282 |
283 | This MCP server provides tools for:
284 | - Viewing and managing messages (push notifications, emails, SMS)
285 | - Managing users and subscriptions
286 | - Viewing and managing segments
287 | - Creating and managing templates
288 | - Viewing app information
289 | - Managing multiple OneSignal applications
290 |
291 | Make sure you have set the appropriate environment variables in your .env file.
292 | """
293 |
294 | # === App Management Tools ===
295 |
296 | @mcp.tool()
297 | async def list_apps() -> str:
298 | """List all configured OneSignal apps in this server."""
299 | if not app_configs:
300 | return "No apps configured. Use add_app to add a new app configuration."
301 |
302 | current_app = get_current_app()
303 |
304 | result = ["Configured OneSignal Apps:"]
305 | for key, app in app_configs.items():
306 | current_marker = " (current)" if current_app and key == current_app_key else ""
307 | result.append(f"- {key}: {app.name} (App ID: {app.app_id}){current_marker}")
308 |
309 | return "\n".join(result)
310 |
311 | @mcp.tool()
312 | async def add_app(key: str, app_id: str, api_key: str, name: str = None) -> str:
313 | """Add a new OneSignal app configuration locally.
314 |
315 | Args:
316 | key: Unique identifier for this app configuration
317 | app_id: OneSignal App ID
318 | api_key: OneSignal REST API Key
319 | name: Display name for the app (optional)
320 | """
321 | if not key or not app_id or not api_key:
322 | return "Error: All parameters (key, app_id, api_key) are required."
323 |
324 | if key in app_configs:
325 | return f"Error: App key '{key}' already exists. Use a different key or update_app to modify it."
326 |
327 | add_app_config(key, app_id, api_key, name)
328 |
329 | # If this is the first app, set it as current
330 | global current_app_key
331 | if len(app_configs) == 1:
332 | current_app_key = key
333 |
334 | return f"Successfully added app '{key}' with name '{name or key}'."
335 |
336 | @mcp.tool()
337 | async def update_local_app_config(key: str, app_id: str = None, api_key: str = None, name: str = None) -> str:
338 | """Update an existing local OneSignal app configuration.
339 |
340 | Args:
341 | key: The key of the app configuration to update locally
342 | app_id: New OneSignal App ID (optional)
343 | api_key: New OneSignal REST API Key (optional)
344 | name: New display name for the app (optional)
345 | """
346 | if key not in app_configs:
347 | return f"Error: App key '{key}' not found."
348 |
349 | app = app_configs[key]
350 | updated = []
351 |
352 | if app_id:
353 | app.app_id = app_id
354 | updated.append("App ID")
355 | if api_key:
356 | app.api_key = api_key
357 | updated.append("API Key")
358 | if name:
359 | app.name = name
360 | updated.append("Name")
361 |
362 | if not updated:
363 | return "No changes were made. Specify at least one parameter to update."
364 |
365 | logger.info(f"Updated app '{key}': {', '.join(updated)}")
366 | return f"Successfully updated app '{key}': {', '.join(updated)}."
367 |
368 | @mcp.tool()
369 | async def remove_app(key: str) -> str:
370 | """Remove a local OneSignal app configuration.
371 |
372 | Args:
373 | key: The key of the app configuration to remove locally
374 | """
375 | if key not in app_configs:
376 | return f"Error: App key '{key}' not found."
377 |
378 | global current_app_key
379 | if current_app_key == key:
380 | if len(app_configs) > 1:
381 | # Set current to another app
382 | other_keys = [k for k in app_configs.keys() if k != key]
383 | current_app_key = other_keys[0]
384 | logger.info(f"Current app changed to '{current_app_key}' after removing '{key}'")
385 | else:
386 | current_app_key = None
387 | logger.warning("No current app set after removing the only app configuration")
388 |
389 | del app_configs[key]
390 | logger.info(f"Removed app configuration '{key}'")
391 |
392 | return f"Successfully removed app '{key}'."
393 |
394 | @mcp.tool()
395 | async def switch_app(key: str) -> str:
396 | """Switch the current app to use for API requests.
397 |
398 | Args:
399 | key: The key of the app configuration to use
400 | """
401 | if key not in app_configs:
402 | return f"Error: App key '{key}' not found. Available apps: {', '.join(app_configs.keys()) or 'None'}"
403 |
404 | global current_app_key
405 | current_app_key = key
406 | app = app_configs[key]
407 |
408 | return f"Switched to app '{key}' ({app.name})."
409 |
410 | # === Message Management Tools ===
411 |
412 | @mcp.tool()
413 | 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]:
414 | """Send a new push notification through OneSignal.
415 |
416 | Args:
417 | title: Notification title.
418 | message: Notification message content.
419 | segments: List of segments to include (e.g., ["Subscribed Users"]).
420 | external_ids: List of external user IDs to target.
421 | data: Additional data to include with the notification (optional).
422 | """
423 | app_config = get_current_app()
424 | if not app_config:
425 | return {"error": "No app currently selected. Use switch_app to select an app."}
426 |
427 | if not segments and not external_ids:
428 | segments = ["Subscribed Users"] # Default if no target specified
429 |
430 | notification_data = {
431 | "app_id": app_config.app_id,
432 | "contents": {"en": message},
433 | "headings": {"en": title},
434 | "target_channel": "push"
435 | }
436 |
437 | if segments:
438 | notification_data["included_segments"] = segments
439 | if external_ids:
440 | # Assuming make_onesignal_request handles converting list to JSON
441 | notification_data["include_external_user_ids"] = external_ids
442 |
443 | if data:
444 | notification_data["data"] = data
445 |
446 | # This endpoint uses app-specific REST API Key
447 | result = await make_onesignal_request("notifications", method="POST", data=notification_data, use_org_key=False)
448 |
449 | return result
450 |
451 | @mcp.tool()
452 | async def view_messages(limit: int = 20, offset: int = 0, kind: int = None) -> Dict[str, Any]:
453 | """View recent messages sent through OneSignal.
454 |
455 | Args:
456 | limit: Maximum number of messages to return (default: 20, max: 50)
457 | offset: Result offset for pagination (default: 0)
458 | kind: Filter by message type (0=Dashboard, 1=API, 3=Automated) (optional)
459 | """
460 | app_config = get_current_app()
461 | if not app_config:
462 | return {"error": "No app currently selected. Use switch_app to select an app."}
463 |
464 | params = {"limit": min(limit, 50), "offset": offset}
465 | if kind is not None:
466 | params["kind"] = kind
467 |
468 | # This endpoint uses app-specific REST API Key
469 | result = await make_onesignal_request("notifications", method="GET", params=params, use_org_key=False)
470 |
471 | # Return the raw JSON result for flexibility
472 | return result
473 |
474 | @mcp.tool()
475 | async def view_message_details(message_id: str) -> Dict[str, Any]:
476 | """Get detailed information about a specific message.
477 |
478 | Args:
479 | message_id: The ID of the message to retrieve details for
480 | """
481 | app_config = get_current_app()
482 | if not app_config:
483 | return {"error": "No app currently selected. Use switch_app to select an app."}
484 |
485 | # This endpoint uses app-specific REST API Key
486 | result = await make_onesignal_request(f"notifications/{message_id}", method="GET", use_org_key=False)
487 |
488 | # Return the raw JSON result
489 | return result
490 |
491 | @mcp.tool()
492 | async def view_message_history(message_id: str, event: str) -> Dict[str, Any]:
493 | """View the history / recipients of a message based on events.
494 |
495 | Args:
496 | message_id: The ID of the message.
497 | event: The event type to track (e.g., 'sent', 'clicked').
498 | """
499 | app_config = get_current_app()
500 | if not app_config:
501 | return {"error": "No app currently selected. Use switch_app to select an app."}
502 |
503 | data = {
504 | "app_id": app_config.app_id,
505 | "events": event,
506 | "email": get_current_app().name + "[email protected]" # Requires an email to send the CSV report
507 | }
508 |
509 | # Endpoint uses REST API Key
510 | result = await make_onesignal_request(f"notifications/{message_id}/history", method="POST", data=data, use_org_key=False)
511 | return result
512 |
513 | @mcp.tool()
514 | async def cancel_message(message_id: str) -> Dict[str, Any]:
515 | """Cancel a scheduled message that hasn't been delivered yet.
516 |
517 | Args:
518 | message_id: The ID of the message to cancel
519 | """
520 | app_config = get_current_app()
521 | if not app_config:
522 | return {"error": "No app currently selected. Use switch_app to select an app."}
523 |
524 | # This endpoint uses app-specific REST API Key
525 | result = await make_onesignal_request(f"notifications/{message_id}", method="DELETE", use_org_key=False)
526 |
527 | return result
528 |
529 | # === Segment Management Tools ===
530 |
531 | @mcp.tool()
532 | async def view_segments() -> str:
533 | """List all segments available in your OneSignal app."""
534 | app_config = get_current_app()
535 | if not app_config:
536 | return "No app currently selected. Use switch_app to select an app."
537 |
538 | # This endpoint requires app_id in the URL path
539 | endpoint = f"apps/{app_config.app_id}/segments"
540 | result = await make_onesignal_request(endpoint, method="GET", use_org_key=False)
541 |
542 | # Check if result is a dictionary with an error
543 | if isinstance(result, dict) and "error" in result:
544 | return f"Error retrieving segments: {result['error']}"
545 |
546 | # Handle different response formats
547 | if isinstance(result, dict):
548 | # Some endpoints return segments in a wrapper object
549 | segments = result.get("segments", [])
550 | elif isinstance(result, list):
551 | # Direct list of segments
552 | segments = result
553 | else:
554 | return f"Unexpected response format: {type(result)}"
555 |
556 | if not segments:
557 | return "No segments found."
558 |
559 | output = "Segments:\n\n"
560 |
561 | for segment in segments:
562 | if isinstance(segment, dict):
563 | output += f"ID: {segment.get('id')}\n"
564 | output += f"Name: {segment.get('name')}\n"
565 | output += f"Created: {segment.get('created_at')}\n"
566 | output += f"Updated: {segment.get('updated_at')}\n"
567 | output += f"Active: {segment.get('is_active', False)}\n"
568 | output += f"Read Only: {segment.get('read_only', False)}\n\n"
569 |
570 | return output
571 |
572 | @mcp.tool()
573 | async def create_segment(name: str, filters: str) -> str:
574 | """Create a new segment in your OneSignal app.
575 |
576 | Args:
577 | name: Name of the segment
578 | filters: JSON string representing the filters for this segment
579 | (e.g., '[{"field":"tag","key":"level","relation":"=","value":"10"}]')
580 | """
581 | try:
582 | parsed_filters = json.loads(filters)
583 | except json.JSONDecodeError:
584 | return "Error: The filters parameter must be a valid JSON string."
585 |
586 | data = {
587 | "name": name,
588 | "filters": parsed_filters
589 | }
590 |
591 | endpoint = f"apps/{get_current_app().app_id}/segments"
592 | result = await make_onesignal_request(endpoint, method="POST", data=data)
593 |
594 | if "error" in result:
595 | return f"Error creating segment: {result['error']}"
596 |
597 | return f"Segment '{name}' created successfully with ID: {result.get('id')}"
598 |
599 | @mcp.tool()
600 | async def delete_segment(segment_id: str) -> str:
601 | """Delete a segment from your OneSignal app.
602 |
603 | Args:
604 | segment_id: ID of the segment to delete
605 | """
606 | endpoint = f"apps/{get_current_app().app_id}/segments/{segment_id}"
607 | result = await make_onesignal_request(endpoint, method="DELETE")
608 |
609 | if "error" in result:
610 | return f"Error deleting segment: {result['error']}"
611 |
612 | return f"Segment '{segment_id}' deleted successfully"
613 |
614 | # === Template Management Tools ===
615 |
616 | @mcp.tool()
617 | async def view_templates() -> str:
618 | """List all templates available in your OneSignal app."""
619 | app_config = get_current_app()
620 | if not app_config:
621 | return "No app currently selected. Use switch_app to select an app."
622 |
623 | # This endpoint requires app_id in the URL path
624 | endpoint = f"apps/{app_config.app_id}/templates"
625 | result = await make_onesignal_request(endpoint, method="GET", use_org_key=False)
626 |
627 | if "error" in result:
628 | return f"Error retrieving templates: {result['error']}"
629 |
630 | templates = result.get("templates", [])
631 |
632 | if not templates:
633 | return "No templates found."
634 |
635 | output = "Templates:\n\n"
636 |
637 | for template in templates:
638 | output += f"ID: {template.get('id')}\n"
639 | output += f"Name: {template.get('name')}\n"
640 | output += f"Created: {template.get('created_at')}\n"
641 | output += f"Updated: {template.get('updated_at')}\n\n"
642 |
643 | return output
644 |
645 | @mcp.tool()
646 | async def view_template_details(template_id: str) -> str:
647 | """Get detailed information about a specific template.
648 |
649 | Args:
650 | template_id: The ID of the template to retrieve details for
651 | """
652 | params = {"app_id": get_current_app().app_id}
653 | result = await make_onesignal_request(f"templates/{template_id}", method="GET", params=params)
654 |
655 | if "error" in result:
656 | return f"Error fetching template details: {result['error']}"
657 |
658 | # Format the template details in a readable way
659 | heading = result.get("headings", {}).get("en", "No heading") if isinstance(result.get("headings"), dict) else "No heading"
660 | content = result.get("contents", {}).get("en", "No content") if isinstance(result.get("contents"), dict) else "No content"
661 |
662 | details = [
663 | f"ID: {result.get('id')}",
664 | f"Name: {result.get('name')}",
665 | f"Title: {heading}",
666 | f"Message: {content}",
667 | f"Platform: {result.get('platform')}",
668 | f"Created: {result.get('created_at')}"
669 | ]
670 |
671 | return "\n".join(details)
672 |
673 | @mcp.tool()
674 | async def create_template(name: str, title: str, message: str) -> str:
675 | """Create a new template in your OneSignal app.
676 |
677 | Args:
678 | name: Name of the template
679 | title: Title/heading of the template
680 | message: Content/message of the template
681 | """
682 | app_config = get_current_app()
683 | if not app_config:
684 | return "No app currently selected. Use switch_app to select an app."
685 |
686 | data = {
687 | "name": name,
688 | "headings": {"en": title},
689 | "contents": {"en": message}
690 | }
691 |
692 | # This endpoint requires app_id in the URL path
693 | endpoint = f"apps/{app_config.app_id}/templates"
694 | result = await make_onesignal_request(endpoint, method="POST", data=data)
695 |
696 | if "error" in result:
697 | return f"Error creating template: {result['error']}"
698 |
699 | return f"Template '{name}' created successfully with ID: {result.get('id')}"
700 |
701 | # === App Information Tools ===
702 |
703 | @mcp.tool()
704 | async def view_app_details() -> str:
705 | """Get detailed information about the configured OneSignal app."""
706 | app_config = get_current_app()
707 | if not app_config:
708 | return "No app currently selected. Use switch_app to select an app."
709 |
710 | # This endpoint requires the app_id in the URL and Organization API Key
711 | result = await make_onesignal_request(f"apps/{app_config.app_id}", method="GET", use_org_key=True)
712 |
713 | if "error" in result:
714 | return f"Error retrieving app details: {result['error']}"
715 |
716 | output = f"ID: {result.get('id')}\n"
717 | output += f"Name: {result.get('name')}\n"
718 | output += f"Created: {result.get('created_at')}\n"
719 | output += f"Updated: {result.get('updated_at')}\n"
720 | output += f"GCM: {'Configured' if result.get('gcm_key') else 'Not Configured'}\n"
721 | output += f"APNS: {'Configured' if result.get('apns_env') else 'Not Configured'}\n"
722 | output += f"Chrome: {'Configured' if result.get('chrome_web_key') else 'Not Configured'}\n"
723 | output += f"Safari: {'Configured' if result.get('safari_site_origin') else 'Not Configured'}\n"
724 | output += f"Email: {'Configured' if result.get('email_marketing') else 'Not Configured'}\n"
725 | output += f"SMS: {'Configured' if result.get('sms_marketing') else 'Not Configured'}\n"
726 |
727 | return output
728 |
729 | @mcp.tool()
730 | async def view_apps() -> str:
731 | """List all OneSignal applications for the organization (requires Organization API Key)."""
732 | result = await make_onesignal_request("apps", method="GET", use_org_key=True)
733 |
734 | if "error" in result:
735 | if "401" in str(result["error"]) or "403" in str(result["error"]):
736 | return ("Error: Your Organization API Key is either not configured or doesn't have permission to view all apps. "
737 | "Make sure you've set the ONESIGNAL_ORG_API_KEY environment variable with a valid Organization API Key. "
738 | "Organization API Keys can be found in your OneSignal dashboard under Organizations > Keys & IDs.")
739 | return f"Error fetching applications: {result['error']}"
740 |
741 | if not result:
742 | return "No applications found."
743 |
744 | apps_info = []
745 | for app in result:
746 | apps_info.append(
747 | f"ID: {app.get('id')}\n"
748 | f"Name: {app.get('name')}\n"
749 | f"GCM: {'Configured' if app.get('gcm_key') else 'Not Configured'}\n"
750 | f"APNS: {'Configured' if app.get('apns_env') else 'Not Configured'}\n"
751 | f"Created: {app.get('created_at')}"
752 | )
753 |
754 | return "Applications:\n\n" + "\n\n".join(apps_info)
755 |
756 | # === Organization-level Tools ===
757 |
758 | @mcp.tool()
759 | async def create_app(name: str, site_name: str = None) -> str:
760 | """Create a new OneSignal application at the organization level (requires Organization API Key).
761 |
762 | Args:
763 | name: Name of the new application
764 | site_name: Optional name of the website for the application
765 | """
766 | data = {
767 | "name": name
768 | }
769 |
770 | if site_name:
771 | data["site_name"] = site_name
772 |
773 | result = await make_onesignal_request("apps", method="POST", data=data, use_org_key=True)
774 |
775 | if "error" in result:
776 | if "401" in str(result["error"]) or "403" in str(result["error"]):
777 | return ("Error: Your Organization API Key is either not configured or doesn't have permission to create apps. "
778 | "Make sure you've set the ONESIGNAL_ORG_API_KEY environment variable with a valid Organization API Key.")
779 | return f"Error creating application: {result['error']}"
780 |
781 | return f"Application '{name}' created successfully with ID: {result.get('id')}"
782 |
783 | @mcp.tool()
784 | async def update_app(app_id: str, name: str = None, site_name: str = None) -> str:
785 | """Update an existing OneSignal application at the organization level (requires Organization API Key).
786 |
787 | Args:
788 | app_id: ID of the app to update
789 | name: New name for the application (optional)
790 | site_name: New site name for the application (optional)
791 | """
792 | data = {}
793 |
794 | if name:
795 | data["name"] = name
796 |
797 | if site_name:
798 | data["site_name"] = site_name
799 |
800 | if not data:
801 | return "Error: No update parameters provided. Specify at least one parameter to update."
802 |
803 | result = await make_onesignal_request(f"apps/{app_id}", method="PUT", data=data, use_org_key=True)
804 |
805 | if "error" in result:
806 | if "401" in str(result["error"]) or "403" in str(result["error"]):
807 | return ("Error: Your Organization API Key is either not configured or doesn't have permission to update apps. "
808 | "Make sure you've set the ONESIGNAL_ORG_API_KEY environment variable with a valid Organization API Key.")
809 | return f"Error updating application: {result['error']}"
810 |
811 | return f"Application '{app_id}' updated successfully"
812 |
813 | @mcp.tool()
814 | async def view_app_api_keys(app_id: str) -> str:
815 | """View API keys for a specific OneSignal app (requires Organization API Key).
816 |
817 | Args:
818 | app_id: The ID of the app to retrieve API keys for
819 | """
820 | result = await make_onesignal_request(f"apps/{app_id}/auth/tokens", use_org_key=True)
821 |
822 | if "error" in result:
823 | if "401" in str(result["error"]) or "403" in str(result["error"]):
824 | return ("Error: Your Organization API Key is either not configured or doesn't have permission to view API keys. "
825 | "Make sure you've set the ONESIGNAL_ORG_API_KEY environment variable with a valid Organization API Key.")
826 | return f"Error fetching API keys: {result['error']}"
827 |
828 | if not result.get("tokens", []):
829 | return f"No API keys found for app ID: {app_id}"
830 |
831 | keys_info = []
832 | for key in result.get("tokens", []):
833 | keys_info.append(
834 | f"ID: {key.get('id')}\n"
835 | f"Name: {key.get('name')}\n"
836 | f"Created: {key.get('created_at')}\n"
837 | f"Updated: {key.get('updated_at')}\n"
838 | f"IP Allowlist Mode: {key.get('ip_allowlist_mode', 'disabled')}"
839 | )
840 |
841 | return f"API Keys for App {app_id}:\n\n" + "\n\n".join(keys_info)
842 |
843 | @mcp.tool()
844 | async def create_app_api_key(app_id: str, name: str) -> str:
845 | """Create a new API key for a specific OneSignal app (requires Organization API Key).
846 |
847 | Args:
848 | app_id: The ID of the app to create an API key for
849 | name: Name for the new API key
850 | """
851 | data = {
852 | "name": name
853 | }
854 |
855 | result = await make_onesignal_request(f"apps/{app_id}/auth/tokens", method="POST", data=data, use_org_key=True)
856 |
857 | if "error" in result:
858 | if "401" in str(result["error"]) or "403" in str(result["error"]):
859 | return ("Error: Your Organization API Key is either not configured or doesn't have permission to create API keys. "
860 | "Make sure you've set the ONESIGNAL_ORG_API_KEY environment variable with a valid Organization API Key.")
861 | return f"Error creating API key: {result['error']}"
862 |
863 | # Format the API key details for display
864 | key_details = (
865 | f"API Key '{name}' created successfully!\n\n"
866 | f"Key ID: {result.get('id')}\n"
867 | f"Token: {result.get('token')}\n\n"
868 | f"IMPORTANT: Save this token now! You won't be able to see the full token again."
869 | )
870 |
871 | return key_details
872 |
873 | # === User Management Tools ===
874 |
875 | @mcp.tool()
876 | async def create_user(name: str = None, email: str = None, external_id: str = None, tags: Dict[str, str] = None) -> Dict[str, Any]:
877 | """Create a new user in OneSignal.
878 |
879 | Args:
880 | name: User's name (optional)
881 | email: User's email address (optional)
882 | external_id: External user ID for identification (optional)
883 | tags: Additional user tags/properties (optional)
884 | """
885 | app_config = get_current_app()
886 | if not app_config:
887 | return {"error": "No app currently selected. Use switch_app to select an app."}
888 |
889 | data = {}
890 | if name:
891 | data["name"] = name
892 | if email:
893 | data["email"] = email
894 | if external_id:
895 | data["external_user_id"] = external_id
896 | if tags:
897 | data["tags"] = tags
898 |
899 | result = await make_onesignal_request("users", method="POST", data=data)
900 | return result
901 |
902 | @mcp.tool()
903 | async def view_user(user_id: str) -> Dict[str, Any]:
904 | """Get detailed information about a specific user.
905 |
906 | Args:
907 | user_id: The OneSignal User ID to retrieve details for
908 | """
909 | app_config = get_current_app()
910 | if not app_config:
911 | return {"error": "No app currently selected. Use switch_app to select an app."}
912 |
913 | result = await make_onesignal_request(f"users/{user_id}", method="GET")
914 | return result
915 |
916 | @mcp.tool()
917 | async def update_user(user_id: str, name: str = None, email: str = None, tags: Dict[str, str] = None) -> Dict[str, Any]:
918 | """Update an existing user's information.
919 |
920 | Args:
921 | user_id: The OneSignal User ID to update
922 | name: New name for the user (optional)
923 | email: New email address (optional)
924 | tags: New or updated tags/properties (optional)
925 | """
926 | app_config = get_current_app()
927 | if not app_config:
928 | return {"error": "No app currently selected. Use switch_app to select an app."}
929 |
930 | data = {}
931 | if name:
932 | data["name"] = name
933 | if email:
934 | data["email"] = email
935 | if tags:
936 | data["tags"] = tags
937 |
938 | if not data:
939 | return {"error": "No update parameters provided"}
940 |
941 | result = await make_onesignal_request(f"users/{user_id}", method="PATCH", data=data)
942 | return result
943 |
944 | @mcp.tool()
945 | async def delete_user(user_id: str) -> Dict[str, Any]:
946 | """Delete a user and all their subscriptions.
947 |
948 | Args:
949 | user_id: The OneSignal User ID to delete
950 | """
951 | app_config = get_current_app()
952 | if not app_config:
953 | return {"error": "No app currently selected. Use switch_app to select an app."}
954 |
955 | result = await make_onesignal_request(f"users/{user_id}", method="DELETE")
956 | return result
957 |
958 | @mcp.tool()
959 | async def view_user_identity(user_id: str) -> Dict[str, Any]:
960 | """Get user identity information.
961 |
962 | Args:
963 | user_id: The OneSignal User ID to retrieve identity for
964 | """
965 | app_config = get_current_app()
966 | if not app_config:
967 | return {"error": "No app currently selected. Use switch_app to select an app."}
968 |
969 | result = await make_onesignal_request(f"users/{user_id}/identity", method="GET")
970 | return result
971 |
972 | @mcp.tool()
973 | async def create_or_update_alias(user_id: str, alias_label: str, alias_id: str) -> Dict[str, Any]:
974 | """Create or update a user alias.
975 |
976 | Args:
977 | user_id: The OneSignal User ID
978 | alias_label: The type/label of the alias (e.g., "email", "phone", "external")
979 | alias_id: The alias identifier value
980 | """
981 | app_config = get_current_app()
982 | if not app_config:
983 | return {"error": "No app currently selected. Use switch_app to select an app."}
984 |
985 | data = {
986 | "alias": {
987 | alias_label: alias_id
988 | }
989 | }
990 |
991 | result = await make_onesignal_request(f"users/{user_id}/identity", method="PATCH", data=data)
992 | return result
993 |
994 | @mcp.tool()
995 | async def delete_alias(user_id: str, alias_label: str) -> Dict[str, Any]:
996 | """Delete a user alias.
997 |
998 | Args:
999 | user_id: The OneSignal User ID
1000 | alias_label: The type/label of the alias to delete
1001 | """
1002 | app_config = get_current_app()
1003 | if not app_config:
1004 | return {"error": "No app currently selected. Use switch_app to select an app."}
1005 |
1006 | result = await make_onesignal_request(f"users/{user_id}/identity/{alias_label}", method="DELETE")
1007 | return result
1008 |
1009 | # === Subscription Management Tools ===
1010 |
1011 | @mcp.tool()
1012 | async def create_subscription(user_id: str, subscription_type: str, identifier: str) -> Dict[str, Any]:
1013 | """Create a new subscription for a user.
1014 |
1015 | Args:
1016 | user_id: The OneSignal User ID
1017 | subscription_type: Type of subscription ("email", "sms", "push")
1018 | identifier: Email address or phone number for the subscription
1019 | """
1020 | app_config = get_current_app()
1021 | if not app_config:
1022 | return {"error": "No app currently selected. Use switch_app to select an app."}
1023 |
1024 | data = {
1025 | "subscription": {
1026 | "type": subscription_type,
1027 | "identifier": identifier
1028 | }
1029 | }
1030 |
1031 | result = await make_onesignal_request(f"users/{user_id}/subscriptions", method="POST", data=data)
1032 | return result
1033 |
1034 | @mcp.tool()
1035 | async def update_subscription(user_id: str, subscription_id: str, enabled: bool = None) -> Dict[str, Any]:
1036 | """Update a user's subscription.
1037 |
1038 | Args:
1039 | user_id: The OneSignal User ID
1040 | subscription_id: The ID of the subscription to update
1041 | enabled: Whether the subscription should be enabled or disabled (optional)
1042 | """
1043 | app_config = get_current_app()
1044 | if not app_config:
1045 | return {"error": "No app currently selected. Use switch_app to select an app."}
1046 |
1047 | data = {}
1048 | if enabled is not None:
1049 | data["enabled"] = enabled
1050 |
1051 | result = await make_onesignal_request(f"users/{user_id}/subscriptions/{subscription_id}", method="PATCH", data=data)
1052 | return result
1053 |
1054 | @mcp.tool()
1055 | async def delete_subscription(user_id: str, subscription_id: str) -> Dict[str, Any]:
1056 | """Delete a user's subscription.
1057 |
1058 | Args:
1059 | user_id: The OneSignal User ID
1060 | subscription_id: The ID of the subscription to delete
1061 | """
1062 | app_config = get_current_app()
1063 | if not app_config:
1064 | return {"error": "No app currently selected. Use switch_app to select an app."}
1065 |
1066 | result = await make_onesignal_request(f"users/{user_id}/subscriptions/{subscription_id}", method="DELETE")
1067 | return result
1068 |
1069 | @mcp.tool()
1070 | async def transfer_subscription(user_id: str, subscription_id: str, new_user_id: str) -> Dict[str, Any]:
1071 | """Transfer a subscription from one user to another.
1072 |
1073 | Args:
1074 | user_id: The current OneSignal User ID
1075 | subscription_id: The ID of the subscription to transfer
1076 | new_user_id: The OneSignal User ID to transfer the subscription to
1077 | """
1078 | app_config = get_current_app()
1079 | if not app_config:
1080 | return {"error": "No app currently selected. Use switch_app to select an app."}
1081 |
1082 | data = {
1083 | "new_user_id": new_user_id
1084 | }
1085 |
1086 | result = await make_onesignal_request(f"users/{user_id}/subscriptions/{subscription_id}/transfer", method="PATCH", data=data)
1087 | return result
1088 |
1089 | @mcp.tool()
1090 | async def unsubscribe_email(token: str) -> Dict[str, Any]:
1091 | """Unsubscribe an email subscription using an unsubscribe token.
1092 |
1093 | Args:
1094 | token: The unsubscribe token from the email
1095 | """
1096 | app_config = get_current_app()
1097 | if not app_config:
1098 | return {"error": "No app currently selected. Use switch_app to select an app."}
1099 |
1100 | data = {
1101 | "token": token
1102 | }
1103 |
1104 | result = await make_onesignal_request("email/unsubscribe", method="POST", data=data)
1105 | return result
1106 |
1107 | # === NEW: Email & SMS Messaging Tools ===
1108 |
1109 | @mcp.tool()
1110 | async def send_email(subject: str, body: str, email_body: str = None,
1111 | include_emails: List[str] = None, segments: List[str] = None,
1112 | external_ids: List[str] = None, template_id: str = None) -> Dict[str, Any]:
1113 | """Send an email through OneSignal.
1114 |
1115 | Args:
1116 | subject: Email subject line
1117 | body: Plain text email content
1118 | email_body: HTML email content (optional)
1119 | include_emails: List of specific email addresses to target
1120 | segments: List of segments to include
1121 | external_ids: List of external user IDs to target
1122 | template_id: Email template ID to use
1123 | """
1124 | app_config = get_current_app()
1125 | if not app_config:
1126 | return {"error": "No app currently selected. Use switch_app to select an app."}
1127 |
1128 | email_data = {
1129 | "app_id": app_config.app_id,
1130 | "email_subject": subject,
1131 | "email_body": email_body or body,
1132 | "target_channel": "email"
1133 | }
1134 |
1135 | # Set targeting
1136 | if include_emails:
1137 | email_data["include_emails"] = include_emails
1138 | elif external_ids:
1139 | email_data["include_external_user_ids"] = external_ids
1140 | elif segments:
1141 | email_data["included_segments"] = segments
1142 | else:
1143 | email_data["included_segments"] = ["Subscribed Users"]
1144 |
1145 | if template_id:
1146 | email_data["template_id"] = template_id
1147 |
1148 | result = await make_onesignal_request("notifications", method="POST", data=email_data)
1149 | return result
1150 |
1151 | @mcp.tool()
1152 | async def send_sms(message: str, phone_numbers: List[str] = None,
1153 | segments: List[str] = None, external_ids: List[str] = None,
1154 | media_url: str = None) -> Dict[str, Any]:
1155 | """Send an SMS/MMS through OneSignal.
1156 |
1157 | Args:
1158 | message: SMS message content
1159 | phone_numbers: List of phone numbers in E.164 format
1160 | segments: List of segments to include
1161 | external_ids: List of external user IDs to target
1162 | media_url: URL for MMS media attachment
1163 | """
1164 | app_config = get_current_app()
1165 | if not app_config:
1166 | return {"error": "No app currently selected. Use switch_app to select an app."}
1167 |
1168 | sms_data = {
1169 | "app_id": app_config.app_id,
1170 | "contents": {"en": message},
1171 | "target_channel": "sms"
1172 | }
1173 |
1174 | if phone_numbers:
1175 | sms_data["include_phone_numbers"] = phone_numbers
1176 | elif external_ids:
1177 | sms_data["include_external_user_ids"] = external_ids
1178 | elif segments:
1179 | sms_data["included_segments"] = segments
1180 | else:
1181 | return {"error": "SMS requires phone_numbers, external_ids, or segments"}
1182 |
1183 | if media_url:
1184 | sms_data["mms_media_url"] = media_url
1185 |
1186 | result = await make_onesignal_request("notifications", method="POST", data=sms_data)
1187 | return result
1188 |
1189 | @mcp.tool()
1190 | async def send_transactional_message(channel: str, content: Dict[str, str],
1191 | recipients: Dict[str, Any], template_id: str = None,
1192 | custom_data: Dict[str, Any] = None) -> Dict[str, Any]:
1193 | """Send a transactional message (immediate delivery, no scheduling).
1194 |
1195 | Args:
1196 | channel: Channel to send on ("push", "email", "sms")
1197 | content: Message content (format depends on channel)
1198 | recipients: Targeting information (include_external_user_ids, include_emails, etc.)
1199 | template_id: Template ID to use
1200 | custom_data: Custom data to include
1201 | """
1202 | app_config = get_current_app()
1203 | if not app_config:
1204 | return {"error": "No app currently selected. Use switch_app to select an app."}
1205 |
1206 | message_data = {
1207 | "app_id": app_config.app_id,
1208 | "target_channel": channel,
1209 | "is_transactional": True
1210 | }
1211 |
1212 | # Set content based on channel
1213 | if channel == "email":
1214 | message_data["email_subject"] = content.get("subject", "")
1215 | message_data["email_body"] = content.get("body", "")
1216 | else:
1217 | message_data["contents"] = content
1218 |
1219 | # Set recipients
1220 | message_data.update(recipients)
1221 |
1222 | if template_id:
1223 | message_data["template_id"] = template_id
1224 |
1225 | if custom_data:
1226 | message_data["data"] = custom_data
1227 |
1228 | result = await make_onesignal_request("notifications", method="POST", data=message_data)
1229 | return result
1230 |
1231 | # === NEW: Enhanced Template Management ===
1232 |
1233 | @mcp.tool()
1234 | async def update_template(template_id: str, name: str = None,
1235 | title: str = None, message: str = None) -> Dict[str, Any]:
1236 | """Update an existing template.
1237 |
1238 | Args:
1239 | template_id: ID of the template to update
1240 | name: New name for the template
1241 | title: New title/heading for the template
1242 | message: New content/message for the template
1243 | """
1244 | data = {}
1245 |
1246 | if name:
1247 | data["name"] = name
1248 | if title:
1249 | data["headings"] = {"en": title}
1250 | if message:
1251 | data["contents"] = {"en": message}
1252 |
1253 | if not data:
1254 | return {"error": "No update parameters provided"}
1255 |
1256 | result = await make_onesignal_request(f"templates/{template_id}",
1257 | method="PATCH", data=data)
1258 | return result
1259 |
1260 | @mcp.tool()
1261 | async def delete_template(template_id: str) -> Dict[str, Any]:
1262 | """Delete a template from your OneSignal app.
1263 |
1264 | Args:
1265 | template_id: ID of the template to delete
1266 | """
1267 | result = await make_onesignal_request(f"templates/{template_id}",
1268 | method="DELETE")
1269 | if "error" not in result:
1270 | return {"success": f"Template '{template_id}' deleted successfully"}
1271 | return result
1272 |
1273 | @mcp.tool()
1274 | async def copy_template_to_app(template_id: str, target_app_id: str,
1275 | new_name: str = None) -> Dict[str, Any]:
1276 | """Copy a template to another OneSignal app.
1277 |
1278 | Args:
1279 | template_id: ID of the template to copy
1280 | target_app_id: ID of the app to copy the template to
1281 | new_name: Optional new name for the copied template
1282 | """
1283 | data = {"app_id": target_app_id}
1284 |
1285 | if new_name:
1286 | data["name"] = new_name
1287 |
1288 | result = await make_onesignal_request(f"templates/{template_id}/copy",
1289 | method="POST", data=data)
1290 | return result
1291 |
1292 | # === NEW: Live Activities (iOS) ===
1293 |
1294 | @mcp.tool()
1295 | async def start_live_activity(activity_id: str, push_token: str,
1296 | subscription_id: str, activity_attributes: Dict[str, Any],
1297 | content_state: Dict[str, Any]) -> Dict[str, Any]:
1298 | """Start a new iOS Live Activity.
1299 |
1300 | Args:
1301 | activity_id: Unique identifier for the activity
1302 | push_token: Push token for the Live Activity
1303 | subscription_id: Subscription ID for the user
1304 | activity_attributes: Static attributes for the activity
1305 | content_state: Initial dynamic content state
1306 | """
1307 | data = {
1308 | "activity_id": activity_id,
1309 | "push_token": push_token,
1310 | "subscription_id": subscription_id,
1311 | "activity_attributes": activity_attributes,
1312 | "content_state": content_state
1313 | }
1314 |
1315 | result = await make_onesignal_request(f"live_activities/{activity_id}/start",
1316 | method="POST", data=data)
1317 | return result
1318 |
1319 | @mcp.tool()
1320 | async def update_live_activity(activity_id: str, name: str, event: str,
1321 | content_state: Dict[str, Any],
1322 | dismissal_date: int = None, priority: int = None,
1323 | sound: str = None) -> Dict[str, Any]:
1324 | """Update an existing iOS Live Activity.
1325 |
1326 | Args:
1327 | activity_id: ID of the activity to update
1328 | name: Name identifier for the update
1329 | event: Event type ("update" or "end")
1330 | content_state: Updated dynamic content state
1331 | dismissal_date: Unix timestamp for automatic dismissal
1332 | priority: Notification priority (5-10)
1333 | sound: Sound file name for the update
1334 | """
1335 | data = {
1336 | "name": name,
1337 | "event": event,
1338 | "content_state": content_state
1339 | }
1340 |
1341 | if dismissal_date:
1342 | data["dismissal_date"] = dismissal_date
1343 | if priority:
1344 | data["priority"] = priority
1345 | if sound:
1346 | data["sound"] = sound
1347 |
1348 | result = await make_onesignal_request(f"live_activities/{activity_id}/update",
1349 | method="POST", data=data)
1350 | return result
1351 |
1352 | @mcp.tool()
1353 | async def end_live_activity(activity_id: str, subscription_id: str,
1354 | dismissal_date: int = None, priority: int = None) -> Dict[str, Any]:
1355 | """End an iOS Live Activity.
1356 |
1357 | Args:
1358 | activity_id: ID of the activity to end
1359 | subscription_id: Subscription ID associated with the activity
1360 | dismissal_date: Unix timestamp for dismissal
1361 | priority: Notification priority (5-10)
1362 | """
1363 | data = {
1364 | "subscription_id": subscription_id,
1365 | "event": "end"
1366 | }
1367 |
1368 | if dismissal_date:
1369 | data["dismissal_date"] = dismissal_date
1370 | if priority:
1371 | data["priority"] = priority
1372 |
1373 | result = await make_onesignal_request(f"live_activities/{activity_id}/end",
1374 | method="POST", data=data)
1375 | return result
1376 |
1377 | # === NEW: Analytics & Outcomes ===
1378 |
1379 | @mcp.tool()
1380 | async def view_outcomes(outcome_names: List[str], outcome_time_range: str = None,
1381 | outcome_platforms: List[str] = None,
1382 | outcome_attribution: str = None) -> Dict[str, Any]:
1383 | """View outcomes data for your OneSignal app.
1384 |
1385 | Args:
1386 | outcome_names: List of outcome names to fetch data for
1387 | outcome_time_range: Time range for data (e.g., "1d", "1mo")
1388 | outcome_platforms: Filter by platforms (e.g., ["ios", "android"])
1389 | outcome_attribution: Attribution model ("direct" or "influenced")
1390 | """
1391 | app_config = get_current_app()
1392 | if not app_config:
1393 | return {"error": "No app currently selected. Use switch_app to select an app."}
1394 |
1395 | params = {"outcome_names": outcome_names}
1396 |
1397 | if outcome_time_range:
1398 | params["outcome_time_range"] = outcome_time_range
1399 | if outcome_platforms:
1400 | params["outcome_platforms"] = outcome_platforms
1401 | if outcome_attribution:
1402 | params["outcome_attribution"] = outcome_attribution
1403 |
1404 | result = await make_onesignal_request(f"apps/{app_config.app_id}/outcomes",
1405 | method="GET", params=params)
1406 | return result
1407 |
1408 | # === NEW: Export Functions ===
1409 |
1410 | @mcp.tool()
1411 | async def export_messages_csv(start_date: str = None, end_date: str = None,
1412 | event_types: List[str] = None) -> Dict[str, Any]:
1413 | """Export messages/notifications data to CSV.
1414 |
1415 | Args:
1416 | start_date: Start date for export (ISO 8601 format)
1417 | end_date: End date for export (ISO 8601 format)
1418 | event_types: List of event types to export
1419 | """
1420 | data = {}
1421 |
1422 | if start_date:
1423 | data["start_date"] = start_date
1424 | if end_date:
1425 | data["end_date"] = end_date
1426 | if event_types:
1427 | data["event_types"] = event_types
1428 |
1429 | result = await make_onesignal_request("notifications/csv_export",
1430 | method="POST", data=data, use_org_key=True)
1431 | return result
1432 |
1433 | # === NEW: API Key Management ===
1434 |
1435 | @mcp.tool()
1436 | async def delete_app_api_key(app_id: str, key_id: str) -> Dict[str, Any]:
1437 | """Delete an API key from a specific OneSignal app.
1438 |
1439 | Args:
1440 | app_id: The ID of the app
1441 | key_id: The ID of the API key to delete
1442 | """
1443 | result = await make_onesignal_request(f"apps/{app_id}/auth/tokens/{key_id}",
1444 | method="DELETE", use_org_key=True)
1445 | if "error" not in result:
1446 | return {"success": f"API key '{key_id}' deleted successfully"}
1447 | return result
1448 |
1449 | @mcp.tool()
1450 | async def update_app_api_key(app_id: str, key_id: str, name: str = None,
1451 | scopes: List[str] = None) -> Dict[str, Any]:
1452 | """Update an API key for a specific OneSignal app.
1453 |
1454 | Args:
1455 | app_id: The ID of the app
1456 | key_id: The ID of the API key to update
1457 | name: New name for the API key
1458 | scopes: New list of permission scopes
1459 | """
1460 | data = {}
1461 |
1462 | if name:
1463 | data["name"] = name
1464 | if scopes:
1465 | data["scopes"] = scopes
1466 |
1467 | if not data:
1468 | return {"error": "No update parameters provided"}
1469 |
1470 | result = await make_onesignal_request(f"apps/{app_id}/auth/tokens/{key_id}",
1471 | method="PATCH", data=data, use_org_key=True)
1472 | return result
1473 |
1474 | @mcp.tool()
1475 | async def rotate_app_api_key(app_id: str, key_id: str) -> Dict[str, Any]:
1476 | """Rotate an API key (generate new token while keeping permissions).
1477 |
1478 | Args:
1479 | app_id: The ID of the app
1480 | key_id: The ID of the API key to rotate
1481 | """
1482 | result = await make_onesignal_request(f"apps/{app_id}/auth/tokens/{key_id}/rotate",
1483 | method="POST", use_org_key=True)
1484 | if "error" not in result:
1485 | return {
1486 | "success": f"API key rotated successfully",
1487 | "new_token": result.get("token"),
1488 | "warning": "Save the new token now! You won't be able to see it again."
1489 | }
1490 | return result
1491 |
1492 | # Run the server
1493 | if __name__ == "__main__":
1494 | # Run the server
1495 | mcp.run()
```