Microsoft SQL Server MCP Server

Instructions

Implementation Reference

• src/mssql_mcp_server/server.py:218-266 (handler)

  The @app.call_tool() handler function that implements the execute_sql tool logic. Connects to MSSQL using pymssql, executes the provided SQL query, handles SELECT vs non-SELECT queries, and returns results or affected rows as TextContent.
  @app.call_tool() async def call_tool(name: str, arguments: dict) -> list[TextContent]: """Execute SQL commands.""" config = get_db_config() command = get_command() logger.info(f"Calling tool: {name} with arguments: {arguments}") if name != command: raise ValueError(f"Unknown tool: {name}") query = arguments.get("query") if not query: raise ValueError("Query is required") try: conn = pymssql.connect(**config) cursor = 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.extend([table[0] for table in tables]) cursor.close() 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))] # Non-SELECT queries else: conn.commit() affected_rows = cursor.rowcount cursor.close() 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)}")]
• src/mssql_mcp_server/server.py:196-216 (registration)

  The @app.list_tools() function that registers the 'execute_sql' tool (name from get_command()), including its description and inputSchema requiring a 'query' string.
  @app.list_tools() async def list_tools() -> list[Tool]: """List available SQL Server tools.""" command = get_command() logger.info("Listing tools...") return [ Tool( name=command, description="Execute an SQL query on the SQL Server", inputSchema={ "type": "object", "properties": { "query": { "type": "string", "description": "The SQL query to execute" } }, "required": ["query"] } ) ]
• src/mssql_mcp_server/server.py:204-215 (schema)

  The inputSchema for the execute_sql tool, defining a required 'query' property of type string.
  description="Execute an SQL query on the SQL Server", inputSchema={ "type": "object", "properties": { "query": { "type": "string", "description": "The SQL query to execute" } }, "required": ["query"] } )
• src/mssql_mcp_server/server.py:100-103 (helper)

  get_command() function that returns the tool name, defaulting to 'execute_sql' from MSSQL_COMMAND env var.
  def get_command(): """Get the command to execute SQL queries.""" return os.getenv("MSSQL_COMMAND", "execute_sql")
• src/mssql_mcp_server/server.py:32-99 (helper)

  get_db_config() helper that constructs the pymssql connection config from environment variables, handling various auth, encryption, LocalDB, etc.
  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"Using server: {server}") # Handle LocalDB connections (Issue #6) # LocalDB format: (localdb)\instancename if server.startswith("(localdb)\\"): # For LocalDB, pymssql needs special formatting # Convert (localdb)\MSSQLLocalDB to localhost\MSSQLLocalDB with dynamic port 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: try: 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"]: config["tds_version"] = "7.4" # Required for Azure SQL # Azure SQL requires encryption - use connection string format for pymssql 2.3+ # This improves upon TDS-only approach by being more explicit if os.getenv("MSSQL_ENCRYPT", "true").lower() == "true": config["server"] += ";Encrypt=yes;TrustServerCertificate=no" else: # For non-Azure connections, respect the MSSQL_ENCRYPT setting # Use connection string format in addition to TDS version for better compatibility 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 # 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"]: logger.error("MSSQL_DATABASE is required") raise ValueError("Missing required database configuration") # Remove user and password for Windows auth config.pop("user", None) config.pop("password", None) logger.info("Using Windows Authentication") 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("MSSQL_USER, MSSQL_PASSWORD, and MSSQL_DATABASE are required") raise ValueError("Missing required database configuration") return config