|
2 | 2 | import logging |
3 | 3 | import os |
4 | 4 | from typing import Any, Dict, List, Optional |
| 5 | +import re |
5 | 6 |
|
6 | 7 | logger = logging.getLogger(__name__) |
7 | 8 |
|
@@ -50,16 +51,29 @@ def exists(self) -> bool: |
50 | 51 | """Check if mcp.json exists""" |
51 | 52 | return os.path.exists(self.config_path) |
52 | 53 |
|
| 54 | + @staticmethod |
| 55 | + def validate_server_name(name: str) -> None: |
| 56 | + """ |
| 57 | + Validate the server name. |
| 58 | +
|
| 59 | + The server name must only contain letters (a-z, A-Z), numbers (0-9), and hyphens (-). |
| 60 | + Raises a ValueError if the name is invalid. |
| 61 | + """ |
| 62 | + if not re.match(r'^[a-zA-Z0-9-]+$', name): |
| 63 | + raise ValueError(f'Invalid server name "{name}": only letters, numbers, and hyphens are allowed.') |
| 64 | + |
| 65 | + |
53 | 66 | def _load_config(self) -> None: |
54 | 67 | """Load and process MCP configuration.""" |
55 | 68 | try: |
56 | 69 | with open(self.config_path, "r") as f: |
57 | 70 | self._raw_config = json.load(f) |
58 | 71 |
|
59 | 72 | servers_config = self._raw_config.get("servers", {}) |
60 | | - self._servers = { |
61 | | - name: McpServer(name, config) for name, config in servers_config.items() |
62 | | - } |
| 73 | + self._servers = {} |
| 74 | + for name in servers_config.keys(): |
| 75 | + self.validate_server_name(name) |
| 76 | + self._servers[name] = McpServer(name, servers_config[name]) |
63 | 77 |
|
64 | 78 | except json.JSONDecodeError as e: |
65 | 79 | logger.error(f"Invalid JSON in {self.config_path}: {str(e)}") |
|
0 commit comments