Coverage for src/quickmcp/registry.py: 93%
120 statements
« prev ^ index » next coverage.py v7.10.4, created at 2025-08-20 20:56 +0200
1"""
2QuickMCP Server Registry - Register and discover QuickMCP servers
3"""
5import json
6import os
7from pathlib import Path
8from typing import Dict, List, Optional, Any
9from dataclasses import dataclass, asdict
10import subprocess
11import sys
14@dataclass
15class ServerRegistration:
16 """Registration info for a QuickMCP server."""
17 name: str
18 description: str
19 command: List[str] # Command to run the server
20 working_dir: Optional[str] = None
21 tool_prefix: Optional[str] = None
22 capabilities: Optional[Dict[str, Any]] = None
23 metadata: Optional[Dict[str, Any]] = None
25 def to_dict(self) -> Dict[str, Any]:
26 """Convert to dictionary."""
27 return {k: v for k, v in asdict(self).items() if v is not None}
29 @classmethod
30 def from_dict(cls, data: Dict[str, Any]) -> "ServerRegistration":
31 """Create from dictionary."""
32 return cls(**data)
35class ServerRegistry:
36 """Registry for QuickMCP servers that can be launched via stdio."""
38 def __init__(self, registry_path: Optional[Path] = None):
39 """
40 Initialize the server registry.
42 Args:
43 registry_path: Path to registry file (defaults to ~/.quickmcp/registry.json)
44 """
45 if registry_path is None:
46 home = Path.home()
47 registry_dir = home / ".quickmcp"
48 registry_dir.mkdir(parents=True, exist_ok=True)
49 registry_path = registry_dir / "registry.json"
51 self.registry_path = Path(registry_path)
52 self.servers: Dict[str, ServerRegistration] = {}
53 self.load()
55 def load(self) -> None:
56 """Load the registry from disk."""
57 if self.registry_path.exists():
58 try:
59 with open(self.registry_path, 'r') as f:
60 data = json.load(f)
61 self.servers = {
62 name: ServerRegistration.from_dict(info)
63 for name, info in data.items()
64 }
65 except Exception as e:
66 print(f"Warning: Could not load registry: {e}", file=sys.stderr)
67 self.servers = {}
68 else:
69 self.servers = {}
71 def save(self) -> None:
72 """Save the registry to disk."""
73 try:
74 with open(self.registry_path, 'w') as f:
75 data = {
76 name: server.to_dict()
77 for name, server in self.servers.items()
78 }
79 json.dump(data, f, indent=2)
80 except Exception as e:
81 print(f"Warning: Could not save registry: {e}", file=sys.stderr)
83 def register(self, server: ServerRegistration) -> None:
84 """
85 Register a server.
87 Args:
88 server: Server registration info
89 """
90 self.servers[server.name] = server
91 self.save()
93 def unregister(self, name: str) -> bool:
94 """
95 Unregister a server.
97 Args:
98 name: Server name
100 Returns:
101 True if server was unregistered, False if not found
102 """
103 if name in self.servers:
104 del self.servers[name]
105 self.save()
106 return True
107 return False
109 def get(self, name: str) -> Optional[ServerRegistration]:
110 """
111 Get a server registration.
113 Args:
114 name: Server name
116 Returns:
117 Server registration or None if not found
118 """
119 return self.servers.get(name)
121 def list(self) -> List[ServerRegistration]:
122 """
123 List all registered servers.
125 Returns:
126 List of server registrations
127 """
128 return list(self.servers.values())
130 def to_gleitzeit_config(self) -> Dict[str, Any]:
131 """
132 Convert registry to Gleitzeit MCP configuration format.
134 Returns:
135 Configuration dictionary for Gleitzeit
136 """
137 servers = []
138 for server in self.servers.values():
139 config = {
140 "name": server.name,
141 "connection_type": "stdio",
142 "command": server.command,
143 "auto_start": True,
144 }
146 if server.working_dir:
147 config["working_dir"] = server.working_dir
149 if server.tool_prefix:
150 config["tool_prefix"] = server.tool_prefix
152 if server.metadata:
153 config.update(server.metadata)
155 servers.append(config)
157 return {
158 "mcp": {
159 "auto_discover": True,
160 "servers": servers
161 }
162 }
164 def auto_discover(self, search_paths: Optional[List[Path]] = None) -> List[ServerRegistration]:
165 """
166 Auto-discover QuickMCP servers in the filesystem.
168 Args:
169 search_paths: Paths to search (defaults to current directory and common locations)
171 Returns:
172 List of discovered servers
173 """
174 if search_paths is None:
175 search_paths = [
176 Path.cwd(),
177 Path.home() / "quickmcp",
178 Path.home() / "mcp-servers",
179 Path("/usr/local/share/quickmcp"),
180 ]
182 discovered = []
184 for base_path in search_paths:
185 if not base_path.exists():
186 continue
188 # Look for Python files that might be QuickMCP servers
189 for py_file in base_path.rglob("*.py"):
190 if self._is_quickmcp_server(py_file):
191 # Extract server info from the file
192 info = self._extract_server_info(py_file)
193 if info:
194 discovered.append(info)
196 return discovered
198 def _is_quickmcp_server(self, file_path: Path) -> bool:
199 """Check if a Python file is a QuickMCP server."""
200 try:
201 with open(file_path, 'r') as f:
202 content = f.read(1000) # Read first 1000 chars
203 return "QuickMCPServer" in content or "from quickmcp import" in content
204 except:
205 return False
207 def _extract_server_info(self, file_path: Path) -> Optional[ServerRegistration]:
208 """Extract server information from a Python file."""
209 try:
210 # Try to run the file with --info flag to get metadata
211 result = subprocess.run(
212 [sys.executable, str(file_path), "--info"],
213 capture_output=True,
214 text=True,
215 timeout=2
216 )
218 if result.returncode == 0:
219 # Parse JSON output
220 info = json.loads(result.stdout)
221 return ServerRegistration(
222 name=info.get("name", file_path.stem),
223 description=info.get("description", ""),
224 command=[sys.executable, str(file_path)],
225 working_dir=str(file_path.parent),
226 capabilities=info.get("capabilities"),
227 metadata=info.get("metadata")
228 )
229 except:
230 pass
232 # Fallback: create basic registration
233 return ServerRegistration(
234 name=file_path.stem,
235 description=f"QuickMCP server: {file_path.name}",
236 command=[sys.executable, str(file_path)],
237 working_dir=str(file_path.parent)
238 )
241def register_server(
242 name: str,
243 command: List[str],
244 description: str = "",
245 working_dir: Optional[str] = None,
246 tool_prefix: Optional[str] = None,
247 **metadata
248) -> None:
249 """
250 Convenience function to register a server.
252 Args:
253 name: Server name
254 command: Command to run the server
255 description: Server description
256 working_dir: Working directory
257 tool_prefix: Tool prefix for Gleitzeit
258 **metadata: Additional metadata
259 """
260 registry = ServerRegistry()
261 server = ServerRegistration(
262 name=name,
263 description=description,
264 command=command,
265 working_dir=working_dir,
266 tool_prefix=tool_prefix,
267 metadata=metadata if metadata else None
268 )
269 registry.register(server)
270 print(f"Registered server: {name}")
273def list_servers() -> List[ServerRegistration]:
274 """
275 List all registered servers.
277 Returns:
278 List of server registrations
279 """
280 registry = ServerRegistry()
281 return registry.list()
284def export_gleitzeit_config(output_path: Optional[Path] = None) -> None:
285 """
286 Export registry as Gleitzeit configuration.
288 Args:
289 output_path: Output path (defaults to stdout)
290 """
291 registry = ServerRegistry()
292 config = registry.to_gleitzeit_config()
294 import yaml
295 yaml_str = yaml.dump(config, default_flow_style=False)
297 if output_path:
298 with open(output_path, 'w') as f:
299 f.write(yaml_str)
300 print(f"Exported configuration to {output_path}")
301 else:
302 print(yaml_str)