Import an MCP server from a file. Args: file: Path to the file server_object: Optional object name in format "module:object" or just "object" Returns: The server object
(file: Path, server_object: str | None = None)
| 117 | |
| 118 | |
| 119 | def _import_server(file: Path, server_object: str | None = None): # pragma: no cover |
| 120 | """Import an MCP server from a file. |
| 121 | |
| 122 | Args: |
| 123 | file: Path to the file |
| 124 | server_object: Optional object name in format "module:object" or just "object" |
| 125 | |
| 126 | Returns: |
| 127 | The server object |
| 128 | """ |
| 129 | # Add parent directory to Python path so imports can be resolved |
| 130 | file_dir = str(file.parent) |
| 131 | if file_dir not in sys.path: |
| 132 | sys.path.insert(0, file_dir) |
| 133 | |
| 134 | # Import the module |
| 135 | spec = importlib.util.spec_from_file_location("server_module", file) |
| 136 | if not spec or not spec.loader: |
| 137 | logger.error("Could not load module", extra={"file": str(file)}) |
| 138 | sys.exit(1) |
| 139 | |
| 140 | module = importlib.util.module_from_spec(spec) |
| 141 | spec.loader.exec_module(module) |
| 142 | |
| 143 | def _check_server_object(server_object: Any, object_name: str): |
| 144 | """Helper function to check that the server object is supported |
| 145 | |
| 146 | Args: |
| 147 | server_object: The server object to check. |
| 148 | |
| 149 | Returns: |
| 150 | True if it's supported. |
| 151 | """ |
| 152 | if not isinstance(server_object, MCPServer): |
| 153 | logger.error(f"The server object {object_name} is of type {type(server_object)} (expecting {MCPServer}).") |
| 154 | if isinstance(server_object, LowLevelServer): |
| 155 | logger.warning("Note that only MCPServer is supported. Low level Server class is not yet supported.") |
| 156 | return False |
| 157 | return True |
| 158 | |
| 159 | # If no object specified, try common server names |
| 160 | if not server_object: |
| 161 | # Look for the most common server object names |
| 162 | for name in ["mcp", "server", "app"]: |
| 163 | if hasattr(module, name): |
| 164 | if not _check_server_object(getattr(module, name), f"{file}:{name}"): |
| 165 | logger.error(f"Ignoring object '{file}:{name}' as it's not a valid server object") |
| 166 | continue |
| 167 | return getattr(module, name) |
| 168 | |
| 169 | logger.error( |
| 170 | f"No server object found in {file}. Please either:\n" |
| 171 | "1. Use a standard variable name (mcp, server, or app)\n" |
| 172 | "2. Specify the object name with file:object syntax" |
| 173 | "3. If the server creates the MCPServer object within main() " |
| 174 | " or another function, refactor the MCPServer object to be a " |
| 175 | " global variable named mcp, server, or app.", |
| 176 | extra={"file": str(file)}, |