# Directory Structure
```
├── .github
│ └── workflows
│ └── ci.yml
├── .gitignore
├── docs
│ ├── client_configuration.md
│ ├── configuration.md
│ ├── examples.md
│ ├── README.md
│ ├── troubleshooting.md
│ └── usage.md
├── LICENSE
├── pyproject.toml
├── README.md
├── setup.py
├── src
│ ├── __init__.py
│ ├── mcp
│ │ ├── __init__.py
│ │ ├── protocol.py
│ │ └── validation.py
│ ├── notification
│ │ ├── __init__.py
│ │ ├── macos.py
│ │ ├── manager.py
│ │ ├── platform.py
│ │ └── toast.py
│ └── server
│ ├── __init__.py
│ ├── commands.py
│ └── connection.py
└── tests
├── __init__.py
├── test_commands.py
├── test_connection.py
├── test_macos.py
├── test_manager.py
├── test_protocol.py
└── test_toast.py
```
# Files
--------------------------------------------------------------------------------
/.gitignore:
--------------------------------------------------------------------------------
```
# Byte-compiled / optimized / DLL files
__pycache__/
*.py[cod]
*$py.class
# C extensions
*.so
# Distribution / packaging
.Python
build/
develop-eggs/
dist/
downloads/
eggs/
.eggs/
lib/
lib64/
parts/
sdist/
var/
wheels/
share/python-wheels/
*.egg-info/
.installed.cfg
*.egg
MANIFEST
# PyInstaller
# Usually these files are written by a python script from a template
# before PyInstaller builds the exe, so as to inject date/other infos into it.
*.manifest
*.spec
# Installer logs
pip-log.txt
pip-delete-this-directory.txt
# Unit test / coverage reports
htmlcov/
.tox/
.nox/
.coverage
.coverage.*
.cache
nosetests.xml
coverage.xml
*.cover
*.py,cover
.hypothesis/
.pytest_cache/
cover/
# Translations
*.mo
*.pot
# Django stuff:
*.log
local_settings.py
db.sqlite3
db.sqlite3-journal
# Flask stuff:
instance/
.webassets-cache
# Scrapy stuff:
.scrapy
# Sphinx documentation
docs/_build/
# PyBuilder
.pybuilder/
target/
# Jupyter Notebook
.ipynb_checkpoints
# IPython
profile_default/
ipython_config.py
# pyenv
# For a library or package, you might want to ignore these files since the code is
# intended to run in multiple environments; otherwise, check them in:
# .python-version
# pipenv
# According to pypa/pipenv#598, it is recommended to include Pipfile.lock in version control.
# However, in case of collaboration, if having platform-specific dependencies or dependencies
# having no cross-platform support, pipenv may install dependencies that don't work, or not
# install all needed dependencies.
#Pipfile.lock
# UV
# Similar to Pipfile.lock, it is generally recommended to include uv.lock in version control.
# This is especially recommended for binary packages to ensure reproducibility, and is more
# commonly ignored for libraries.
#uv.lock
# poetry
# Similar to Pipfile.lock, it is generally recommended to include poetry.lock in version control.
# This is especially recommended for binary packages to ensure reproducibility, and is more
# commonly ignored for libraries.
# https://python-poetry.org/docs/basic-usage/#commit-your-poetrylock-file-to-version-control
#poetry.lock
# pdm
# Similar to Pipfile.lock, it is generally recommended to include pdm.lock in version control.
#pdm.lock
# pdm stores project-wide configurations in .pdm.toml, but it is recommended to not include it
# in version control.
# https://pdm.fming.dev/latest/usage/project/#working-with-version-control
.pdm.toml
.pdm-python
.pdm-build/
# PEP 582; used by e.g. github.com/David-OConnor/pyflow and github.com/pdm-project/pdm
__pypackages__/
# Celery stuff
celerybeat-schedule
celerybeat.pid
# SageMath parsed files
*.sage.py
# Environments
.env
.venv
env/
venv/
ENV/
env.bak/
venv.bak/
# Spyder project settings
.spyderproject
.spyproject
# Rope project settings
.ropeproject
# mkdocs documentation
/site
# mypy
.mypy_cache/
.dmypy.json
dmypy.json
# Pyre type checker
.pyre/
# pytype static type analyzer
.pytype/
# Cython debug symbols
cython_debug/
# PyCharm
# JetBrains specific template is maintained in a separate JetBrains.gitignore that can
# be found at https://github.com/github/gitignore/blob/main/Global/JetBrains.gitignore
# and can be added to the global gitignore or merged into this file. For a more nuclear
# option (not recommended) you can uncomment the following to ignore the entire idea folder.
#.idea/
# Ruff stuff:
.ruff_cache/
# PyPI configuration file
.pypirc
```
--------------------------------------------------------------------------------
/src/__init__.py:
--------------------------------------------------------------------------------
```python
"""
toast-mcp-server package.
"""
```
--------------------------------------------------------------------------------
/src/notification/__init__.py:
--------------------------------------------------------------------------------
```python
"""
Notification system package.
"""
```
--------------------------------------------------------------------------------
/src/server/__init__.py:
--------------------------------------------------------------------------------
```python
"""
Server implementation package.
"""
```
--------------------------------------------------------------------------------
/tests/__init__.py:
--------------------------------------------------------------------------------
```python
"""
Test package for toast-mcp-server.
"""
```
--------------------------------------------------------------------------------
/src/mcp/__init__.py:
--------------------------------------------------------------------------------
```python
"""
MCP protocol implementation package.
"""
```
--------------------------------------------------------------------------------
/setup.py:
--------------------------------------------------------------------------------
```python
from setuptools import setup, find_packages
setup(
name="toast-mcp-server",
version="0.1.0",
packages=find_packages(),
install_requires=[
"win10toast;platform_system=='Windows'",
],
python_requires=">=3.8",
)
```
--------------------------------------------------------------------------------
/pyproject.toml:
--------------------------------------------------------------------------------
```toml
[build-system]
requires = ["setuptools>=42", "wheel"]
build-backend = "setuptools.build_meta"
[project]
name = "toast-mcp-server"
version = "0.1.0"
description = "A Model Context Protocol (MCP) server with Windows 10 and macOS desktop notifications support"
readme = "README.md"
requires-python = ">=3.8"
license = {text = "MIT"}
dependencies = [
"win10toast;platform_system=='Windows'",
]
[tool.pytest.ini_options]
testpaths = ["tests"]
python_files = "test_*.py"
```
--------------------------------------------------------------------------------
/src/notification/platform.py:
--------------------------------------------------------------------------------
```python
"""
Platform detection utilities for toast-mcp-server.
This module provides utilities for detecting the current platform
and selecting the appropriate notification system.
"""
import logging
import platform
from typing import Dict, Any, Optional, List, Union, Callable
from src.mcp.protocol import NotificationType
logger = logging.getLogger(__name__)
def is_windows() -> bool:
"""
Check if the current platform is Windows.
Returns:
True if the current platform is Windows, False otherwise
"""
return platform.system() == "Windows"
def is_macos() -> bool:
"""
Check if the current platform is macOS.
Returns:
True if the current platform is macOS, False otherwise
"""
return platform.system() == "Darwin"
def is_linux() -> bool:
"""
Check if the current platform is Linux.
Returns:
True if the current platform is Linux, False otherwise
"""
return platform.system() == "Linux"
def get_platform_name() -> str:
"""
Get the name of the current platform.
Returns:
Name of the current platform ("windows", "macos", "linux", or "unknown")
"""
if is_windows():
return "windows"
elif is_macos():
return "macos"
elif is_linux():
return "linux"
else:
return "unknown"
```
--------------------------------------------------------------------------------
/.github/workflows/ci.yml:
--------------------------------------------------------------------------------
```yaml
name: CI/CD
on:
push:
branches: [ main, devin/* ]
pull_request:
branches: [ main ]
jobs:
test:
runs-on: ubuntu-latest
strategy:
matrix:
python-version: [3.8, 3.9, '3.10']
steps:
- uses: actions/checkout@v3
- name: Set up Python ${{ matrix.python-version }}
uses: actions/setup-python@v4
with:
python-version: ${{ matrix.python-version }}
- name: Install dependencies
run: |
python -m pip install --upgrade pip
if [ -f requirements.txt ]; then pip install -r requirements.txt; fi
pip install pytest pytest-cov flake8
pip install -e .
- name: Lint with flake8
run: |
# stop the build if there are Python syntax errors or undefined names
flake8 . --count --select=E9,F63,F7,F82 --show-source --statistics
# exit-zero treats all errors as warnings
flake8 . --count --exit-zero --max-complexity=10 --max-line-length=127 --statistics
- name: Test with pytest
run: |
pytest --cov=src tests/
build:
runs-on: ubuntu-latest
needs: test
if: github.event_name == 'push' && github.ref == 'refs/heads/main'
steps:
- uses: actions/checkout@v3
- name: Set up Python
uses: actions/setup-python@v4
with:
python-version: '3.10'
- name: Install dependencies
run: |
python -m pip install --upgrade pip
if [ -f requirements.txt ]; then pip install -r requirements.txt; fi
pip install build
- name: Build package
run: |
python -m build
- name: Archive production artifacts
uses: actions/upload-artifact@v3
with:
name: dist
path: |
dist/
deploy:
runs-on: ubuntu-latest
needs: build
if: github.event_name == 'push' && github.ref == 'refs/heads/main' && startsWith(github.event.head_commit.message, 'Release')
steps:
- uses: actions/checkout@v3
- name: Download artifacts
uses: actions/download-artifact@v3
with:
name: dist
path: dist
- name: Set up Python
uses: actions/setup-python@v4
with:
python-version: '3.10'
- name: Install dependencies
run: |
python -m pip install --upgrade pip
pip install twine
- name: Create GitHub Release
id: create_release
uses: actions/create-release@v1
env:
GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}
with:
tag_name: v${{ github.run_number }}
release_name: Release v${{ github.run_number }}
body: |
Automated release from CI/CD pipeline
draft: false
prerelease: false
```
--------------------------------------------------------------------------------
/tests/test_toast.py:
--------------------------------------------------------------------------------
```python
"""
Tests for the Windows 10 Toast Notification implementation.
"""
import unittest
from unittest.mock import patch, MagicMock
from src.notification.toast import (
ToastNotificationManager, NotificationFactory,
show_notification, NotificationType
)
class TestToastNotification(unittest.TestCase):
"""Test cases for the Windows 10 Toast Notification implementation."""
@patch('src.notification.toast.ToastNotifier')
def test_toast_notification_manager_init(self, mock_toaster):
"""Test initializing the toast notification manager."""
manager = ToastNotificationManager()
self.assertIsNotNone(manager.toaster)
@patch('src.notification.toast.ToastNotifier')
def test_show_notification(self, mock_toaster):
"""Test showing a notification."""
mock_instance = MagicMock()
mock_toaster.return_value = mock_instance
manager = ToastNotificationManager()
result = manager.show_notification(
title="Test Title",
message="Test Message",
notification_type=NotificationType.INFO,
duration=5
)
mock_instance.show_toast.assert_called_once()
self.assertTrue(result)
args, kwargs = mock_instance.show_toast.call_args
self.assertEqual(kwargs["title"], "Test Title")
self.assertEqual(kwargs["msg"], "Test Message")
self.assertEqual(kwargs["duration"], 5)
self.assertTrue(kwargs["threaded"])
@patch('src.notification.toast.ToastNotifier')
def test_show_notification_with_exception(self, mock_toaster):
"""Test showing a notification with an exception."""
mock_instance = MagicMock()
mock_instance.show_toast.side_effect = Exception("Test exception")
mock_toaster.return_value = mock_instance
manager = ToastNotificationManager()
result = manager.show_notification(
title="Test Title",
message="Test Message"
)
self.assertFalse(result)
@patch('src.notification.toast.ToastNotificationManager')
def test_notification_factory(self, mock_manager_class):
"""Test the notification factory."""
mock_manager = MagicMock()
mock_manager_class.return_value = mock_manager
mock_manager.show_notification.return_value = True
factory = NotificationFactory()
result = factory.create_info_notification("Info Title", "Info Message")
self.assertTrue(result)
mock_manager.show_notification.assert_called_with(
title="Info Title",
message="Info Message",
notification_type=NotificationType.INFO,
duration=5
)
result = factory.create_warning_notification("Warning Title", "Warning Message")
self.assertTrue(result)
mock_manager.show_notification.assert_called_with(
title="Warning Title",
message="Warning Message",
notification_type=NotificationType.WARNING,
duration=7
)
result = factory.create_error_notification("Error Title", "Error Message")
self.assertTrue(result)
mock_manager.show_notification.assert_called_with(
title="Error Title",
message="Error Message",
notification_type=NotificationType.ERROR,
duration=10
)
result = factory.create_success_notification("Success Title", "Success Message")
self.assertTrue(result)
mock_manager.show_notification.assert_called_with(
title="Success Title",
message="Success Message",
notification_type=NotificationType.SUCCESS,
duration=5
)
@patch('src.notification.toast.notification_factory')
def test_show_notification_helper(self, mock_factory):
"""Test the show_notification helper function."""
mock_factory.create_info_notification.return_value = True
mock_factory.create_warning_notification.return_value = True
mock_factory.create_error_notification.return_value = True
mock_factory.create_success_notification.return_value = True
result = show_notification("Info Title", "Info Message", "info")
self.assertTrue(result)
mock_factory.create_info_notification.assert_called_with("Info Title", "Info Message", 5)
result = show_notification("Warning Title", "Warning Message", "warning")
self.assertTrue(result)
mock_factory.create_warning_notification.assert_called_with("Warning Title", "Warning Message", 5)
result = show_notification("Error Title", "Error Message", "error")
self.assertTrue(result)
mock_factory.create_error_notification.assert_called_with("Error Title", "Error Message", 5)
result = show_notification("Success Title", "Success Message", "success")
self.assertTrue(result)
mock_factory.create_success_notification.assert_called_with("Success Title", "Success Message", 5)
result = show_notification("Invalid Title", "Invalid Message", "invalid")
self.assertTrue(result)
mock_factory.create_info_notification.assert_called_with("Invalid Title", "Invalid Message", 5)
if __name__ == "__main__":
unittest.main()
```
--------------------------------------------------------------------------------
/src/mcp/validation.py:
--------------------------------------------------------------------------------
```python
"""
Validation utilities for MCP protocol messages.
This module provides functions for validating MCP messages and their contents
to ensure they conform to the protocol specification.
"""
import re
from typing import Dict, Any, List, Optional, Tuple, Union
from .protocol import NotificationType, MessageType
def validate_notification(data: Dict[str, Any]) -> Tuple[bool, Optional[str]]:
"""
Validate a notification message.
Args:
data: The notification message data to validate
Returns:
A tuple of (is_valid, error_message)
"""
if "title" not in data:
return False, "Missing required field: title"
if "message" not in data:
return False, "Missing required field: message"
if not isinstance(data["title"], str):
return False, "Title must be a string"
if len(data["title"]) > 100:
return False, "Title exceeds maximum length of 100 characters"
if not isinstance(data["message"], str):
return False, "Message must be a string"
if len(data["message"]) > 1000:
return False, "Message exceeds maximum length of 1000 characters"
if "notification_type" in data:
try:
NotificationType(data["notification_type"])
except ValueError:
valid_types = [t.value for t in NotificationType]
return False, f"Invalid notification type. Must be one of: {', '.join(valid_types)}"
if "duration" in data:
if not isinstance(data["duration"], int):
return False, "Duration must be an integer"
if data["duration"] < 1 or data["duration"] > 60:
return False, "Duration must be between 1 and 60 seconds"
if "client_id" in data:
if not isinstance(data["client_id"], str):
return False, "Client ID must be a string"
if not re.match(r'^[a-zA-Z0-9_\-\.]{1,50}$', data["client_id"]):
return False, "Client ID contains invalid characters or exceeds maximum length"
if "icon" in data and not isinstance(data["icon"], str):
return False, "Icon must be a string"
if "actions" in data:
if not isinstance(data["actions"], list):
return False, "Actions must be a list"
for i, action in enumerate(data["actions"]):
if not isinstance(action, dict):
return False, f"Action at index {i} must be a dictionary"
if "id" not in action:
return False, f"Action at index {i} missing required field: id"
if "text" not in action:
return False, f"Action at index {i} missing required field: text"
if not isinstance(action["id"], str) or not isinstance(action["text"], str):
return False, f"Action id and text must be strings"
return True, None
def validate_message(message_type: MessageType, data: Dict[str, Any]) -> Tuple[bool, Optional[str]]:
"""
Validate a message based on its type.
Args:
message_type: The type of message to validate
data: The message data to validate
Returns:
A tuple of (is_valid, error_message)
"""
if message_type == MessageType.NOTIFICATION:
return validate_notification(data)
elif message_type == MessageType.RESPONSE:
if "success" not in data:
return False, "Response message missing required field: success"
if not isinstance(data["success"], bool):
return False, "Success field must be a boolean"
return True, None
elif message_type == MessageType.ERROR:
if "code" not in data:
return False, "Error message missing required field: code"
if "message" not in data:
return False, "Error message missing required field: message"
if not isinstance(data["code"], int):
return False, "Error code must be an integer"
if not isinstance(data["message"], str):
return False, "Error message must be a string"
return True, None
elif message_type in (MessageType.PING, MessageType.PONG):
return True, None
return False, f"Unknown message type: {message_type}"
def validate_message_format(data: Dict[str, Any]) -> Tuple[bool, Optional[str], Optional[MessageType]]:
"""
Validate the format of a message.
Args:
data: The message data to validate
Returns:
A tuple of (is_valid, error_message, message_type)
"""
if not isinstance(data, dict):
return False, "Message must be a dictionary", None
if "type" not in data:
return False, "Message missing required field: type", None
try:
message_type = MessageType(data["type"])
except ValueError:
valid_types = [t.value for t in MessageType]
return False, f"Invalid message type. Must be one of: {', '.join(valid_types)}", None
if "data" not in data:
data["data"] = {} # Add empty data for types that don't require it
if not isinstance(data["data"], dict):
return False, "Message data must be a dictionary", None
is_valid, error_message = validate_message(message_type, data["data"])
return is_valid, error_message, message_type
```
--------------------------------------------------------------------------------
/tests/test_protocol.py:
--------------------------------------------------------------------------------
```python
"""
Tests for the MCP protocol implementation.
"""
import unittest
import json
from src.mcp.protocol import (
MCPMessage, NotificationMessage, ResponseMessage, ErrorMessage,
PingMessage, PongMessage, MessageType, NotificationType, parse_message
)
from src.mcp.validation import validate_message_format, validate_notification
class TestMCPProtocol(unittest.TestCase):
"""Test cases for the MCP protocol implementation."""
def test_notification_message_creation(self):
"""Test creating a notification message."""
notification = NotificationMessage(
title="Test Notification",
message="This is a test notification",
notification_type=NotificationType.INFO,
duration=10,
client_id="test-client",
icon="info-icon",
actions=[{"id": "action1", "text": "Click Me"}]
)
self.assertEqual(notification.msg_type, MessageType.NOTIFICATION)
self.assertEqual(notification.data["title"], "Test Notification")
self.assertEqual(notification.data["message"], "This is a test notification")
self.assertEqual(notification.data["notification_type"], "info")
self.assertEqual(notification.data["duration"], 10)
self.assertEqual(notification.data["client_id"], "test-client")
self.assertEqual(notification.data["icon"], "info-icon")
self.assertEqual(len(notification.data["actions"]), 1)
self.assertEqual(notification.data["actions"][0]["id"], "action1")
def test_notification_message_serialization(self):
"""Test serializing a notification message to JSON."""
notification = NotificationMessage(
title="Test Notification",
message="This is a test notification"
)
json_str = notification.to_json()
data = json.loads(json_str)
self.assertEqual(data["type"], "notification")
self.assertEqual(data["data"]["title"], "Test Notification")
self.assertEqual(data["data"]["message"], "This is a test notification")
self.assertEqual(data["data"]["notification_type"], "info") # Default type
def test_notification_message_deserialization(self):
"""Test deserializing a notification message from JSON."""
json_str = json.dumps({
"type": "notification",
"data": {
"title": "Test Notification",
"message": "This is a test notification",
"notification_type": "warning",
"duration": 15
}
})
message = MCPMessage.from_json(json_str)
self.assertIsInstance(message, NotificationMessage)
self.assertEqual(message.data["title"], "Test Notification")
self.assertEqual(message.data["message"], "This is a test notification")
self.assertEqual(message.data["notification_type"], "warning")
self.assertEqual(message.data["duration"], 15)
def test_response_message(self):
"""Test creating and serializing a response message."""
response = ResponseMessage(
success=True,
message="Operation completed successfully",
data={"id": 123}
)
self.assertEqual(response.msg_type, MessageType.RESPONSE)
self.assertTrue(response.data["success"])
self.assertEqual(response.data["message"], "Operation completed successfully")
self.assertEqual(response.data["data"]["id"], 123)
json_str = response.to_json()
data = json.loads(json_str)
self.assertEqual(data["type"], "response")
self.assertTrue(data["data"]["success"])
def test_error_message(self):
"""Test creating and serializing an error message."""
error = ErrorMessage(
error_code=404,
error_message="Resource not found",
details={"resource_id": "abc123"}
)
self.assertEqual(error.msg_type, MessageType.ERROR)
self.assertEqual(error.data["code"], 404)
self.assertEqual(error.data["message"], "Resource not found")
self.assertEqual(error.data["details"]["resource_id"], "abc123")
json_str = error.to_json()
data = json.loads(json_str)
self.assertEqual(data["type"], "error")
self.assertEqual(data["data"]["code"], 404)
def test_ping_pong_messages(self):
"""Test creating ping and pong messages."""
ping = PingMessage()
pong = PongMessage()
self.assertEqual(ping.msg_type, MessageType.PING)
self.assertEqual(pong.msg_type, MessageType.PONG)
ping_json = ping.to_json()
pong_json = pong.to_json()
ping_data = json.loads(ping_json)
pong_data = json.loads(pong_json)
self.assertEqual(ping_data["type"], "ping")
self.assertEqual(pong_data["type"], "pong")
def test_parse_message(self):
"""Test parsing messages from different formats."""
json_str = json.dumps({
"type": "notification",
"data": {
"title": "Test",
"message": "Test message"
}
})
message1 = parse_message(json_str)
self.assertIsInstance(message1, NotificationMessage)
dict_data = {
"type": "response",
"data": {
"success": True
}
}
message2 = parse_message(dict_data)
self.assertIsInstance(message2, ResponseMessage)
def test_validation(self):
"""Test message validation."""
valid_notification = {
"type": "notification",
"data": {
"title": "Valid Title",
"message": "Valid message content",
"notification_type": "info",
"duration": 5
}
}
is_valid, error, msg_type = validate_message_format(valid_notification)
self.assertTrue(is_valid)
self.assertIsNone(error)
self.assertEqual(msg_type, MessageType.NOTIFICATION)
invalid_notification = {
"type": "notification",
"data": {
"message": "Message without title"
}
}
is_valid, error, msg_type = validate_message_format(invalid_notification)
self.assertFalse(is_valid)
self.assertIn("Missing required field: title", error)
valid_data = {
"title": "Test",
"message": "Test message",
"duration": 10
}
is_valid, error = validate_notification(valid_data)
self.assertTrue(is_valid)
self.assertIsNone(error)
invalid_data = {
"title": "Test",
"message": "Test message",
"duration": 100 # Too long
}
is_valid, error = validate_notification(invalid_data)
self.assertFalse(is_valid)
self.assertIn("Duration must be between 1 and 60 seconds", error)
if __name__ == "__main__":
unittest.main()
```
--------------------------------------------------------------------------------
/src/notification/toast.py:
--------------------------------------------------------------------------------
```python
"""
Windows 10 Toast Notification implementation for toast-mcp-server.
This module provides functionality to display Windows 10 toast notifications
using the win10toast library.
"""
import logging
from typing import Dict, Any, Optional, List, Union
from win10toast import ToastNotifier
from src.mcp.protocol import NotificationType
logger = logging.getLogger(__name__)
class ToastNotificationManager:
"""
Manager for Windows 10 Toast Notifications.
This class handles the creation and display of Windows 10 toast notifications
using the win10toast library.
"""
def __init__(self):
"""Initialize the toast notification manager."""
self.toaster = ToastNotifier()
logger.info("Toast notification manager initialized")
def show_notification(self,
title: str,
message: str,
notification_type: NotificationType = NotificationType.INFO,
duration: int = 5,
icon_path: Optional[str] = None,
threaded: bool = True) -> bool:
"""
Show a Windows 10 toast notification.
Args:
title: Title of the notification
message: Content of the notification
notification_type: Type of notification (info, warning, error, success)
duration: Duration to display the notification in seconds
icon_path: Path to the icon file to display with the notification
threaded: Whether to show the notification in a separate thread
Returns:
True if the notification was successfully displayed, False otherwise
"""
if not icon_path:
icon_path = self._get_default_icon(notification_type)
logger.debug(f"Showing notification: {title} ({notification_type.value})")
try:
self.toaster.show_toast(
title=title,
msg=message,
icon_path=icon_path,
duration=duration,
threaded=threaded
)
return True
except Exception as e:
logger.error(f"Failed to show notification: {str(e)}")
return False
def _get_default_icon(self, notification_type: NotificationType) -> str:
"""
Get the default icon path for a notification type.
Args:
notification_type: Type of notification
Returns:
Path to the default icon for the notification type
"""
icons = {
NotificationType.INFO: "icons/info.ico",
NotificationType.WARNING: "icons/warning.ico",
NotificationType.ERROR: "icons/error.ico",
NotificationType.SUCCESS: "icons/success.ico"
}
return icons.get(notification_type, "icons/default.ico")
class NotificationFactory:
"""
Factory for creating notifications based on notification type.
This class provides methods for creating and displaying different types
of notifications with appropriate default settings.
"""
def __init__(self):
"""Initialize the notification factory."""
self.toast_manager = ToastNotificationManager()
def create_info_notification(self, title: str, message: str, duration: int = 5) -> bool:
"""
Create and show an information notification.
Args:
title: Title of the notification
message: Content of the notification
duration: Duration to display the notification in seconds
Returns:
True if the notification was successfully displayed, False otherwise
"""
return self.toast_manager.show_notification(
title=title,
message=message,
notification_type=NotificationType.INFO,
duration=duration
)
def create_warning_notification(self, title: str, message: str, duration: int = 7) -> bool:
"""
Create and show a warning notification.
Args:
title: Title of the notification
message: Content of the notification
duration: Duration to display the notification in seconds
Returns:
True if the notification was successfully displayed, False otherwise
"""
return self.toast_manager.show_notification(
title=title,
message=message,
notification_type=NotificationType.WARNING,
duration=duration
)
def create_error_notification(self, title: str, message: str, duration: int = 10) -> bool:
"""
Create and show an error notification.
Args:
title: Title of the notification
message: Content of the notification
duration: Duration to display the notification in seconds
Returns:
True if the notification was successfully displayed, False otherwise
"""
return self.toast_manager.show_notification(
title=title,
message=message,
notification_type=NotificationType.ERROR,
duration=duration
)
def create_success_notification(self, title: str, message: str, duration: int = 5) -> bool:
"""
Create and show a success notification.
Args:
title: Title of the notification
message: Content of the notification
duration: Duration to display the notification in seconds
Returns:
True if the notification was successfully displayed, False otherwise
"""
return self.toast_manager.show_notification(
title=title,
message=message,
notification_type=NotificationType.SUCCESS,
duration=duration
)
notification_factory = NotificationFactory()
def show_notification(title: str, message: str, notification_type: str = "info", duration: int = 5) -> bool:
"""
Show a notification with the specified parameters.
This is a convenience function for showing notifications without directly
interacting with the NotificationFactory or ToastNotificationManager classes.
Args:
title: Title of the notification
message: Content of the notification
notification_type: Type of notification ("info", "warning", "error", "success")
duration: Duration to display the notification in seconds
Returns:
True if the notification was successfully displayed, False otherwise
"""
try:
notification_type_enum = NotificationType(notification_type)
except ValueError:
logger.warning(f"Invalid notification type: {notification_type}, using INFO")
notification_type_enum = NotificationType.INFO
if notification_type_enum == NotificationType.INFO:
return notification_factory.create_info_notification(title, message, duration)
elif notification_type_enum == NotificationType.WARNING:
return notification_factory.create_warning_notification(title, message, duration)
elif notification_type_enum == NotificationType.ERROR:
return notification_factory.create_error_notification(title, message, duration)
elif notification_type_enum == NotificationType.SUCCESS:
return notification_factory.create_success_notification(title, message, duration)
return notification_factory.create_info_notification(title, message, duration)
```
--------------------------------------------------------------------------------
/src/notification/manager.py:
--------------------------------------------------------------------------------
```python
"""
Unified notification manager for toast-mcp-server.
This module provides a unified interface for displaying notifications
on different platforms, automatically selecting the appropriate
notification system based on the current platform.
"""
import logging
from typing import Dict, Any, Optional, List, Union
from src.mcp.protocol import NotificationType
from src.notification.platform import is_windows, is_macos, get_platform_name
logger = logging.getLogger(__name__)
class NotificationManager:
"""
Unified notification manager for multiple platforms.
This class provides a unified interface for displaying notifications
on different platforms, automatically selecting the appropriate
notification system based on the current platform.
"""
def __init__(self):
"""Initialize the notification manager."""
self._platform = get_platform_name()
self._notification_system = self._get_notification_system()
logger.info(f"Notification manager initialized for platform: {self._platform}")
def _get_notification_system(self):
"""
Get the appropriate notification system for the current platform.
Returns:
Notification system for the current platform
"""
if is_windows():
from src.notification.toast import ToastNotificationManager
return ToastNotificationManager()
elif is_macos():
from src.notification.macos import MacOSNotificationManager
return MacOSNotificationManager()
else:
logger.warning(f"No notification system available for platform: {self._platform}")
return None
def show_notification(self,
title: str,
message: str,
notification_type: NotificationType = NotificationType.INFO,
duration: int = 5,
**kwargs) -> bool:
"""
Show a notification on the current platform.
Args:
title: Title of the notification
message: Content of the notification
notification_type: Type of notification (info, warning, error, success)
duration: Duration to display the notification in seconds (ignored on some platforms)
**kwargs: Additional platform-specific parameters
Returns:
True if the notification was successfully displayed, False otherwise
"""
if not self._notification_system:
logger.error(f"Cannot show notification on unsupported platform: {self._platform}")
return False
logger.debug(f"Showing notification on {self._platform}: {title} ({notification_type.value})")
try:
if is_windows():
return self._notification_system.show_notification(
title=title,
message=message,
notification_type=notification_type,
duration=duration,
**kwargs
)
elif is_macos():
subtitle = kwargs.get("subtitle")
sound = kwargs.get("sound", True)
return self._notification_system.show_notification(
title=title,
message=message,
notification_type=notification_type,
duration=duration,
subtitle=subtitle,
sound=sound
)
else:
return False
except Exception as e:
logger.error(f"Failed to show notification: {str(e)}")
return False
class NotificationFactory:
"""
Factory for creating notifications based on notification type.
This class provides methods for creating and displaying different types
of notifications with appropriate default settings.
"""
def __init__(self):
"""Initialize the notification factory."""
self.notification_manager = NotificationManager()
def create_info_notification(self, title: str, message: str, **kwargs) -> bool:
"""
Create and show an information notification.
Args:
title: Title of the notification
message: Content of the notification
**kwargs: Additional platform-specific parameters
Returns:
True if the notification was successfully displayed, False otherwise
"""
return self.notification_manager.show_notification(
title=title,
message=message,
notification_type=NotificationType.INFO,
**kwargs
)
def create_warning_notification(self, title: str, message: str, **kwargs) -> bool:
"""
Create and show a warning notification.
Args:
title: Title of the notification
message: Content of the notification
**kwargs: Additional platform-specific parameters
Returns:
True if the notification was successfully displayed, False otherwise
"""
return self.notification_manager.show_notification(
title=title,
message=message,
notification_type=NotificationType.WARNING,
**kwargs
)
def create_error_notification(self, title: str, message: str, **kwargs) -> bool:
"""
Create and show an error notification.
Args:
title: Title of the notification
message: Content of the notification
**kwargs: Additional platform-specific parameters
Returns:
True if the notification was successfully displayed, False otherwise
"""
return self.notification_manager.show_notification(
title=title,
message=message,
notification_type=NotificationType.ERROR,
**kwargs
)
def create_success_notification(self, title: str, message: str, **kwargs) -> bool:
"""
Create and show a success notification.
Args:
title: Title of the notification
message: Content of the notification
**kwargs: Additional platform-specific parameters
Returns:
True if the notification was successfully displayed, False otherwise
"""
return self.notification_manager.show_notification(
title=title,
message=message,
notification_type=NotificationType.SUCCESS,
**kwargs
)
notification_factory = NotificationFactory()
def show_notification(title: str, message: str, notification_type: str = "info", **kwargs) -> bool:
"""
Show a notification with the specified parameters.
This is a convenience function for showing notifications without directly
interacting with the NotificationFactory or platform-specific notification classes.
Args:
title: Title of the notification
message: Content of the notification
notification_type: Type of notification ("info", "warning", "error", "success")
**kwargs: Additional platform-specific parameters
Returns:
True if the notification was successfully displayed, False otherwise
"""
try:
notification_type_enum = NotificationType(notification_type)
except ValueError:
logger.warning(f"Invalid notification type: {notification_type}, using INFO")
notification_type_enum = NotificationType.INFO
if notification_type_enum == NotificationType.INFO:
return notification_factory.create_info_notification(title, message, **kwargs)
elif notification_type_enum == NotificationType.WARNING:
return notification_factory.create_warning_notification(title, message, **kwargs)
elif notification_type_enum == NotificationType.ERROR:
return notification_factory.create_error_notification(title, message, **kwargs)
elif notification_type_enum == NotificationType.SUCCESS:
return notification_factory.create_success_notification(title, message, **kwargs)
return notification_factory.create_info_notification(title, message, **kwargs)
```
--------------------------------------------------------------------------------
/src/server/commands.py:
--------------------------------------------------------------------------------
```python
"""
Command processing system for toast-mcp-server.
This module handles the processing of commands received from clients,
including command registration, validation, and execution.
"""
import logging
import asyncio
from typing import Dict, Any, Optional, Callable, Awaitable, List, Union, Tuple
from src.mcp.protocol import (
MCPMessage, ResponseMessage, ErrorMessage, MessageType
)
from src.notification.toast import show_notification
logger = logging.getLogger(__name__)
CommandHandler = Callable[[Dict[str, Any], str], Awaitable[Tuple[bool, str, Optional[Dict[str, Any]]]]]
CommandValidator = Callable[[Dict[str, Any]], Tuple[bool, Optional[str]]]
class Command:
"""
Represents a command that can be executed by the server.
Commands are registered with the CommandProcessor and can be executed
when received from clients.
"""
def __init__(self,
name: str,
handler: CommandHandler,
validator: Optional[CommandValidator] = None,
description: str = "",
requires_auth: bool = False):
"""
Initialize a new command.
Args:
name: Name of the command
handler: Function to handle the command execution
validator: Optional function to validate command parameters
description: Description of the command
requires_auth: Whether the command requires authentication
"""
self.name = name
self.handler = handler
self.validator = validator
self.description = description
self.requires_auth = requires_auth
async def execute(self, params: Dict[str, Any], client_id: str) -> Tuple[bool, str, Optional[Dict[str, Any]]]:
"""
Execute the command.
Args:
params: Parameters for the command
client_id: ID of the client executing the command
Returns:
Tuple of (success, message, data)
"""
if self.validator:
is_valid, error = self.validator(params)
if not is_valid:
return False, f"Invalid parameters: {error}", None
try:
return await self.handler(params, client_id)
except Exception as e:
logger.error(f"Error executing command {self.name}: {str(e)}")
return False, f"Error executing command: {str(e)}", None
class CommandProcessor:
"""
Processes commands received from clients.
This class manages the registration and execution of commands,
as well as the handling of command responses.
"""
def __init__(self):
"""Initialize the command processor."""
self.commands: Dict[str, Command] = {}
self.authenticated_clients: List[str] = []
logger.info("Command processor initialized")
def register_command(self, command: Command) -> None:
"""
Register a command with the processor.
Args:
command: The command to register
"""
self.commands[command.name] = command
logger.debug(f"Registered command: {command.name}")
def register_commands(self, commands: List[Command]) -> None:
"""
Register multiple commands with the processor.
Args:
commands: List of commands to register
"""
for command in commands:
self.register_command(command)
async def process_command(self,
command_name: str,
params: Dict[str, Any],
client_id: str) -> Tuple[bool, str, Optional[Dict[str, Any]]]:
"""
Process a command.
Args:
command_name: Name of the command to execute
params: Parameters for the command
client_id: ID of the client executing the command
Returns:
Tuple of (success, message, data)
"""
if command_name not in self.commands:
return False, f"Unknown command: {command_name}", None
command = self.commands[command_name]
if command.requires_auth and client_id not in self.authenticated_clients:
return False, "Authentication required", None
return await command.execute(params, client_id)
def authenticate_client(self, client_id: str) -> None:
"""
Mark a client as authenticated.
Args:
client_id: ID of the client to authenticate
"""
if client_id not in self.authenticated_clients:
self.authenticated_clients.append(client_id)
logger.debug(f"Client authenticated: {client_id}")
def deauthenticate_client(self, client_id: str) -> None:
"""
Remove a client's authentication.
Args:
client_id: ID of the client to deauthenticate
"""
if client_id in self.authenticated_clients:
self.authenticated_clients.remove(client_id)
logger.debug(f"Client deauthenticated: {client_id}")
async def handle_show_notification(params: Dict[str, Any], client_id: str) -> Tuple[bool, str, Optional[Dict[str, Any]]]:
"""
Handle the 'show_notification' command.
Args:
params: Parameters for the command
client_id: ID of the client executing the command
Returns:
Tuple of (success, message, data)
"""
title = params.get("title", "")
message = params.get("message", "")
notification_type = params.get("type", "info")
duration = params.get("duration", 5)
success = show_notification(title, message, notification_type, duration)
if success:
return True, "Notification displayed successfully", None
else:
return False, "Failed to display notification", None
async def handle_list_commands(params: Dict[str, Any], client_id: str) -> Tuple[bool, str, Optional[Dict[str, Any]]]:
"""
Handle the 'list_commands' command.
Args:
params: Parameters for the command
client_id: ID of the client executing the command
Returns:
Tuple of (success, message, data)
"""
commands = [
{"name": "show_notification", "description": "Display a Windows 10 toast notification"},
{"name": "list_commands", "description": "List available commands"}
]
return True, "Commands retrieved successfully", {"commands": commands}
def validate_show_notification(params: Dict[str, Any]) -> Tuple[bool, Optional[str]]:
"""
Validate parameters for the 'show_notification' command.
Args:
params: Parameters to validate
Returns:
Tuple of (is_valid, error_message)
"""
if "title" not in params:
return False, "Missing required parameter: title"
if "message" not in params:
return False, "Missing required parameter: message"
if not isinstance(params.get("title"), str):
return False, "Title must be a string"
if not isinstance(params.get("message"), str):
return False, "Message must be a string"
if "type" in params and params["type"] not in ["info", "warning", "error", "success"]:
return False, "Invalid notification type"
if "duration" in params and not isinstance(params["duration"], int):
return False, "Duration must be an integer"
return True, None
DEFAULT_COMMANDS = [
Command(
name="show_notification",
handler=handle_show_notification,
validator=validate_show_notification,
description="Display a Windows 10 toast notification"
),
Command(
name="list_commands",
handler=handle_list_commands,
description="List available commands"
)
]
command_processor = CommandProcessor()
command_processor.register_commands(DEFAULT_COMMANDS)
async def process_command_message(message_data: Dict[str, Any], client_id: str) -> MCPMessage:
"""
Process a command message and return a response.
Args:
message_data: Dictionary containing the command message data
client_id: ID of the client sending the command
Returns:
Response message to send back to the client
"""
command_name = message_data.get("command")
params = message_data.get("params", {})
if not command_name:
return ErrorMessage(400, "Missing command name")
success, message, data = await command_processor.process_command(command_name, params, client_id)
if success:
return ResponseMessage(True, message, data)
else:
return ErrorMessage(400, message)
```
--------------------------------------------------------------------------------
/tests/test_connection.py:
--------------------------------------------------------------------------------
```python
"""
Tests for the client connection management implementation.
"""
import unittest
import asyncio
import json
from unittest.mock import patch, MagicMock, AsyncMock
from src.server.connection import (
ClientConnection, ConnectionManager, run_server
)
from src.mcp.protocol import (
MCPMessage, NotificationMessage, ResponseMessage, ErrorMessage,
PingMessage, PongMessage, MessageType, NotificationType
)
class TestClientConnection(unittest.TestCase):
"""Test cases for the ClientConnection class."""
def setUp(self):
"""Set up test fixtures."""
self.reader = AsyncMock(spec=asyncio.StreamReader)
self.writer = AsyncMock(spec=asyncio.StreamWriter)
self.server = MagicMock(spec=ConnectionManager)
self.writer.get_extra_info.return_value = ('127.0.0.1', 12345)
self.client = ClientConnection(
self.reader, self.writer, "test_client", self.server
)
@patch('src.server.connection.logger')
async def test_handle_connection(self, mock_logger):
"""Test handling a client connection."""
self.reader.readline.side_effect = [
b'{"type": "ping", "data": {}}\n',
b''
]
await self.client.handle()
self.reader.readline.assert_called()
self.assertFalse(self.client.connected)
self.server.remove_client.assert_called_once_with("test_client")
self.writer.close.assert_called_once()
self.writer.wait_closed.assert_called_once()
@patch('src.server.connection.logger')
async def test_process_message_ping(self, mock_logger):
"""Test processing a ping message."""
await self.client._process_message('{"type": "ping", "data": {}}')
self.writer.write.assert_called_once()
written_data = self.writer.write.call_args[0][0].decode()
self.assertIn('"type": "pong"', written_data)
@patch('src.server.connection.show_notification')
@patch('src.server.connection.logger')
async def test_process_message_notification(self, mock_logger, mock_show_notification):
"""Test processing a notification message."""
mock_show_notification.return_value = True
await self.client._process_message(
'{"type": "notification", "data": {"title": "Test", "message": "Test message"}}'
)
mock_show_notification.assert_called_once_with(
"Test", "Test message", "info", 5
)
self.writer.write.assert_called_once()
written_data = self.writer.write.call_args[0][0].decode()
self.assertIn('"type": "response"', written_data)
self.assertIn('"success": true', written_data)
@patch('src.server.connection.logger')
async def test_process_message_invalid_json(self, mock_logger):
"""Test processing an invalid JSON message."""
await self.client._process_message('invalid json')
self.writer.write.assert_called_once()
written_data = self.writer.write.call_args[0][0].decode()
self.assertIn('"type": "error"', written_data)
self.assertIn('"code": 400', written_data)
@patch('src.server.connection.logger')
async def test_send_message(self, mock_logger):
"""Test sending a message to the client."""
message = PingMessage()
await self.client.send_message(message)
self.writer.write.assert_called_once()
self.writer.drain.assert_called_once()
written_data = self.writer.write.call_args[0][0].decode()
self.assertIn('"type": "ping"', written_data)
@patch('src.server.connection.logger')
async def test_close(self, mock_logger):
"""Test closing the client connection."""
await self.client.close()
self.assertFalse(self.client.connected)
self.server.remove_client.assert_called_once_with("test_client")
self.writer.close.assert_called_once()
self.writer.wait_closed.assert_called_once()
self.server.remove_client.reset_mock()
self.writer.close.reset_mock()
self.writer.wait_closed.reset_mock()
await self.client.close()
self.server.remove_client.assert_not_called()
self.writer.close.assert_not_called()
self.writer.wait_closed.assert_not_called()
class TestConnectionManager(unittest.TestCase):
"""Test cases for the ConnectionManager class."""
def setUp(self):
"""Set up test fixtures."""
self.manager = ConnectionManager("127.0.0.1", 8765)
@patch('src.server.connection.asyncio.start_server')
@patch('src.server.connection.logger')
async def test_start_server(self, mock_logger, mock_start_server):
"""Test starting the server."""
mock_server = AsyncMock()
mock_server.sockets = [MagicMock()]
mock_server.sockets[0].getsockname.return_value = ('127.0.0.1', 8765)
mock_start_server.return_value = mock_server
task = asyncio.create_task(self.manager.start())
await asyncio.sleep(0.1)
task.cancel()
try:
await task
except asyncio.CancelledError:
pass
mock_start_server.assert_called_once_with(
self.manager._handle_new_connection, "127.0.0.1", 8765
)
@patch('src.server.connection.ClientConnection')
@patch('src.server.connection.asyncio.create_task')
@patch('src.server.connection.logger')
async def test_handle_new_connection(self, mock_logger, mock_create_task, mock_client_connection):
"""Test handling a new connection."""
reader = AsyncMock(spec=asyncio.StreamReader)
writer = AsyncMock(spec=asyncio.StreamWriter)
mock_client = MagicMock()
mock_client_connection.return_value = mock_client
await self.manager._handle_new_connection(reader, writer)
mock_client_connection.assert_called_once_with(
reader, writer, "client_1", self.manager
)
self.assertEqual(len(self.manager.clients), 1)
self.assertEqual(self.manager.clients["client_1"], mock_client)
mock_create_task.assert_called_once_with(mock_client.handle())
@patch('src.server.connection.logger')
def test_remove_client(self, mock_logger):
"""Test removing a client."""
self.manager.clients["test_client"] = MagicMock()
self.manager.remove_client("test_client")
self.assertEqual(len(self.manager.clients), 0)
self.manager.remove_client("non_existent_client")
@patch('src.server.connection.logger')
async def test_broadcast(self, mock_logger):
"""Test broadcasting a message to all clients."""
client1 = MagicMock()
client1.send_message = AsyncMock()
client2 = MagicMock()
client2.send_message = AsyncMock()
self.manager.clients["client1"] = client1
self.manager.clients["client2"] = client2
message = PingMessage()
await self.manager.broadcast(message)
client1.send_message.assert_called_once_with(message)
client2.send_message.assert_called_once_with(message)
client1.send_message.reset_mock()
client2.send_message.reset_mock()
await self.manager.broadcast(message, exclude="client1")
client1.send_message.assert_not_called()
client2.send_message.assert_called_once_with(message)
@patch('src.server.connection.logger')
async def test_stop(self, mock_logger):
"""Test stopping the server."""
client1 = MagicMock()
client1.close = AsyncMock()
client2 = MagicMock()
client2.close = AsyncMock()
self.manager.clients["client1"] = client1
self.manager.clients["client2"] = client2
self.manager.server = MagicMock()
self.manager.server.wait_closed = AsyncMock()
await self.manager.stop()
client1.close.assert_called_once()
client2.close.assert_called_once()
self.manager.server.close.assert_called_once()
self.manager.server.wait_closed.assert_called_once()
@patch('src.server.connection.ConnectionManager')
@patch('src.server.connection.logging')
async def test_run_server(mock_logging, mock_manager_class):
"""Test running the server."""
mock_manager = AsyncMock()
mock_manager_class.return_value = mock_manager
await run_server("127.0.0.1", 8765)
mock_logging.basicConfig.assert_called_once()
mock_manager_class.assert_called_once_with("127.0.0.1", 8765)
mock_manager.start.assert_called_once()
if __name__ == "__main__":
unittest.main()
```
--------------------------------------------------------------------------------
/tests/test_macos.py:
--------------------------------------------------------------------------------
```python
"""
Tests for the macOS Notification implementation.
"""
import unittest
from unittest.mock import patch, MagicMock
import platform
import subprocess
from src.notification.macos import (
MacOSNotificationManager, MacOSNotificationFactory,
show_macos_notification, NotificationType
)
class TestMacOSNotification(unittest.TestCase):
"""Test cases for the macOS Notification implementation."""
@patch('src.notification.macos.platform.system')
def test_is_macos(self, mock_system):
"""Test platform detection."""
mock_system.return_value = "Darwin"
manager = MacOSNotificationManager()
self.assertTrue(manager._is_macos())
mock_system.return_value = "Linux"
manager = MacOSNotificationManager()
self.assertFalse(manager._is_macos())
@patch('src.notification.macos.platform.system')
@patch('src.notification.macos.subprocess.run')
def test_show_notification(self, mock_run, mock_system):
"""Test showing a notification."""
mock_system.return_value = "Darwin"
mock_process = MagicMock()
mock_process.returncode = 0
mock_run.return_value = mock_process
manager = MacOSNotificationManager()
result = manager.show_notification(
title="Test Title",
message="Test Message",
notification_type=NotificationType.INFO
)
self.assertTrue(result)
mock_run.assert_called_once()
args, kwargs = mock_run.call_args
self.assertEqual(args[0][0], "osascript")
self.assertEqual(args[0][1], "-e")
self.assertIn("display notification", args[0][2])
self.assertIn("Test Message", args[0][2])
self.assertIn("Test Title", args[0][2])
@patch('src.notification.macos.platform.system')
@patch('src.notification.macos.subprocess.run')
def test_show_notification_with_subtitle(self, mock_run, mock_system):
"""Test showing a notification with a subtitle."""
mock_system.return_value = "Darwin"
mock_process = MagicMock()
mock_process.returncode = 0
mock_run.return_value = mock_process
manager = MacOSNotificationManager()
result = manager.show_notification(
title="Test Title",
message="Test Message",
subtitle="Test Subtitle",
notification_type=NotificationType.INFO
)
self.assertTrue(result)
mock_run.assert_called_once()
args, kwargs = mock_run.call_args
self.assertIn("subtitle", args[0][2])
self.assertIn("Test Subtitle", args[0][2])
@patch('src.notification.macos.platform.system')
@patch('src.notification.macos.subprocess.run')
def test_show_notification_without_sound(self, mock_run, mock_system):
"""Test showing a notification without sound."""
mock_system.return_value = "Darwin"
mock_process = MagicMock()
mock_process.returncode = 0
mock_run.return_value = mock_process
manager = MacOSNotificationManager()
result = manager.show_notification(
title="Test Title",
message="Test Message",
sound=False,
notification_type=NotificationType.INFO
)
self.assertTrue(result)
mock_run.assert_called_once()
args, kwargs = mock_run.call_args
self.assertNotIn("sound name", args[0][2])
@patch('src.notification.macos.platform.system')
@patch('src.notification.macos.subprocess.run')
def test_show_notification_on_non_macos(self, mock_run, mock_system):
"""Test showing a notification on a non-macOS platform."""
mock_system.return_value = "Linux"
manager = MacOSNotificationManager()
result = manager.show_notification(
title="Test Title",
message="Test Message",
notification_type=NotificationType.INFO
)
self.assertFalse(result)
mock_run.assert_not_called()
@patch('src.notification.macos.platform.system')
@patch('src.notification.macos.subprocess.run')
def test_show_notification_with_subprocess_error(self, mock_run, mock_system):
"""Test showing a notification with a subprocess error."""
mock_system.return_value = "Darwin"
mock_run.side_effect = subprocess.SubprocessError("Test error")
manager = MacOSNotificationManager()
result = manager.show_notification(
title="Test Title",
message="Test Message",
notification_type=NotificationType.INFO
)
self.assertFalse(result)
@patch('src.notification.macos.platform.system')
@patch('src.notification.macos.subprocess.run')
def test_show_notification_with_return_code_error(self, mock_run, mock_system):
"""Test showing a notification with a return code error."""
mock_system.return_value = "Darwin"
mock_process = MagicMock()
mock_process.returncode = 1
mock_process.stderr = "Test error"
mock_run.return_value = mock_process
manager = MacOSNotificationManager()
result = manager.show_notification(
title="Test Title",
message="Test Message",
notification_type=NotificationType.INFO
)
self.assertFalse(result)
@patch('src.notification.macos.MacOSNotificationManager')
def test_notification_factory(self, mock_manager_class):
"""Test the notification factory."""
mock_manager = MagicMock()
mock_manager_class.return_value = mock_manager
mock_manager.show_notification.return_value = True
factory = MacOSNotificationFactory()
result = factory.create_info_notification("Info Title", "Info Message", "Info Subtitle")
self.assertTrue(result)
mock_manager.show_notification.assert_called_with(
title="Info Title",
message="Info Message",
notification_type=NotificationType.INFO,
subtitle="Info Subtitle"
)
result = factory.create_warning_notification("Warning Title", "Warning Message")
self.assertTrue(result)
mock_manager.show_notification.assert_called_with(
title="Warning Title",
message="Warning Message",
notification_type=NotificationType.WARNING,
subtitle=None
)
result = factory.create_error_notification("Error Title", "Error Message")
self.assertTrue(result)
mock_manager.show_notification.assert_called_with(
title="Error Title",
message="Error Message",
notification_type=NotificationType.ERROR,
subtitle=None
)
result = factory.create_success_notification("Success Title", "Success Message")
self.assertTrue(result)
mock_manager.show_notification.assert_called_with(
title="Success Title",
message="Success Message",
notification_type=NotificationType.SUCCESS,
subtitle=None
)
@patch('src.notification.macos.macos_notification_factory')
def test_show_macos_notification_helper(self, mock_factory):
"""Test the show_macos_notification helper function."""
mock_factory.create_info_notification.return_value = True
mock_factory.create_warning_notification.return_value = True
mock_factory.create_error_notification.return_value = True
mock_factory.create_success_notification.return_value = True
result = show_macos_notification("Info Title", "Info Message", "info", "Info Subtitle")
self.assertTrue(result)
mock_factory.create_info_notification.assert_called_with("Info Title", "Info Message", "Info Subtitle")
result = show_macos_notification("Warning Title", "Warning Message", "warning")
self.assertTrue(result)
mock_factory.create_warning_notification.assert_called_with("Warning Title", "Warning Message", None)
result = show_macos_notification("Error Title", "Error Message", "error")
self.assertTrue(result)
mock_factory.create_error_notification.assert_called_with("Error Title", "Error Message", None)
result = show_macos_notification("Success Title", "Success Message", "success")
self.assertTrue(result)
mock_factory.create_success_notification.assert_called_with("Success Title", "Success Message", None)
result = show_macos_notification("Invalid Title", "Invalid Message", "invalid")
self.assertTrue(result)
mock_factory.create_info_notification.assert_called_with("Invalid Title", "Invalid Message", None)
if __name__ == "__main__":
unittest.main()
```
--------------------------------------------------------------------------------
/src/mcp/protocol.py:
--------------------------------------------------------------------------------
```python
"""
MCP (Model Context Protocol) implementation for toast-mcp-server.
This module defines the protocol specification and message handling for
communication between MCP clients and the notification server.
"""
import json
import logging
from enum import Enum
from typing import Dict, Any, Optional, List, Union
logger = logging.getLogger(__name__)
class MessageType(Enum):
"""Enum defining the types of messages in the MCP protocol."""
NOTIFICATION = "notification"
RESPONSE = "response"
ERROR = "error"
PING = "ping"
PONG = "pong"
class NotificationType(Enum):
"""Enum defining the types of notifications supported."""
INFO = "info"
WARNING = "warning"
ERROR = "error"
SUCCESS = "success"
class MCPMessage:
"""Base class for MCP protocol messages."""
def __init__(self, msg_type: MessageType, data: Dict[str, Any] = None):
"""
Initialize a new MCP message.
Args:
msg_type: The type of message
data: Optional data payload for the message
"""
self.msg_type = msg_type
self.data = data or {}
def to_dict(self) -> Dict[str, Any]:
"""Convert the message to a dictionary representation."""
return {
"type": self.msg_type.value,
"data": self.data
}
def to_json(self) -> str:
"""Convert the message to a JSON string."""
return json.dumps(self.to_dict())
@classmethod
def from_dict(cls, data: Dict[str, Any]) -> 'MCPMessage':
"""
Create a message from a dictionary.
Args:
data: Dictionary containing message data
Returns:
An MCPMessage instance
Raises:
ValueError: If the message type is invalid or required fields are missing
"""
if "type" not in data:
raise ValueError("Message missing required 'type' field")
try:
msg_type = MessageType(data["type"])
except ValueError:
raise ValueError(f"Invalid message type: {data['type']}")
msg_data = data.get("data", {})
if msg_type == MessageType.NOTIFICATION:
return NotificationMessage.from_dict(data)
elif msg_type == MessageType.RESPONSE:
return ResponseMessage.from_dict(data)
elif msg_type == MessageType.ERROR:
return ErrorMessage.from_dict(data)
elif msg_type == MessageType.PING:
return PingMessage()
elif msg_type == MessageType.PONG:
return PongMessage()
return cls(msg_type, msg_data)
@classmethod
def from_json(cls, json_str: str) -> 'MCPMessage':
"""
Create a message from a JSON string.
Args:
json_str: JSON string containing message data
Returns:
An MCPMessage instance
Raises:
ValueError: If the JSON is invalid or the message format is incorrect
"""
try:
data = json.loads(json_str)
except json.JSONDecodeError:
raise ValueError("Invalid JSON format")
return cls.from_dict(data)
class NotificationMessage(MCPMessage):
"""Message for sending notifications to the server."""
def __init__(self,
title: str,
message: str,
notification_type: NotificationType = NotificationType.INFO,
duration: int = 5,
client_id: str = None,
icon: str = None,
actions: List[Dict[str, str]] = None):
"""
Initialize a new notification message.
Args:
title: Title of the notification
message: Content of the notification
notification_type: Type of notification (info, warning, error, success)
duration: Duration to display the notification in seconds
client_id: Optional identifier for the client sending the notification
icon: Optional icon to display with the notification
actions: Optional list of actions that can be taken on the notification
"""
data = {
"title": title,
"message": message,
"notification_type": notification_type.value,
"duration": duration
}
if client_id:
data["client_id"] = client_id
if icon:
data["icon"] = icon
if actions:
data["actions"] = actions
super().__init__(MessageType.NOTIFICATION, data)
@classmethod
def from_dict(cls, data: Dict[str, Any]) -> 'NotificationMessage':
"""Create a NotificationMessage from a dictionary."""
msg_data = data.get("data", {})
if "title" not in msg_data or "message" not in msg_data:
raise ValueError("Notification message missing required fields: title and message")
try:
notification_type = NotificationType(msg_data.get("notification_type", "info"))
except ValueError:
notification_type = NotificationType.INFO
logger.warning(f"Invalid notification type: {msg_data.get('notification_type')}, using INFO")
return cls(
title=msg_data["title"],
message=msg_data["message"],
notification_type=notification_type,
duration=msg_data.get("duration", 5),
client_id=msg_data.get("client_id"),
icon=msg_data.get("icon"),
actions=msg_data.get("actions")
)
class ResponseMessage(MCPMessage):
"""Message for server responses to clients."""
def __init__(self, success: bool, message: str = None, data: Dict[str, Any] = None):
"""
Initialize a new response message.
Args:
success: Whether the operation was successful
message: Optional message describing the result
data: Optional additional data to include in the response
"""
response_data = {
"success": success
}
if message:
response_data["message"] = message
if data:
response_data["data"] = data
super().__init__(MessageType.RESPONSE, response_data)
@classmethod
def from_dict(cls, data: Dict[str, Any]) -> 'ResponseMessage':
"""Create a ResponseMessage from a dictionary."""
msg_data = data.get("data", {})
if "success" not in msg_data:
raise ValueError("Response message missing required field: success")
return cls(
success=msg_data["success"],
message=msg_data.get("message"),
data=msg_data.get("data")
)
class ErrorMessage(MCPMessage):
"""Message for error responses."""
def __init__(self, error_code: int, error_message: str, details: Any = None):
"""
Initialize a new error message.
Args:
error_code: Numeric error code
error_message: Human-readable error message
details: Optional additional error details
"""
error_data = {
"code": error_code,
"message": error_message
}
if details:
error_data["details"] = details
super().__init__(MessageType.ERROR, error_data)
@classmethod
def from_dict(cls, data: Dict[str, Any]) -> 'ErrorMessage':
"""Create an ErrorMessage from a dictionary."""
msg_data = data.get("data", {})
if "code" not in msg_data or "message" not in msg_data:
raise ValueError("Error message missing required fields: code and message")
return cls(
error_code=msg_data["code"],
error_message=msg_data["message"],
details=msg_data.get("details")
)
class PingMessage(MCPMessage):
"""Message for ping requests."""
def __init__(self):
"""Initialize a new ping message."""
super().__init__(MessageType.PING)
class PongMessage(MCPMessage):
"""Message for pong responses."""
def __init__(self):
"""Initialize a new pong message."""
super().__init__(MessageType.PONG)
def parse_message(data: Union[str, Dict[str, Any]]) -> MCPMessage:
"""
Parse a message from either a JSON string or a dictionary.
Args:
data: JSON string or dictionary containing message data
Returns:
An MCPMessage instance
Raises:
ValueError: If the message format is invalid
"""
if isinstance(data, str):
return MCPMessage.from_json(data)
elif isinstance(data, dict):
return MCPMessage.from_dict(data)
else:
raise ValueError(f"Unsupported message format: {type(data)}")
```
--------------------------------------------------------------------------------
/src/notification/macos.py:
--------------------------------------------------------------------------------
```python
"""
macOS Notification implementation for toast-mcp-server.
This module provides functionality to display macOS notifications
using the osascript command to interact with the macOS Notification Center.
"""
import logging
import subprocess
import platform
from typing import Dict, Any, Optional, List, Union
from src.mcp.protocol import NotificationType
logger = logging.getLogger(__name__)
class MacOSNotificationManager:
"""
Manager for macOS Notifications.
This class handles the creation and display of macOS notifications
using the osascript command to interact with the Notification Center.
"""
def __init__(self):
"""Initialize the macOS notification manager."""
if not self._is_macos():
logger.warning("MacOSNotificationManager initialized on non-macOS platform")
logger.info("macOS notification manager initialized")
def _is_macos(self) -> bool:
"""
Check if the current platform is macOS.
Returns:
True if the current platform is macOS, False otherwise
"""
return platform.system() == "Darwin"
def show_notification(self,
title: str,
message: str,
notification_type: NotificationType = NotificationType.INFO,
duration: int = 5,
sound: bool = True,
subtitle: Optional[str] = None) -> bool:
"""
Show a macOS notification.
Args:
title: Title of the notification
message: Content of the notification
notification_type: Type of notification (info, warning, error, success)
duration: Duration parameter is ignored on macOS (included for API compatibility)
sound: Whether to play a sound with the notification
subtitle: Optional subtitle for the notification
Returns:
True if the notification was successfully displayed, False otherwise
"""
if not self._is_macos():
logger.error("Cannot show macOS notification on non-macOS platform")
return False
logger.debug(f"Showing notification: {title} ({notification_type.value})")
try:
script = self._build_notification_script(title, message, subtitle, sound)
result = subprocess.run(
["osascript", "-e", script],
capture_output=True,
text=True,
check=True
)
if result.returncode == 0:
return True
else:
logger.error(f"Failed to show notification: {result.stderr}")
return False
except subprocess.SubprocessError as e:
logger.error(f"Failed to show notification: {str(e)}")
return False
except Exception as e:
logger.error(f"Unexpected error showing notification: {str(e)}")
return False
def _build_notification_script(self,
title: str,
message: str,
subtitle: Optional[str] = None,
sound: bool = True) -> str:
"""
Build the AppleScript command for displaying a notification.
Args:
title: Title of the notification
message: Content of the notification
subtitle: Optional subtitle for the notification
sound: Whether to play a sound with the notification
Returns:
AppleScript command string
"""
title_escaped = title.replace('"', '\\"')
message_escaped = message.replace('"', '\\"')
script = f'display notification "{message_escaped}" with title "{title_escaped}"'
if subtitle:
subtitle_escaped = subtitle.replace('"', '\\"')
script += f' subtitle "{subtitle_escaped}"'
if sound:
script += " sound name \"Ping\""
return script
class MacOSNotificationFactory:
"""
Factory for creating macOS notifications based on notification type.
This class provides methods for creating and displaying different types
of notifications with appropriate default settings.
"""
def __init__(self):
"""Initialize the notification factory."""
self.notification_manager = MacOSNotificationManager()
def create_info_notification(self, title: str, message: str, subtitle: Optional[str] = None) -> bool:
"""
Create and show an information notification.
Args:
title: Title of the notification
message: Content of the notification
subtitle: Optional subtitle for the notification
Returns:
True if the notification was successfully displayed, False otherwise
"""
return self.notification_manager.show_notification(
title=title,
message=message,
notification_type=NotificationType.INFO,
subtitle=subtitle
)
def create_warning_notification(self, title: str, message: str, subtitle: Optional[str] = None) -> bool:
"""
Create and show a warning notification.
Args:
title: Title of the notification
message: Content of the notification
subtitle: Optional subtitle for the notification
Returns:
True if the notification was successfully displayed, False otherwise
"""
return self.notification_manager.show_notification(
title=title,
message=message,
notification_type=NotificationType.WARNING,
subtitle=subtitle
)
def create_error_notification(self, title: str, message: str, subtitle: Optional[str] = None) -> bool:
"""
Create and show an error notification.
Args:
title: Title of the notification
message: Content of the notification
subtitle: Optional subtitle for the notification
Returns:
True if the notification was successfully displayed, False otherwise
"""
return self.notification_manager.show_notification(
title=title,
message=message,
notification_type=NotificationType.ERROR,
subtitle=subtitle
)
def create_success_notification(self, title: str, message: str, subtitle: Optional[str] = None) -> bool:
"""
Create and show a success notification.
Args:
title: Title of the notification
message: Content of the notification
subtitle: Optional subtitle for the notification
Returns:
True if the notification was successfully displayed, False otherwise
"""
return self.notification_manager.show_notification(
title=title,
message=message,
notification_type=NotificationType.SUCCESS,
subtitle=subtitle
)
macos_notification_factory = MacOSNotificationFactory()
def show_macos_notification(title: str, message: str, notification_type: str = "info", subtitle: Optional[str] = None) -> bool:
"""
Show a macOS notification with the specified parameters.
This is a convenience function for showing notifications without directly
interacting with the MacOSNotificationFactory or MacOSNotificationManager classes.
Args:
title: Title of the notification
message: Content of the notification
notification_type: Type of notification ("info", "warning", "error", "success")
subtitle: Optional subtitle for the notification
Returns:
True if the notification was successfully displayed, False otherwise
"""
try:
notification_type_enum = NotificationType(notification_type)
except ValueError:
logger.warning(f"Invalid notification type: {notification_type}, using INFO")
notification_type_enum = NotificationType.INFO
if notification_type_enum == NotificationType.INFO:
return macos_notification_factory.create_info_notification(title, message, subtitle)
elif notification_type_enum == NotificationType.WARNING:
return macos_notification_factory.create_warning_notification(title, message, subtitle)
elif notification_type_enum == NotificationType.ERROR:
return macos_notification_factory.create_error_notification(title, message, subtitle)
elif notification_type_enum == NotificationType.SUCCESS:
return macos_notification_factory.create_success_notification(title, message, subtitle)
return macos_notification_factory.create_info_notification(title, message, subtitle)
```
--------------------------------------------------------------------------------
/tests/test_manager.py:
--------------------------------------------------------------------------------
```python
"""
Tests for the unified notification manager.
"""
import unittest
from unittest.mock import patch, MagicMock
import platform
from src.notification.manager import (
NotificationManager, NotificationFactory,
show_notification, NotificationType
)
from src.notification.platform import is_windows, is_macos
class TestNotificationManager(unittest.TestCase):
"""Test cases for the NotificationManager class."""
@patch('src.notification.manager.get_platform_name')
@patch('src.notification.manager.is_windows')
@patch('src.notification.manager.is_macos')
def test_init_windows(self, mock_is_macos, mock_is_windows, mock_get_platform):
"""Test initializing the notification manager on Windows."""
mock_get_platform.return_value = "windows"
mock_is_windows.return_value = True
mock_is_macos.return_value = False
with patch('src.notification.manager.ToastNotificationManager') as mock_toast:
manager = NotificationManager()
mock_toast.assert_called_once()
@patch('src.notification.manager.get_platform_name')
@patch('src.notification.manager.is_windows')
@patch('src.notification.manager.is_macos')
def test_init_macos(self, mock_is_macos, mock_is_windows, mock_get_platform):
"""Test initializing the notification manager on macOS."""
mock_get_platform.return_value = "macos"
mock_is_windows.return_value = False
mock_is_macos.return_value = True
with patch('src.notification.manager.MacOSNotificationManager') as mock_macos:
manager = NotificationManager()
mock_macos.assert_called_once()
@patch('src.notification.manager.get_platform_name')
@patch('src.notification.manager.is_windows')
@patch('src.notification.manager.is_macos')
def test_init_unsupported(self, mock_is_macos, mock_is_windows, mock_get_platform):
"""Test initializing the notification manager on an unsupported platform."""
mock_get_platform.return_value = "linux"
mock_is_windows.return_value = False
mock_is_macos.return_value = False
manager = NotificationManager()
self.assertIsNone(manager._notification_system)
@patch('src.notification.manager.get_platform_name')
@patch('src.notification.manager.is_windows')
@patch('src.notification.manager.is_macos')
def test_show_notification_windows(self, mock_is_macos, mock_is_windows, mock_get_platform):
"""Test showing a notification on Windows."""
mock_get_platform.return_value = "windows"
mock_is_windows.return_value = True
mock_is_macos.return_value = False
mock_system = MagicMock()
mock_system.show_notification.return_value = True
with patch('src.notification.manager.ToastNotificationManager', return_value=mock_system):
manager = NotificationManager()
result = manager.show_notification(
title="Test Title",
message="Test Message",
notification_type=NotificationType.INFO,
duration=5,
icon_path="test.ico"
)
self.assertTrue(result)
mock_system.show_notification.assert_called_once_with(
title="Test Title",
message="Test Message",
notification_type=NotificationType.INFO,
duration=5,
icon_path="test.ico"
)
@patch('src.notification.manager.get_platform_name')
@patch('src.notification.manager.is_windows')
@patch('src.notification.manager.is_macos')
def test_show_notification_macos(self, mock_is_macos, mock_is_windows, mock_get_platform):
"""Test showing a notification on macOS."""
mock_get_platform.return_value = "macos"
mock_is_windows.return_value = False
mock_is_macos.return_value = True
mock_system = MagicMock()
mock_system.show_notification.return_value = True
with patch('src.notification.manager.MacOSNotificationManager', return_value=mock_system):
manager = NotificationManager()
result = manager.show_notification(
title="Test Title",
message="Test Message",
notification_type=NotificationType.INFO,
duration=5,
subtitle="Test Subtitle",
sound=True
)
self.assertTrue(result)
mock_system.show_notification.assert_called_once_with(
title="Test Title",
message="Test Message",
notification_type=NotificationType.INFO,
duration=5,
subtitle="Test Subtitle",
sound=True
)
@patch('src.notification.manager.get_platform_name')
@patch('src.notification.manager.is_windows')
@patch('src.notification.manager.is_macos')
def test_show_notification_unsupported(self, mock_is_macos, mock_is_windows, mock_get_platform):
"""Test showing a notification on an unsupported platform."""
mock_get_platform.return_value = "linux"
mock_is_windows.return_value = False
mock_is_macos.return_value = False
manager = NotificationManager()
result = manager.show_notification(
title="Test Title",
message="Test Message",
notification_type=NotificationType.INFO
)
self.assertFalse(result)
@patch('src.notification.manager.get_platform_name')
@patch('src.notification.manager.is_windows')
@patch('src.notification.manager.is_macos')
def test_show_notification_exception(self, mock_is_macos, mock_is_windows, mock_get_platform):
"""Test showing a notification that raises an exception."""
mock_get_platform.return_value = "windows"
mock_is_windows.return_value = True
mock_is_macos.return_value = False
mock_system = MagicMock()
mock_system.show_notification.side_effect = Exception("Test exception")
with patch('src.notification.manager.ToastNotificationManager', return_value=mock_system):
manager = NotificationManager()
result = manager.show_notification(
title="Test Title",
message="Test Message",
notification_type=NotificationType.INFO
)
self.assertFalse(result)
class TestNotificationFactory(unittest.TestCase):
"""Test cases for the NotificationFactory class."""
@patch('src.notification.manager.NotificationManager')
def test_notification_factory(self, mock_manager_class):
"""Test the notification factory."""
mock_manager = MagicMock()
mock_manager_class.return_value = mock_manager
mock_manager.show_notification.return_value = True
factory = NotificationFactory()
result = factory.create_info_notification("Info Title", "Info Message", param="value")
self.assertTrue(result)
mock_manager.show_notification.assert_called_with(
title="Info Title",
message="Info Message",
notification_type=NotificationType.INFO,
param="value"
)
result = factory.create_warning_notification("Warning Title", "Warning Message")
self.assertTrue(result)
mock_manager.show_notification.assert_called_with(
title="Warning Title",
message="Warning Message",
notification_type=NotificationType.WARNING
)
result = factory.create_error_notification("Error Title", "Error Message")
self.assertTrue(result)
mock_manager.show_notification.assert_called_with(
title="Error Title",
message="Error Message",
notification_type=NotificationType.ERROR
)
result = factory.create_success_notification("Success Title", "Success Message")
self.assertTrue(result)
mock_manager.show_notification.assert_called_with(
title="Success Title",
message="Success Message",
notification_type=NotificationType.SUCCESS
)
@patch('src.notification.manager.notification_factory')
class TestShowNotification(unittest.TestCase):
"""Test cases for the show_notification helper function."""
def test_show_notification(self, mock_factory):
"""Test the show_notification helper function."""
mock_factory.create_info_notification.return_value = True
mock_factory.create_warning_notification.return_value = True
mock_factory.create_error_notification.return_value = True
mock_factory.create_success_notification.return_value = True
result = show_notification("Info Title", "Info Message", "info", param="value")
self.assertTrue(result)
mock_factory.create_info_notification.assert_called_with("Info Title", "Info Message", param="value")
result = show_notification("Warning Title", "Warning Message", "warning")
self.assertTrue(result)
mock_factory.create_warning_notification.assert_called_with("Warning Title", "Warning Message")
result = show_notification("Error Title", "Error Message", "error")
self.assertTrue(result)
mock_factory.create_error_notification.assert_called_with("Error Title", "Error Message")
result = show_notification("Success Title", "Success Message", "success")
self.assertTrue(result)
mock_factory.create_success_notification.assert_called_with("Success Title", "Success Message")
result = show_notification("Invalid Title", "Invalid Message", "invalid")
self.assertTrue(result)
mock_factory.create_info_notification.assert_called_with("Invalid Title", "Invalid Message")
if __name__ == "__main__":
unittest.main()
```
--------------------------------------------------------------------------------
/src/server/connection.py:
--------------------------------------------------------------------------------
```python
"""
Client connection management for toast-mcp-server.
This module handles client connections, including connection establishment,
message handling, and connection termination.
"""
import asyncio
import logging
import json
from typing import Dict, Set, Any, Optional, Callable, Awaitable, List
from src.mcp.protocol import (
MCPMessage, NotificationMessage, ResponseMessage, ErrorMessage,
PingMessage, PongMessage, MessageType, parse_message
)
from src.mcp.validation import validate_message_format
from src.notification.toast import show_notification
logger = logging.getLogger(__name__)
class ClientConnection:
"""
Represents a connection to a client.
This class handles the communication with a single client, including
receiving and sending messages.
"""
def __init__(self, reader: asyncio.StreamReader, writer: asyncio.StreamWriter,
client_id: str, server: 'ConnectionManager'):
"""
Initialize a new client connection.
Args:
reader: Stream reader for receiving data from the client
writer: Stream writer for sending data to the client
client_id: Unique identifier for the client
server: Reference to the connection manager
"""
self.reader = reader
self.writer = writer
self.client_id = client_id
self.server = server
self.connected = True
self.addr = writer.get_extra_info('peername')
logger.info(f"Client connected: {self.client_id} from {self.addr}")
async def handle(self) -> None:
"""
Handle the client connection.
This method continuously reads messages from the client and processes them
until the connection is closed.
"""
try:
while self.connected:
data = await self.reader.readline()
if not data:
break
await self._process_message(data.decode().strip())
except asyncio.CancelledError:
logger.info(f"Connection handling cancelled for client: {self.client_id}")
except Exception as e:
logger.error(f"Error handling client {self.client_id}: {str(e)}")
finally:
await self.close()
async def _process_message(self, data: str) -> None:
"""
Process a message received from the client.
Args:
data: JSON string containing the message data
"""
logger.debug(f"Received message from {self.client_id}: {data}")
try:
message_data = json.loads(data)
is_valid, error, msg_type = validate_message_format(message_data)
if not is_valid:
await self.send_error(400, f"Invalid message format: {error}")
return
if msg_type == MessageType.NOTIFICATION:
await self._handle_notification(message_data)
elif msg_type == MessageType.PING:
await self._handle_ping()
else:
await self.send_error(400, f"Unsupported message type: {msg_type.value}")
except json.JSONDecodeError:
await self.send_error(400, "Invalid JSON format")
except Exception as e:
logger.error(f"Error processing message from {self.client_id}: {str(e)}")
await self.send_error(500, "Internal server error")
async def _handle_notification(self, message_data: Dict[str, Any]) -> None:
"""
Handle a notification message.
Args:
message_data: Dictionary containing the notification message data
"""
try:
notification = NotificationMessage.from_dict(message_data)
title = notification.data["title"]
message = notification.data["message"]
notification_type = notification.data.get("notification_type", "info")
duration = notification.data.get("duration", 5)
success = show_notification(title, message, notification_type, duration)
if success:
await self.send_response(True, "Notification displayed successfully")
else:
await self.send_response(False, "Failed to display notification")
except Exception as e:
logger.error(f"Error handling notification from {self.client_id}: {str(e)}")
await self.send_error(500, f"Error handling notification: {str(e)}")
async def _handle_ping(self) -> None:
"""Handle a ping message by sending a pong response."""
try:
pong = PongMessage()
await self.send_message(pong)
except Exception as e:
logger.error(f"Error handling ping from {self.client_id}: {str(e)}")
await self.send_error(500, f"Error handling ping: {str(e)}")
async def send_message(self, message: MCPMessage) -> None:
"""
Send a message to the client.
Args:
message: The message to send
"""
try:
json_str = message.to_json() + "\n"
self.writer.write(json_str.encode())
await self.writer.drain()
logger.debug(f"Sent message to {self.client_id}: {message.msg_type.value}")
except Exception as e:
logger.error(f"Error sending message to {self.client_id}: {str(e)}")
await self.close()
async def send_response(self, success: bool, message: str = None, data: Dict[str, Any] = None) -> None:
"""
Send a response message to the client.
Args:
success: Whether the operation was successful
message: Optional message describing the result
data: Optional additional data to include in the response
"""
response = ResponseMessage(success, message, data)
await self.send_message(response)
async def send_error(self, error_code: int, error_message: str, details: Any = None) -> None:
"""
Send an error message to the client.
Args:
error_code: Numeric error code
error_message: Human-readable error message
details: Optional additional error details
"""
error = ErrorMessage(error_code, error_message, details)
await self.send_message(error)
async def close(self) -> None:
"""Close the client connection."""
if not self.connected:
return
self.connected = False
try:
self.writer.close()
await self.writer.wait_closed()
self.server.remove_client(self.client_id)
logger.info(f"Client disconnected: {self.client_id}")
except Exception as e:
logger.error(f"Error closing connection for {self.client_id}: {str(e)}")
class ConnectionManager:
"""
Manages client connections to the server.
This class handles the creation and management of client connections,
including accepting new connections and broadcasting messages to clients.
"""
def __init__(self, host: str = "127.0.0.1", port: int = 8765):
"""
Initialize the connection manager.
Args:
host: Host address to bind to
port: Port to listen on
"""
self.host = host
self.port = port
self.clients: Dict[str, ClientConnection] = {}
self.server = None
self.next_client_id = 1
logger.info(f"Connection manager initialized with host={host}, port={port}")
async def start(self) -> None:
"""Start the server and begin accepting connections."""
try:
self.server = await asyncio.start_server(
self._handle_new_connection, self.host, self.port
)
addr = self.server.sockets[0].getsockname()
logger.info(f"Server started on {addr}")
async with self.server:
await self.server.serve_forever()
except Exception as e:
logger.error(f"Error starting server: {str(e)}")
raise
async def _handle_new_connection(self, reader: asyncio.StreamReader, writer: asyncio.StreamWriter) -> None:
"""
Handle a new client connection.
Args:
reader: Stream reader for receiving data from the client
writer: Stream writer for sending data to the client
"""
client_id = f"client_{self.next_client_id}"
self.next_client_id += 1
client = ClientConnection(reader, writer, client_id, self)
self.clients[client_id] = client
asyncio.create_task(client.handle())
def remove_client(self, client_id: str) -> None:
"""
Remove a client from the connection manager.
Args:
client_id: ID of the client to remove
"""
if client_id in self.clients:
del self.clients[client_id]
logger.debug(f"Removed client: {client_id}")
async def broadcast(self, message: MCPMessage, exclude: Optional[str] = None) -> None:
"""
Broadcast a message to all connected clients.
Args:
message: The message to broadcast
exclude: Optional client ID to exclude from the broadcast
"""
for client_id, client in list(self.clients.items()):
if exclude and client_id == exclude:
continue
try:
await client.send_message(message)
except Exception as e:
logger.error(f"Error broadcasting to {client_id}: {str(e)}")
async def stop(self) -> None:
"""Stop the server and close all client connections."""
logger.info("Stopping server...")
for client_id, client in list(self.clients.items()):
try:
await client.close()
except Exception as e:
logger.error(f"Error closing client {client_id}: {str(e)}")
if self.server:
self.server.close()
await self.server.wait_closed()
logger.info("Server stopped")
async def run_server(host: str = "127.0.0.1", port: int = 8765) -> None:
"""
Run the MCP server.
Args:
host: Host address to bind to
port: Port to listen on
"""
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
manager = ConnectionManager(host, port)
await manager.start()
```
--------------------------------------------------------------------------------
/tests/test_commands.py:
--------------------------------------------------------------------------------
```python
"""
Tests for the command processing system.
"""
import unittest
import asyncio
from unittest.mock import patch, MagicMock, AsyncMock
from src.server.commands import (
Command, CommandProcessor, process_command_message,
handle_show_notification, handle_list_commands,
validate_show_notification, DEFAULT_COMMANDS
)
from src.mcp.protocol import (
ResponseMessage, ErrorMessage
)
class TestCommand(unittest.TestCase):
"""Test cases for the Command class."""
async def test_execute_command(self):
"""Test executing a command."""
mock_handler = AsyncMock()
mock_handler.return_value = (True, "Success", {"data": "value"})
command = Command(
name="test_command",
handler=mock_handler,
description="Test command"
)
success, message, data = await command.execute({}, "test_client")
mock_handler.assert_called_once_with({}, "test_client")
self.assertTrue(success)
self.assertEqual(message, "Success")
self.assertEqual(data, {"data": "value"})
async def test_execute_command_with_validator(self):
"""Test executing a command with a validator."""
mock_handler = AsyncMock()
mock_handler.return_value = (True, "Success", {"data": "value"})
mock_validator = MagicMock()
mock_validator.return_value = (True, None)
command = Command(
name="test_command",
handler=mock_handler,
validator=mock_validator,
description="Test command"
)
success, message, data = await command.execute({"param": "value"}, "test_client")
mock_validator.assert_called_once_with({"param": "value"})
mock_handler.assert_called_once_with({"param": "value"}, "test_client")
self.assertTrue(success)
self.assertEqual(message, "Success")
self.assertEqual(data, {"data": "value"})
async def test_execute_command_with_invalid_params(self):
"""Test executing a command with invalid parameters."""
mock_handler = AsyncMock()
mock_validator = MagicMock()
mock_validator.return_value = (False, "Invalid parameter")
command = Command(
name="test_command",
handler=mock_handler,
validator=mock_validator,
description="Test command"
)
success, message, data = await command.execute({}, "test_client")
mock_validator.assert_called_once_with({})
mock_handler.assert_not_called()
self.assertFalse(success)
self.assertEqual(message, "Invalid parameters: Invalid parameter")
self.assertIsNone(data)
async def test_execute_command_with_exception(self):
"""Test executing a command that raises an exception."""
mock_handler = AsyncMock()
mock_handler.side_effect = Exception("Test exception")
command = Command(
name="test_command",
handler=mock_handler,
description="Test command"
)
success, message, data = await command.execute({}, "test_client")
mock_handler.assert_called_once_with({}, "test_client")
self.assertFalse(success)
self.assertEqual(message, "Error executing command: Test exception")
self.assertIsNone(data)
class TestCommandProcessor(unittest.TestCase):
"""Test cases for the CommandProcessor class."""
def setUp(self):
"""Set up test fixtures."""
self.processor = CommandProcessor()
def test_register_command(self):
"""Test registering a command."""
command = Command(
name="test_command",
handler=AsyncMock(),
description="Test command"
)
self.processor.register_command(command)
self.assertIn("test_command", self.processor.commands)
self.assertEqual(self.processor.commands["test_command"], command)
def test_register_commands(self):
"""Test registering multiple commands."""
command1 = Command(
name="command1",
handler=AsyncMock(),
description="Command 1"
)
command2 = Command(
name="command2",
handler=AsyncMock(),
description="Command 2"
)
self.processor.register_commands([command1, command2])
self.assertIn("command1", self.processor.commands)
self.assertIn("command2", self.processor.commands)
self.assertEqual(self.processor.commands["command1"], command1)
self.assertEqual(self.processor.commands["command2"], command2)
async def test_process_command(self):
"""Test processing a command."""
mock_handler = AsyncMock()
mock_handler.return_value = (True, "Success", {"data": "value"})
command = Command(
name="test_command",
handler=mock_handler,
description="Test command"
)
self.processor.register_command(command)
success, message, data = await self.processor.process_command(
"test_command", {"param": "value"}, "test_client"
)
mock_handler.assert_called_once_with({"param": "value"}, "test_client")
self.assertTrue(success)
self.assertEqual(message, "Success")
self.assertEqual(data, {"data": "value"})
async def test_process_unknown_command(self):
"""Test processing an unknown command."""
success, message, data = await self.processor.process_command(
"unknown_command", {}, "test_client"
)
self.assertFalse(success)
self.assertEqual(message, "Unknown command: unknown_command")
self.assertIsNone(data)
async def test_process_command_requiring_auth(self):
"""Test processing a command that requires authentication."""
mock_handler = AsyncMock()
command = Command(
name="auth_command",
handler=mock_handler,
description="Auth command",
requires_auth=True
)
self.processor.register_command(command)
success, message, data = await self.processor.process_command(
"auth_command", {}, "test_client"
)
mock_handler.assert_not_called()
self.assertFalse(success)
self.assertEqual(message, "Authentication required")
self.assertIsNone(data)
self.processor.authenticate_client("test_client")
success, message, data = await self.processor.process_command(
"auth_command", {}, "test_client"
)
mock_handler.assert_called_once_with({}, "test_client")
def test_authenticate_deauthenticate_client(self):
"""Test authenticating and deauthenticating a client."""
self.processor.authenticate_client("test_client")
self.assertIn("test_client", self.processor.authenticated_clients)
self.processor.authenticate_client("test_client")
self.assertEqual(self.processor.authenticated_clients.count("test_client"), 1)
self.processor.deauthenticate_client("test_client")
self.assertNotIn("test_client", self.processor.authenticated_clients)
self.processor.deauthenticate_client("non_existent_client")
class TestCommandHandlers(unittest.TestCase):
"""Test cases for the command handlers."""
@patch('src.server.commands.show_notification')
async def test_handle_show_notification(self, mock_show_notification):
"""Test the show_notification command handler."""
mock_show_notification.return_value = True
success, message, data = await handle_show_notification(
{
"title": "Test Title",
"message": "Test Message",
"type": "info",
"duration": 5
},
"test_client"
)
mock_show_notification.assert_called_once_with(
"Test Title", "Test Message", "info", 5
)
self.assertTrue(success)
self.assertEqual(message, "Notification displayed successfully")
self.assertIsNone(data)
mock_show_notification.reset_mock()
mock_show_notification.return_value = False
success, message, data = await handle_show_notification(
{
"title": "Test Title",
"message": "Test Message"
},
"test_client"
)
self.assertFalse(success)
self.assertEqual(message, "Failed to display notification")
self.assertIsNone(data)
async def test_handle_list_commands(self):
"""Test the list_commands command handler."""
success, message, data = await handle_list_commands({}, "test_client")
self.assertTrue(success)
self.assertEqual(message, "Commands retrieved successfully")
self.assertIsInstance(data, dict)
self.assertIn("commands", data)
self.assertIsInstance(data["commands"], list)
self.assertTrue(len(data["commands"]) > 0)
def test_validate_show_notification(self):
"""Test the show_notification validator."""
is_valid, error = validate_show_notification({
"title": "Test Title",
"message": "Test Message",
"type": "info",
"duration": 5
})
self.assertTrue(is_valid)
self.assertIsNone(error)
is_valid, error = validate_show_notification({
"message": "Test Message"
})
self.assertFalse(is_valid)
self.assertEqual(error, "Missing required parameter: title")
is_valid, error = validate_show_notification({
"title": "Test Title"
})
self.assertFalse(is_valid)
self.assertEqual(error, "Missing required parameter: message")
is_valid, error = validate_show_notification({
"title": "Test Title",
"message": "Test Message",
"type": "invalid"
})
self.assertFalse(is_valid)
self.assertEqual(error, "Invalid notification type")
is_valid, error = validate_show_notification({
"title": "Test Title",
"message": "Test Message",
"duration": "5" # Should be an integer
})
self.assertFalse(is_valid)
self.assertEqual(error, "Duration must be an integer")
@patch('src.server.commands.command_processor')
class TestProcessCommandMessage(unittest.TestCase):
"""Test cases for the process_command_message function."""
async def test_process_command_message(self, mock_processor):
"""Test processing a command message."""
mock_processor.process_command = AsyncMock()
mock_processor.process_command.return_value = (True, "Success", {"data": "value"})
message = await process_command_message(
{
"command": "test_command",
"params": {"param": "value"}
},
"test_client"
)
mock_processor.process_command.assert_called_once_with(
"test_command", {"param": "value"}, "test_client"
)
self.assertIsInstance(message, ResponseMessage)
self.assertTrue(message.data["success"])
self.assertEqual(message.data["message"], "Success")
self.assertEqual(message.data["data"], {"data": "value"})
async def test_process_command_message_failure(self, mock_processor):
"""Test processing a command message that fails."""
mock_processor.process_command = AsyncMock()
mock_processor.process_command.return_value = (False, "Failure", None)
message = await process_command_message(
{
"command": "test_command",
"params": {"param": "value"}
},
"test_client"
)
self.assertIsInstance(message, ErrorMessage)
self.assertEqual(message.data["code"], 400)
self.assertEqual(message.data["message"], "Failure")
async def test_process_command_message_missing_command(self, mock_processor):
"""Test processing a command message with a missing command name."""
message = await process_command_message(
{
"params": {"param": "value"}
},
"test_client"
)
mock_processor.process_command.assert_not_called()
self.assertIsInstance(message, ErrorMessage)
self.assertEqual(message.data["code"], 400)
self.assertEqual(message.data["message"], "Missing command name")
class TestDefaultCommands(unittest.TestCase):
"""Test cases for the default commands."""
def test_default_commands(self):
"""Test that default commands are defined."""
self.assertIsInstance(DEFAULT_COMMANDS, list)
self.assertTrue(len(DEFAULT_COMMANDS) > 0)
for command in DEFAULT_COMMANDS:
self.assertIsInstance(command, Command)
if __name__ == "__main__":
unittest.main()
```