Skip to content

Commit e984a82

Browse files
author
LittleCoinCoin
committed
feat: implement consolidated MCPServerConfig Pydantic model
Add consolidated MCPServerConfig model supporting both local and remote server configurations with proper Pydantic v2 validation patterns. Key features: - Single model replacing separate local/remote server configs - Cross-field validation ensuring either command or URL (not both) - Field combination validation (args with command, headers with URL) - Environment data models with corrected single-server-per-package structure - Configuration result models for operation tracking Eliminates redundant HostServerConfig class and future extension fields (timeout, retry_attempts, ssl_verify) as specified in v2 requirements. Follows organizational standards for Pydantic v2 compatibility with @field_validator and @model_validator decorators.
1 parent c5858ff commit e984a82

File tree

1 file changed

+299
-0
lines changed

1 file changed

+299
-0
lines changed

hatch/mcp_host_config/models.py

Lines changed: 299 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,299 @@
1+
"""
2+
Consolidated Pydantic models for MCP host configuration management.
3+
4+
This module provides the core data models for MCP server configuration,
5+
environment data structures, and host configuration management following
6+
the v2 design specification with consolidated MCPServerConfig model.
7+
"""
8+
9+
from pydantic import BaseModel, Field, field_validator, model_validator
10+
from typing import Dict, List, Optional, Union
11+
from datetime import datetime
12+
from pathlib import Path
13+
from enum import Enum
14+
import logging
15+
16+
logger = logging.getLogger(__name__)
17+
18+
19+
class MCPHostType(str, Enum):
20+
"""Enumeration of supported MCP host types."""
21+
CLAUDE_DESKTOP = "claude-desktop"
22+
CLAUDE_CODE = "claude-code"
23+
VSCODE = "vscode"
24+
CURSOR = "cursor"
25+
LMSTUDIO = "lmstudio"
26+
GEMINI = "gemini"
27+
28+
29+
class MCPServerConfig(BaseModel):
30+
"""Consolidated MCP server configuration supporting local and remote servers."""
31+
32+
# Server identification
33+
name: Optional[str] = Field(None, description="Server name for identification")
34+
35+
# Local server configuration (Pattern A: Command-Based)
36+
command: Optional[str] = Field(None, description="Executable path/name for local servers")
37+
args: Optional[List[str]] = Field(None, description="Command arguments for local servers")
38+
env: Optional[Dict[str, str]] = Field(None, description="Environment variables for local servers")
39+
40+
# Remote server configuration (Pattern B: URL-Based)
41+
url: Optional[str] = Field(None, description="Server endpoint URL for remote servers")
42+
headers: Optional[Dict[str, str]] = Field(None, description="HTTP headers for remote servers")
43+
44+
@model_validator(mode='after')
45+
def validate_server_type(self):
46+
"""Validate that either local or remote configuration is provided, not both."""
47+
command = self.command
48+
url = self.url
49+
50+
if not command and not url:
51+
raise ValueError("Either 'command' (local server) or 'url' (remote server) must be provided")
52+
53+
if command and url:
54+
raise ValueError("Cannot specify both 'command' and 'url' - choose local or remote server")
55+
56+
return self
57+
58+
@field_validator('command')
59+
@classmethod
60+
def validate_command_not_empty(cls, v):
61+
"""Validate command is not empty when provided."""
62+
if v is not None and not v.strip():
63+
raise ValueError("Command cannot be empty")
64+
return v.strip() if v else v
65+
66+
@field_validator('url')
67+
@classmethod
68+
def validate_url_format(cls, v):
69+
"""Validate URL format when provided."""
70+
if v is not None:
71+
if not v.startswith(('http://', 'https://')):
72+
raise ValueError("URL must start with http:// or https://")
73+
return v
74+
75+
@model_validator(mode='after')
76+
def validate_field_combinations(self):
77+
"""Validate field combinations for local vs remote servers."""
78+
# Validate args are only provided with command
79+
if self.args is not None and self.command is None:
80+
raise ValueError("'args' can only be specified with 'command' for local servers")
81+
82+
# Validate env is only provided with command
83+
if self.env is not None and self.command is None:
84+
raise ValueError("'env' can only be specified with 'command' for local servers")
85+
86+
# Validate headers are only provided with URL
87+
if self.headers is not None and self.url is None:
88+
raise ValueError("'headers' can only be specified with 'url' for remote servers")
89+
90+
return self
91+
92+
@property
93+
def is_local_server(self) -> bool:
94+
"""Check if this is a local server configuration."""
95+
return self.command is not None
96+
97+
@property
98+
def is_remote_server(self) -> bool:
99+
"""Check if this is a remote server configuration."""
100+
return self.url is not None
101+
102+
class Config:
103+
"""Pydantic configuration."""
104+
extra = "forbid" # Prevent additional fields for strict validation
105+
json_encoders = {
106+
Path: str
107+
}
108+
109+
110+
class HostConfigurationMetadata(BaseModel):
111+
"""Metadata for host configuration tracking."""
112+
config_path: str = Field(..., description="Path to host configuration file")
113+
configured_at: datetime = Field(..., description="Initial configuration timestamp")
114+
last_synced: datetime = Field(..., description="Last synchronization timestamp")
115+
116+
@field_validator('config_path')
117+
@classmethod
118+
def validate_config_path_not_empty(cls, v):
119+
"""Validate config path is not empty."""
120+
if not v.strip():
121+
raise ValueError("Config path cannot be empty")
122+
return v.strip()
123+
124+
125+
class PackageHostConfiguration(BaseModel):
126+
"""Host configuration for a single package (corrected structure)."""
127+
config_path: str = Field(..., description="Path to host configuration file")
128+
configured_at: datetime = Field(..., description="Initial configuration timestamp")
129+
last_synced: datetime = Field(..., description="Last synchronization timestamp")
130+
server_config: MCPServerConfig = Field(..., description="Server configuration for this host")
131+
132+
@field_validator('config_path')
133+
@classmethod
134+
def validate_config_path_format(cls, v):
135+
"""Validate config path format."""
136+
if not v.strip():
137+
raise ValueError("Config path cannot be empty")
138+
return v.strip()
139+
140+
141+
class EnvironmentPackageEntry(BaseModel):
142+
"""Package entry within environment with corrected MCP structure."""
143+
name: str = Field(..., description="Package name")
144+
version: str = Field(..., description="Package version")
145+
type: str = Field(..., description="Package type (hatch, mcp_standalone, etc.)")
146+
source: str = Field(..., description="Package source")
147+
installed_at: datetime = Field(..., description="Installation timestamp")
148+
configured_hosts: Dict[str, PackageHostConfiguration] = Field(
149+
default_factory=dict,
150+
description="Host configurations for this package's MCP server"
151+
)
152+
153+
@field_validator('name')
154+
@classmethod
155+
def validate_package_name(cls, v):
156+
"""Validate package name format."""
157+
if not v.strip():
158+
raise ValueError("Package name cannot be empty")
159+
# Allow standard package naming patterns
160+
if not v.replace('-', '').replace('_', '').replace('.', '').isalnum():
161+
raise ValueError(f"Invalid package name format: {v}")
162+
return v.strip()
163+
164+
@field_validator('configured_hosts')
165+
@classmethod
166+
def validate_host_names(cls, v):
167+
"""Validate host names are supported."""
168+
supported_hosts = {
169+
'claude-desktop', 'claude-code', 'vscode',
170+
'cursor', 'lmstudio', 'gemini'
171+
}
172+
for host_name in v.keys():
173+
if host_name not in supported_hosts:
174+
raise ValueError(f"Unsupported host: {host_name}. Supported: {supported_hosts}")
175+
return v
176+
177+
178+
class EnvironmentData(BaseModel):
179+
"""Complete environment data structure with corrected MCP integration."""
180+
name: str = Field(..., description="Environment name")
181+
description: str = Field(..., description="Environment description")
182+
created_at: datetime = Field(..., description="Environment creation timestamp")
183+
packages: List[EnvironmentPackageEntry] = Field(
184+
default_factory=list,
185+
description="Packages installed in this environment"
186+
)
187+
python_environment: bool = Field(True, description="Whether this is a Python environment")
188+
python_env: Dict = Field(default_factory=dict, description="Python environment data")
189+
190+
@field_validator('name')
191+
@classmethod
192+
def validate_environment_name(cls, v):
193+
"""Validate environment name format."""
194+
if not v.strip():
195+
raise ValueError("Environment name cannot be empty")
196+
return v.strip()
197+
198+
def get_mcp_packages(self) -> List[EnvironmentPackageEntry]:
199+
"""Get packages that have MCP server configurations."""
200+
return [pkg for pkg in self.packages if pkg.configured_hosts]
201+
202+
def get_standalone_mcp_package(self) -> Optional[EnvironmentPackageEntry]:
203+
"""Get the standalone MCP servers package if it exists."""
204+
for pkg in self.packages:
205+
if pkg.name == "__standalone_mcp_servers__":
206+
return pkg
207+
return None
208+
209+
def add_standalone_mcp_server(self, server_name: str, host_config: PackageHostConfiguration):
210+
"""Add a standalone MCP server configuration."""
211+
standalone_pkg = self.get_standalone_mcp_package()
212+
213+
if standalone_pkg is None:
214+
# Create standalone package entry
215+
standalone_pkg = EnvironmentPackageEntry(
216+
name="__standalone_mcp_servers__",
217+
version="1.0.0",
218+
type="mcp_standalone",
219+
source="user_configured",
220+
installed_at=datetime.now(),
221+
configured_hosts={}
222+
)
223+
self.packages.append(standalone_pkg)
224+
225+
# Add host configuration (single server per package constraint)
226+
for host_name, config in host_config.items():
227+
standalone_pkg.configured_hosts[host_name] = config
228+
229+
230+
class HostConfiguration(BaseModel):
231+
"""Host configuration file structure using consolidated MCPServerConfig."""
232+
servers: Dict[str, MCPServerConfig] = Field(
233+
default_factory=dict,
234+
description="Configured MCP servers"
235+
)
236+
237+
@field_validator('servers')
238+
@classmethod
239+
def validate_servers_not_empty_when_present(cls, v):
240+
"""Validate servers dict structure."""
241+
for server_name, config in v.items():
242+
if not isinstance(config, (dict, MCPServerConfig)):
243+
raise ValueError(f"Invalid server config for {server_name}")
244+
return v
245+
246+
def add_server(self, name: str, config: MCPServerConfig):
247+
"""Add server configuration."""
248+
self.servers[name] = config
249+
250+
def remove_server(self, name: str) -> bool:
251+
"""Remove server configuration."""
252+
if name in self.servers:
253+
del self.servers[name]
254+
return True
255+
return False
256+
257+
class Config:
258+
"""Pydantic configuration."""
259+
arbitrary_types_allowed = True
260+
extra = "allow" # Allow additional host-specific fields
261+
262+
263+
class ConfigurationResult(BaseModel):
264+
"""Result of a configuration operation."""
265+
success: bool = Field(..., description="Whether operation succeeded")
266+
hostname: str = Field(..., description="Target hostname")
267+
server_name: Optional[str] = Field(None, description="Server name if applicable")
268+
backup_created: bool = Field(False, description="Whether backup was created")
269+
backup_path: Optional[Path] = Field(None, description="Path to backup file")
270+
error_message: Optional[str] = Field(None, description="Error message if failed")
271+
272+
@model_validator(mode='after')
273+
def validate_result_consistency(self):
274+
"""Validate result consistency."""
275+
if not self.success and not self.error_message:
276+
raise ValueError("Error message required when success=False")
277+
278+
return self
279+
280+
281+
class SyncResult(BaseModel):
282+
"""Result of environment synchronization operation."""
283+
success: bool = Field(..., description="Whether overall sync succeeded")
284+
results: List[ConfigurationResult] = Field(..., description="Individual host results")
285+
servers_synced: int = Field(..., description="Total servers synchronized")
286+
hosts_updated: int = Field(..., description="Number of hosts updated")
287+
288+
@property
289+
def failed_hosts(self) -> List[str]:
290+
"""Get list of hosts that failed synchronization."""
291+
return [r.hostname for r in self.results if not r.success]
292+
293+
@property
294+
def success_rate(self) -> float:
295+
"""Calculate success rate percentage."""
296+
if not self.results:
297+
return 0.0
298+
successful = len([r for r in self.results if r.success])
299+
return (successful / len(self.results)) * 100.0

0 commit comments

Comments
 (0)