MCP Notify Server

Instructions

Implementation Reference

• src/mcp_server_notify/server.py:118-159 (handler)

  Core handler function that sends the system notification using Apprise (cross-platform notifier), falling back to plyer library, logging/printing as last resort, and plays sound if requested using SoundPlayer.
  def _send_notification(self, request: NotificationRequest): logger.debug("request: %s", request) notification_sent = False # 尝试使用 Apprise 发送 if self.apprise and len(self.apprise) > 0: try: logger.debug("尝试使用 Apprise 发送通知...") result = self.apprise.notify(title=request.title, body=request.message) if result: logger.info("Apprise 通知发送成功") notification_sent = True else: logger.warning("Apprise 通知发送失败") except Exception as e: logger.error(f"Apprise 通知异常: {str(e)}") # 如果 Apprise 失败，尝试 plyer if not notification_sent: try: logger.debug("尝试使用 plyer 发送通知...") notification.notify( title=request.title, message=request.message, timeout=request.timeout, app_name="mcp-notification", ) logger.info("plyer 通知发送成功") notification_sent = True except Exception as e: logger.error(f"plyer 通知失败: {str(e)}") # 如果两种方法都失败或在容器环境中，使用日志输出 if not notification_sent or os.path.exists("/.dockerenv"): logger.info(f"通知：{request.title} - {request.message}") print(f"Notification: {request.title} - {request.message}") # 播放声音 if request.play_sound: logger.debug("播放通知声音...") self.sound_player.play()
• src/mcp_server_notify/schemas.py:3-17 (schema)

  Pydantic model defining the input schema for the 'send_notification' tool, including title (str), message (str), play_sound (bool, default True), and timeout (int, default 60).
  class NotificationRequest(BaseModel): """Request schema for sending a system notification title: str - Notification title message: str - Notification message play_sound: bool - Whether to play a sound timeout: int - Notification timeout in seconds """ title: str message: str play_sound: bool = True timeout: int = 60 # seconds
• src/mcp_server_notify/server.py:80-86 (registration)

  Registers the 'send_notification' tool in the MCP server's list_tools() method, providing name, description, and input schema.
  return [ Tool( name="send_notification", description="Send system notification with optional sound", inputSchema=NotificationRequest.model_json_schema(), ) ]
• src/mcp_server_notify/server.py:88-112 (handler)

  MCP server tool dispatcher that validates input using NotificationRequest schema, calls the _send_notification handler if name matches, and returns success/error TextContent.
  @self.server.call_tool() async def call_tool(name: str, arguments: dict) -> list[TextContent]: """Call a tool name: str - Tool name arguments: dict - Tool arguments """ logger.info(f"Calling tool: {name} with arguments: {arguments}") if name != "send_notification": return [TextContent(type="text", text="Invalid tool name")] try: logger.debug("Validating request...") req = NotificationRequest(**arguments) logger.debug("Sending notification...") self._send_notification(req) logger.info("Notification sent successfully") return [TextContent(type="text", text="Notification sent successfully")] except ValidationError as e: logger.error(f"Validation error: {str(e)}") return [TextContent(type="text", text=f"Invalid request: {str(e)}")] except Exception as e: logger.error(f"Error: {str(e)}") return [TextContent(type="text", text=f"Error: {str(e)}")]
• src/mcp_server_notify/sound.py:11-114 (helper)

  Cross-platform SoundPlayer class used by the handler to play notification sounds, supporting Windows (winsound), macOS (afplay), Linux (paplay/aplay), and fallback (pygame).
  class SoundPlayer: # 这是一个简短的WAV格式"叮"声音的base64编码 _DEFAULT_SOUND = """ UklGRnQFAABXQVZFZm10IBAAAAABAAEARKwAAIhYAQACABAAZGF0YVAFAACAjYyLhoWHioyQlZyp tcDJ0dbd5+rs7erm3tfRyMG5sKeekIeBfHNrW0g6KRoQBvLj1se1pJR/Z043GAT27OfgzLSLYkMl BuzPsotqUSIMx6V+VTYVAdy0ilcxEwLZsoRNKxAJ78ulcD4hFR7+4L+ATC0cKTcvGQT03MLAqYFU PUBOTzQQ58Sjg2VMWGpjSiYR++fo5NW7mXBVRDE0NyIL8efn29CuimRPOhwbJiUN8ebg1MazmX1r YVlORDcpGQkB/P/9+O7hyLamlYd5Z1hJNyYWCf7s28vAuLO0tLS1tLW4vcPK0NXZ2dnY1dPQz87P 0dPW2Nzg4+Xp6+3u7/Dx8vLz8/T19fb19PPy8vLy8vP09fb3+Pn6+vv7+/v7+/v7+/v7+/v7+vr5 +Pj39/f39/f3+Pn5+vr7+/z8/fz8/Pz8+/v7+/v7+/v7+/v7+/z8/Pz8/Pz8/Pz8/Pz8/Pz8/f39 /f39/f7+/v///////////////////////////////////////wAAAAAAAAAAAAAAAAAAAAAAAAAA AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA AAAAAA== """ def __init__(self): self.system = platform.system() self.sound_data = base64.b64decode(self._DEFAULT_SOUND.strip()) def play(self): """跨平台播放声音""" try: if self.system == 'Windows': logger.debug("Playing sound on Windows...") self._play_windows() elif self.system == 'Darwin': logger.debug("Playing sound on MacOS...") self._play_macos() elif self.system == 'Linux': logger.debug("Playing sound on Linux...") if os.path.exists('/.dockerenv'): logger.debug("Playing sound on Linux in container...") self._play_linux_simple() else: logger.debug("Playing sound on Linux...") self._play_linux() else: logger.debug("Playing sound on fallback...") self._play_fallback() except Exception as e: logger.error(f"Sound playback failed: {str(e)}") print(f"Sound playback failed: {str(e)}") def _play_windows(self): import winsound with tempfile.NamedTemporaryFile(suffix='.wav', delete=False) as f: f.write(self.sound_data) f.close() winsound.PlaySound(f.name, winsound.SND_FILENAME) os.unlink(f.name) def _play_macos(self): with tempfile.NamedTemporaryFile(suffix='.wav') as f: f.write(self.sound_data) f.flush() os.system(f'afplay "{f.name}"') def _play_linux(self): players = ['paplay', 'aplay', 'play'] with tempfile.NamedTemporaryFile(suffix='.wav', delete=False) as f: try: f.write(self.sound_data) f.flush() for player in players: if os.system(f'which {player} > /dev/null 2>&1') == 0: logger.debug(f"Playing sound with {player}...") os.system(f'{player} "{f.name}"') break finally: try: os.unlink(f.name) except: pass def _play_linux_simple(self): with tempfile.NamedTemporaryFile(suffix='.wav') as f: f.write(self.sound_data) f.flush() os.system(f'aplay "{f.name}" 2>/dev/null') def _play_fallback(self): try: import pygame pygame.mixer.init() sound = pygame.mixer.Sound(io.BytesIO(self.sound_data)) logger.debug("Playing sound with pygame...") sound.play() time.sleep(1) except ImportError: logger.error("No sound playback available")