|
1 | 1 | import json |
2 | 2 | import logging |
3 | 3 | import os |
| 4 | +import re |
4 | 5 | from typing import Any |
5 | 6 |
|
6 | 7 | logger = logging.getLogger(__name__) |
@@ -66,16 +67,29 @@ def exists(self) -> bool: |
66 | 67 | """Check if mcp.json exists""" |
67 | 68 | return os.path.exists(self.config_path) |
68 | 69 |
|
| 70 | + @staticmethod |
| 71 | + def validate_server_name(name: str) -> None: |
| 72 | + """ |
| 73 | + Validate the server name. |
| 74 | +
|
| 75 | + The server name must only contain letters (a-z, A-Z), numbers (0-9), and hyphens (-). |
| 76 | + Raises a ValueError if the name is invalid. |
| 77 | + """ |
| 78 | + if not re.match(r'^[a-zA-Z0-9-]+$', name): |
| 79 | + raise ValueError(f'Invalid server name "{name}": only letters, numbers, and hyphens are allowed.') |
| 80 | + |
| 81 | + |
69 | 82 | def _load_config(self) -> None: |
70 | 83 | """Load and process MCP configuration.""" |
71 | 84 | try: |
72 | 85 | with open(self.config_path, "r") as f: |
73 | 86 | self._raw_config = json.load(f) |
74 | 87 |
|
75 | 88 | servers_config = self._raw_config.get("servers", {}) |
76 | | - self._servers = { |
77 | | - name: McpServer(name, config) for name, config in servers_config.items() |
78 | | - } |
| 89 | + self._servers = {} |
| 90 | + for name in servers_config.keys(): |
| 91 | + self.validate_server_name(name) |
| 92 | + self._servers[name] = McpServer(name, servers_config[name]) |
79 | 93 |
|
80 | 94 | except json.JSONDecodeError as e: |
81 | 95 | logger.error(f"Invalid JSON in {self.config_path}: {str(e)}") |
|
0 commit comments