diff --git a/src/scitex/notification/README.md b/src/scitex/notification/README.md
deleted file mode 100644
index 138f725cc..000000000
--- a/src/scitex/notification/README.md
+++ /dev/null
@@ -1,205 +0,0 @@
-# scitex.notify
-
-Multi-backend notification system for SciTeX. Sends alerts via audio (TTS), phone calls (Twilio), SMS, email, desktop notifications, Emacs, browser popups, and webhooks.
-
-## Quick Start
-
-```python
-import scitex
-
-# Alert with automatic fallback (audio -> emacs -> matplotlib -> email)
-scitex.notify.alert("Task complete!")
-
-# Phone call via Twilio
-scitex.notify.call("Wake up! Your experiment finished.")
-
-# SMS via Twilio
-scitex.notify.sms("Build finished!")
-
-# Specify backend explicitly
-scitex.notify.alert("Error in pipeline", backend="email", level="error")
-
-# Multiple backends
-scitex.notify.alert("Critical failure", backend=["audio", "email", "twilio"])
-
-# Check available backends
-scitex.notify.available_backends()
-# ['audio', 'emacs', 'twilio', ...]
-```
-
-## Backends
-
-| Backend | Description | Requirements |
-|---------|-------------|--------------|
-| `audio` | Text-to-Speech | `scitex-audio` package |
-| `twilio` | Phone call / SMS | Twilio account + env vars |
-| `email` | SMTP email | SMTP server config |
-| `emacs` | Minibuffer message | Running Emacs server |
-| `desktop` | System notification | Windows/macOS |
-| `matplotlib` | Visual popup | `matplotlib` |
-| `playwright` | Browser popup | `playwright` |
-| `webhook` | HTTP POST | Webhook URL |
-
-## Phone Calls (Twilio)
-
-### Setup
-
-```bash
-export SCITEX_NOTIFY_TWILIO_SID="ACxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx"
-export SCITEX_NOTIFY_TWILIO_TOKEN="your_auth_token"
-export SCITEX_NOTIFY_TWILIO_FROM="+1xxxxxxxxxx" # Your Twilio number
-export SCITEX_NOTIFY_TWILIO_TO="+61xxxxxxxxxx" # Your phone number
-```
-
-### Usage
-
-```python
-import scitex
-
-# Simple call
-scitex.notify.call("Build finished!")
-
-# Call twice to bypass iOS silent mode (30s apart)
-scitex.notify.call("Wake up!", repeat=2)
-
-# Call with Twilio Studio Flow
-scitex.notify.call("Alert!", flow_sid="FWxxxxxxx")
-```
-
-### Bypassing Silent Mode (iOS)
-
-To receive calls while in Do Not Disturb / silent mode:
-
-1. Save the Twilio number as a contact (e.g., "SciTeX Alert")
-2. **Settings -> Focus -> Do Not Disturb -> Allow Repeated Calls** -> ON
-3. Use `repeat=2` -- the second call within 3 minutes bypasses silent mode
-
-## SMS (Twilio)
-
-Uses the same Twilio env vars as phone calls.
-
-```python
-import scitex
-
-# Simple SMS
-scitex.notify.sms("Build finished!")
-
-# With title prefix
-scitex.notify.sms("Pipeline error on node 3", title="SciTeX Alert")
-
-# Override destination
-scitex.notify.sms("Urgent!", to_number="+61400000000")
-```
-
-CLI:
-
-```bash
-scitex notify sms "Build finished!"
-scitex notify sms "Alert!" --to +61400000000
-scitex notify sms "Error" --title "SciTeX"
-```
-
-## Configuration
-
-### Environment Variables
-
-| Variable | Description | Default |
-|----------|-------------|---------|
-| `SCITEX_NOTIFY_DEFAULT_BACKEND` | Default backend | `audio` |
-| `SCITEX_NOTIFY_TWILIO_SID` | Twilio Account SID | -- |
-| `SCITEX_NOTIFY_TWILIO_TOKEN` | Twilio Auth Token | -- |
-| `SCITEX_NOTIFY_TWILIO_FROM` | Twilio phone number | -- |
-| `SCITEX_NOTIFY_TWILIO_TO` | Destination phone | -- |
-| `SCITEX_NOTIFY_TWILIO_FLOW` | Studio Flow SID | -- |
-
-### YAML Config (`~/.scitex/config.yaml`)
-
-```yaml
-notify:
- default_backend: audio
- backend_priority:
- - audio
- - emacs
- - email
- level_backends:
- info: [audio]
- warning: [audio, emacs]
- error: [audio, emacs, email]
- critical: [audio, emacs, email, twilio]
-```
-
-## Fallback Priority
-
-When no backend is specified, `alert()` tries backends in order until one succeeds:
-
-1. **audio** -- TTS (fast, non-blocking)
-2. **emacs** -- Minibuffer message
-3. **matplotlib** -- Visual popup
-4. **playwright** -- Browser popup
-5. **email** -- Email (slowest, most reliable)
-
-Note: `twilio` is never in the fallback chain -- phone calls are explicit only via `call()` or `backend="twilio"`.
-
-## API Reference
-
-```python
-# Send notification with fallback
-scitex.notify.alert(
- message: str,
- title: str = None,
- backend: str | list[str] = None,
- level: str = "info", # info, warning, error, critical
- fallback: bool = True,
- **kwargs,
-) -> bool
-
-# Make phone call (no fallback)
-scitex.notify.call(
- message: str,
- title: str = None,
- level: str = "info",
- to_number: str = None, # Override default
- repeat: int = 1, # Call multiple times (bypass silent mode)
- **kwargs,
-) -> bool
-
-# Send SMS (no fallback)
-scitex.notify.sms(
- message: str,
- title: str = None, # Prepended to message
- to_number: str = None, # Override default
- **kwargs,
-) -> bool
-
-# Async versions
-await scitex.notify.alert_async(...)
-await scitex.notify.call_async(...)
-await scitex.notify.sms_async(...)
-
-# List available backends
-scitex.notify.available_backends() -> list[str]
-```
-
-## CLI Commands
-
-```bash
-scitex notify send "Task complete!" # Auto-fallback
-scitex notify send "Error" --backend email --level error # Specific backend
-scitex notify call "Wake up!" --repeat 2 # Phone call
-scitex notify sms "Build finished!" # SMS
-scitex notify backends # List backends
-scitex notify config # Show config
-scitex notify --help-recursive # All help
-```
-
-## MCP Tools
-
-Available via `scitex mcp serve`:
-
-| Tool | Description |
-|------|-------------|
-| `notify_send` | Send notification via backend(s) |
-| `notify_call` | Make a phone call via Twilio |
-| `notify_sms` | Send an SMS via Twilio |
-| `notify_backends` | List all backends and availability |
-| `notify_config` | Get current configuration |
diff --git a/src/scitex/notification/__init__.py b/src/scitex/notification/__init__.py
index ca841b026..abdc1272a 100755
--- a/src/scitex/notification/__init__.py
+++ b/src/scitex/notification/__init__.py
@@ -1,30 +1,13 @@
-#!/usr/bin/env python3
-"""SciTeX Notification — thin wrapper delegating to scitex-notification package.
+"""SciTeX notification — thin compatibility shim for scitex-notification."""
-All notification logic lives in the standalone scitex-notification package.
-This module re-exports the public API.
-"""
+import sys as _sys
-from scitex_notification import (
- DEFAULT_FALLBACK_ORDER,
- alert,
- alert_async,
- available_backends,
- call,
- call_async,
- sms,
- sms_async,
-)
+try:
+ import scitex_notification as _real
+except ImportError as _e:
+ raise ImportError(
+ "scitex.notification requires the 'scitex-notification' package. "
+ "Install with: pip install scitex[notification] (or: pip install scitex-notification)"
+ ) from _e
-__all__ = [
- "alert",
- "alert_async",
- "call",
- "call_async",
- "sms",
- "sms_async",
- "available_backends",
- "DEFAULT_FALLBACK_ORDER",
-]
-
-# EOF
+_sys.modules[__name__] = _real
diff --git a/src/scitex/notification/_backends.py b/src/scitex/notification/_backends.py
deleted file mode 100755
index 3fbf16bfb..000000000
--- a/src/scitex/notification/_backends.py
+++ /dev/null
@@ -1,37 +0,0 @@
-#!/usr/bin/env python3
-# Timestamp: "2026-01-13 (ywatanabe)"
-# File: /home/ywatanabe/proj/scitex-code/src/scitex/ui/_backends.py
-
-"""Re-export from _backends package for backward compatibility."""
-
-from ._backends import (
- BACKENDS,
- AudioBackend,
- BaseNotifyBackend,
- DesktopBackend,
- EmailBackend,
- MatplotlibBackend,
- NotifyLevel,
- NotifyResult,
- PlaywrightBackend,
- WebhookBackend,
- available_backends,
- get_backend,
-)
-
-__all__ = [
- "NotifyLevel",
- "NotifyResult",
- "BaseNotifyBackend",
- "AudioBackend",
- "EmailBackend",
- "DesktopBackend",
- "WebhookBackend",
- "MatplotlibBackend",
- "PlaywrightBackend",
- "BACKENDS",
- "get_backend",
- "available_backends",
-]
-
-# EOF
diff --git a/src/scitex/notification/_backends/__init__.py b/src/scitex/notification/_backends/__init__.py
deleted file mode 100755
index ab741a78e..000000000
--- a/src/scitex/notification/_backends/__init__.py
+++ /dev/null
@@ -1,69 +0,0 @@
-#!/usr/bin/env python3
-# Timestamp: "2026-01-13 (ywatanabe)"
-# File: /home/ywatanabe/proj/scitex-code/src/scitex/ui/_backends/__init__.py
-
-"""Notification backend registry and utilities."""
-
-from __future__ import annotations
-
-from ._audio import AudioBackend
-from ._desktop import DesktopBackend
-from ._emacs import EmacsBackend
-from ._email import EmailBackend
-from ._matplotlib import MatplotlibBackend
-from ._playwright import PlaywrightBackend
-from ._twilio import TwilioBackend
-from ._types import BaseNotifyBackend, NotifyLevel, NotifyResult
-from ._webhook import WebhookBackend
-
-__all__ = [
- "NotifyLevel",
- "NotifyResult",
- "BaseNotifyBackend",
- "AudioBackend",
- "EmailBackend",
- "DesktopBackend",
- "EmacsBackend",
- "WebhookBackend",
- "MatplotlibBackend",
- "PlaywrightBackend",
- "TwilioBackend",
- "BACKENDS",
- "get_backend",
- "available_backends",
-]
-
-# Registry of available backends
-BACKENDS: dict[str, type[BaseNotifyBackend]] = {
- "audio": AudioBackend,
- "email": EmailBackend,
- "desktop": DesktopBackend,
- "emacs": EmacsBackend,
- "webhook": WebhookBackend,
- "matplotlib": MatplotlibBackend,
- "playwright": PlaywrightBackend,
- "twilio": TwilioBackend,
-}
-
-
-def get_backend(name: str, **kwargs) -> BaseNotifyBackend:
- """Get a notification backend by name."""
- if name not in BACKENDS:
- raise ValueError(f"Unknown backend: {name}. Available: {list(BACKENDS.keys())}")
- return BACKENDS[name](**kwargs)
-
-
-def available_backends() -> list[str]:
- """Return list of available notification backends."""
- available = []
- for name, cls in BACKENDS.items():
- try:
- backend = cls()
- if backend.is_available():
- available.append(name)
- except Exception:
- pass
- return available
-
-
-# EOF
diff --git a/src/scitex/notification/_backends/_audio.py b/src/scitex/notification/_backends/_audio.py
deleted file mode 100755
index da0e329ab..000000000
--- a/src/scitex/notification/_backends/_audio.py
+++ /dev/null
@@ -1,86 +0,0 @@
-#!/usr/bin/env python3
-# Timestamp: "2026-01-13 (ywatanabe)"
-# File: /home/ywatanabe/proj/scitex-code/src/scitex/ui/_backends/_audio.py
-
-"""Audio notification backend via TTS."""
-
-from __future__ import annotations
-
-import asyncio
-from datetime import datetime
-from typing import Optional
-
-from ._types import BaseNotifyBackend, NotifyLevel, NotifyResult
-
-
-class AudioBackend(BaseNotifyBackend):
- """Audio notification via scitex.audio TTS."""
-
- name = "audio"
-
- def __init__(
- self,
- backend: str = "gtts",
- speed: float = 1.5,
- rate: int = 180,
- ):
- self.tts_backend = backend
- self.speed = speed
- self.rate = rate
-
- def is_available(self) -> bool:
- try:
- from scitex.audio import available_backends
-
- return len(available_backends()) > 0
- except ImportError:
- return False
-
- async def send(
- self,
- message: str,
- title: Optional[str] = None,
- level: NotifyLevel = NotifyLevel.INFO,
- **kwargs,
- ) -> NotifyResult:
- try:
- from scitex.audio import speak
-
- # Prepend title if provided
- full_message = f"{title}. {message}" if title else message
-
- # Add urgency prefix for critical/error levels
- if level == NotifyLevel.CRITICAL:
- full_message = f"Critical alert! {full_message}"
- elif level == NotifyLevel.ERROR:
- full_message = f"Error. {full_message}"
-
- # Run TTS in executor to not block
- loop = asyncio.get_event_loop()
- await loop.run_in_executor(
- None,
- lambda: speak(
- full_message,
- backend=kwargs.get("tts_backend", self.tts_backend),
- speed=kwargs.get("speed", self.speed),
- rate=kwargs.get("rate", self.rate),
- ),
- )
-
- return NotifyResult(
- success=True,
- backend=self.name,
- message=message,
- timestamp=datetime.now().isoformat(),
- )
- except Exception as e:
- return NotifyResult(
- success=False,
- backend=self.name,
- message=message,
- timestamp=datetime.now().isoformat(),
- error=str(e),
- )
-
-
-# EOF
diff --git a/src/scitex/notification/_backends/_config.py b/src/scitex/notification/_backends/_config.py
deleted file mode 100755
index 0fde9455e..000000000
--- a/src/scitex/notification/_backends/_config.py
+++ /dev/null
@@ -1,270 +0,0 @@
-#!/usr/bin/env python3
-# Timestamp: "2026-01-13 (ywatanabe)"
-# File: /home/ywatanabe/proj/scitex-code/src/scitex/ui/_backends/_config.py
-
-"""Configuration for notification backends using scitex.config.
-
-Priority resolution (same as ScitexConfig):
- direct → config (YAML) → env → default
-
-Configuration sources:
-1. YAML file: ~/.scitex/config.yaml or custom path via SCITEX_NOTIFY_CONFIG
-2. Environment variables: SCITEX_NOTIFY_* (SCITEX_UI_* for backward compat)
-
-Example YAML (in default.yaml or custom file):
- ui:
- default_backend: audio
- backend_priority:
- - audio
- - desktop
- - email
- level_backends:
- info: [audio]
- warning: [audio, desktop]
- error: [audio, desktop, email]
- critical: [audio, desktop, email]
- timeouts:
- matplotlib: 5.0
- playwright: 5.0
-
-Environment variables:
- SCITEX_NOTIFY_CONFIG: Path to custom UI config file
- SCITEX_NOTIFY_DEFAULT_BACKEND: audio (preferred)
- SCITEX_UI_CONFIG: (deprecated) use SCITEX_NOTIFY_CONFIG
- SCITEX_UI_DEFAULT_BACKEND: (deprecated) use SCITEX_NOTIFY_DEFAULT_BACKEND
- SCITEX_NOTIFY_BACKEND_PRIORITY: audio,desktop,email (comma-separated)
- SCITEX_NOTIFY_INFO_BACKENDS: audio (comma-separated)
- SCITEX_NOTIFY_WARNING_BACKENDS: audio,desktop
- SCITEX_NOTIFY_ERROR_BACKENDS: audio,desktop,email
- SCITEX_NOTIFY_CRITICAL_BACKENDS: audio,desktop,email
-"""
-
-from __future__ import annotations
-
-import importlib.util
-import os
-from functools import lru_cache
-from typing import Optional
-
-from ._types import NotifyLevel
-
-# Backend package requirements
-BACKEND_PACKAGES = {
- "audio": None, # Uses MCP or pyttsx3 (optional)
- "desktop": None, # Uses PowerShell on WSL (no package needed)
- "emacs": None, # Uses emacsclient (no Python package needed)
- "matplotlib": "matplotlib",
- "playwright": "playwright",
- "email": None, # Uses stdlib smtplib
- "webhook": None, # Uses stdlib urllib
-}
-
-
-@lru_cache(maxsize=16)
-def is_package_available(package: str) -> bool:
- """Check if a Python package is available."""
- if package is None:
- return True
- return importlib.util.find_spec(package) is not None
-
-
-def is_backend_available(backend: str) -> bool:
- """Check if a backend's required packages are available."""
- package = BACKEND_PACKAGES.get(backend)
- return is_package_available(package)
-
-
-# Default configuration (used if not in YAML)
-DEFAULT_CONFIG = {
- "default_backend": "audio",
- "backend_priority": [
- "audio",
- "emacs",
- "desktop",
- "matplotlib",
- "playwright",
- "email",
- "webhook",
- ],
- "level_backends": {
- "info": ["audio"],
- "warning": ["audio", "emacs"],
- "error": ["audio", "emacs", "desktop", "email"],
- "critical": ["audio", "emacs", "desktop", "matplotlib", "email"],
- },
- "timeouts": {
- "matplotlib": 5.0,
- "playwright": 5.0,
- },
-}
-
-
-class UIConfig:
- """Configuration manager for scitex.notify using ScitexConfig pattern."""
-
- _instance: Optional[UIConfig] = None
- _config: dict
-
- def __new__(cls, config_path: Optional[str] = None):
- # Allow creating new instance with custom path
- if config_path is not None:
- instance = super().__new__(cls)
- instance._config = {}
- instance._config_path = config_path
- instance._load_config()
- return instance
-
- # Otherwise use singleton
- if cls._instance is None:
- cls._instance = super().__new__(cls)
- cls._instance._config = {}
- cls._instance._config_path = None
- cls._instance._load_config()
- return cls._instance
-
- def _load_config(self):
- """Load configuration from ScitexConfig and environment."""
- self._config = DEFAULT_CONFIG.copy()
- self._config["level_backends"] = DEFAULT_CONFIG["level_backends"].copy()
- self._config["timeouts"] = DEFAULT_CONFIG["timeouts"].copy()
-
- # Try to load from ScitexConfig (integrates with default.yaml)
- try:
- from scitex.config import get_config
-
- # Support custom config path via env var or constructor
- # Check SCITEX_NOTIFY_CONFIG first, fall back to SCITEX_UI_CONFIG
- config_path = (
- self._config_path
- or os.getenv("SCITEX_NOTIFY_CONFIG")
- or os.getenv("SCITEX_UI_CONFIG")
- )
- scitex_config = get_config(config_path)
-
- # Get UI section from config
- ui_config = scitex_config.get_nested("ui") or {}
-
- if ui_config:
- # Update default_backend
- if "default_backend" in ui_config:
- self._config["default_backend"] = ui_config["default_backend"]
-
- # Update backend_priority
- if "backend_priority" in ui_config:
- self._config["backend_priority"] = ui_config["backend_priority"]
-
- # Update level_backends
- if "level_backends" in ui_config:
- for level, backends in ui_config["level_backends"].items():
- self._config["level_backends"][level] = backends
-
- # Update timeouts
- if "timeouts" in ui_config:
- self._config["timeouts"].update(ui_config["timeouts"])
-
- except ImportError:
- pass # scitex.config not available
- except Exception:
- pass # Config loading failed
-
- # Override with environment variables (env has lowest priority after config)
- self._load_env_overrides()
-
- def _load_env_overrides(self):
- """Load environment variable overrides.
-
- Checks SCITEX_NOTIFY_* first, falls back to SCITEX_UI_* for backward compat.
- """
- default_backend = os.getenv("SCITEX_NOTIFY_DEFAULT_BACKEND") or os.getenv(
- "SCITEX_UI_DEFAULT_BACKEND"
- )
- if default_backend:
- self._config["default_backend"] = default_backend
-
- backend_priority = os.getenv("SCITEX_NOTIFY_BACKEND_PRIORITY") or os.getenv(
- "SCITEX_UI_BACKEND_PRIORITY"
- )
- if backend_priority:
- self._config["backend_priority"] = backend_priority.split(",")
-
- # Level-specific backends from env
- for level in ["info", "warning", "error", "critical"]:
- level_upper = level.upper()
- env_val = os.getenv(f"SCITEX_NOTIFY_{level_upper}_BACKENDS") or os.getenv(
- f"SCITEX_UI_{level_upper}_BACKENDS"
- )
- if env_val:
- self._config["level_backends"][level] = env_val.split(",")
-
- # Timeouts from env
- for backend in ["matplotlib", "playwright"]:
- backend_upper = backend.upper()
- env_val = os.getenv(f"SCITEX_NOTIFY_TIMEOUT_{backend_upper}") or os.getenv(
- f"SCITEX_UI_TIMEOUT_{backend_upper}"
- )
- if env_val:
- try:
- self._config["timeouts"][backend] = float(env_val)
- except ValueError:
- pass
-
- @property
- def default_backend(self) -> str:
- return self._config.get("default_backend", "audio")
-
- @property
- def backend_priority(self) -> list[str]:
- return self._config.get("backend_priority", ["audio"])
-
- def get_available_backend_priority(self) -> list[str]:
- """Get backend priority filtered by package availability."""
- return [b for b in self.backend_priority if is_backend_available(b)]
-
- def get_backends_for_level(self, level: NotifyLevel) -> list[str]:
- """Get configured backends for a notification level."""
- level_backends = self._config.get("level_backends", {})
- return level_backends.get(level.value, [self.default_backend])
-
- def get_available_backends_for_level(self, level: NotifyLevel) -> list[str]:
- """Get backends for level filtered by package availability."""
- backends = self.get_backends_for_level(level)
- return [b for b in backends if is_backend_available(b)]
-
- def get_first_available_backend(self) -> str:
- """Get first available backend from priority list."""
- for backend in self.backend_priority:
- if is_backend_available(backend):
- return backend
- return self.default_backend
-
- def get_timeout(self, backend: str) -> float:
- """Get timeout for a backend."""
- timeouts = self._config.get("timeouts", {})
- value = timeouts.get(backend, 5.0)
- return float(value) if value is not None else 5.0
-
- def reload(self):
- """Reload configuration from files."""
- self._load_config()
-
- @classmethod
- def reset(cls):
- """Reset singleton instance (useful for testing)."""
- cls._instance = None
-
-
-def get_config(config_path: Optional[str] = None) -> UIConfig:
- """Get the UI configuration instance.
-
- Parameters
- ----------
- config_path : str, optional
- Path to custom config file. If provided, creates new instance.
- Otherwise returns cached singleton.
- """
- if config_path:
- return UIConfig(config_path)
- return UIConfig()
-
-
-# EOF
diff --git a/src/scitex/notification/_backends/_desktop.py b/src/scitex/notification/_backends/_desktop.py
deleted file mode 100755
index 500f244b6..000000000
--- a/src/scitex/notification/_backends/_desktop.py
+++ /dev/null
@@ -1,184 +0,0 @@
-#!/usr/bin/env python3
-# Timestamp: "2026-01-13 (ywatanabe)"
-# File: /home/ywatanabe/proj/scitex-code/src/scitex/ui/_backends/_desktop.py
-
-"""Desktop notification backend (Linux notify-send, WSL PowerShell toast)."""
-
-from __future__ import annotations
-
-import asyncio
-import shutil
-import subprocess
-import tempfile
-from datetime import datetime
-from typing import Optional
-
-from ._types import BaseNotifyBackend, NotifyLevel, NotifyResult
-
-
-class DesktopBackend(BaseNotifyBackend):
- """Desktop notification via native OS APIs.
-
- Supports:
- - Linux: notify-send
- - WSL/Windows: PowerShell toast notifications
- """
-
- name = "desktop"
-
- def _is_wsl(self) -> bool:
- """Check if running in WSL."""
- try:
- with open("/proc/version") as f:
- return "microsoft" in f.read().lower()
- except Exception:
- return False
-
- def _has_powershell(self) -> bool:
- """Check if PowerShell is available."""
- return shutil.which("powershell.exe") is not None
-
- def is_available(self) -> bool:
- # Check for notify-send (Linux)
- if shutil.which("notify-send") is not None:
- return True
- # Check for PowerShell (WSL/Windows)
- if self._is_wsl() and self._has_powershell():
- return True
- return False
-
- async def send(
- self,
- message: str,
- title: Optional[str] = None,
- level: NotifyLevel = NotifyLevel.INFO,
- **kwargs,
- ) -> NotifyResult:
- try:
- title = title or "SciTeX"
-
- # Use PowerShell for WSL
- if self._is_wsl() and self._has_powershell():
- return await self._send_windows_toast(message, title, level)
-
- # Use notify-send for Linux
- if shutil.which("notify-send"):
- return await self._send_notify_send(message, title, level)
-
- return NotifyResult(
- success=False,
- backend=self.name,
- message=message,
- timestamp=datetime.now().isoformat(),
- error="No desktop notification method available",
- )
- except Exception as e:
- return NotifyResult(
- success=False,
- backend=self.name,
- message=message,
- timestamp=datetime.now().isoformat(),
- error=str(e),
- )
-
- async def _send_notify_send(
- self, message: str, title: str, level: NotifyLevel
- ) -> NotifyResult:
- """Send notification via notify-send (Linux)."""
- urgency_map = {
- NotifyLevel.INFO: "normal",
- NotifyLevel.WARNING: "normal",
- NotifyLevel.ERROR: "critical",
- NotifyLevel.CRITICAL: "critical",
- }
-
- cmd = [
- "notify-send",
- "-u",
- urgency_map.get(level, "normal"),
- title,
- message,
- ]
-
- loop = asyncio.get_event_loop()
- await loop.run_in_executor(
- None,
- lambda: subprocess.run(cmd, capture_output=True, timeout=5),
- )
-
- return NotifyResult(
- success=True,
- backend=self.name,
- message=message,
- timestamp=datetime.now().isoformat(),
- )
-
- async def _send_windows_toast(
- self, message: str, title: str, level: NotifyLevel
- ) -> NotifyResult:
- """Send Windows toast notification via PowerShell."""
- import os as _os
-
- # Escape for XML
- title_escaped = (
- title.replace("&", "&").replace("<", "<").replace(">", ">")
- )
- message_escaped = (
- message.replace("&", "&").replace("<", "<").replace(">", ">")
- )
-
- ps_script = f"""[Windows.UI.Notifications.ToastNotificationManager, Windows.UI.Notifications, ContentType = WindowsRuntime] | Out-Null
-[Windows.Data.Xml.Dom.XmlDocument, Windows.Data.Xml.Dom.XmlDocument, ContentType = WindowsRuntime] | Out-Null
-$template = '{title_escaped}{message_escaped}'
-$xml = New-Object Windows.Data.Xml.Dom.XmlDocument
-$xml.LoadXml($template)
-$toast = [Windows.UI.Notifications.ToastNotification]::new($xml)
-[Windows.UI.Notifications.ToastNotificationManager]::CreateToastNotifier("SciTeX").Show($toast)
-"""
-
- # Write to temp file
- with tempfile.NamedTemporaryFile(mode="w", suffix=".ps1", delete=False) as f:
- f.write(ps_script)
- ps_file = f.name
-
- try:
- # Use Popen to avoid blocking
- loop = asyncio.get_event_loop()
- await loop.run_in_executor(
- None,
- lambda: subprocess.Popen(
- [
- "powershell.exe",
- "-NoProfile",
- "-NonInteractive",
- "-ExecutionPolicy",
- "Bypass",
- "-File",
- ps_file,
- ],
- stdout=subprocess.DEVNULL,
- stderr=subprocess.DEVNULL,
- stdin=subprocess.DEVNULL,
- ),
- )
-
- # Give PowerShell time to read the file
- await asyncio.sleep(0.5)
-
- return NotifyResult(
- success=True,
- backend=self.name,
- message=message,
- timestamp=datetime.now().isoformat(),
- details={"method": "windows_toast"},
- )
- finally:
- # Clean up temp file after delay
- await asyncio.sleep(1)
- try:
- _os.unlink(ps_file)
- except Exception:
- pass
-
-
-# EOF
diff --git a/src/scitex/notification/_backends/_emacs.py b/src/scitex/notification/_backends/_emacs.py
deleted file mode 100755
index 350cf69ee..000000000
--- a/src/scitex/notification/_backends/_emacs.py
+++ /dev/null
@@ -1,210 +0,0 @@
-#!/usr/bin/env python3
-# Timestamp: "2026-01-13 (ywatanabe)"
-# File: /home/ywatanabe/proj/scitex-code/src/scitex/ui/_backends/_emacs.py
-
-"""Emacs notification backend using emacsclient."""
-
-from __future__ import annotations
-
-import asyncio
-import shutil
-import subprocess
-from datetime import datetime
-from typing import Optional
-
-from ._types import BaseNotifyBackend, NotifyLevel, NotifyResult
-
-
-class EmacsBackend(BaseNotifyBackend):
- """Notification via Emacs using emacsclient.
-
- Displays notifications in Emacs minibuffer or as alerts.
- Supports different display methods:
- - popup: temporary popup buffer (default, most noticeable)
- - minibuffer: message function
- - alert: alert.el package
- - notifications: notifications.el (desktop notifications from Emacs)
- """
-
- name = "emacs"
-
- def __init__(self, method: str = "popup", timeout: float = 5.0):
- """Initialize Emacs backend.
-
- Parameters
- ----------
- method : str
- Notification method: 'popup', 'minibuffer', 'alert', or 'notifications'
- timeout : float
- Display timeout for visual methods
- """
- self.method = method
- self.timeout = timeout
-
- def is_available(self) -> bool:
- """Check if emacsclient is available."""
- return shutil.which("emacsclient") is not None
-
- def _escape_elisp_string(self, s: str) -> str:
- """Escape a string for use in elisp."""
- return s.replace("\\", "\\\\").replace('"', '\\"').replace("\n", "\\n")
-
- def _get_face_for_level(self, level: NotifyLevel) -> str:
- """Get Emacs face name for notification level."""
- faces = {
- NotifyLevel.INFO: "success",
- NotifyLevel.WARNING: "warning",
- NotifyLevel.ERROR: "error",
- NotifyLevel.CRITICAL: "error",
- }
- return faces.get(level, "default")
-
- async def send(
- self,
- message: str,
- title: Optional[str] = None,
- level: NotifyLevel = NotifyLevel.INFO,
- **kwargs,
- ) -> NotifyResult:
- """Send notification via Emacs."""
- try:
- method = kwargs.get("method", self.method)
- timeout = kwargs.get("timeout", self.timeout)
-
- # Escape strings for elisp
- msg_escaped = self._escape_elisp_string(message)
- title_escaped = self._escape_elisp_string(title or "SciTeX")
- face = self._get_face_for_level(level)
-
- # Build elisp command based on method
- if method == "popup":
- # Popup buffer - most noticeable
- level_colors = {
- NotifyLevel.INFO: "#98C379", # green
- NotifyLevel.WARNING: "#E5C07B", # yellow
- NotifyLevel.ERROR: "#E06C75", # red
- NotifyLevel.CRITICAL: "#E06C75", # red
- }
- color = level_colors.get(level, "#98C379")
- elisp = f"""
- (let* ((buf (get-buffer-create "*SciTeX Alert*"))
- (timeout {int(timeout)}))
- (with-current-buffer buf
- (erase-buffer)
- (insert (propertize "\\n ╔══════════════════════════════════════╗\\n"
- 'face '(:foreground "{color}" :weight bold)))
- (insert (propertize " ║ SciTeX Alert ║\\n"
- 'face '(:foreground "{color}" :weight bold)))
- (insert (propertize " ╠══════════════════════════════════════╣\\n"
- 'face '(:foreground "{color}")))
- (insert (propertize (format " ║ [%s] %s\\n" "{level.value.upper()}" "{msg_escaped}")
- 'face '(:foreground "{color}")))
- (insert (propertize " ╚══════════════════════════════════════╝\\n"
- 'face '(:foreground "{color}")))
- (goto-char (point-min)))
- (display-buffer buf
- '((display-buffer-in-side-window)
- (side . bottom)
- (window-height . 8)))
- (run-at-time timeout nil
- (lambda ()
- (when-let ((win (get-buffer-window buf t)))
- (delete-window win))
- (kill-buffer buf)))
- (message "[SciTeX] %s" "{msg_escaped}"))
- """
- elif method == "alert":
- # Use alert.el package (if installed)
- severity_map = {
- NotifyLevel.INFO: "normal",
- NotifyLevel.WARNING: "moderate",
- NotifyLevel.ERROR: "high",
- NotifyLevel.CRITICAL: "urgent",
- }
- severity = severity_map.get(level, "normal")
- elisp = f"""
- (if (fboundp 'alert)
- (alert "{msg_escaped}"
- :title "{title_escaped}"
- :severity '{severity}
- :timeout {int(timeout)})
- (message "[%s] %s: %s" "{level.value.upper()}" "{title_escaped}" "{msg_escaped}"))
- """
- elif method == "notifications":
- # Use notifications.el (requires D-Bus)
- urgency_map = {
- NotifyLevel.INFO: "normal",
- NotifyLevel.WARNING: "normal",
- NotifyLevel.ERROR: "critical",
- NotifyLevel.CRITICAL: "critical",
- }
- urgency = urgency_map.get(level, "normal")
- elisp = f"""
- (if (fboundp 'notifications-notify)
- (notifications-notify
- :title "{title_escaped}"
- :body "{msg_escaped}"
- :urgency '{urgency}
- :timeout {int(timeout * 1000)})
- (message "[%s] %s: %s" "{level.value.upper()}" "{title_escaped}" "{msg_escaped}"))
- """
- else:
- # Default: minibuffer message with face
- elisp = f"""
- (let ((msg (propertize "[{level.value.upper()}] {title_escaped}: {msg_escaped}" 'face '{face})))
- (message "%s" msg))
- """
-
- # Clean up elisp (remove extra whitespace)
- elisp = " ".join(elisp.split())
-
- # Execute via emacsclient
- cmd = ["emacsclient", "--eval", elisp]
-
- loop = asyncio.get_event_loop()
- result = await loop.run_in_executor(
- None,
- lambda: subprocess.run(
- cmd,
- capture_output=True,
- text=True,
- timeout=10,
- ),
- )
-
- if result.returncode == 0:
- return NotifyResult(
- success=True,
- backend=self.name,
- message=message,
- timestamp=datetime.now().isoformat(),
- details={"method": method, "elisp_result": result.stdout.strip()},
- )
- else:
- return NotifyResult(
- success=False,
- backend=self.name,
- message=message,
- timestamp=datetime.now().isoformat(),
- error=result.stderr.strip() or "emacsclient failed",
- )
-
- except subprocess.TimeoutExpired:
- return NotifyResult(
- success=False,
- backend=self.name,
- message=message,
- timestamp=datetime.now().isoformat(),
- error="emacsclient timed out",
- )
- except Exception as e:
- return NotifyResult(
- success=False,
- backend=self.name,
- message=message,
- timestamp=datetime.now().isoformat(),
- error=str(e),
- )
-
-
-# EOF
diff --git a/src/scitex/notification/_backends/_email.py b/src/scitex/notification/_backends/_email.py
deleted file mode 100755
index 36eb7d921..000000000
--- a/src/scitex/notification/_backends/_email.py
+++ /dev/null
@@ -1,86 +0,0 @@
-#!/usr/bin/env python3
-# Timestamp: "2026-01-13 (ywatanabe)"
-# File: /home/ywatanabe/proj/scitex-code/src/scitex/ui/_backends/_email.py
-
-"""Email notification backend."""
-
-from __future__ import annotations
-
-import asyncio
-import os
-from datetime import datetime
-from typing import Optional
-
-from ._types import BaseNotifyBackend, NotifyLevel, NotifyResult
-
-
-class EmailBackend(BaseNotifyBackend):
- """Email notification via scitex.utils._email."""
-
- name = "email"
-
- def __init__(
- self,
- recipient: Optional[str] = None,
- sender: Optional[str] = None,
- ):
- self.recipient = (
- recipient
- or os.getenv("SCITEX_UI_EMAIL_NOTIFICATION_TO")
- or os.getenv("SCITEX_NOTIFY_EMAIL_TO")
- )
- self.sender = (
- sender
- or os.getenv("SCITEX_UI_EMAIL_NOTIFICATION_FROM")
- or os.getenv("SCITEX_NOTIFY_EMAIL_FROM")
- )
-
- def is_available(self) -> bool:
- return bool(
- os.getenv("SCITEX_SCHOLAR_EMAIL_NOREPLY") # New name
- or os.getenv("SCITEX_SCHOLAR_FROM_EMAIL_ADDRESS") # Deprecated
- or os.getenv("SCITEX_EMAIL_NOREPLY") # Global
- or os.getenv("SCITEX_EMAIL_AGENT") # Fallback
- ) and bool(
- os.getenv("SCITEX_SCHOLAR_EMAIL_PASSWORD") # New name
- or os.getenv("SCITEX_SCHOLAR_FROM_EMAIL_PASSWORD") # Deprecated
- or os.getenv("SCITEX_EMAIL_PASSWORD") # Fallback
- )
-
- async def send(
- self,
- message: str,
- title: Optional[str] = None,
- level: NotifyLevel = NotifyLevel.INFO,
- **kwargs,
- ) -> NotifyResult:
- try:
- from scitex.utils._notify import notify as email_notify
-
- loop = asyncio.get_event_loop()
- await loop.run_in_executor(
- None,
- lambda: email_notify(
- subject=title or f"[SciTeX] {level.value.upper()}",
- message=message,
- recipient_email=kwargs.get("recipient", self.recipient),
- ),
- )
-
- return NotifyResult(
- success=True,
- backend=self.name,
- message=message,
- timestamp=datetime.now().isoformat(),
- )
- except Exception as e:
- return NotifyResult(
- success=False,
- backend=self.name,
- message=message,
- timestamp=datetime.now().isoformat(),
- error=str(e),
- )
-
-
-# EOF
diff --git a/src/scitex/notification/_backends/_matplotlib.py b/src/scitex/notification/_backends/_matplotlib.py
deleted file mode 100755
index 954d68062..000000000
--- a/src/scitex/notification/_backends/_matplotlib.py
+++ /dev/null
@@ -1,118 +0,0 @@
-#!/usr/bin/env python3
-# Timestamp: "2026-01-13 (ywatanabe)"
-# File: /home/ywatanabe/proj/scitex-code/src/scitex/ui/_backends/_matplotlib.py
-
-"""Matplotlib visual notification backend."""
-
-from __future__ import annotations
-
-import asyncio
-from datetime import datetime
-from typing import Optional
-
-from ._types import BaseNotifyBackend, NotifyLevel, NotifyResult
-
-
-class MatplotlibBackend(BaseNotifyBackend):
- """Visual notification via matplotlib popup window."""
-
- name = "matplotlib"
-
- def __init__(self, timeout: float = 5.0):
- self.timeout = timeout
-
- def is_available(self) -> bool:
- try:
- import matplotlib # noqa: F401
-
- return True
- except ImportError:
- return False
-
- async def send(
- self,
- message: str,
- title: Optional[str] = None,
- level: NotifyLevel = NotifyLevel.INFO,
- **kwargs,
- ) -> NotifyResult:
- try:
- import matplotlib.pyplot as plt
-
- # Color based on level
- colors = {
- NotifyLevel.INFO: "#2196F3",
- NotifyLevel.WARNING: "#FF9800",
- NotifyLevel.ERROR: "#F44336",
- NotifyLevel.CRITICAL: "#9C27B0",
- }
- color = colors.get(level, "#2196F3")
-
- # Create figure
- fig, ax = plt.subplots(figsize=(6, 2))
- ax.set_xlim(0, 1)
- ax.set_ylim(0, 1)
- ax.axis("off")
-
- # Background
- fig.patch.set_facecolor(color)
-
- # Title and message
- ax.text(
- 0.5,
- 0.7,
- title or "SciTeX",
- ha="center",
- va="center",
- fontsize=16,
- fontweight="bold",
- color="white",
- )
- ax.text(
- 0.5,
- 0.35,
- message,
- ha="center",
- va="center",
- fontsize=12,
- color="white",
- wrap=True,
- )
-
- # Show non-blocking
- plt.ion()
- plt.show(block=False)
-
- # Force render
- fig.canvas.draw()
- fig.canvas.flush_events()
-
- # Auto-close after timeout, keeping GUI responsive
- timeout = kwargs.get("timeout", self.timeout)
- elapsed = 0.0
- interval = 0.1 # Check every 100ms
- while elapsed < timeout:
- await asyncio.sleep(interval)
- fig.canvas.flush_events()
- elapsed += interval
-
- plt.close(fig)
-
- return NotifyResult(
- success=True,
- backend=self.name,
- message=message,
- timestamp=datetime.now().isoformat(),
- details={"timeout": timeout},
- )
- except Exception as e:
- return NotifyResult(
- success=False,
- backend=self.name,
- message=message,
- timestamp=datetime.now().isoformat(),
- error=str(e),
- )
-
-
-# EOF
diff --git a/src/scitex/notification/_backends/_playwright.py b/src/scitex/notification/_backends/_playwright.py
deleted file mode 100755
index 7baea2cf7..000000000
--- a/src/scitex/notification/_backends/_playwright.py
+++ /dev/null
@@ -1,109 +0,0 @@
-#!/usr/bin/env python3
-# Timestamp: "2026-01-13 (ywatanabe)"
-# File: /home/ywatanabe/proj/scitex-code/src/scitex/ui/_backends/_playwright.py
-
-"""Playwright browser notification backend."""
-
-from __future__ import annotations
-
-import asyncio
-from datetime import datetime
-from typing import Optional
-
-from ._types import BaseNotifyBackend, NotifyLevel, NotifyResult
-
-
-class PlaywrightBackend(BaseNotifyBackend):
- """Browser notification via Playwright."""
-
- name = "playwright"
-
- def __init__(self, timeout: float = 5.0):
- self.timeout = timeout
-
- def is_available(self) -> bool:
- try:
- from playwright.async_api import async_playwright # noqa: F401
-
- return True
- except ImportError:
- return False
-
- async def send(
- self,
- message: str,
- title: Optional[str] = None,
- level: NotifyLevel = NotifyLevel.INFO,
- **kwargs,
- ) -> NotifyResult:
- try:
- from playwright.async_api import async_playwright
-
- # Color based on level
- colors = {
- NotifyLevel.INFO: "#2196F3",
- NotifyLevel.WARNING: "#FF9800",
- NotifyLevel.ERROR: "#F44336",
- NotifyLevel.CRITICAL: "#9C27B0",
- }
- color = colors.get(level, "#2196F3")
-
- # Escape HTML
- title_safe = (title or "SciTeX").replace("<", "<").replace(">", ">")
- message_safe = message.replace("<", "<").replace(">", ">")
-
- html = f"""
-
-
-
-
-
-
- {title_safe}
- {message_safe}
-
-
- """
-
- async with async_playwright() as p:
- browser = await p.chromium.launch(headless=False)
- page = await browser.new_page()
- await page.set_viewport_size({"width": 400, "height": 150})
- await page.set_content(html)
-
- timeout = kwargs.get("timeout", self.timeout)
- await asyncio.sleep(timeout)
- await browser.close()
-
- return NotifyResult(
- success=True,
- backend=self.name,
- message=message,
- timestamp=datetime.now().isoformat(),
- details={"timeout": timeout},
- )
- except Exception as e:
- return NotifyResult(
- success=False,
- backend=self.name,
- message=message,
- timestamp=datetime.now().isoformat(),
- error=str(e),
- )
-
-
-# EOF
diff --git a/src/scitex/notification/_backends/_twilio.py b/src/scitex/notification/_backends/_twilio.py
deleted file mode 100755
index da101bf68..000000000
--- a/src/scitex/notification/_backends/_twilio.py
+++ /dev/null
@@ -1,320 +0,0 @@
-#!/usr/bin/env python3
-# File: /home/ywatanabe/proj/scitex-python/src/scitex/notify/_backends/_twilio.py
-
-"""Twilio phone call notification backend.
-
-Makes an actual phone call to wake up the user.
-Supports both direct TwiML calls and Studio Flow executions.
-
-Environment Variables:
- SCITEX_NOTIFY_TWILIO_SID: Twilio Account SID
- SCITEX_NOTIFY_TWILIO_TOKEN: Twilio Auth Token
- SCITEX_NOTIFY_TWILIO_FROM: Twilio phone number (e.g., +1234567890)
- SCITEX_NOTIFY_TWILIO_TO: Destination phone number (e.g., +8190xxxx)
- SCITEX_NOTIFY_TWILIO_FLOW: Studio Flow SID (optional, e.g., FWxxxxxxx)
-"""
-
-from __future__ import annotations
-
-import asyncio
-import os
-from datetime import datetime
-from typing import Optional
-
-from ._types import BaseNotifyBackend, NotifyLevel, NotifyResult
-
-
-class TwilioBackend(BaseNotifyBackend):
- """Phone call notification via Twilio."""
-
- name = "twilio"
-
- def __init__(
- self,
- account_sid: Optional[str] = None,
- auth_token: Optional[str] = None,
- from_number: Optional[str] = None,
- to_number: Optional[str] = None,
- flow_sid: Optional[str] = None,
- repeat: int = 1,
- ):
- self.account_sid = account_sid or os.getenv("SCITEX_NOTIFY_TWILIO_SID", "")
- self.auth_token = auth_token or os.getenv("SCITEX_NOTIFY_TWILIO_TOKEN", "")
- self.from_number = from_number or os.getenv("SCITEX_NOTIFY_TWILIO_FROM", "")
- self.to_number = to_number or os.getenv("SCITEX_NOTIFY_TWILIO_TO", "")
- self.flow_sid = flow_sid or os.getenv("SCITEX_NOTIFY_TWILIO_FLOW", "")
- self.repeat = repeat
-
- def is_available(self) -> bool:
- return bool(
- self.account_sid and self.auth_token and self.from_number and self.to_number
- )
-
- async def send(
- self,
- message: str,
- title: Optional[str] = None,
- level: NotifyLevel = NotifyLevel.INFO,
- **kwargs,
- ) -> NotifyResult:
- try:
- to_number = kwargs.get("to_number") or self.to_number
- from_number = kwargs.get("from_number") or self.from_number
- flow_sid = kwargs.get("flow_sid") or self.flow_sid
- repeat = kwargs.get("repeat") or self.repeat
-
- if not all([self.account_sid, self.auth_token, from_number, to_number]):
- raise ValueError(
- "Twilio requires: account_sid, auth_token, from_number, to_number. "
- "Set SCITEX_NOTIFY_TWILIO_SID/TOKEN/FROM/TO env vars."
- )
-
- loop = asyncio.get_event_loop()
-
- for attempt in range(max(1, repeat)):
- if attempt > 0:
- # Wait 30s between calls (iOS "Repeated Calls" needs
- # same number within 3 min to bypass silent mode)
- await asyncio.sleep(30)
-
- if flow_sid:
- await loop.run_in_executor(
- None,
- lambda: _execute_flow(
- self.account_sid,
- self.auth_token,
- flow_sid,
- from_number,
- to_number,
- ),
- )
- else:
- full_message = f"{title}. {message}" if title else message
- if level == NotifyLevel.CRITICAL:
- full_message = f"Critical alert! {full_message}"
- elif level == NotifyLevel.ERROR:
- full_message = f"Error. {full_message}"
-
- twiml = (
- f""
- f''
- f"{_escape_xml(full_message)}"
- f''
- f''
- f"{_escape_xml(full_message)}"
- f""
- )
-
- await loop.run_in_executor(
- None,
- lambda: _make_call(
- self.account_sid,
- self.auth_token,
- from_number,
- to_number,
- twiml,
- ),
- )
-
- return NotifyResult(
- success=True,
- backend=self.name,
- message=message,
- timestamp=datetime.now().isoformat(),
- details={
- "to": to_number,
- "flow": flow_sid or "direct",
- "repeat": repeat,
- },
- )
- except Exception as e:
- return NotifyResult(
- success=False,
- backend=self.name,
- message=message,
- timestamp=datetime.now().isoformat(),
- error=str(e),
- )
-
-
-def _twilio_request(url: str, account_sid: str, auth_token: str, data: bytes):
- """Make an authenticated Twilio API request."""
- import base64
- import json
- import urllib.request
-
- credentials = base64.b64encode(f"{account_sid}:{auth_token}".encode()).decode(
- "ascii"
- )
-
- req = urllib.request.Request(
- url,
- data=data,
- headers={
- "Authorization": f"Basic {credentials}",
- "Content-Type": "application/x-www-form-urlencoded",
- },
- )
-
- resp = urllib.request.urlopen(req, timeout=30)
- return json.loads(resp.read().decode())
-
-
-def _execute_flow(
- account_sid: str,
- auth_token: str,
- flow_sid: str,
- from_number: str,
- to_number: str,
-) -> None:
- """Execute a Twilio Studio Flow (no SDK dependency)."""
- import urllib.parse
-
- url = f"https://studio.twilio.com/v2/Flows/{flow_sid}/Executions"
- data = urllib.parse.urlencode(
- {
- "To": to_number,
- "From": from_number,
- }
- ).encode("utf-8")
-
- result = _twilio_request(url, account_sid, auth_token, data)
- if result.get("status") == "failed":
- raise RuntimeError(f"Twilio flow failed: {result.get('message', 'unknown')}")
-
-
-def _make_call(
- account_sid: str,
- auth_token: str,
- from_number: str,
- to_number: str,
- twiml: str,
-) -> None:
- """Make a Twilio call using the REST API (no SDK dependency)."""
- import urllib.parse
-
- url = f"https://api.twilio.com/2010-04-01/Accounts/{account_sid}/Calls.json"
- data = urllib.parse.urlencode(
- {
- "To": to_number,
- "From": from_number,
- "Twiml": twiml,
- }
- ).encode("utf-8")
-
- result = _twilio_request(url, account_sid, auth_token, data)
- if result.get("status") in ("failed", "canceled"):
- raise RuntimeError(f"Twilio call failed: {result.get('message', 'unknown')}")
-
-
-def _send_sms(
- account_sid: str,
- auth_token: str,
- from_number: str,
- to_number: str,
- body: str,
-) -> dict:
- """Send an SMS via Twilio REST API (no SDK dependency)."""
- import urllib.parse
-
- url = f"https://api.twilio.com/2010-04-01/Accounts/{account_sid}/Messages.json"
- data = urllib.parse.urlencode(
- {
- "To": to_number,
- "From": from_number,
- "Body": body,
- }
- ).encode("utf-8")
-
- result = _twilio_request(url, account_sid, auth_token, data)
- if result.get("status") == "failed":
- raise RuntimeError(f"Twilio SMS failed: {result.get('message', 'unknown')}")
- return result
-
-
-async def send_sms(
- message: str,
- title: Optional[str] = None,
- to_number: Optional[str] = None,
- from_number: Optional[str] = None,
- account_sid: Optional[str] = None,
- auth_token: Optional[str] = None,
-) -> NotifyResult:
- """Send an SMS message via Twilio.
-
- Parameters
- ----------
- message : str
- SMS body text
- title : str, optional
- Prepended to message if provided
- to_number : str, optional
- Override SCITEX_NOTIFY_TWILIO_TO
- from_number : str, optional
- Override SCITEX_NOTIFY_TWILIO_FROM
- account_sid : str, optional
- Override SCITEX_NOTIFY_TWILIO_SID
- auth_token : str, optional
- Override SCITEX_NOTIFY_TWILIO_TOKEN
-
- Returns
- -------
- NotifyResult
- """
- sid = account_sid or os.getenv("SCITEX_NOTIFY_TWILIO_SID", "")
- token = auth_token or os.getenv("SCITEX_NOTIFY_TWILIO_TOKEN", "")
- from_num = from_number or os.getenv("SCITEX_NOTIFY_TWILIO_FROM", "")
- to_num = to_number or os.getenv("SCITEX_NOTIFY_TWILIO_TO", "")
-
- if not all([sid, token, from_num, to_num]):
- return NotifyResult(
- success=False,
- backend="twilio_sms",
- message=message,
- timestamp=datetime.now().isoformat(),
- error=(
- "Twilio SMS requires: account_sid, auth_token, from_number, to_number. "
- "Set SCITEX_NOTIFY_TWILIO_SID/TOKEN/FROM/TO env vars."
- ),
- )
-
- try:
- body = f"{title}: {message}" if title else message
- loop = asyncio.get_event_loop()
- result = await loop.run_in_executor(
- None,
- lambda: _send_sms(sid, token, from_num, to_num, body),
- )
- return NotifyResult(
- success=True,
- backend="twilio_sms",
- message=message,
- timestamp=datetime.now().isoformat(),
- details={
- "to": to_num,
- "sid": result.get("sid", ""),
- "status": result.get("status", ""),
- },
- )
- except Exception as e:
- return NotifyResult(
- success=False,
- backend="twilio_sms",
- message=message,
- timestamp=datetime.now().isoformat(),
- error=str(e),
- )
-
-
-def _escape_xml(text: str) -> str:
- """Escape XML special characters for TwiML."""
- return (
- text.replace("&", "&")
- .replace("<", "<")
- .replace(">", ">")
- .replace('"', """)
- .replace("'", "'")
- )
-
-
-# EOF
diff --git a/src/scitex/notification/_backends/_types.py b/src/scitex/notification/_backends/_types.py
deleted file mode 100755
index 696d68468..000000000
--- a/src/scitex/notification/_backends/_types.py
+++ /dev/null
@@ -1,58 +0,0 @@
-#!/usr/bin/env python3
-# Timestamp: "2026-01-13 (ywatanabe)"
-# File: /home/ywatanabe/proj/scitex-code/src/scitex/ui/_backends/_types.py
-
-"""Core types for notification backends."""
-
-from __future__ import annotations
-
-from abc import ABC, abstractmethod
-from dataclasses import dataclass
-from enum import Enum
-from typing import Optional
-
-
-class NotifyLevel(Enum):
- """Notification urgency levels."""
-
- INFO = "info"
- WARNING = "warning"
- ERROR = "error"
- CRITICAL = "critical"
-
-
-@dataclass
-class NotifyResult:
- """Result of a notification attempt."""
-
- success: bool
- backend: str
- message: str
- timestamp: str
- error: Optional[str] = None
- details: Optional[dict] = None
-
-
-class BaseNotifyBackend(ABC):
- """Base class for notification backends."""
-
- name: str = "base"
-
- @abstractmethod
- async def send(
- self,
- message: str,
- title: Optional[str] = None,
- level: NotifyLevel = NotifyLevel.INFO,
- **kwargs,
- ) -> NotifyResult:
- """Send a notification."""
- pass
-
- @abstractmethod
- def is_available(self) -> bool:
- """Check if this backend is available."""
- pass
-
-
-# EOF
diff --git a/src/scitex/notification/_backends/_webhook.py b/src/scitex/notification/_backends/_webhook.py
deleted file mode 100755
index a582a3ae9..000000000
--- a/src/scitex/notification/_backends/_webhook.py
+++ /dev/null
@@ -1,81 +0,0 @@
-#!/usr/bin/env python3
-# Timestamp: "2026-01-13 (ywatanabe)"
-# File: /home/ywatanabe/proj/scitex-code/src/scitex/ui/_backends/_webhook.py
-
-"""Webhook notification backend for Slack, Discord, etc."""
-
-from __future__ import annotations
-
-import asyncio
-import json
-import os
-import urllib.request
-from datetime import datetime
-from typing import Optional
-
-from ._types import BaseNotifyBackend, NotifyLevel, NotifyResult
-
-
-class WebhookBackend(BaseNotifyBackend):
- """Webhook notification for Slack, Discord, etc."""
-
- name = "webhook"
-
- def __init__(self, url: Optional[str] = None):
- self.url = (
- url
- or os.getenv("SCITEX_UI_WEBHOOK_URL")
- or os.getenv("SCITEX_NOTIFY_WEBHOOK_URL")
- )
-
- def is_available(self) -> bool:
- return bool(self.url)
-
- async def send(
- self,
- message: str,
- title: Optional[str] = None,
- level: NotifyLevel = NotifyLevel.INFO,
- **kwargs,
- ) -> NotifyResult:
- try:
- url = kwargs.get("url", self.url)
- if not url:
- raise ValueError("No webhook URL configured")
-
- # Format for Slack/Discord compatibility
- payload = {
- "text": f"*{title or 'SciTeX'}*\n{message}",
- "content": f"**{title or 'SciTeX'}**\n{message}",
- }
-
- data = json.dumps(payload).encode("utf-8")
- req = urllib.request.Request(
- url,
- data=data,
- headers={"Content-Type": "application/json"},
- )
-
- loop = asyncio.get_event_loop()
- await loop.run_in_executor(
- None,
- lambda: urllib.request.urlopen(req, timeout=10),
- )
-
- return NotifyResult(
- success=True,
- backend=self.name,
- message=message,
- timestamp=datetime.now().isoformat(),
- )
- except Exception as e:
- return NotifyResult(
- success=False,
- backend=self.name,
- message=message,
- timestamp=datetime.now().isoformat(),
- error=str(e),
- )
-
-
-# EOF
diff --git a/src/scitex/notification/_mcp/__init__.py b/src/scitex/notification/_mcp/__init__.py
deleted file mode 100755
index 4ab69c49c..000000000
--- a/src/scitex/notification/_mcp/__init__.py
+++ /dev/null
@@ -1,23 +0,0 @@
-#!/usr/bin/env python3
-# Timestamp: "2026-01-13 (ywatanabe)"
-# File: /home/ywatanabe/proj/scitex-code/src/scitex/ui/_mcp/__init__.py
-
-"""MCP handlers and schemas for scitex.notification server."""
-
-from .handlers import (
- available_backends_handler,
- get_config_handler,
- list_backends_handler,
- notify_handler,
-)
-from .tool_schemas import get_tool_schemas
-
-__all__ = [
- "get_tool_schemas",
- "notify_handler",
- "list_backends_handler",
- "available_backends_handler",
- "get_config_handler",
-]
-
-# EOF
diff --git a/src/scitex/notification/_mcp/handlers.py b/src/scitex/notification/_mcp/handlers.py
deleted file mode 100755
index a4a56ed2d..000000000
--- a/src/scitex/notification/_mcp/handlers.py
+++ /dev/null
@@ -1,262 +0,0 @@
-#!/usr/bin/env python3
-# Timestamp: "2026-01-13 (ywatanabe)"
-# File: /home/ywatanabe/proj/scitex-code/src/scitex/ui/_mcp/handlers.py
-
-"""MCP handlers for scitex-notify server."""
-
-from __future__ import annotations
-
-from datetime import datetime
-from typing import Optional
-
-__all__ = [
- "notify_handler",
- "notify_by_level_handler",
- "list_backends_handler",
- "available_backends_handler",
- "get_config_handler",
-]
-
-
-async def notify_handler(
- message: str,
- title: Optional[str] = None,
- level: str = "info",
- backend: Optional[str] = None,
- backends: Optional[list[str]] = None,
- timeout: float = 5.0,
- **kwargs,
-) -> dict:
- """Send notification via specified backend(s)."""
- from .._backends import BACKENDS, NotifyLevel, get_backend
- from .._backends._config import get_config
-
- try:
- # Determine notification level
- try:
- notify_level = NotifyLevel(level.lower())
- except ValueError:
- notify_level = NotifyLevel.INFO
-
- # Determine backends to use
- config = get_config()
- if backends:
- backend_list = backends
- elif backend:
- backend_list = [backend]
- else:
- backend_list = [config.default_backend]
-
- results = []
- success_count = 0
-
- for backend_name in backend_list:
- try:
- if backend_name not in BACKENDS:
- results.append(
- {
- "backend": backend_name,
- "success": False,
- "error": f"Unknown backend: {backend_name}",
- }
- )
- continue
-
- b = get_backend(backend_name, **kwargs)
- result = await b.send(
- message,
- title=title,
- level=notify_level,
- timeout=timeout,
- **kwargs,
- )
-
- results.append(
- {
- "backend": backend_name,
- "success": result.success,
- "error": result.error,
- "details": result.details,
- }
- )
-
- if result.success:
- success_count += 1
-
- except Exception as e:
- results.append(
- {
- "backend": backend_name,
- "success": False,
- "error": str(e),
- }
- )
-
- return {
- "success": success_count > 0,
- "message": message,
- "title": title,
- "level": level,
- "backends_used": backend_list,
- "results": results,
- "success_count": success_count,
- "total_count": len(backend_list),
- "timestamp": datetime.now().isoformat(),
- }
-
- except Exception as e:
- return {
- "success": False,
- "error": str(e),
- "timestamp": datetime.now().isoformat(),
- }
-
-
-async def notify_by_level_handler(
- message: str,
- title: Optional[str] = None,
- level: str = "info",
-) -> dict:
- """Send notification using backends configured for the level."""
- from .._backends import NotifyLevel
- from .._backends._config import get_config
-
- try:
- # Determine notification level
- try:
- notify_level = NotifyLevel(level.lower())
- except ValueError:
- notify_level = NotifyLevel.INFO
-
- # Get backends configured for this level
- config = get_config()
- backend_list = config.get_available_backends_for_level(notify_level)
-
- if not backend_list:
- backend_list = [config.default_backend]
-
- # Use notify_handler with determined backends
- return await notify_handler(
- message=message,
- title=title,
- level=level,
- backends=backend_list,
- )
-
- except Exception as e:
- return {
- "success": False,
- "error": str(e),
- "timestamp": datetime.now().isoformat(),
- }
-
-
-async def list_backends_handler() -> dict:
- """List all notification backends with their status."""
- from .._backends import BACKENDS
- from .._backends._config import is_backend_available
-
- try:
- backends_info = []
-
- for name, cls in BACKENDS.items():
- try:
- backend = cls()
- is_available = backend.is_available()
- pkg_available = is_backend_available(name)
-
- backends_info.append(
- {
- "name": name,
- "available": is_available,
- "package_available": pkg_available,
- "class": cls.__name__,
- }
- )
- except Exception as e:
- backends_info.append(
- {
- "name": name,
- "available": False,
- "error": str(e),
- }
- )
-
- return {
- "success": True,
- "backends": backends_info,
- "total_count": len(backends_info),
- "available_count": sum(1 for b in backends_info if b.get("available")),
- "timestamp": datetime.now().isoformat(),
- }
-
- except Exception as e:
- return {
- "success": False,
- "error": str(e),
- "timestamp": datetime.now().isoformat(),
- }
-
-
-async def available_backends_handler() -> dict:
- """Get list of currently available backends."""
- from .._backends import available_backends
-
- try:
- available = available_backends()
-
- return {
- "success": True,
- "available_backends": available,
- "count": len(available),
- "timestamp": datetime.now().isoformat(),
- }
-
- except Exception as e:
- return {
- "success": False,
- "error": str(e),
- "timestamp": datetime.now().isoformat(),
- }
-
-
-async def get_config_handler() -> dict:
- """Get current notification configuration."""
- from .._backends._config import UIConfig, get_config
-
- try:
- # Reset to get fresh config
- UIConfig.reset()
- config = get_config()
-
- return {
- "success": True,
- "config": {
- "default_backend": config.default_backend,
- "backend_priority": config.backend_priority,
- "available_priority": config.get_available_backend_priority(),
- "first_available": config.get_first_available_backend(),
- "level_backends": {
- "info": config._config.get("level_backends", {}).get("info", []),
- "warning": config._config.get("level_backends", {}).get(
- "warning", []
- ),
- "error": config._config.get("level_backends", {}).get("error", []),
- "critical": config._config.get("level_backends", {}).get(
- "critical", []
- ),
- },
- "timeouts": config._config.get("timeouts", {}),
- },
- "timestamp": datetime.now().isoformat(),
- }
-
- except Exception as e:
- return {
- "success": False,
- "error": str(e),
- "timestamp": datetime.now().isoformat(),
- }
-
-
-# EOF
diff --git a/src/scitex/notification/_mcp/tool_schemas.py b/src/scitex/notification/_mcp/tool_schemas.py
deleted file mode 100755
index 2c0abf736..000000000
--- a/src/scitex/notification/_mcp/tool_schemas.py
+++ /dev/null
@@ -1,107 +0,0 @@
-#!/usr/bin/env python3
-# Timestamp: "2026-01-13 (ywatanabe)"
-# File: /home/ywatanabe/proj/scitex-code/src/scitex/ui/_mcp/tool_schemas.py
-
-"""Tool schemas for the scitex-notify MCP server."""
-
-from __future__ import annotations
-
-import mcp.types as types
-
-__all__ = ["get_tool_schemas"]
-
-
-def get_tool_schemas() -> list[types.Tool]:
- """Return all tool schemas for the notification MCP server."""
- return [
- types.Tool(
- name="notify",
- description=(
- "Send a notification via configured backends (audio, desktop, email, "
- "matplotlib, playwright, webhook). Supports multi-backend delivery "
- "and notification levels (info, warning, error, critical)."
- ),
- inputSchema={
- "type": "object",
- "properties": {
- "message": {
- "type": "string",
- "description": "The notification message to send",
- },
- "title": {
- "type": "string",
- "description": "Optional notification title",
- },
- "level": {
- "type": "string",
- "description": "Notification urgency level",
- "enum": ["info", "warning", "error", "critical"],
- "default": "info",
- },
- "backend": {
- "type": "string",
- "description": (
- "Backend to use (audio, desktop, email, matplotlib, "
- "playwright, webhook). If not specified, uses default from config."
- ),
- },
- "backends": {
- "type": "array",
- "items": {"type": "string"},
- "description": "Multiple backends to use simultaneously",
- },
- "timeout": {
- "type": "number",
- "description": "Timeout for visual backends (matplotlib, playwright)",
- "default": 5.0,
- },
- },
- "required": ["message"],
- },
- ),
- types.Tool(
- name="notify_by_level",
- description=(
- "Send notification using backends configured for a specific level. "
- "Uses level_backends config (e.g., critical -> audio + desktop + email)."
- ),
- inputSchema={
- "type": "object",
- "properties": {
- "message": {
- "type": "string",
- "description": "The notification message to send",
- },
- "title": {
- "type": "string",
- "description": "Optional notification title",
- },
- "level": {
- "type": "string",
- "description": "Notification level (determines which backends to use)",
- "enum": ["info", "warning", "error", "critical"],
- "default": "info",
- },
- },
- "required": ["message"],
- },
- ),
- types.Tool(
- name="list_notification_backends",
- description="List all notification backends and their status",
- inputSchema={"type": "object", "properties": {}},
- ),
- types.Tool(
- name="available_notification_backends",
- description="Get list of currently available (working) notification backends",
- inputSchema={"type": "object", "properties": {}},
- ),
- types.Tool(
- name="get_notification_config",
- description="Get current notification configuration (priority, level_backends, timeouts)",
- inputSchema={"type": "object", "properties": {}},
- ),
- ]
-
-
-# EOF
diff --git a/src/scitex/notification/_skills/SKILL.md b/src/scitex/notification/_skills/SKILL.md
deleted file mode 100644
index 51aad2be1..000000000
--- a/src/scitex/notification/_skills/SKILL.md
+++ /dev/null
@@ -1,42 +0,0 @@
----
-name: stx.notification
-description: Multi-backend notification system for alerts, calls, and SMS with automatic fallback ordering.
----
-
-# stx.notification
-
-The `stx.notification` module provides a multi-backend notification system for scientific workflow alerts. It supports desktop notifications, SMS, and phone calls with configurable fallback ordering when backends are unavailable.
-
-## Python API
-
-```python
-import scitex as stx
-
-# Send an alert notification (tries backends in order)
-stx.notification.alert("Experiment complete! Accuracy: 0.95")
-await stx.notification.alert_async("Training finished")
-
-# Make a phone call
-stx.notification.call("+1-555-0123", "Experiment failed, check logs")
-await stx.notification.call_async("+1-555-0123", "message")
-
-# Send SMS
-stx.notification.sms("+1-555-0123", "Results saved to output/")
-await stx.notification.sms_async("+1-555-0123", "message")
-
-# Check available backends
-backends = stx.notification.available_backends()
-
-# Default fallback order
-print(stx.notification.DEFAULT_FALLBACK_ORDER)
-```
-
-## Key Features
-
-- `alert(message)` / `alert_async(message)` — send notification via best available backend
-- `call(number, message)` / `call_async` — automated phone call notification
-- `sms(number, message)` / `sms_async` — SMS notification
-- `available_backends()` — list installed and configured notification backends
-- `DEFAULT_FALLBACK_ORDER` — configured priority order of backends
-- Thin re-export over `scitex-notification` standalone package
-- Both sync and async interfaces for all notification types
diff --git a/src/scitex/notification/_skills/alert.md b/src/scitex/notification/_skills/alert.md
deleted file mode 100644
index 1d499edfe..000000000
--- a/src/scitex/notification/_skills/alert.md
+++ /dev/null
@@ -1,66 +0,0 @@
----
-description: Send notifications to all available backends with alert() and alert_async(). Supports message, title, and urgency level parameters.
----
-
-# Alerts
-
-## alert
-
-Send a notification through all configured backends.
-
-```python
-alert(
- message: str,
- title: str = "SciTeX",
- urgency: str = "normal", # "low" | "normal" | "critical"
-) -> None
-```
-
-Iterates through `DEFAULT_FALLBACK_ORDER` and delivers via every backend that is currently reachable.
-
-**Examples**
-
-```python
-import scitex as stx
-
-# Notify when a long training run finishes
-stx.notification.alert("Training complete. Loss = 0.023")
-
-# With a custom title and urgency
-stx.notification.alert(
- "GPU memory exceeded threshold!",
- title="Resource Warning",
- urgency="critical",
-)
-```
-
-Typical usage at the end of a `@stx.session` script:
-
-```python
-@stx.session
-def main(CONFIG=stx.INJECTED, logger=stx.INJECTED):
- # ... long computation ...
- stx.notification.alert("main.py finished", title="SciTeX Session")
- return 0
-```
-
----
-
-## alert_async
-
-Async variant for use inside `asyncio` event loops.
-
-```python
-await alert_async(message, title="SciTeX", urgency="normal")
-```
-
-```python
-import asyncio
-import scitex as stx
-
-async def run():
- # ... async work ...
- await stx.notification.alert_async("Async job complete.")
-
-asyncio.run(run())
-```
diff --git a/src/scitex/notification/_skills/backends.md b/src/scitex/notification/_skills/backends.md
deleted file mode 100644
index 1eef7c573..000000000
--- a/src/scitex/notification/_skills/backends.md
+++ /dev/null
@@ -1,47 +0,0 @@
----
-description: Discover available notification backends with available_backends() and inspect the DEFAULT_FALLBACK_ORDER list that controls delivery sequence.
----
-
-# Notification Backends
-
-## available_backends
-
-Return a list of backend names whose dependencies are satisfied.
-
-```python
-available_backends() -> list[str]
-```
-
-```python
-import scitex as stx
-
-print(stx.notification.available_backends())
-# ['desktop', 'audio', 'emacs'] — varies by environment
-```
-
----
-
-## DEFAULT_FALLBACK_ORDER
-
-Module-level list defining the sequence in which backends are tried when calling `alert()`.
-
-Built-in backend identifiers:
-
-| Backend | Description |
-|---------|-------------|
-| `"desktop"` | OS desktop notification (`notify-send` / `osascript`) |
-| `"audio"` | Audio chime via `scitex-audio` |
-| `"emacs"` | Emacs message buffer notification |
-| `"email"` | SMTP email (requires `SCITEX_EMAIL_*` env vars) |
-| `"webhook"` | HTTP POST webhook |
-| `"matplotlib"` | Matplotlib figure popup |
-| `"playwright"` | Browser-based notification |
-| `"twilio"` | SMS/voice via Twilio |
-
-```python
-import scitex as stx
-
-print(stx.notification.DEFAULT_FALLBACK_ORDER)
-# Modify the order for your environment:
-# stx.notification.DEFAULT_FALLBACK_ORDER = ["audio", "desktop"]
-```
diff --git a/src/scitex/notification/_skills/voice-sms.md b/src/scitex/notification/_skills/voice-sms.md
deleted file mode 100644
index 321a71fcd..000000000
--- a/src/scitex/notification/_skills/voice-sms.md
+++ /dev/null
@@ -1,60 +0,0 @@
----
-description: Make phone calls with call() / call_async() and send SMS with sms() / sms_async() via Twilio credentials stored as environment variables.
----
-
-# Voice and SMS
-
-Requires Twilio credentials set as environment variables:
-- `SCITEX_TWILIO_ACCOUNT_SID`
-- `SCITEX_TWILIO_AUTH_TOKEN`
-- `SCITEX_TWILIO_FROM_NUMBER`
-- `SCITEX_TWILIO_TO_NUMBER`
-
----
-
-## call / call_async
-
-Place a voice call that reads a message.
-
-```python
-call(message: str, to: str | None = None, repeat: int | None = None) -> None
-await call_async(message: str, to: str | None = None, repeat: int | None = None)
-```
-
-- `to` defaults to `SCITEX_TWILIO_TO_NUMBER` if not provided.
-- `repeat` defaults to `$SCITEX_NOTIFICATION_PHONE_CALL_N_REPEAT` (default: `1`). Set to `1` if iOS Emergency Bypass is configured; `2` triggers iOS "Repeated Calls" bypass.
-
-**Example**
-
-```python
-import scitex as stx
-
-# Call the configured recipient
-stx.notification.call("Experiment finished. Check results.")
-
-# Call twice to bypass iOS silent mode
-stx.notification.call("Wake up!", repeat=2)
-```
-
----
-
-## sms / sms_async
-
-Send a text message.
-
-```python
-sms(message: str, to: str | None = None) -> None
-await sms_async(message: str, to: str | None = None)
-```
-
-**Examples**
-
-```python
-import scitex as stx
-
-stx.notification.sms("Pipeline succeeded. 412 samples processed.")
-
-# Async usage
-import asyncio
-asyncio.run(stx.notification.sms_async("GPU job done."))
-```
diff --git a/src/scitex/notification/mcp_server.py b/src/scitex/notification/mcp_server.py
deleted file mode 100755
index 02216fd9a..000000000
--- a/src/scitex/notification/mcp_server.py
+++ /dev/null
@@ -1,164 +0,0 @@
-#!/usr/bin/env python3
-# Timestamp: "2026-01-13 (ywatanabe)"
-# File: /home/ywatanabe/proj/scitex-code/src/scitex/ui/mcp_server.py
-
-"""MCP Server for SciTeX Notifications - Multi-backend alert system.
-
-.. deprecated::
- This standalone server is deprecated. Use the unified scitex MCP server:
- CLI: scitex serve
- Python: from scitex.mcp_server import run_server
-
-Supports: audio, desktop, email, matplotlib, playwright, webhook backends.
-"""
-
-from __future__ import annotations
-
-import warnings
-
-warnings.warn(
- "scitex.notification.mcp_server is deprecated. Use 'scitex serve' or "
- "'from scitex.mcp_server import run_server' for the unified MCP server.",
- DeprecationWarning,
- stacklevel=2,
-)
-
-import asyncio
-from datetime import datetime
-
-# Graceful MCP dependency handling
-try:
- import mcp.types as types
- from mcp.server import NotificationOptions, Server
- from mcp.server.models import InitializationOptions
- from mcp.server.stdio import stdio_server
-
- MCP_AVAILABLE = True
-except ImportError:
- MCP_AVAILABLE = False
- types = None # type: ignore
- Server = None # type: ignore
- NotificationOptions = None # type: ignore
- InitializationOptions = None # type: ignore
- stdio_server = None # type: ignore
-
-__all__ = ["NotifyServer", "main", "MCP_AVAILABLE"]
-
-
-class NotifyServer:
- """MCP Server for multi-backend notifications."""
-
- def __init__(self):
- self.server = Server("scitex-ui")
- self._notification_count: int = 0
- self.setup_handlers()
-
- def setup_handlers(self):
- """Set up MCP server handlers."""
- from ._mcp.handlers import (
- available_backends_handler,
- get_config_handler,
- list_backends_handler,
- notify_by_level_handler,
- notify_handler,
- )
- from ._mcp.tool_schemas import get_tool_schemas
-
- @self.server.list_tools()
- async def handle_list_tools():
- return get_tool_schemas()
-
- @self.server.call_tool()
- async def handle_call_tool(name: str, arguments: dict):
- if name == "notify":
- self._notification_count += 1
- return await notify_handler(**arguments)
- elif name == "notify_by_level":
- self._notification_count += 1
- return await notify_by_level_handler(**arguments)
- elif name == "list_notification_backends":
- return await list_backends_handler()
- elif name == "available_notification_backends":
- return await available_backends_handler()
- elif name == "get_notification_config":
- return await get_config_handler()
- else:
- raise ValueError(f"Unknown tool: {name}")
-
- @self.server.list_resources()
- async def handle_list_resources():
- # Return notification statistics as a resource
- return [
- types.Resource(
- uri="notify://stats",
- name="Notification Statistics",
- description="Current notification session statistics",
- mimeType="application/json",
- )
- ]
-
- @self.server.read_resource()
- async def handle_read_resource(uri: str):
- if uri == "notify://stats":
- from ._backends import available_backends
- from ._backends._config import get_config
-
- config = get_config()
- stats = {
- "total_notifications": self._notification_count,
- "available_backends": available_backends(),
- "default_backend": config.default_backend,
- "timestamp": datetime.now().isoformat(),
- }
- import json
-
- return types.ResourceContent(
- uri=uri,
- mimeType="application/json",
- content=json.dumps(stats, indent=2),
- )
- raise ValueError(f"Unknown resource: {uri}")
-
-
-async def _run_server():
- """Run the MCP server (internal)."""
- server = NotifyServer()
- async with stdio_server() as (read_stream, write_stream):
- await server.server.run(
- read_stream,
- write_stream,
- InitializationOptions(
- server_name="scitex-ui",
- server_version="0.1.0",
- capabilities=server.server.get_capabilities(
- notification_options=NotificationOptions(),
- experimental_capabilities={},
- ),
- ),
- )
-
-
-def main():
- """Run the MCP server."""
- if not MCP_AVAILABLE:
- import sys
-
- print("=" * 60)
- print("MCP Server 'scitex-ui' requires the 'mcp' package.")
- print()
- print("Install with:")
- print(" pip install mcp")
- print()
- print("Or install scitex with MCP support:")
- print(" pip install scitex[mcp]")
- print("=" * 60)
- sys.exit(1)
-
- asyncio.run(_run_server())
-
-
-if __name__ == "__main__":
- main()
-
-
-# EOF