Skip to content

Commit b424520

Browse files
author
LittleCoinCoin
committed
feat: implement decorator-based strategy registration system
Add MCPHostRegistry with @register_host_strategy decorator for automatic strategy discovery following established Hatchling patterns. Key components: - MCPHostRegistry: Central registry with decorator-based registration - @register_host_strategy: Decorator for automatic strategy registration - MCPHostStrategy: Abstract base class with inheritance validation - MCPHostConfigurationManager: Core manager with backup integration - Family-based host organization (Claude, Cursor, Independent) Features: - Singleton instance management for registered strategies - Inheritance validation ensuring proper MCPHostStrategy subclassing - Host availability detection and family mappings - Integration with backup system for atomic operations - Environment synchronization capabilities Replaces manual registration patterns with automatic discovery, reducing maintenance overhead and improving code organization.
1 parent e984a82 commit b424520

File tree

1 file changed

+309
-0
lines changed

1 file changed

+309
-0
lines changed
Lines changed: 309 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,309 @@
1+
"""
2+
MCP host configuration management with decorator-based strategy registration.
3+
4+
This module provides the core host management infrastructure including
5+
decorator-based strategy registration following Hatchling patterns,
6+
host registry, and configuration manager with consolidated model support.
7+
"""
8+
9+
from typing import Dict, List, Type, Optional, Callable, Any
10+
from pathlib import Path
11+
import json
12+
import logging
13+
14+
from .models import (
15+
MCPHostType, MCPServerConfig, HostConfiguration, EnvironmentData,
16+
ConfigurationResult, SyncResult
17+
)
18+
19+
logger = logging.getLogger(__name__)
20+
21+
22+
class MCPHostRegistry:
23+
"""Registry for MCP host strategies with decorator-based registration."""
24+
25+
_strategies: Dict[MCPHostType, Type["MCPHostStrategy"]] = {}
26+
_instances: Dict[MCPHostType, "MCPHostStrategy"] = {}
27+
_family_mappings: Dict[str, List[MCPHostType]] = {
28+
"claude": [MCPHostType.CLAUDE_DESKTOP, MCPHostType.CLAUDE_CODE],
29+
"cursor": [MCPHostType.CURSOR, MCPHostType.LMSTUDIO]
30+
}
31+
32+
@classmethod
33+
def register(cls, host_type: MCPHostType):
34+
"""Decorator to register a host strategy class."""
35+
def decorator(strategy_class: Type["MCPHostStrategy"]):
36+
if not issubclass(strategy_class, MCPHostStrategy):
37+
raise ValueError(f"Strategy class {strategy_class.__name__} must inherit from MCPHostStrategy")
38+
39+
if host_type in cls._strategies:
40+
logger.warning(f"Overriding existing strategy for {host_type}: {cls._strategies[host_type].__name__} -> {strategy_class.__name__}")
41+
42+
cls._strategies[host_type] = strategy_class
43+
logger.debug(f"Registered MCP host strategy '{host_type}' -> {strategy_class.__name__}")
44+
return strategy_class
45+
return decorator
46+
47+
@classmethod
48+
def get_strategy(cls, host_type: MCPHostType) -> "MCPHostStrategy":
49+
"""Get strategy instance for host type."""
50+
if host_type not in cls._strategies:
51+
available = list(cls._strategies.keys())
52+
raise ValueError(f"Unknown host type: '{host_type}'. Available: {available}")
53+
54+
if host_type not in cls._instances:
55+
cls._instances[host_type] = cls._strategies[host_type]()
56+
57+
return cls._instances[host_type]
58+
59+
@classmethod
60+
def detect_available_hosts(cls) -> List[MCPHostType]:
61+
"""Detect available hosts on the system."""
62+
available_hosts = []
63+
for host_type, strategy_class in cls._strategies.items():
64+
try:
65+
strategy = cls.get_strategy(host_type)
66+
if strategy.is_host_available():
67+
available_hosts.append(host_type)
68+
except Exception:
69+
# Host detection failed, skip
70+
continue
71+
return available_hosts
72+
73+
@classmethod
74+
def get_family_hosts(cls, family: str) -> List[MCPHostType]:
75+
"""Get all hosts in a strategy family."""
76+
return cls._family_mappings.get(family, [])
77+
78+
@classmethod
79+
def get_host_config_path(cls, host_type: MCPHostType) -> Optional[Path]:
80+
"""Get configuration path for host type."""
81+
strategy = cls.get_strategy(host_type)
82+
return strategy.get_config_path()
83+
84+
85+
def register_host_strategy(host_type: MCPHostType) -> Callable:
86+
"""Convenience decorator for registering host strategies."""
87+
return MCPHostRegistry.register(host_type)
88+
89+
90+
class MCPHostStrategy:
91+
"""Abstract base class for host configuration strategies."""
92+
93+
def get_config_path(self) -> Optional[Path]:
94+
"""Get configuration file path for this host."""
95+
raise NotImplementedError("Subclasses must implement get_config_path")
96+
97+
def is_host_available(self) -> bool:
98+
"""Check if host is available on system."""
99+
raise NotImplementedError("Subclasses must implement is_host_available")
100+
101+
def read_configuration(self) -> HostConfiguration:
102+
"""Read and parse host configuration."""
103+
raise NotImplementedError("Subclasses must implement read_configuration")
104+
105+
def write_configuration(self, config: HostConfiguration,
106+
no_backup: bool = False) -> bool:
107+
"""Write configuration to host file."""
108+
raise NotImplementedError("Subclasses must implement write_configuration")
109+
110+
def validate_server_config(self, server_config: MCPServerConfig) -> bool:
111+
"""Validate server configuration for this host."""
112+
raise NotImplementedError("Subclasses must implement validate_server_config")
113+
114+
def get_config_key(self) -> str:
115+
"""Get the root configuration key for MCP servers."""
116+
return "mcpServers" # Default for most platforms
117+
118+
119+
class MCPHostConfigurationManager:
120+
"""Central manager for MCP host configuration operations."""
121+
122+
def __init__(self, backup_manager: Optional[Any] = None):
123+
self.host_registry = MCPHostRegistry
124+
self.backup_manager = backup_manager or self._create_default_backup_manager()
125+
126+
def _create_default_backup_manager(self):
127+
"""Create default backup manager."""
128+
try:
129+
from .backup import MCPHostConfigBackupManager
130+
return MCPHostConfigBackupManager()
131+
except ImportError:
132+
logger.warning("Backup manager not available")
133+
return None
134+
135+
def configure_server(self, server_config: MCPServerConfig,
136+
hostname: str, no_backup: bool = False) -> ConfigurationResult:
137+
"""Configure MCP server on specified host."""
138+
try:
139+
host_type = MCPHostType(hostname)
140+
strategy = self.host_registry.get_strategy(host_type)
141+
142+
# Validate server configuration for this host
143+
if not strategy.validate_server_config(server_config):
144+
return ConfigurationResult(
145+
success=False,
146+
hostname=hostname,
147+
error_message=f"Server configuration invalid for {hostname}"
148+
)
149+
150+
# Read current configuration
151+
current_config = strategy.read_configuration()
152+
153+
# Create backup if requested
154+
backup_path = None
155+
if not no_backup and self.backup_manager:
156+
config_path = strategy.get_config_path()
157+
if config_path and config_path.exists():
158+
backup_result = self.backup_manager.create_backup(config_path, hostname)
159+
if backup_result.success:
160+
backup_path = backup_result.backup_path
161+
162+
# Add server to configuration
163+
server_name = getattr(server_config, 'name', 'default_server')
164+
current_config.add_server(server_name, server_config)
165+
166+
# Write updated configuration
167+
success = strategy.write_configuration(current_config, no_backup=no_backup)
168+
169+
return ConfigurationResult(
170+
success=success,
171+
hostname=hostname,
172+
server_name=server_name,
173+
backup_created=backup_path is not None,
174+
backup_path=backup_path
175+
)
176+
177+
except Exception as e:
178+
return ConfigurationResult(
179+
success=False,
180+
hostname=hostname,
181+
error_message=str(e)
182+
)
183+
184+
def remove_server(self, server_name: str, hostname: str,
185+
no_backup: bool = False) -> ConfigurationResult:
186+
"""Remove MCP server from specified host."""
187+
try:
188+
host_type = MCPHostType(hostname)
189+
strategy = self.host_registry.get_strategy(host_type)
190+
191+
# Read current configuration
192+
current_config = strategy.read_configuration()
193+
194+
# Check if server exists
195+
if server_name not in current_config.servers:
196+
return ConfigurationResult(
197+
success=False,
198+
hostname=hostname,
199+
server_name=server_name,
200+
error_message=f"Server '{server_name}' not found in {hostname} configuration"
201+
)
202+
203+
# Create backup if requested
204+
backup_path = None
205+
if not no_backup and self.backup_manager:
206+
config_path = strategy.get_config_path()
207+
if config_path and config_path.exists():
208+
backup_result = self.backup_manager.create_backup(config_path, hostname)
209+
if backup_result.success:
210+
backup_path = backup_result.backup_path
211+
212+
# Remove server from configuration
213+
current_config.remove_server(server_name)
214+
215+
# Write updated configuration
216+
success = strategy.write_configuration(current_config, no_backup=no_backup)
217+
218+
return ConfigurationResult(
219+
success=success,
220+
hostname=hostname,
221+
server_name=server_name,
222+
backup_created=backup_path is not None,
223+
backup_path=backup_path
224+
)
225+
226+
except Exception as e:
227+
return ConfigurationResult(
228+
success=False,
229+
hostname=hostname,
230+
server_name=server_name,
231+
error_message=str(e)
232+
)
233+
234+
def sync_environment_to_hosts(self, env_data: EnvironmentData,
235+
target_hosts: Optional[List[str]] = None,
236+
no_backup: bool = False) -> SyncResult:
237+
"""Synchronize environment MCP data to host configurations."""
238+
if target_hosts is None:
239+
target_hosts = [host.value for host in self.host_registry.detect_available_hosts()]
240+
241+
results = []
242+
servers_synced = 0
243+
244+
for hostname in target_hosts:
245+
try:
246+
host_type = MCPHostType(hostname)
247+
strategy = self.host_registry.get_strategy(host_type)
248+
249+
# Collect all MCP servers for this host from environment
250+
host_servers = {}
251+
for package in env_data.get_mcp_packages():
252+
if hostname in package.configured_hosts:
253+
host_config = package.configured_hosts[hostname]
254+
# Use package name as server name (single server per package)
255+
host_servers[package.name] = host_config.server_config
256+
257+
if not host_servers:
258+
# No servers to sync for this host
259+
results.append(ConfigurationResult(
260+
success=True,
261+
hostname=hostname,
262+
error_message="No servers to sync"
263+
))
264+
continue
265+
266+
# Read current host configuration
267+
current_config = strategy.read_configuration()
268+
269+
# Create backup if requested
270+
backup_path = None
271+
if not no_backup and self.backup_manager:
272+
config_path = strategy.get_config_path()
273+
if config_path and config_path.exists():
274+
backup_result = self.backup_manager.create_backup(config_path, hostname)
275+
if backup_result.success:
276+
backup_path = backup_result.backup_path
277+
278+
# Update configuration with environment servers
279+
for server_name, server_config in host_servers.items():
280+
current_config.add_server(server_name, server_config)
281+
servers_synced += 1
282+
283+
# Write updated configuration
284+
success = strategy.write_configuration(current_config, no_backup=no_backup)
285+
286+
results.append(ConfigurationResult(
287+
success=success,
288+
hostname=hostname,
289+
backup_created=backup_path is not None,
290+
backup_path=backup_path
291+
))
292+
293+
except Exception as e:
294+
results.append(ConfigurationResult(
295+
success=False,
296+
hostname=hostname,
297+
error_message=str(e)
298+
))
299+
300+
# Calculate summary statistics
301+
successful_results = [r for r in results if r.success]
302+
hosts_updated = len(successful_results)
303+
304+
return SyncResult(
305+
success=hosts_updated > 0,
306+
results=results,
307+
servers_synced=servers_synced,
308+
hosts_updated=hosts_updated
309+
)

0 commit comments

Comments
 (0)