platform_registry.py
1 """ 2 Platform Adapter Registry 3 4 Allows platform adapters (built-in and plugin) to self-register so the gateway 5 can discover and instantiate them without hardcoded if/elif chains. 6 7 Built-in adapters continue to use the existing if/elif in _create_adapter() 8 for now. Plugin adapters register here via PluginContext.register_platform() 9 and are looked up first -- if nothing is found the gateway falls through to 10 the legacy code path. 11 12 Usage (plugin side): 13 14 from gateway.platform_registry import platform_registry, PlatformEntry 15 16 platform_registry.register(PlatformEntry( 17 name="irc", 18 label="IRC", 19 adapter_factory=lambda cfg: IRCAdapter(cfg), 20 check_fn=check_requirements, 21 validate_config=lambda cfg: bool(cfg.extra.get("server")), 22 required_env=["IRC_SERVER"], 23 install_hint="pip install irc", 24 )) 25 26 Usage (gateway side): 27 28 adapter = platform_registry.create_adapter("irc", platform_config) 29 """ 30 31 import logging 32 from dataclasses import dataclass, field 33 from typing import Any, Callable, Optional 34 35 logger = logging.getLogger(__name__) 36 37 38 @dataclass 39 class PlatformEntry: 40 """Metadata and factory for a single platform adapter.""" 41 42 # Identifier used in config.yaml (e.g. "irc", "viber"). 43 name: str 44 45 # Human-readable label (e.g. "IRC", "Viber"). 46 label: str 47 48 # Factory callable: receives a PlatformConfig, returns an adapter instance. 49 # Using a factory instead of a bare class lets plugins do custom init 50 # (e.g. passing extra kwargs, wrapping in try/except). 51 adapter_factory: Callable[[Any], Any] 52 53 # Returns True when the platform's dependencies are available. 54 check_fn: Callable[[], bool] 55 56 # Optional: given a PlatformConfig, is it properly configured? 57 # If None, the registry skips config validation and lets the adapter 58 # fail at connect() time with a descriptive error. 59 validate_config: Optional[Callable[[Any], bool]] = None 60 61 # Optional: given a PlatformConfig, is the platform connected/enabled? 62 # Used by ``GatewayConfig.get_connected_platforms()`` and setup UI status. 63 # If None, falls back to ``validate_config`` or ``check_fn``. 64 is_connected: Optional[Callable[[Any], bool]] = None 65 66 # Env vars this platform needs (for ``hermes setup`` display). 67 required_env: list = field(default_factory=list) 68 69 # Hint shown when check_fn returns False. 70 install_hint: str = "" 71 72 # Optional setup function for interactive configuration. 73 # Signature: () -> None (prompts user, saves env vars). 74 # If None, falls back to _setup_standard_platform (needs token_var + vars) 75 # or a generic "set these env vars" display. 76 setup_fn: Optional[Callable[[], None]] = None 77 78 # "builtin" or "plugin" 79 source: str = "plugin" 80 81 # Name of the plugin manifest that registered this entry (empty for 82 # built-ins). Used by ``hermes gateway setup`` to auto-enable the 83 # owning plugin when the user configures its platform. 84 plugin_name: str = "" 85 86 # ── Auth env var names (for _is_user_authorized integration) ── 87 # E.g. "IRC_ALLOWED_USERS" — checked for comma-separated user IDs. 88 allowed_users_env: str = "" 89 # E.g. "IRC_ALLOW_ALL_USERS" — if truthy, all users authorized. 90 allow_all_env: str = "" 91 92 # ── Message limits ── 93 # Max message length for smart-chunking. 0 = no limit. 94 max_message_length: int = 0 95 96 # ── Privacy ── 97 # If True, session descriptions redact PII (phone numbers, etc.) 98 pii_safe: bool = False 99 100 # ── Display ── 101 # Emoji for CLI/gateway display (e.g. "💬") 102 emoji: str = "🔌" 103 104 # Whether this platform should appear in _UPDATE_ALLOWED_PLATFORMS 105 # (allows /update command from this platform). 106 allow_update_command: bool = True 107 108 # ── LLM guidance ── 109 # Platform hint injected into the system prompt (e.g. "You are on IRC. 110 # Do not use markdown."). Empty string = no hint. 111 platform_hint: str = "" 112 113 114 class PlatformRegistry: 115 """Central registry of platform adapters. 116 117 Thread-safe for reads (dict lookups are atomic under GIL). 118 Writes happen at startup during sequential discovery. 119 """ 120 121 def __init__(self) -> None: 122 self._entries: dict[str, PlatformEntry] = {} 123 124 def register(self, entry: PlatformEntry) -> None: 125 """Register a platform adapter entry. 126 127 If an entry with the same name exists, it is replaced (last writer 128 wins -- this lets plugins override built-in adapters if desired). 129 """ 130 if entry.name in self._entries: 131 prev = self._entries[entry.name] 132 logger.info( 133 "Platform '%s' re-registered (was %s, now %s)", 134 entry.name, 135 prev.source, 136 entry.source, 137 ) 138 self._entries[entry.name] = entry 139 logger.debug("Registered platform adapter: %s (%s)", entry.name, entry.source) 140 141 def unregister(self, name: str) -> bool: 142 """Remove a platform entry. Returns True if it existed.""" 143 return self._entries.pop(name, None) is not None 144 145 def get(self, name: str) -> Optional[PlatformEntry]: 146 """Look up a platform entry by name.""" 147 return self._entries.get(name) 148 149 def all_entries(self) -> list[PlatformEntry]: 150 """Return all registered platform entries.""" 151 return list(self._entries.values()) 152 153 def plugin_entries(self) -> list[PlatformEntry]: 154 """Return only plugin-registered platform entries.""" 155 return [e for e in self._entries.values() if e.source == "plugin"] 156 157 def is_registered(self, name: str) -> bool: 158 return name in self._entries 159 160 def create_adapter(self, name: str, config: Any) -> Optional[Any]: 161 """Create an adapter instance for the given platform name. 162 163 Returns None if: 164 - No entry registered for *name* 165 - check_fn() returns False (missing deps) 166 - validate_config() returns False (misconfigured) 167 - The factory raises an exception 168 """ 169 entry = self._entries.get(name) 170 if entry is None: 171 return None 172 173 if not entry.check_fn(): 174 hint = f" ({entry.install_hint})" if entry.install_hint else "" 175 logger.warning( 176 "Platform '%s' requirements not met%s", 177 entry.label, 178 hint, 179 ) 180 return None 181 182 if entry.validate_config is not None: 183 try: 184 if not entry.validate_config(config): 185 logger.warning( 186 "Platform '%s' config validation failed", 187 entry.label, 188 ) 189 return None 190 except Exception as e: 191 logger.warning( 192 "Platform '%s' config validation error: %s", 193 entry.label, 194 e, 195 ) 196 return None 197 198 try: 199 adapter = entry.adapter_factory(config) 200 return adapter 201 except Exception as e: 202 logger.error( 203 "Failed to create adapter for platform '%s': %s", 204 entry.label, 205 e, 206 exc_info=True, 207 ) 208 return None 209 210 211 # Module-level singleton 212 platform_registry = PlatformRegistry()