From 2b379e5134aa37b7269cdc3cc50b71bdca2e73aa Mon Sep 17 00:00:00 2001 From: aaalrajhi Date: Thu, 26 Mar 2026 15:29:34 +0300 Subject: [PATCH 1/2] feat: add multi-host support for connecting to multiple SQL Servers Allow a single MCP server instance to connect to multiple SQL Server hosts. Connections are configured via indexed env vars (MSSQL_HOST__*) or a JSON config file (MSSQL_CONNECTIONS_FILE). In multi-host mode, execute_sql gains a required 'connection' parameter and a new list_connections tool is exposed. Fully backward compatible - existing single-host setups (MSSQL_SERVER/USER/etc.) continue to work unchanged. --- README.md | 114 ++++++ src/mssql_mcp_server/server.py | 542 ++++++++++++++++++++++------ tests/test_config.py | 640 +++++++++++++++++++++++++++------ tests/test_server.py | 138 ++++++- 4 files changed, 1206 insertions(+), 228 deletions(-) diff --git a/README.md b/README.md index 816458a..afbc8e0 100644 --- a/README.md +++ b/README.md @@ -16,6 +16,7 @@ A Model Context Protocol (MCP) server for secure SQL Server database access thro - 🔐 Multiple authentication methods (SQL, Windows, Azure AD) - 🏢 LocalDB and Azure SQL support - 🔌 Custom port configuration +- 🌐 **Multi-host support** — connect to multiple SQL Servers from a single MCP instance ## Quick Start @@ -72,6 +73,119 @@ MSSQL_PORT=1433 # Custom port (default: 1433) MSSQL_ENCRYPT=true # Force encryption ``` +### Multi-Host Configuration + +Connect to **multiple SQL Servers** from a single MCP server instance. Each +connection has its own name, credentials, and settings. + +When multi-host mode is active the `execute_sql` tool gains a required +`connection` parameter, and a new `list_connections` tool becomes available. + +#### Option 1 — Indexed environment variables (recommended) + +Define hosts with `MSSQL_HOST__*` env vars. Each index is a separate +connection: + +```json +{ + "mcpServers": { + "mssql": { + "command": "uvx", + "args": ["microsoft_sql_server_mcp"], + "env": { + "MSSQL_HOST_1_NAME": "prod", + "MSSQL_HOST_1_SERVER": "prod.example.com", + "MSSQL_HOST_1_DATABASE": "proddb", + "MSSQL_HOST_1_USER": "admin", + "MSSQL_HOST_1_PASSWORD": "secret", + + "MSSQL_HOST_2_NAME": "dev", + "MSSQL_HOST_2_SERVER": "dev.example.com", + "MSSQL_HOST_2_DATABASE": "devdb", + "MSSQL_HOST_2_USER": "dev", + "MSSQL_HOST_2_PASSWORD": "devpass", + "MSSQL_HOST_2_PORT": "1434" + } + } + } +} +``` + +Indexes don't have to be contiguous — `1, 2, 5, 10` works fine. + +#### Option 2 — JSON config file + +Point `MSSQL_CONNECTIONS_FILE` to a JSON file on disk: + +```json +{ + "mcpServers": { + "mssql": { + "command": "uvx", + "args": ["microsoft_sql_server_mcp"], + "env": { + "MSSQL_CONNECTIONS_FILE": "/path/to/connections.json" + } + } + } +} +``` + +Where `connections.json` looks like: + +```json +[ + { + "name": "prod", + "server": "prod.example.com", + "database": "proddb", + "user": "admin", + "password": "secret" + }, + { + "name": "dev", + "server": "dev.example.com", + "database": "devdb", + "user": "dev", + "password": "devpass", + "port": "1434" + }, + { + "name": "local", + "server": "localhost", + "database": "localdb", + "windows_auth": "true" + } +] +``` + +#### Connection fields + +Each host supports these fields (as env var suffixes or JSON keys): + +| Field / Suffix | Required | Default | Description | +|------------------|----------|-------------|------------------------------------------| +| `NAME` | Yes | — | Unique connection identifier | +| `SERVER` | No | `localhost` | SQL Server hostname | +| `DATABASE` | Yes | — | Database name | +| `USER` | Cond. | — | Username (required for SQL auth) | +| `PASSWORD` | Cond. | — | Password (required for SQL auth) | +| `PORT` | No | `1433` | TCP port | +| `ENCRYPT` | No | `false` | Enable encryption (`true`/`false`) | +| `WINDOWS_AUTH` | No | `false` | Use Windows authentication | + +#### Resolution order + +1. `MSSQL_HOST__*` env vars (highest priority) +2. `MSSQL_CONNECTIONS_FILE` JSON file +3. Legacy single-host env vars (`MSSQL_SERVER`, `MSSQL_USER`, etc.) + +#### Backward compatibility + +If no `MSSQL_HOST_*` env vars or `MSSQL_CONNECTIONS_FILE` are set, the server +falls back to the original single-host environment variables — **no changes +needed for existing setups**. + ## Alternative Installation Methods ### Using pip diff --git a/src/mssql_mcp_server/server.py b/src/mssql_mcp_server/server.py index c4d304b..27cae54 100644 --- a/src/mssql_mcp_server/server.py +++ b/src/mssql_mcp_server/server.py @@ -1,4 +1,5 @@ import asyncio +import json import logging import os import re @@ -9,19 +10,19 @@ # Configure logging logging.basicConfig( - level=logging.INFO, - format='%(asctime)s - %(name)s - %(levelname)s - %(message)s' + level=logging.INFO, format="%(asctime)s - %(name)s - %(levelname)s - %(message)s" ) logger = logging.getLogger("mssql_mcp_server") + def validate_table_name(table_name: str) -> str: """Validate and escape table name to prevent SQL injection.""" # Allow only alphanumeric, underscore, and dot (for schema.table) - if not re.match(r'^[a-zA-Z0-9_]+(\.[a-zA-Z0-9_]+)?$', table_name): + if not re.match(r"^[a-zA-Z0-9_]+(\.[a-zA-Z0-9_]+)?$", table_name): raise ValueError(f"Invalid table name: {table_name}") - + # Split schema and table if present - parts = table_name.split('.') + parts = table_name.split(".") if len(parts) == 2: # Escape both schema and table name return f"[{parts[0]}].[{parts[1]}]" @@ -29,13 +30,16 @@ def validate_table_name(table_name: str) -> str: # Just table name return f"[{table_name}]" + def get_db_config(): """Get database configuration from environment variables.""" # Basic configuration server = os.getenv("MSSQL_SERVER", "localhost") - logger.info(f"MSSQL_SERVER environment variable: {os.getenv('MSSQL_SERVER', 'NOT SET')}") + logger.info( + f"MSSQL_SERVER environment variable: {os.getenv('MSSQL_SERVER', 'NOT SET')}" + ) logger.info(f"Using server: {server}") - + # Handle LocalDB connections (Issue #6) # LocalDB format: (localdb)\instancename if server.startswith("(localdb)\\"): @@ -44,14 +48,14 @@ def get_db_config(): instance_name = server.replace("(localdb)\\", "") server = f".\\{instance_name}" logger.info(f"Detected LocalDB connection, converted to: {server}") - + config = { "server": server, "user": os.getenv("MSSQL_USER"), "password": os.getenv("MSSQL_PASSWORD"), "database": os.getenv("MSSQL_DATABASE"), "port": os.getenv("MSSQL_PORT", "1433"), # Default MSSQL port - } + } # Port support (Issue #8) port = os.getenv("MSSQL_PORT") if port: @@ -59,7 +63,7 @@ def get_db_config(): config["port"] = int(port) except ValueError: logger.warning(f"Invalid MSSQL_PORT value: {port}. Using default port.") - + # Encryption settings for Azure SQL (Issue #11) # Check if we're connecting to Azure SQL if config["server"] and ".database.windows.net" in config["server"]: @@ -74,11 +78,13 @@ def get_db_config(): encrypt_str = os.getenv("MSSQL_ENCRYPT", "false") if encrypt_str.lower() == "true": config["tds_version"] = "7.4" # Keep existing TDS approach - config["server"] += ";Encrypt=yes;TrustServerCertificate=yes" # Add explicit setting - + config["server"] += ( + ";Encrypt=yes;TrustServerCertificate=yes" # Add explicit setting + ) + # Windows Authentication support (Issue #7) use_windows_auth = os.getenv("MSSQL_WINDOWS_AUTH", "false").lower() == "true" - + if use_windows_auth: # For Windows authentication, user and password are not required if not config["database"]: @@ -91,202 +97,532 @@ def get_db_config(): else: # SQL Authentication - user and password are required if not all([config["user"], config["password"], config["database"]]): - logger.error("Missing required database configuration. Please check environment variables:") + logger.error( + "Missing required database configuration. Please check environment variables:" + ) logger.error("MSSQL_USER, MSSQL_PASSWORD, and MSSQL_DATABASE are required") raise ValueError("Missing required database configuration") - + + return config + + +def build_connection_config(conn: dict) -> dict: + """Build a pymssql-compatible config dict from a connection definition. + + Accepts a dict with keys: server, database, user, password, port, + encrypt, windows_auth. Applies the same Azure/LocalDB/encryption + logic as get_db_config() so every connection is treated uniformly. + """ + server = conn.get("server", "localhost") + + # Handle LocalDB connections + if server.startswith("(localdb)\\"): + instance_name = server.replace("(localdb)\\", "") + server = f".\\{instance_name}" + logger.info(f"Detected LocalDB connection, converted to: {server}") + + config: dict = { + "server": server, + "database": conn.get("database"), + } + + # Port + port = conn.get("port") + if port is not None: + try: + config["port"] = int(port) + except (ValueError, TypeError): + logger.warning(f"Invalid port value: {port}. Skipping.") + + # Encryption / Azure + if config["server"] and ".database.windows.net" in config["server"]: + config["tds_version"] = "7.4" + encrypt = str(conn.get("encrypt", "true")).lower() + if encrypt == "true": + config["server"] += ";Encrypt=yes;TrustServerCertificate=no" + else: + encrypt = str(conn.get("encrypt", "false")).lower() + if encrypt == "true": + config["tds_version"] = "7.4" + config["server"] += ";Encrypt=yes;TrustServerCertificate=yes" + + # Authentication + use_windows_auth = str(conn.get("windows_auth", "false")).lower() == "true" + + if use_windows_auth: + if not config["database"]: + raise ValueError("Missing required database configuration") + logger.info("Using Windows Authentication") + else: + user = conn.get("user") + password = conn.get("password") + if not all([user, password, config["database"]]): + raise ValueError( + "Missing required database configuration " + "(user, password, and database are required for SQL authentication)" + ) + config["user"] = user + config["password"] = password + return config + +def _scan_host_env_vars() -> list[dict] | None: + """Scan for ``MSSQL_HOST__*`` indexed environment variables. + + Returns a list of connection dicts (one per host index found) or + ``None`` when no ``MSSQL_HOST_*`` variables exist. + + Expected env var pattern:: + + MSSQL_HOST_1_NAME=prod + MSSQL_HOST_1_SERVER=prod.example.com + MSSQL_HOST_1_DATABASE=proddb + MSSQL_HOST_1_USER=admin + MSSQL_HOST_1_PASSWORD=secret + MSSQL_HOST_1_PORT=1433 + MSSQL_HOST_1_ENCRYPT=false + MSSQL_HOST_1_WINDOWS_AUTH=false + + Indexes don't have to be contiguous — any positive integer works. + """ + # Collect all unique host indexes present in the environment. + prefix = "MSSQL_HOST_" + indexes: set[int] = set() + for key in os.environ: + if key.startswith(prefix): + # e.g. MSSQL_HOST_1_NAME → parts = ["MSSQL", "HOST", "1", "NAME"] + parts = key.split("_") + if len(parts) >= 4: + try: + indexes.add(int(parts[2])) + except ValueError: + continue + + if not indexes: + return None + + # Map of allowed suffixes → connection-dict key + field_map = { + "NAME": "name", + "SERVER": "server", + "DATABASE": "database", + "USER": "user", + "PASSWORD": "password", + "PORT": "port", + "ENCRYPT": "encrypt", + "WINDOWS_AUTH": "windows_auth", + } + + hosts: list[dict] = [] + for idx in sorted(indexes): + entry: dict = {} + for suffix, field in field_map.items(): + value = os.getenv(f"{prefix}{idx}_{suffix}") + if value is not None: + entry[field] = value + if entry: + hosts.append(entry) + + return hosts if hosts else None + + +def get_all_connections() -> dict[str, dict]: + """Return a mapping of connection-name → pymssql config dict. + + Resolution order: + 1. ``MSSQL_HOST__*`` indexed environment variables. + 2. ``MSSQL_CONNECTIONS_FILE`` – path to a JSON file with an array of + connection objects (each must have a ``name`` key). + 3. Fall back to the legacy single-connection env vars via + ``get_db_config()`` (connection name = ``"default"``). + """ + raw: list[dict] | None = None + + # 1. Indexed env vars (MSSQL_HOST_1_*, MSSQL_HOST_2_*, …) + raw = _scan_host_env_vars() + if raw is not None: + logger.info(f"Loaded {len(raw)} connection(s) from MSSQL_HOST_* env vars") + + # 2. File-based config + if raw is None: + connections_file = os.getenv("MSSQL_CONNECTIONS_FILE") + if connections_file: + try: + with open(connections_file, "r", encoding="utf-8") as fh: + raw = json.load(fh) + logger.info(f"Loaded connections from file: {connections_file}") + except (OSError, json.JSONDecodeError) as exc: + logger.error( + f"Failed to load connections file '{connections_file}': {exc}" + ) + raise ValueError(f"Invalid connections file: {exc}") from exc + + # 3. Legacy single-host fallback + if raw is None: + return {"default": get_db_config()} + + # Validate & build + if not isinstance(raw, list) or len(raw) == 0: + raise ValueError("MSSQL connections config must be a non-empty JSON array") + + connections: dict[str, dict] = {} + for idx, entry in enumerate(raw): + if not isinstance(entry, dict): + raise ValueError(f"Connection entry at index {idx} is not a JSON object") + name = entry.get("name") + if not name or not isinstance(name, str): + raise ValueError( + f"Connection entry at index {idx} is missing a valid 'name' field" + ) + if name in connections: + raise ValueError(f"Duplicate connection name: '{name}'") + connections[name] = build_connection_config(entry) + + return connections + + +def is_multi_host() -> bool: + """Return True when multi-host configuration is active.""" + return bool( + _scan_host_env_vars() is not None or os.getenv("MSSQL_CONNECTIONS_FILE") + ) + + +def get_connection_config(connection_name: str) -> dict: + """Resolve a single connection by name.""" + conns = get_all_connections() + if connection_name not in conns: + available = ", ".join(sorted(conns.keys())) + raise ValueError( + f"Unknown connection '{connection_name}'. " + f"Available connections: {available}" + ) + return conns[connection_name] + + def get_command(): """Get the command to execute SQL queries.""" return os.getenv("MSSQL_COMMAND", "execute_sql") + def is_select_query(query: str) -> bool: """ Check if a query is a SELECT statement, accounting for comments. Handles both single-line (--) and multi-line (/* */) SQL comments. """ # Remove multi-line comments /* ... */ - query_cleaned = re.sub(r'/\*.*?\*/', '', query, flags=re.DOTALL) - + query_cleaned = re.sub(r"/\*.*?\*/", "", query, flags=re.DOTALL) + # Remove single-line comments -- ... - lines = query_cleaned.split('\n') + lines = query_cleaned.split("\n") cleaned_lines = [] for line in lines: # Find -- comment marker and remove everything after it - comment_pos = line.find('--') + comment_pos = line.find("--") if comment_pos != -1: line = line[:comment_pos] cleaned_lines.append(line) - - query_cleaned = '\n'.join(cleaned_lines) - + + query_cleaned = "\n".join(cleaned_lines) + # Get the first non-empty word after stripping whitespace first_word = query_cleaned.strip().split()[0] if query_cleaned.strip() else "" return first_word.upper() == "SELECT" + # Initialize server app = Server("mssql_mcp_server") + @app.list_resources() async def list_resources() -> list[Resource]: - """List SQL Server tables as resources.""" - config = get_db_config() - try: - conn = pymssql.connect(**config) - cursor = conn.cursor() - # Query to get user tables from the current database - cursor.execute(""" - SELECT TABLE_NAME - FROM INFORMATION_SCHEMA.TABLES - WHERE TABLE_TYPE = 'BASE TABLE' - """) - tables = cursor.fetchall() - logger.info(f"Found tables: {tables}") - - resources = [] - for table in tables: - resources.append( - Resource( - uri=f"mssql://{table[0]}/data", - name=f"Table: {table[0]}", - mimeType="text/plain", - description=f"Data in table: {table[0]}" + """List SQL Server tables as resources. + + In multi-host mode the URI contains the connection name so that + ``read_resource`` can route to the correct server: + mssql:////data + In single-host (legacy) mode the original format is preserved: + mssql://
/data + """ + connections = get_all_connections() + multi = is_multi_host() + resources: list[Resource] = [] + + for conn_name, config in connections.items(): + try: + db_conn = pymssql.connect(**config) + cursor = db_conn.cursor() + cursor.execute(""" + SELECT TABLE_NAME + FROM INFORMATION_SCHEMA.TABLES + WHERE TABLE_TYPE = 'BASE TABLE' + """) + tables = cursor.fetchall() + logger.info(f"[{conn_name}] Found tables: {tables}") + + for table in tables: + if multi: + uri = f"mssql://{conn_name}/{table[0]}/data" + label = f"[{conn_name}] Table: {table[0]}" + desc = f"Data in table {table[0]} on connection '{conn_name}'" + else: + uri = f"mssql://{table[0]}/data" + label = f"Table: {table[0]}" + desc = f"Data in table: {table[0]}" + + resources.append( + Resource( + uri=uri, + name=label, + mimeType="text/plain", + description=desc, + ) ) - ) - cursor.close() - conn.close() - return resources - except Exception as e: - logger.error(f"Failed to list resources: {str(e)}") - return [] + cursor.close() + db_conn.close() + except Exception as e: + logger.error(f"[{conn_name}] Failed to list resources: {str(e)}") + + return resources + @app.read_resource() async def read_resource(uri: AnyUrl) -> str: - """Read table contents.""" - config = get_db_config() + """Read table contents. + + URI formats: + Single-host: mssql://
/data + Multi-host: mssql:///
/data + """ uri_str = str(uri) logger.info(f"Reading resource: {uri_str}") - + if not uri_str.startswith("mssql://"): raise ValueError(f"Invalid URI scheme: {uri_str}") - - parts = uri_str[8:].split('/') - table = parts[0] - + + parts = uri_str[8:].split("/") + multi = is_multi_host() + + if multi: + # mssql:///
/data → parts = [conn, table, "data"] + if len(parts) < 2: + raise ValueError(f"Invalid multi-host URI: {uri_str}") + conn_name = parts[0] + table = parts[1] + config = get_connection_config(conn_name) + else: + # mssql://
/data → parts = [table, "data"] + table = parts[0] + config = get_db_config() + try: - # Validate table name to prevent SQL injection safe_table = validate_table_name(table) - - conn = pymssql.connect(**config) - cursor = conn.cursor() - # Use TOP 100 for MSSQL (equivalent to LIMIT in MySQL) + + db_conn = pymssql.connect(**config) + cursor = db_conn.cursor() cursor.execute(f"SELECT TOP 100 * FROM {safe_table}") columns = [desc[0] for desc in cursor.description] rows = cursor.fetchall() result = [",".join(map(str, row)) for row in rows] cursor.close() - conn.close() + db_conn.close() return "\n".join([",".join(columns)] + result) - + except Exception as e: logger.error(f"Database error reading resource {uri}: {str(e)}") raise RuntimeError(f"Database error: {str(e)}") + @app.list_tools() async def list_tools() -> list[Tool]: - """List available SQL Server tools.""" + """List available SQL Server tools. + + In multi-host mode an extra ``connection`` parameter is required on the + execute tool so the caller can target a specific server, and a + ``list_connections`` tool is exposed. + """ command = get_command() + multi = is_multi_host() logger.info("Listing tools...") - return [ + + # -- execute_sql (or custom command name) -------------------------------- + properties: dict = { + "query": { + "type": "string", + "description": "The SQL query to execute", + } + } + required = ["query"] + + if multi: + connections = get_all_connections() + conn_names = sorted(connections.keys()) + properties["connection"] = { + "type": "string", + "description": ( + "Name of the database connection to use. " + f"Available: {', '.join(conn_names)}" + ), + "enum": conn_names, + } + required.append("connection") + execute_description = ( + "Execute an SQL query on the specified SQL Server connection" + ) + else: + execute_description = "Execute an SQL query on the SQL Server" + + tools: list[Tool] = [ Tool( name=command, - description="Execute an SQL query on the SQL Server", + description=execute_description, inputSchema={ "type": "object", - "properties": { - "query": { - "type": "string", - "description": "The SQL query to execute" - } - }, - "required": ["query"] - } + "properties": properties, + "required": required, + }, ) ] + # -- list_connections (multi-host only) ----------------------------------- + if multi: + tools.append( + Tool( + name="list_connections", + description="List all configured database connections", + inputSchema={ + "type": "object", + "properties": {}, + }, + ) + ) + + return tools + + @app.call_tool() async def call_tool(name: str, arguments: dict) -> list[TextContent]: - """Execute SQL commands.""" - config = get_db_config() + """Execute SQL commands or list connections.""" command = get_command() + multi = is_multi_host() logger.info(f"Calling tool: {name} with arguments: {arguments}") - + + # -- list_connections ---------------------------------------------------- + if name == "list_connections": + connections = get_all_connections() + lines: list[str] = [] + for cname, cfg in connections.items(): + server_info = cfg.get("server", "?") + if "port" in cfg: + server_info += f":{cfg['port']}" + db = cfg.get("database", "?") + user = cfg.get("user", "Windows Auth") + lines.append(f"- {cname}: {server_info}/{db} (user: {user})") + return [ + TextContent( + type="text", text="Configured connections:\n" + "\n".join(lines) + ) + ] + + # -- execute_sql (or custom command) ------------------------------------- if name != command: raise ValueError(f"Unknown tool: {name}") - + query = arguments.get("query") if not query: raise ValueError("Query is required") - + + # Resolve the target connection config + if multi: + connection_name = arguments.get("connection") + if not connection_name: + raise ValueError( + "The 'connection' parameter is required in multi-host mode" + ) + config = get_connection_config(connection_name) + else: + config = get_db_config() + try: - conn = pymssql.connect(**config) - cursor = conn.cursor() + db_conn = pymssql.connect(**config) + cursor = db_conn.cursor() cursor.execute(query) - + # Special handling for table listing if is_select_query(query) and "INFORMATION_SCHEMA.TABLES" in query.upper(): tables = cursor.fetchall() - result = ["Tables_in_" + config["database"]] # Header + result = ["Tables_in_" + config.get("database", "unknown")] result.extend([table[0] for table in tables]) cursor.close() - conn.close() + db_conn.close() return [TextContent(type="text", text="\n".join(result))] - + # Regular SELECT queries elif is_select_query(query): columns = [desc[0] for desc in cursor.description] rows = cursor.fetchall() result = [",".join(map(str, row)) for row in rows] cursor.close() - conn.close() - return [TextContent(type="text", text="\n".join([",".join(columns)] + result))] - + db_conn.close() + return [ + TextContent(type="text", text="\n".join([",".join(columns)] + result)) + ] + # Non-SELECT queries else: - conn.commit() + db_conn.commit() affected_rows = cursor.rowcount cursor.close() - conn.close() - return [TextContent(type="text", text=f"Query executed successfully. Rows affected: {affected_rows}")] - + db_conn.close() + return [ + TextContent( + type="text", + text=f"Query executed successfully. Rows affected: {affected_rows}", + ) + ] + except Exception as e: logger.error(f"Error executing SQL '{query}': {e}") return [TextContent(type="text", text=f"Error executing query: {str(e)}")] + async def main(): """Main entry point to run the MCP server.""" from mcp.server.stdio import stdio_server - + logger.info("Starting MSSQL MCP server...") - config = get_db_config() + # Log connection info without exposing sensitive data - server_info = config['server'] - if 'port' in config: - server_info += f":{config['port']}" - user_info = config.get('user', 'Windows Auth') - logger.info(f"Database config: {server_info}/{config['database']} as {user_info}") - + connections = get_all_connections() + multi = is_multi_host() + + if multi: + logger.info(f"Multi-host mode: {len(connections)} connection(s) configured") + else: + logger.info("Single-host mode (legacy env vars)") + + for conn_name, config in connections.items(): + server_info = config.get("server", "?") + if "port" in config: + server_info += f":{config['port']}" + user_info = config.get("user", "Windows Auth") + logger.info( + f" [{conn_name}] {server_info}/{config.get('database', '?')} " + f"as {user_info}" + ) + async with stdio_server() as (read_stream, write_stream): try: await app.run( read_stream, write_stream, - app.create_initialization_options() + app.create_initialization_options(), ) except Exception as e: logger.error(f"Server error: {str(e)}", exc_info=True) raise + if __name__ == "__main__": asyncio.run(main()) diff --git a/tests/test_config.py b/tests/test_config.py index bbebf03..4c6a4af 100644 --- a/tests/test_config.py +++ b/tests/test_config.py @@ -1,168 +1,588 @@ """Test database configuration and environment variable handling.""" + +import json import pytest import os +import tempfile from unittest.mock import patch -from mssql_mcp_server.server import get_db_config, validate_table_name +from mssql_mcp_server.server import ( + get_db_config, + validate_table_name, + build_connection_config, + get_all_connections, + is_multi_host, + get_connection_config, +) class TestDatabaseConfiguration: """Test database configuration from environment variables.""" - + def test_default_configuration(self): """Test default configuration values.""" - with patch.dict(os.environ, { - 'MSSQL_USER': 'testuser', - 'MSSQL_PASSWORD': 'testpass', - 'MSSQL_DATABASE': 'testdb' - }, clear=True): + with patch.dict( + os.environ, + { + "MSSQL_USER": "testuser", + "MSSQL_PASSWORD": "testpass", + "MSSQL_DATABASE": "testdb", + }, + clear=True, + ): config = get_db_config() - assert config['server'] == 'localhost' - assert config['user'] == 'testuser' - assert config['password'] == 'testpass' - assert config['database'] == 'testdb' - assert 'port' not in config - + assert config["server"] == "localhost" + assert config["user"] == "testuser" + assert config["password"] == "testpass" + assert config["database"] == "testdb" + assert "port" not in config + def test_custom_server_and_port(self): """Test custom server and port configuration.""" - with patch.dict(os.environ, { - 'MSSQL_SERVER': 'custom-server.com', - 'MSSQL_PORT': '1433', - 'MSSQL_USER': 'testuser', - 'MSSQL_PASSWORD': 'testpass', - 'MSSQL_DATABASE': 'testdb' - }): + with patch.dict( + os.environ, + { + "MSSQL_SERVER": "custom-server.com", + "MSSQL_PORT": "1433", + "MSSQL_USER": "testuser", + "MSSQL_PASSWORD": "testpass", + "MSSQL_DATABASE": "testdb", + }, + ): config = get_db_config() - assert config['server'] == 'custom-server.com' - assert config['port'] == 1433 - + assert config["server"] == "custom-server.com" + assert config["port"] == 1433 + def test_invalid_port(self): """Test invalid port handling.""" - with patch.dict(os.environ, { - 'MSSQL_PORT': 'invalid', - 'MSSQL_USER': 'testuser', - 'MSSQL_PASSWORD': 'testpass', - 'MSSQL_DATABASE': 'testdb' - }): + with patch.dict( + os.environ, + { + "MSSQL_PORT": "invalid", + "MSSQL_USER": "testuser", + "MSSQL_PASSWORD": "testpass", + "MSSQL_DATABASE": "testdb", + }, + ): config = get_db_config() - assert 'port' not in config # Invalid port should be ignored - + assert "port" not in config # Invalid port should be ignored + def test_azure_sql_configuration(self): """Test Azure SQL automatic encryption configuration.""" - with patch.dict(os.environ, { - 'MSSQL_SERVER': 'myserver.database.windows.net', - 'MSSQL_USER': 'testuser', - 'MSSQL_PASSWORD': 'testpass', - 'MSSQL_DATABASE': 'testdb' - }): + with patch.dict( + os.environ, + { + "MSSQL_SERVER": "myserver.database.windows.net", + "MSSQL_USER": "testuser", + "MSSQL_PASSWORD": "testpass", + "MSSQL_DATABASE": "testdb", + }, + ): config = get_db_config() - assert config['encrypt'] == True - assert config['tds_version'] == '7.4' - + assert config["encrypt"] == True + assert config["tds_version"] == "7.4" + def test_localdb_configuration(self): """Test LocalDB connection string conversion.""" - with patch.dict(os.environ, { - 'MSSQL_SERVER': '(localdb)\\MSSQLLocalDB', - 'MSSQL_DATABASE': 'testdb', - 'MSSQL_WINDOWS_AUTH': 'true' - }): + with patch.dict( + os.environ, + { + "MSSQL_SERVER": "(localdb)\\MSSQLLocalDB", + "MSSQL_DATABASE": "testdb", + "MSSQL_WINDOWS_AUTH": "true", + }, + ): config = get_db_config() - assert config['server'] == '.\\MSSQLLocalDB' - assert 'user' not in config - assert 'password' not in config - + assert config["server"] == ".\\MSSQLLocalDB" + assert "user" not in config + assert "password" not in config + def test_windows_authentication(self): """Test Windows authentication configuration.""" - with patch.dict(os.environ, { - 'MSSQL_SERVER': 'localhost', - 'MSSQL_DATABASE': 'testdb', - 'MSSQL_WINDOWS_AUTH': 'true' - }): + with patch.dict( + os.environ, + { + "MSSQL_SERVER": "localhost", + "MSSQL_DATABASE": "testdb", + "MSSQL_WINDOWS_AUTH": "true", + }, + ): config = get_db_config() - assert 'user' not in config - assert 'password' not in config - + assert "user" not in config + assert "password" not in config + def test_missing_required_config_sql_auth(self): """Test missing required configuration for SQL authentication.""" - with patch.dict(os.environ, { - 'MSSQL_SERVER': 'localhost' - }, clear=True): - with pytest.raises(ValueError, match="Missing required database configuration"): + with patch.dict(os.environ, {"MSSQL_SERVER": "localhost"}, clear=True): + with pytest.raises( + ValueError, match="Missing required database configuration" + ): get_db_config() - + def test_missing_database_windows_auth(self): """Test missing database for Windows authentication.""" - with patch.dict(os.environ, { - 'MSSQL_WINDOWS_AUTH': 'true' - }, clear=True): - with pytest.raises(ValueError, match="Missing required database configuration"): + with patch.dict(os.environ, {"MSSQL_WINDOWS_AUTH": "true"}, clear=True): + with pytest.raises( + ValueError, match="Missing required database configuration" + ): get_db_config() - + def test_encryption_settings(self): """Test various encryption settings.""" # Non-Azure with encryption - with patch.dict(os.environ, { - 'MSSQL_SERVER': 'localhost', - 'MSSQL_ENCRYPT': 'true', - 'MSSQL_USER': 'testuser', - 'MSSQL_PASSWORD': 'testpass', - 'MSSQL_DATABASE': 'testdb' - }): + with patch.dict( + os.environ, + { + "MSSQL_SERVER": "localhost", + "MSSQL_ENCRYPT": "true", + "MSSQL_USER": "testuser", + "MSSQL_PASSWORD": "testpass", + "MSSQL_DATABASE": "testdb", + }, + ): config = get_db_config() - assert config['encrypt'] == True - + assert config["encrypt"] == True + # Non-Azure without encryption (default) - with patch.dict(os.environ, { - 'MSSQL_SERVER': 'localhost', - 'MSSQL_USER': 'testuser', - 'MSSQL_PASSWORD': 'testpass', - 'MSSQL_DATABASE': 'testdb' - }): + with patch.dict( + os.environ, + { + "MSSQL_SERVER": "localhost", + "MSSQL_USER": "testuser", + "MSSQL_PASSWORD": "testpass", + "MSSQL_DATABASE": "testdb", + }, + ): config = get_db_config() - assert config['encrypt'] == False + assert config["encrypt"] == False class TestTableNameValidation: """Test SQL table name validation and escaping.""" - + def test_valid_table_names(self): """Test validation of valid table names.""" valid_names = [ - 'users', - 'UserAccounts', - 'user_accounts', - 'table123', - 'dbo.users', - 'schema_name.table_name' + "users", + "UserAccounts", + "user_accounts", + "table123", + "dbo.users", + "schema_name.table_name", ] - + for name in valid_names: escaped = validate_table_name(name) assert escaped is not None - assert '[' in escaped and ']' in escaped - + assert "[" in escaped and "]" in escaped + def test_invalid_table_names(self): """Test rejection of invalid table names.""" invalid_names = [ - 'users; DROP TABLE users', # SQL injection - 'users OR 1=1', # SQL injection - 'users--', # SQL comment - 'users/*comment*/', # SQL comment - 'users\'', # Quote - 'users"', # Double quote - 'schema.name.table', # Too many dots - 'user@table', # Invalid character - 'user#table', # Invalid character - '', # Empty - '.', # Just dot - '..', # Double dot + "users; DROP TABLE users", # SQL injection + "users OR 1=1", # SQL injection + "users--", # SQL comment + "users/*comment*/", # SQL comment + "users'", # Quote + 'users"', # Double quote + "schema.name.table", # Too many dots + "user@table", # Invalid character + "user#table", # Invalid character + "", # Empty + ".", # Just dot + "..", # Double dot ] - + for name in invalid_names: with pytest.raises(ValueError, match="Invalid table name"): validate_table_name(name) - + def test_table_name_escaping(self): """Test proper escaping of table names.""" - assert validate_table_name('users') == '[users]' - assert validate_table_name('dbo.users') == '[dbo].[users]' - assert validate_table_name('my_table_123') == '[my_table_123]' \ No newline at end of file + assert validate_table_name("users") == "[users]" + assert validate_table_name("dbo.users") == "[dbo].[users]" + assert validate_table_name("my_table_123") == "[my_table_123]" + + +class TestBuildConnectionConfig: + """Test build_connection_config() for individual connection dicts.""" + + def test_basic_sql_auth(self): + """Test basic SQL authentication connection config.""" + cfg = build_connection_config( + { + "server": "myhost.example.com", + "database": "mydb", + "user": "admin", + "password": "secret", + } + ) + assert cfg["server"] == "myhost.example.com" + assert cfg["database"] == "mydb" + assert cfg["user"] == "admin" + assert cfg["password"] == "secret" + + def test_port_is_parsed(self): + """Test port is converted to int.""" + cfg = build_connection_config( + { + "server": "host", + "database": "db", + "user": "u", + "password": "p", + "port": "5000", + } + ) + assert cfg["port"] == 5000 + + def test_invalid_port_ignored(self): + """Test invalid port values are silently ignored.""" + cfg = build_connection_config( + { + "server": "host", + "database": "db", + "user": "u", + "password": "p", + "port": "not_a_number", + } + ) + assert "port" not in cfg + + def test_windows_auth(self): + """Test Windows authentication removes user/password.""" + cfg = build_connection_config( + { + "server": "host", + "database": "db", + "windows_auth": "true", + } + ) + assert "user" not in cfg + assert "password" not in cfg + assert cfg["database"] == "db" + + def test_windows_auth_missing_database(self): + """Test Windows auth without database raises.""" + with pytest.raises(ValueError, match="Missing required database"): + build_connection_config( + { + "server": "host", + "windows_auth": "true", + } + ) + + def test_sql_auth_missing_fields(self): + """Test SQL auth missing user/password/database raises.""" + with pytest.raises(ValueError, match="Missing required database"): + build_connection_config({"server": "host"}) + + def test_azure_sql_encryption(self): + """Test Azure SQL auto-encryption.""" + cfg = build_connection_config( + { + "server": "myserver.database.windows.net", + "database": "db", + "user": "u", + "password": "p", + } + ) + assert cfg["tds_version"] == "7.4" + assert "Encrypt=yes" in cfg["server"] + + def test_explicit_encryption(self): + """Test explicit encryption flag on non-Azure.""" + cfg = build_connection_config( + { + "server": "host", + "database": "db", + "user": "u", + "password": "p", + "encrypt": "true", + } + ) + assert cfg["tds_version"] == "7.4" + assert "Encrypt=yes" in cfg["server"] + + def test_localdb_conversion(self): + """Test LocalDB server name conversion.""" + cfg = build_connection_config( + { + "server": "(localdb)\\MSSQLLocalDB", + "database": "db", + "windows_auth": "true", + } + ) + assert cfg["server"] == ".\\MSSQLLocalDB" + + +class TestGetAllConnections: + """Test get_all_connections() resolution logic.""" + + def test_fallback_to_legacy_env_vars(self): + """When no multi-host env vars set, falls back to get_db_config().""" + with patch.dict( + os.environ, + { + "MSSQL_USER": "u", + "MSSQL_PASSWORD": "p", + "MSSQL_DATABASE": "db", + }, + clear=True, + ): + conns = get_all_connections() + assert "default" in conns + assert conns["default"]["user"] == "u" + + def test_indexed_env_vars_single_host(self): + """Test MSSQL_HOST_1_* env vars produce one connection.""" + with patch.dict( + os.environ, + { + "MSSQL_HOST_1_NAME": "prod", + "MSSQL_HOST_1_SERVER": "prod.host", + "MSSQL_HOST_1_DATABASE": "proddb", + "MSSQL_HOST_1_USER": "admin", + "MSSQL_HOST_1_PASSWORD": "s3cret", + }, + clear=True, + ): + conns = get_all_connections() + assert set(conns.keys()) == {"prod"} + assert conns["prod"]["server"] == "prod.host" + assert conns["prod"]["user"] == "admin" + + def test_indexed_env_vars_multiple_hosts(self): + """Test MSSQL_HOST_1_* + MSSQL_HOST_2_* produce two connections.""" + with patch.dict( + os.environ, + { + "MSSQL_HOST_1_NAME": "prod", + "MSSQL_HOST_1_SERVER": "prod.host", + "MSSQL_HOST_1_DATABASE": "proddb", + "MSSQL_HOST_1_USER": "admin", + "MSSQL_HOST_1_PASSWORD": "s3cret", + "MSSQL_HOST_2_NAME": "dev", + "MSSQL_HOST_2_SERVER": "dev.host", + "MSSQL_HOST_2_DATABASE": "devdb", + "MSSQL_HOST_2_USER": "dev", + "MSSQL_HOST_2_PASSWORD": "devpass", + }, + clear=True, + ): + conns = get_all_connections() + assert set(conns.keys()) == {"prod", "dev"} + assert conns["prod"]["server"] == "prod.host" + assert conns["dev"]["database"] == "devdb" + + def test_indexed_env_vars_non_contiguous_indexes(self): + """Indexes don't have to be 1,2,3 — gaps are fine.""" + with patch.dict( + os.environ, + { + "MSSQL_HOST_5_NAME": "five", + "MSSQL_HOST_5_SERVER": "h5", + "MSSQL_HOST_5_DATABASE": "db5", + "MSSQL_HOST_5_USER": "u", + "MSSQL_HOST_5_PASSWORD": "p", + "MSSQL_HOST_10_NAME": "ten", + "MSSQL_HOST_10_SERVER": "h10", + "MSSQL_HOST_10_DATABASE": "db10", + "MSSQL_HOST_10_USER": "u", + "MSSQL_HOST_10_PASSWORD": "p", + }, + clear=True, + ): + conns = get_all_connections() + assert set(conns.keys()) == {"five", "ten"} + + def test_indexed_env_vars_with_optional_fields(self): + """Test port, encrypt, windows_auth are picked up.""" + with patch.dict( + os.environ, + { + "MSSQL_HOST_1_NAME": "custom", + "MSSQL_HOST_1_SERVER": "host", + "MSSQL_HOST_1_DATABASE": "db", + "MSSQL_HOST_1_USER": "u", + "MSSQL_HOST_1_PASSWORD": "p", + "MSSQL_HOST_1_PORT": "5000", + "MSSQL_HOST_1_ENCRYPT": "true", + }, + clear=True, + ): + conns = get_all_connections() + assert conns["custom"]["port"] == 5000 + assert "Encrypt=yes" in conns["custom"]["server"] + + def test_indexed_env_vars_take_precedence_over_file(self): + """MSSQL_HOST_* should win when both file and env vars are set.""" + file_payload = [ + { + "name": "from_file", + "server": "file.host", + "database": "db", + "user": "u", + "password": "p", + }, + ] + with tempfile.NamedTemporaryFile(mode="w", suffix=".json", delete=False) as f: + json.dump(file_payload, f) + f.flush() + fpath = f.name + + try: + with patch.dict( + os.environ, + { + "MSSQL_CONNECTIONS_FILE": fpath, + "MSSQL_HOST_1_NAME": "from_env", + "MSSQL_HOST_1_SERVER": "env.host", + "MSSQL_HOST_1_DATABASE": "db", + "MSSQL_HOST_1_USER": "u", + "MSSQL_HOST_1_PASSWORD": "p", + }, + clear=True, + ): + conns = get_all_connections() + assert "from_env" in conns + assert "from_file" not in conns + finally: + os.unlink(fpath) + + def test_file_based_config(self): + """Test MSSQL_CONNECTIONS_FILE parsing.""" + payload = [ + { + "name": "staging", + "server": "staging.host", + "database": "stgdb", + "user": "stg", + "password": "stgpass", + }, + ] + with tempfile.NamedTemporaryFile(mode="w", suffix=".json", delete=False) as f: + json.dump(payload, f) + f.flush() + fpath = f.name + + try: + with patch.dict( + os.environ, + { + "MSSQL_CONNECTIONS_FILE": fpath, + }, + clear=True, + ): + conns = get_all_connections() + assert "staging" in conns + assert conns["staging"]["server"] == "staging.host" + finally: + os.unlink(fpath) + + def test_duplicate_connection_names_rejected(self): + """Duplicate names in indexed env vars should raise.""" + with patch.dict( + os.environ, + { + "MSSQL_HOST_1_NAME": "dup", + "MSSQL_HOST_1_SERVER": "h1", + "MSSQL_HOST_1_DATABASE": "db", + "MSSQL_HOST_1_USER": "u", + "MSSQL_HOST_1_PASSWORD": "p", + "MSSQL_HOST_2_NAME": "dup", + "MSSQL_HOST_2_SERVER": "h2", + "MSSQL_HOST_2_DATABASE": "db", + "MSSQL_HOST_2_USER": "u", + "MSSQL_HOST_2_PASSWORD": "p", + }, + clear=True, + ): + with pytest.raises(ValueError, match="Duplicate connection name"): + get_all_connections() + + def test_missing_name_rejected(self): + """Host entries without a NAME should raise.""" + with patch.dict( + os.environ, + { + "MSSQL_HOST_1_SERVER": "h1", + "MSSQL_HOST_1_DATABASE": "db", + "MSSQL_HOST_1_USER": "u", + "MSSQL_HOST_1_PASSWORD": "p", + }, + clear=True, + ): + with pytest.raises(ValueError, match="missing a valid 'name'"): + get_all_connections() + + +class TestIsMultiHost: + """Test is_multi_host() detection.""" + + def test_single_host(self): + with patch.dict(os.environ, {}, clear=True): + assert is_multi_host() is False + + def test_multi_host_indexed_env_vars(self): + with patch.dict( + os.environ, + { + "MSSQL_HOST_1_NAME": "x", + "MSSQL_HOST_1_SERVER": "h", + }, + clear=True, + ): + assert is_multi_host() is True + + def test_multi_host_file(self): + with patch.dict( + os.environ, {"MSSQL_CONNECTIONS_FILE": "/some/path"}, clear=True + ): + assert is_multi_host() is True + + def test_legacy_env_vars_not_multi_host(self): + """Plain MSSQL_SERVER/USER/etc. should NOT trigger multi-host.""" + with patch.dict( + os.environ, + { + "MSSQL_SERVER": "localhost", + "MSSQL_USER": "sa", + "MSSQL_PASSWORD": "pass", + "MSSQL_DATABASE": "db", + }, + clear=True, + ): + assert is_multi_host() is False + + +class TestGetConnectionConfig: + """Test get_connection_config() name resolution.""" + + def test_existing_connection(self): + with patch.dict( + os.environ, + { + "MSSQL_HOST_1_NAME": "alpha", + "MSSQL_HOST_1_SERVER": "a.host", + "MSSQL_HOST_1_DATABASE": "adb", + "MSSQL_HOST_1_USER": "u", + "MSSQL_HOST_1_PASSWORD": "p", + }, + clear=True, + ): + cfg = get_connection_config("alpha") + assert cfg["server"] == "a.host" + + def test_unknown_connection_raises(self): + with patch.dict( + os.environ, + { + "MSSQL_HOST_1_NAME": "alpha", + "MSSQL_HOST_1_SERVER": "a.host", + "MSSQL_HOST_1_DATABASE": "adb", + "MSSQL_HOST_1_USER": "u", + "MSSQL_HOST_1_PASSWORD": "p", + }, + clear=True, + ): + with pytest.raises(ValueError, match="Unknown connection 'nope'"): + get_connection_config("nope") diff --git a/tests/test_server.py b/tests/test_server.py index 7e433e9..f2a20a3 100644 --- a/tests/test_server.py +++ b/tests/test_server.py @@ -1,39 +1,147 @@ +import os import pytest -from mssql_mcp_server.server import app, list_tools, list_resources, read_resource, call_tool +from unittest.mock import patch +from mssql_mcp_server.server import ( + app, + list_tools, + list_resources, + read_resource, + call_tool, +) from pydantic import AnyUrl + def test_server_initialization(): """Test that the server initializes correctly.""" assert app.name == "mssql_mcp_server" + @pytest.mark.asyncio async def test_list_tools(): - """Test that list_tools returns expected tools.""" - tools = await list_tools() - assert len(tools) == 1 - assert tools[0].name == "execute_sql" - assert "query" in tools[0].inputSchema["properties"] + """Test that list_tools returns expected tools in single-host mode.""" + with patch.dict( + os.environ, + { + "MSSQL_USER": "u", + "MSSQL_PASSWORD": "p", + "MSSQL_DATABASE": "db", + }, + clear=True, + ): + tools = await list_tools() + assert len(tools) == 1 + assert tools[0].name == "execute_sql" + assert "query" in tools[0].inputSchema["properties"] + assert "connection" not in tools[0].inputSchema["properties"] + + +@pytest.mark.asyncio +async def test_list_tools_multi_host(): + """Test that list_tools exposes connection param + list_connections in multi-host mode.""" + with patch.dict( + os.environ, + { + "MSSQL_HOST_1_NAME": "a", + "MSSQL_HOST_1_SERVER": "h", + "MSSQL_HOST_1_DATABASE": "db", + "MSSQL_HOST_1_USER": "u", + "MSSQL_HOST_1_PASSWORD": "p", + "MSSQL_HOST_2_NAME": "b", + "MSSQL_HOST_2_SERVER": "h", + "MSSQL_HOST_2_DATABASE": "db", + "MSSQL_HOST_2_USER": "u", + "MSSQL_HOST_2_PASSWORD": "p", + }, + clear=True, + ): + tools = await list_tools() + names = [t.name for t in tools] + assert "execute_sql" in names + assert "list_connections" in names + exec_tool = next(t for t in tools if t.name == "execute_sql") + assert "connection" in exec_tool.inputSchema["properties"] + assert "connection" in exec_tool.inputSchema["required"] + assert set(exec_tool.inputSchema["properties"]["connection"]["enum"]) == { + "a", + "b", + } + @pytest.mark.asyncio async def test_call_tool_invalid_name(): """Test calling a tool with an invalid name.""" - with pytest.raises(ValueError, match="Unknown tool"): - await call_tool("invalid_tool", {}) + with patch.dict( + os.environ, + { + "MSSQL_USER": "u", + "MSSQL_PASSWORD": "p", + "MSSQL_DATABASE": "db", + }, + clear=True, + ): + with pytest.raises(ValueError, match="Unknown tool"): + await call_tool("invalid_tool", {}) + @pytest.mark.asyncio async def test_call_tool_missing_query(): """Test calling execute_sql without a query.""" - with pytest.raises(ValueError, match="Query is required"): - await call_tool("execute_sql", {}) + with patch.dict( + os.environ, + { + "MSSQL_USER": "u", + "MSSQL_PASSWORD": "p", + "MSSQL_DATABASE": "db", + }, + clear=True, + ): + with pytest.raises(ValueError, match="Query is required"): + await call_tool("execute_sql", {}) + + +@pytest.mark.asyncio +async def test_call_tool_multi_host_missing_connection(): + """Test that multi-host mode requires connection parameter.""" + with patch.dict( + os.environ, + { + "MSSQL_HOST_1_NAME": "x", + "MSSQL_HOST_1_SERVER": "h", + "MSSQL_HOST_1_DATABASE": "db", + "MSSQL_HOST_1_USER": "u", + "MSSQL_HOST_1_PASSWORD": "p", + }, + clear=True, + ): + with pytest.raises(ValueError, match="'connection' parameter is required"): + await call_tool("execute_sql", {"query": "SELECT 1"}) + + +@pytest.mark.asyncio +async def test_call_tool_list_connections(): + """Test list_connections tool returns connection info.""" + with patch.dict( + os.environ, + { + "MSSQL_HOST_1_NAME": "prod", + "MSSQL_HOST_1_SERVER": "prod.host", + "MSSQL_HOST_1_DATABASE": "proddb", + "MSSQL_HOST_1_USER": "admin", + "MSSQL_HOST_1_PASSWORD": "p", + }, + clear=True, + ): + result = await call_tool("list_connections", {}) + assert len(result) == 1 + assert "prod" in result[0].text + assert "prod.host" in result[0].text + # Skip database-dependent tests if no database connection @pytest.mark.asyncio @pytest.mark.skipif( - not all([ - pytest.importorskip("pymssql"), - pytest.importorskip("mssql_mcp_server") - ]), - reason="SQL Server connection not available" + not all([pytest.importorskip("pymssql"), pytest.importorskip("mssql_mcp_server")]), + reason="SQL Server connection not available", ) async def test_list_resources(): """Test listing resources (requires database connection).""" From ebcb80ec8b7b086805fcb3da92343cd295c5a2e3 Mon Sep 17 00:00:00 2001 From: aaalrajhi Date: Sun, 3 May 2026 11:17:59 +0300 Subject: [PATCH 2/2] fix: percent-encode connection and table names in resource URIs Resource URIs built in multi-host list_resources() embedded the raw connection name as the URI authority (host) and the raw table name as the first path segment, e.g. `mssql://payment gateway - QA/Hash/data`. Pydantic's AnyUrl validator (used by mcp.types.Resource.uri) rejects spaces and other characters that are illegal in the URI authority, causing every resource for spaced connection names to fail validation with `invalid domain character` and surface as `Failed to list resources: 1 validation error for Resource`. As a result, all tables on those connections silently disappeared from the resource list. Percent-encode connection and table names with urllib.parse.quote when building the URI, and decode them with unquote in read_resource so routing still resolves to the original names. Single-host mode also encodes the table name for consistency. --- src/mssql_mcp_server/server.py | 19 ++++++++++++++----- 1 file changed, 14 insertions(+), 5 deletions(-) diff --git a/src/mssql_mcp_server/server.py b/src/mssql_mcp_server/server.py index 27cae54..52a2777 100644 --- a/src/mssql_mcp_server/server.py +++ b/src/mssql_mcp_server/server.py @@ -3,6 +3,7 @@ import logging import os import re +from urllib.parse import quote, unquote import pymssql from mcp.server import Server from mcp.types import Resource, Tool, TextContent @@ -363,11 +364,17 @@ async def list_resources() -> list[Resource]: for table in tables: if multi: - uri = f"mssql://{conn_name}/{table[0]}/data" + # Percent-encode connection and table names so URIs containing + # spaces or other characters illegal in the URI host/path stay + # valid (pydantic AnyUrl rejects raw spaces in the authority). + uri = ( + f"mssql://{quote(conn_name, safe='')}" + f"/{quote(table[0], safe='')}/data" + ) label = f"[{conn_name}] Table: {table[0]}" desc = f"Data in table {table[0]} on connection '{conn_name}'" else: - uri = f"mssql://{table[0]}/data" + uri = f"mssql://{quote(table[0], safe='')}/data" label = f"Table: {table[0]}" desc = f"Data in table: {table[0]}" @@ -406,14 +413,16 @@ async def read_resource(uri: AnyUrl) -> str: if multi: # mssql:///
/data → parts = [conn, table, "data"] + # Connection and table names are percent-encoded by list_resources, + # so decode them before routing/lookup. if len(parts) < 2: raise ValueError(f"Invalid multi-host URI: {uri_str}") - conn_name = parts[0] - table = parts[1] + conn_name = unquote(parts[0]) + table = unquote(parts[1]) config = get_connection_config(conn_name) else: # mssql://
/data → parts = [table, "data"] - table = parts[0] + table = unquote(parts[0]) config = get_db_config() try: