diff --git a/boot.py b/boot.py
index 017a190ad..d63ae368d 100644
--- a/boot.py
+++ b/boot.py
@@ -2,7 +2,8 @@
 
 from .plugin.api import AbstractPlugin
 from .plugin.api import g_plugins
-from .plugin.api import register_plugin
+from .plugin.api import LspPlugin
+from .plugin.api import register_plugin_impl
 from .plugin.code_actions import LspCodeActionsCommand
 from .plugin.code_actions import LspRefactorCommand
 from .plugin.code_actions import LspSourceActionCommand
@@ -180,15 +181,16 @@ def _get_final_subclasses(derived: list[type], results: list[type]) -> None:
 
 
 def _register_all_plugins() -> None:
-    plugin_classes: list[type[AbstractPlugin]] = []
+    plugin_classes: list[type[AbstractPlugin | LspPlugin]] = []
     _get_final_subclasses(AbstractPlugin.__subclasses__(), plugin_classes)
+    _get_final_subclasses(LspPlugin.__subclasses__(), plugin_classes)
     for plugin_class in plugin_classes:
         try:
-            if not plugin_class.name():
+            if issubclass(plugin_class, AbstractPlugin) and not plugin_class.name():
                 continue
         except NotImplementedError:
             continue
-        register_plugin(plugin_class, notify_listener=False)
+        register_plugin_impl(plugin_class, notify_listener=False)
 
 
 def _unregister_all_plugins() -> None:
diff --git a/docs/mkdocs.yml b/docs/mkdocs.yml
index c70b4161e..7595e3f16 100644
--- a/docs/mkdocs.yml
+++ b/docs/mkdocs.yml
@@ -61,6 +61,7 @@ nav:
   - Keyboard Shortcuts: keyboard_shortcuts.md
   - Client Configuration: client_configuration.md
   - Troubleshooting: troubleshooting.md
+  - Migration to LspPlugin: migrating_to_lsp_plugin.md
 
 extra:
   social:
diff --git a/docs/src/migrating_to_lsp_plugin.md b/docs/src/migrating_to_lsp_plugin.md
new file mode 100644
index 000000000..16f253745
--- /dev/null
+++ b/docs/src/migrating_to_lsp_plugin.md
@@ -0,0 +1,418 @@
+# Migrating from AbstractPlugin to LspPlugin
+
+`LspPlugin` is the modern base class for LSP helper packages. It replaces `AbstractPlugin` with a cleaner, context-based API that reduces boilerplate and consolidates the server lifecycle into fewer override points.
+
+!!! note
+    `AbstractPlugin` is still supported. You only need to migrate when you are ready to adopt the new API.
+
+---
+
+## Overview of changes
+
+| AbstractPlugin | LspPlugin |
+|---|---|
+| `name()` | Removed - derived automatically from the package name and exposed as a `name` property |
+| `configuration()` | Removed - settings file located automatically |
+| `storage_path()` | `plugin_storage_path` class attribute (derived automatically) |
+| `needs_update_or_installation()` + `install_or_update()` | `install_async(context)` |
+| `can_start(window, view, folders, config)` | Raise `PluginStartError` from `install_async` (or other `@classmethod`) |
+| `on_pre_start(window, view, folders, config)` | `command(context)`, `working_directory(context)`, `initialization_options(context)` |
+| `on_post_start(window, view, folders, config)` | `__init__(weaksession, context)` |
+| `on_settings_changed(settings: DottedDict)` | `__init__(weaksession, context)` |
+| `is_applicable(view, config)` | `is_applicable(context)` |
+| `additional_variables()` | `additional_variables(context)` |
+| `on_pre_server_command(command, done_callback)` | `on_execute_command(command)` - return a `Promise` instead of invoking a callback |
+| `on_pre_send_request_async(request_id, request)` | `on_pre_send_request_async(request, view)` |
+| `on_server_response_async(method, response)` | `on_server_response_async(response)` |
+| `register_plugin(MyPlugin)` / `unregister_plugin(MyPlugin)` | `MyPlugin.register()` / `MyPlugin.unregister()` - no standalone import needed |
+
+All other instance methods (`on_workspace_configuration`,
+`on_pre_send_notification_async`, `on_server_notification_async`, `on_open_uri_async`,
+`on_session_buffer_changed_async`, `on_selection_modified_async`, `on_session_end_async`)
+are available in `LspPlugin` with the same name and the same signature.
+
+---
+
+## Step-by-step migration
+
+### 1. Change the base class
+
+```python
+# Before
+from LSP.plugin import AbstractPlugin
+
+class LspFoo(AbstractPlugin):
+    ...
+```
+
+```python
+# After
+from LSP.plugin import LspPlugin
+
+class LspFoo(LspPlugin):
+    ...
+```
+
+`LspPlugin` provides `register()` and `unregister()` classmethods, so `register_plugin` and
+`unregister_plugin` **no longer need to be imported or called directly**. Replace them with calls
+on your plugin class:
+
+```python
+# Before
+from LSP.plugin import AbstractPlugin
+from LSP.plugin import register_plugin
+from LSP.plugin import unregister_plugin
+
+class LspFoo(AbstractPlugin):
+    ...
+
+def plugin_loaded() -> None:
+    register_plugin(LspFoo)
+
+def plugin_unloaded() -> None:
+    unregister_plugin(LspFoo)
+```
+
+```python
+# After
+from LSP.plugin import LspPlugin
+
+class LspFoo(LspPlugin):
+    ...
+
+def plugin_loaded() -> None:
+    LspFoo.register()
+
+def plugin_unloaded() -> None:
+    LspFoo.unregister()
+```
+
+---
+
+### 2. Remove `name()` and `configuration()`
+
+`LspPlugin` derives the session name from the top-level package name automatically (i.e. `__module__.split('.')[0]`). The settings file is expected at `Packages/<PackageName>/<PackageName>.sublime-settings`, also without any manual configuration.
+
+Remove both overrides:
+
+```python
+# Before - remove these
+@classmethod
+def name(cls) -> str:
+    return "foo"
+
+@classmethod
+def configuration(cls) -> tuple[sublime.Settings, str]:
+    basename = "LSP-foo.sublime-settings"
+    return sublime.load_settings(basename), f"Packages/LSP-foo/{basename}"
+```
+
+---
+
+### 3. Replace `storage_path()` with `plugin_storage_path`
+
+The storage path is now a class attribute set automatically to `$DATA/Package Storage/<PackageName>`. Replace calls to `cls.storage_path()` with `cls.plugin_storage_path`:
+
+```python
+# Before
+server_dir = os.path.join(cls.storage_path(), cls.name(), "server")
+
+# After
+server_dir = cls.plugin_storage_path / "server"
+```
+
+---
+
+### 4. Merge `needs_update_or_installation` and `install_or_update` into `install_async`
+
+`install_async` is always called before the server starts and runs on a worker thread. Combine your install check and install logic there. To abort startup with a user-visible message, raise `PluginStartError` (this replaces returning a string from `can_start`):
+
+```python
+# Before
+@classmethod
+def needs_update_or_installation(cls) -> bool:
+    return not server_binary().exists()
+
+@classmethod
+def install_or_update(cls) -> None:
+    download_server(server_binary())
+
+@classmethod
+def can_start(cls, window, initiating_view, workspace_folders, configuration) -> str | None:
+    if not server_binary().exists():
+        return "Server binary missing"
+    return None
+```
+
+```python
+# After
+from LSP.plugin import PluginStartError
+
+@classmethod
+def install_async(cls, context: PluginContext) -> None:
+    if not server_binary().exists():
+        download_server(server_binary())
+    if not server_binary().exists():
+        raise PluginStartError("Server binary missing after installation attempt")
+```
+
+---
+
+### 5. Migrate `on_pre_start` overrides
+
+`on_pre_start` was used to customise the command, working directory, and initialization options, often by mutating the passed-in `configuration`. `LspPlugin` provides dedicated override points for each concern:
+
+```python
+# Before
+@classmethod
+def on_pre_start(cls, window, initiating_view, workspace_folders, configuration) -> str | None:
+    configuration.command = [str(server_binary()), "--stdio"]
+    return str(workspace_folders[0].path) if workspace_folders else None
+```
+
+```python
+# After
+@classmethod
+def command(cls, context: PluginContext) -> list[str]:
+    return [str(cls.plugin_storage_path / "server"), "--stdio"]
+
+@classmethod
+def working_directory(cls, context: PluginContext) -> str | None:
+    return context.workspace_folders[0].path if context.workspace_folders else None
+```
+
+For initialization options, override `initialization_options` instead of mutating `configuration.initialization_options` in `on_pre_start`:
+
+```python
+# After
+@classmethod
+def initialization_options(cls, context: PluginContext) -> dict[str, Any]:
+    options = context.configuration.initialization_options.get()
+    options["myCustomKey"] = "value"
+    return options
+```
+
+---
+
+### 6. Replace `on_post_start` with `__init__`
+
+`on_post_start` ran after the subprocess started but before the `initialize` handshake. In `LspPlugin`, `__init__` is called after a successful `initialize` response, which is the more useful point to run post-start logic. The `context` argument gives you access to the same information that was previously passed to `on_post_start`:
+
+```python
+# Before
+@classmethod
+def on_post_start(cls, window, initiating_view, workspace_folders, configuration) -> None:
+    log_start(window, configuration)
+```
+
+```python
+# After
+def __init__(self, weaksession) -> None:
+    super().__init__(weaksession)
+    if session := weaksession():
+        log_start(session.window, session.config)
+```
+
+---
+
+### 7. Remove `on_settings_changed`
+
+`LspPlugin` does not provide an `on_settings_changed` override point. The method has been removed because it was only called once right after sending the `initialize` request and the same result can be achieved by running the logic from `__init__` (called after a successful `initialize` response) for one-time setup, or by using `on_workspace_configuration` to adjust per-request configuration values dynamically.
+
+If you were using `on_settings_changed` to apply a fixed setting at startup, move that logic to `__init__`:
+
+```python
+# After — one-time setup
+def __init__(self, weaksession, context: PluginContext) -> None:
+    super().__init__(weaksession, context)
+    context.configuration.settings.set('foo', 'bar')
+```
+
+If you were using it to adjust configuration returned for `workspace/configuration` requests, use `on_workspace_configuration` instead.
+
+---
+
+### 8. Update `is_applicable` and `additional_variables`
+
+Both methods now receive a single `PluginContext` argument instead of individual parameters. `context.view` and `context.configuration` replace the former `view` and `config` arguments. `additional_variables` now always expects a dict value (default implementation returns empty dict):
+
+```python
+# Before
+@classmethod
+def is_applicable(cls, view: sublime.View, config: ClientConfig) -> bool:
+    return super().is_applicable(view, config) and my_condition(view)
+
+@classmethod
+def additional_variables(cls) -> dict[str, str] | None:
+    return {"server_version": SERVER_VERSION}
+```
+
+```python
+# After
+@classmethod
+def is_applicable(cls, context: PluginContext) -> bool:
+    return super().is_applicable(context) and my_condition(context.view)
+
+@classmethod
+def additional_variables(cls, context: PluginContext) -> dict[str, str]:
+    return {"server_version": SERVER_VERSION}
+```
+
+---
+
+### 9. Replace `on_pre_server_command` with `on_execute_command`
+
+The callback-based approach is replaced by returning a `Promise`:
+
+```python
+# Before
+def on_pre_server_command(self, command: ExecuteCommandParams, done_callback: Callable[[], None]) -> bool:
+    if command["command"] == "foo/bar":
+        handle_command(command)
+        done_callback()
+        return True
+    return False
+```
+
+```python
+# After
+from LSP.plugin import Promise
+
+def on_execute_command(self, command: ExecuteCommandParams) -> Promise[None] | None:
+    if command["command"] == "foo/bar":
+        return Promise.resolve(handle_command(command))
+    return None
+```
+
+---
+
+### 10. Update `on_pre_send_request_async` and `on_server_response_async`
+
+Both methods have had their signatures simplified.
+
+`on_pre_send_request_async` no longer receives the numeric request ID and the `view` argument is now passed explicitly:
+
+```python
+# Before
+def on_pre_send_request_async(self, request_id: int, request: Request) -> None:
+    log(f"[{request_id}] {request.method}")
+```
+
+```python
+# After
+def on_pre_send_request_async(self, request: ClientRequest, view: sublime.View | None) -> None:
+    log(request['method'])
+```
+
+`on_server_response_async` no longer receives the method name separately:
+
+```python
+# Before
+def on_server_response_async(self, method: str, response: Response) -> None:
+    if method == 'textDocument/hover':
+        process(response.result)
+```
+
+```python
+# After
+def on_server_response_async(self, response: ServerResponse) -> None:
+    process(response.get('result'))
+```
+
+---
+
+### 11. Use `@notification_handler` and `@request_handler` for custom messages
+
+`LspPlugin` introduces decorators to handle non-standard server-to-client notifications and requests. These replace manual approach with method names transformed using logic from `method2attr`:
+
+```python
+# Before
+def m__eslint_status(self, params: str) -> None:
+    self.handle_status(notification.params)
+```
+
+```python
+# After
+from LSP.plugin import notification_handler
+
+@notification_handler('eslint/status')
+def on_eslint_status(self, params: str) -> None:
+    self.handle_status(params)
+```
+
+```python
+# Similarly for requests
+from LSP.plugin import request_handler
+
+@request_handler('eslint/openDoc')
+def on_eslint_open_doc(self, params: TextDocumentIdentifier) -> Promise[bool]:
+    ...
+```
+
+---
+
+## Complete before/after example
+
+```python
+# Before - AbstractPlugin
+from LSP.plugin import AbstractPlugin
+from LSP.plugin import register_plugin
+from LSP.plugin import unregister_plugin
+
+
+class LspFoo(AbstractPlugin):
+
+    @classmethod
+    def name(cls) -> str:
+        return "foo"
+
+    @classmethod
+    def needs_update_or_installation(cls) -> bool:
+        return not (cls.storage_path() / "foo" / "server").exists()
+
+    @classmethod
+    def install_or_update(cls) -> None:
+        download_server()
+
+    @classmethod
+    def on_pre_start(cls, window, initiating_view, workspace_folders, configuration) -> str | None:
+        configuration.command = [str(cls.storage_path() / "foo" / "server"), "--stdio"]
+        return None
+
+
+def plugin_loaded() -> None:
+    register_plugin(LspFoo)
+
+
+def plugin_unloaded() -> None:
+    unregister_plugin(LspFoo)
+```
+
+```python
+# After - LspPlugin
+from LSP.plugin import LspPlugin
+from LSP.plugin import PluginContext
+from LSP.plugin import PluginStartError
+
+
+class LspFoo(LspPlugin):
+
+    @classmethod
+    def install_async(cls, context: PluginContext) -> None:
+        if not (cls.plugin_storage_path / "server").exists():
+            download_server()
+        if not (cls.plugin_storage_path / "server").exists():
+            raise PluginStartError("Failed to install foo language server")
+
+    @classmethod
+    def command(cls, context: PluginContext) -> list[str]:
+        return [str(cls.plugin_storage_path / "server"), "--stdio"]
+
+
+def plugin_loaded() -> None:
+    LspFoo.register()
+
+
+def plugin_unloaded() -> None:
+    LspFoo.unregister()
+```
diff --git a/plugin/__init__.py b/plugin/__init__.py
index ad7c30d39..a532e12cc 100644
--- a/plugin/__init__.py
+++ b/plugin/__init__.py
@@ -1,7 +1,10 @@
 from __future__ import annotations
 
 from .api import AbstractPlugin
+from .api import LspPlugin
 from .api import notification_handler
+from .api import PluginContext
+from .api import PluginStartError
 from .api import register_plugin
 from .api import request_handler
 from .api import unregister_plugin
@@ -13,6 +16,7 @@
 from .core.file_watcher import FileWatcherEventType
 from .core.file_watcher import FileWatcherProtocol
 from .core.file_watcher import register_file_watcher_implementation
+from .core.promise import PackagedTask
 from .core.promise import Promise
 from .core.protocol import ClientNotification
 from .core.protocol import ClientRequest
@@ -28,6 +32,7 @@
 from .core.sessions import Session
 from .core.sessions import SessionBufferProtocol
 from .core.sessions import SessionViewProtocol
+from .core.transports import TransportWrapper
 from .core.types import ClientConfig
 from .core.types import DebouncerNonThreadSafe
 from .core.types import matches_pattern
@@ -52,10 +57,14 @@
     'FileWatcherEvent',
     'FileWatcherEventType',
     'FileWatcherProtocol',
+    'LspPlugin',
     'LspTextCommand',
     'LspWindowCommand',
     'MarkdownLangMap',
     'Notification',
+    'PackagedTask',
+    'PluginContext',
+    'PluginStartError',
     'Promise',
     'Request',
     'Response',
@@ -65,6 +74,7 @@
     'Session',
     'SessionBufferProtocol',
     'SessionViewProtocol',
+    'TransportWrapper',
     'WorkspaceFolder',
     '__version__',
     'apply_text_edits',
diff --git a/plugin/api.py b/plugin/api.py
index a3788226d..3b4215ff8 100644
--- a/plugin/api.py
+++ b/plugin/api.py
@@ -6,9 +6,13 @@
 from ..protocol import LSPAny
 from .core.constants import ST_STORAGE_PATH
 from .core.logging import exception_log
+from .core.protocol import ClientNotification
+from .core.protocol import ClientRequest
 from .core.protocol import Notification
 from .core.protocol import Request
 from .core.protocol import Response
+from .core.protocol import ServerNotification
+from .core.protocol import ServerResponse
 from .core.settings import client_configs
 from .core.types import ClientConfig
 from .core.types import method2attr
@@ -17,9 +21,13 @@
 from .core.views import uri_from_view
 from abc import ABC
 from abc import abstractmethod
+from dataclasses import dataclass
 from functools import wraps
+from pathlib import Path
 from typing import Any
 from typing import Callable
+from typing import Final
+from typing import final
 from typing import TYPE_CHECKING
 from typing import TypeVar
 from typing_extensions import deprecated
@@ -27,16 +35,25 @@
 import sublime
 
 if TYPE_CHECKING:
+    from ..protocol import ConfigurationItem
+    from ..protocol import DocumentUri
+    from ..protocol import ExecuteCommandParams
     from .core.collections import DottedDict
     from .core.promise import Promise
+    from .core.protocol import Notification
+    from .core.protocol import Request
     from .core.sessions import Session
     from .core.sessions import SessionBufferProtocol
     from .core.sessions import SessionViewProtocol
+    from .core.transports import TransportWrapper
+    from .core.types import ClientConfig
+    from .core.views import MarkdownLangMap
     from .core.workspace import WorkspaceFolder
-    import weakref
+    from weakref import ref
 
 __all__ = [
     'APIHandler',
+    'PluginStartError',
     'notification_handler',
     'request_handler',
 ]
@@ -48,9 +65,22 @@
 R = TypeVar('R', bound=LSPAny)
 
 
-g_plugins: dict[str, type[AbstractPlugin]] = {}
+g_plugins: dict[str, type[AbstractPlugin | LspPlugin]] = {}
 
 
+class PluginStartError(Exception):
+    """
+    Abort startup with a user-visible message.
+
+    Raise it from `install_async` or other LspPlugin's `@classmethod` to prevent plugin from starting.
+    First argument is the text that will be shown in the status field.
+    """
+
+    def __init__(self, message: str) -> None:
+        super().__init__(message)
+
+
+@deprecated('Use LspPlugin.register() instead')
 def register_plugin(plugin: type[AbstractPlugin], notify_listener: bool = True) -> None:
     """
     Register an LSP plugin in LSP.
@@ -84,6 +114,10 @@ def plugin_unloaded():
     If you need to install supplementary files (e.g. javascript source code that implements the actual server), do so
     in `AbstractPlugin.install_or_update` in a blocking manner, without the use of Python's `threading` module.
     """
+    register_plugin_impl(plugin, notify_listener)
+
+
+def register_plugin_impl(plugin: type[AbstractPlugin | LspPlugin], notify_listener: bool = True) -> None:
     if notify_listener:
         # There is a bug in Sublime Text's `plugin_loaded` callback. When the package is in the list of
         # `"ignored_packages"` in Packages/User/Preferences.sublime-settings, and then removed from that list, the
@@ -96,18 +130,22 @@ def plugin_unloaded():
         _register_plugin_impl(plugin, notify_listener)
 
 
-def _register_plugin_impl(plugin: type[AbstractPlugin], notify_listener: bool) -> None:
-    name = plugin.name()
+def _register_plugin_impl(plugin: type[AbstractPlugin | LspPlugin], notify_listener: bool) -> None:
+    name = plugin.name() if issubclass(plugin, AbstractPlugin) else plugin.name
     if name in g_plugins:
         return
     try:
-        _, settings_path = plugin.configuration()
+        if issubclass(plugin, AbstractPlugin):
+            _, settings_path = plugin.configuration()
+        else:
+            settings_path = f"Packages/{name}/{name}.sublime-settings"
         if client_configs.add_external_config(name, settings_path, notify_listener):
             g_plugins[name] = plugin
     except Exception as ex:
         exception_log(f'Failed to register plugin "{name}"', ex)
 
 
+@deprecated('Use LspPlugin.unregister() instead')
 def unregister_plugin(plugin: type[AbstractPlugin]) -> None:
     """
     Unregister an LSP plugin in LSP.
@@ -116,7 +154,11 @@ def unregister_plugin(plugin: type[AbstractPlugin]) -> None:
     by a user, your language server is shut down for the views that it is attached to. This results in a good user
     experience.
     """
-    name = plugin.name()
+    unregister_plugin_impl(plugin)
+
+
+def unregister_plugin_impl(plugin: type[AbstractPlugin | LspPlugin]) -> None:
+    name = plugin.name() if issubclass(plugin, AbstractPlugin) else plugin.__module__.split('.')[0]
     try:
         g_plugins.pop(name, None)
         client_configs.remove_external_config(name)
@@ -124,7 +166,7 @@ def unregister_plugin(plugin: type[AbstractPlugin]) -> None:
         exception_log(f'Failed to unregister plugin "{name}"', ex)
 
 
-def get_plugin(name: str) -> type[AbstractPlugin] | None:
+def get_plugin(name: str) -> type[AbstractPlugin | LspPlugin] | None:
     return g_plugins.get(name)
 
 
@@ -201,6 +243,329 @@ def wrapper(self: Any, params: P, request_id: int) -> Promise[Response[Any]]:
     return decorator
 
 
+@dataclass
+class PluginContext:
+    """Plugin context information passed to various `LspPlugin` classmethods."""
+
+    configuration: ClientConfig
+    """The resolved `ClientConfig` for this session."""
+    view: sublime.View
+    """The view relevant to the method being called."""
+    window: sublime.Window
+    """The window in which the session is running."""
+    workspace_folders: list[WorkspaceFolder]
+    """The workspace folders active for this session."""
+
+
+class LspPlugin(APIHandler):
+    """
+    Base class for LSP helper packages.
+
+    Subclass this to integrate a language server with LSP. The session name is automatically
+    derived from the top-level package name (i.e. ``__module__.split('.')[0]``), so no manual
+    configuration is needed.
+
+    A minimal integration looks like this:
+
+    ```py
+    from LSP.plugin import LspPlugin
+
+
+    class LspFooPlugin(LspPlugin):
+        pass
+
+
+    def plugin_loaded() -> None:
+        LspFooPlugin.register()
+
+
+    def plugin_unloaded() -> None:
+        LspFooPlugin.unregister()
+    ```
+
+    LSP will look for a settings file at ``Packages/<package_name>/<package_name>.sublime-settings``
+    to read the ``command``, ``selector``, ``schemes``, and other server configuration. Override
+    the classmethods below to customise behaviour beyond what the settings file provides.
+
+    Raise ``PluginStartError`` exception from any of the classmethods (but typically from ``install_async``) to
+    prevent plugin from starting while showing relevant message in the status field.
+
+    Use `@notification_handler` and `@request_handler` decorators to handle non-standard
+    server-to-client notifications and requests respectively.
+    """
+
+    name: Final[str] = ''
+    """
+    The name of the plugin.
+
+    This is the session name and the key of the settings object (in project settings, for example).
+    It is automatically inferred from package name and is not to be changed manually.
+    """
+
+    plugin_storage_path: Final[Path] = Path()  # Path is updated on subclassing this class.
+    """
+    The storage path for the plugin.
+
+    Use this as your directory to install server files. Its path is `$DATA/Package Storage/<Package Name>`.
+    """
+
+    def __init_subclass__(cls, **kwargs: Any) -> None:
+        cls.name = cls.__module__.split('.')[0]  # pyright: ignore[reportAttributeAccessIssue]
+        cls.plugin_storage_path = Path(ST_STORAGE_PATH, cls.name)  # pyright: ignore[reportAttributeAccessIssue]
+
+    @classmethod
+    @final
+    def register(cls) -> None:
+        """
+        Register this plugin with LSP.
+
+        Call this from your `plugin_loaded` callback so that LSP picks up configuration changes when your package
+        is disabled and re-enabled:
+
+        ```py
+        def plugin_loaded() -> None:
+            LspFooPlugin.register()
+        ```
+        """
+        register_plugin_impl(cls)
+
+    @classmethod
+    @final
+    def unregister(cls) -> None:
+        """
+        Unregister this plugin from LSP.
+
+        Call this from your `plugin_unloaded` callback so that the language server is shut down when your package
+        is disabled:
+
+        ```py
+        def plugin_unloaded() -> None:
+            LspFooPlugin.unregister()
+        ```
+        """
+        unregister_plugin_impl(cls)
+
+    @classmethod
+    def is_applicable(cls, context: PluginContext) -> bool:
+        """
+        Determine whether the server should run on the view given by `context.view`.
+
+        The default implementation checks whether the URI scheme and the syntax scope match against the schemes and
+        selector from the settings file. You can override this method for example to dynamically evaluate the applicable
+        selector, or to ignore certain views even when those would match the static config. Please note that no document
+        syncronization messages (textDocument/didOpen, textDocument/didChange, textDocument/didClose, etc.) are sent to
+        the server for ignored views.
+
+        This method is called when the view gets opened. To manually trigger this method again, run the
+        `lsp_check_applicable` TextCommand for the given view and with a `session_name` keyword argument.
+
+        :param      context:           The plugin context.
+        """
+        if (syntax := context.view.syntax()) and (selector := context.configuration.selector.strip()):
+            scheme, _ = parse_uri(uri_from_view(context.view))
+            return scheme in context.configuration.schemes and sublime.score_selector(syntax.scope, selector) > 0
+        return False
+
+    @classmethod
+    def additional_variables(cls, context: PluginContext) -> dict[str, str]:
+        """
+        Return extra template variables to be substituted in ``command``, ``env``, and ``initialization_options``.
+
+        By default includes variables like ``$storage_path``, ``$cache_path``, ``$temp_dir``, ``$home`` and also
+        all variables extracted from the window (the ``window.extract_variables()`` API). Override this method
+        to inject additional variables specific to your plugin.
+
+        :param      context:    The plugin context.
+        :returns:   A dictionary of variable name → value pairs.
+        """
+        return {}
+
+    @classmethod
+    def install_async(cls, context: PluginContext) -> None:
+        """
+        Update or install the server binary if this plugin manages one. Called before the server is started.
+
+        This method runs on a worker thread. Perform any blocking I/O (e.g. downloading a binary,
+        running ``npm install``) directly here without spawning additional threads.
+
+        :param      context:    The plugin context.
+        """
+        pass
+
+    @classmethod
+    def command(cls, context: PluginContext) -> list[str]:
+        """
+        Return the command used to start the language server subprocess.
+
+        The default implementation returns the ``command`` from the settings file after
+        template variable substitution. Override this method to build the command
+        programmatically (e.g. to resolve a binary path at runtime).
+
+        :param      context:    The plugin context.
+        :returns:   A non-empty list where the first element is the executable and the
+                    remaining elements are its arguments.
+        """
+        return context.configuration.command
+
+    @classmethod
+    def initialization_options(cls, context: PluginContext) -> dict[str, Any]:
+        """
+        Return the ``initializationOptions`` sent to the server in the ``initialize`` request.
+
+        The default implementation returns the ``initialization_options`` from the settings
+        file after template variable substitution. Override this method to compute the options
+        dynamically or to merge in runtime values.
+
+        :param      context:    The plugin context.
+        :returns:   A dictionary of initialization options.
+        """
+        return context.configuration.initialization_options.get()
+
+    @classmethod
+    def working_directory(cls, context: PluginContext) -> str | None:
+        """
+        Return the working directory for the language server subprocess.
+
+        The default implementation returns the path of the first workspace folder, or
+        ``None`` when there are no workspace folders. Override this method if the server
+        requires a specific working directory.
+
+        :param      context:    The plugin context.
+        :returns:   An absolute path to use as the working directory, or ``None`` to let the OS choose a default.
+        """
+        return context.workspace_folders[0].path if context.workspace_folders else None
+
+    @classmethod
+    def on_before_initialize(cls, context: PluginContext, transport: TransportWrapper) -> None:
+        """
+        Called after the transport is established but before the LSP ``initialize`` request is sent.
+
+        Override this method when your server requires out-of-band communication that must happen
+        before LSP negotiation begins — for example, sending a proprietary handshake or
+        authentication token over the raw transport.
+
+        Warning:
+            Anything sent via ``transport.send()`` bypasses the LSP message queue. Only use this
+            hook for pre-initialization messages that your server explicitly expects before the
+            ``initialize`` request. Sending arbitrary LSP messages here will corrupt the session.
+
+        :param context:     The plugin context.
+        :param transport:   The live transport connected to the language server process.
+                            Use ``transport.send()`` to write `JSONRPCMessage` messages or ``transport.send_bytes()``
+                            to write byte data.
+        """
+        pass
+
+    @classmethod
+    def markdown_language_id_to_st_syntax_map(cls) -> MarkdownLangMap | None:
+        """
+        Override this method to tweak the syntax highlighting of code blocks in popups from your language server.
+        The returned object should be a dictionary exactly in the form of mdpopup's language_map setting.
+
+        See: https://facelessuser.github.io/sublime-markdown-popups/settings/#mdpopupssublime_user_lang_map
+
+        :returns:   The markdown language map, or None
+        """
+        pass
+
+    def __init__(self, weaksession: ref[Session]) -> None:
+        """
+        Constructs a new instance. Your instance is constructed after a response to the initialize request.
+
+        :param      weaksession:  A weak reference to the Session. You can grab a strong reference through
+                                  self.weaksession(), but don't hold on to that reference.
+        """
+        super().__init__()
+        self.weaksession: ref[Session] = weaksession
+
+    def on_workspace_configuration(self, params: ConfigurationItem, configuration: Any) -> Any:
+        """
+        Override to augment configuration returned for the workspace/configuration request.
+
+        :param      params:         A ConfigurationItem for which configuration is requested.
+        :param      configuration:  The pre-resolved configuration for given params using the settings object or None.
+
+        :returns: The resolved configuration for given params.
+        """
+        return configuration
+
+    def on_execute_command(self, command: ExecuteCommandParams) -> Promise[None] | None:
+        """
+        Intercept a command that is about to be sent to the language server.
+
+        :param    command:        The payload containing a "command" and optionally "arguments".
+
+        :returns: Promise if *YOU* will handle this command plugin-side, None otherwise.
+        """
+        pass
+
+    def on_pre_send_request_async(self, request: ClientRequest, view: sublime.View | None) -> None:
+        """
+        Notifies about a request that is about to be sent to the language server.
+
+        :param    request:     The request object. The request['params'] can be modified by the plugin.
+        :param    view:        The corresponding View if applicable.
+        """
+        pass
+
+    def on_pre_send_notification_async(self, notification: ClientNotification) -> None:
+        """
+        Notifies about a notification that is about to be sent to the language server.
+
+        :param    notification:  The notification object. The notification['params'] can be modified by the plugin.
+        """
+        pass
+
+    def on_server_response_async(self, response: ServerResponse) -> None:
+        """
+        Notifies about a response message that has been received from the language server.
+
+        Only successful responses are passed to this method.
+
+        :param    response:  The response object to the request. The response['result'] field can be modified by the
+                             plugin, before it gets further handled by the LSP package.
+        """
+        pass
+
+    def on_server_notification_async(self, notification: ServerNotification) -> None:
+        """
+        Notifies about a notification message that has been received from the language server.
+
+        :param    notification:  The notification object.
+        """
+        pass
+
+    def on_open_uri_async(self, uri: DocumentUri) -> Promise[sublime.Sheet] | None:
+        """
+        Called when a language server reports to open an URI. If you know how to handle this URI, then return a Promise
+        resolved with `sublime.Sheet` instance.
+        """
+        pass
+
+    def on_session_buffer_changed_async(self, session_buffer: SessionBufferProtocol) -> None:
+        """Called when the context of the session buffer has changed or a new buffer was opened."""
+        pass
+
+    def on_selection_modified_async(self, session_view: SessionViewProtocol) -> None:
+        """Called after the selection has been modified in a view (debounced)."""
+        pass
+
+    def on_session_end_async(self, exit_code: int | None, exception: Exception | None) -> None:
+        """
+        Notifies about the session ending (also if the session has crashed). Provides an opportunity to clean up
+        any stored state or delete references to the session or plugin instance that would otherwise prevent the
+        instance from being garbage-collected.
+
+        If the session hasn't crashed, a shutdown message will be send immediately
+        after this method returns. In this case exit_code and exception are None.
+        If the session has crashed, the exit_code and an optional exception are provided.
+
+        This API is triggered on async thread.
+        """
+        pass
+
+
+@deprecated('Use LspPlugin instead')
 class AbstractPlugin(APIHandler, ABC):
 
     @classmethod
@@ -275,17 +640,11 @@ def is_applicable(cls, view: sublime.View, config: ClientConfig) -> bool:
         :param      view:             The view
         :param      config:           The config
         """
-        if (syntax := view.syntax()) and (selector := cls.selector(view, config).strip()):
-            # TODO: replace `cls.selector(view, config)` with `config.selector` after the next release
+        if (syntax := view.syntax()) and (selector := config.selector.strip()):
             scheme, _ = parse_uri(uri_from_view(view))
             return scheme in config.schemes and sublime.score_selector(syntax.scope, selector) > 0
         return False
 
-    @classmethod
-    @deprecated("Use `is_applicable(view, config)` instead.")
-    def selector(cls, view: sublime.View, config: ClientConfig) -> str:
-        return config.selector
-
     @classmethod
     def additional_variables(cls) -> dict[str, str] | None:
         """In addition to the above variables, add more variables here to be expanded."""
@@ -382,11 +741,6 @@ def on_post_start(cls, window: sublime.Window, initiating_view: sublime.View,
         """
         pass
 
-    @classmethod
-    @deprecated("Use `is_applicable(view, config)` instead.")
-    def should_ignore(cls, view: sublime.View) -> bool:
-        return False
-
     @classmethod
     def markdown_language_id_to_st_syntax_map(cls) -> MarkdownLangMap | None:
         """
@@ -399,7 +753,7 @@ def markdown_language_id_to_st_syntax_map(cls) -> MarkdownLangMap | None:
         """
         return None
 
-    def __init__(self, weaksession: weakref.ref[Session]) -> None:
+    def __init__(self, weaksession: ref[Session]) -> None:
         """
         Constructs a new instance. Your instance is constructed after a response to the initialize request.
 
diff --git a/plugin/core/configurations.py b/plugin/core/configurations.py
index 21b9a1bc3..1f4e8a754 100644
--- a/plugin/core/configurations.py
+++ b/plugin/core/configurations.py
@@ -7,6 +7,7 @@
 from .url import parse_uri
 from .workspace import disable_in_project
 from .workspace import enable_in_project
+from .workspace import WorkspaceFolder
 from abc import ABC
 from abc import abstractmethod
 from collections import deque
@@ -49,7 +50,7 @@ def get_config(self, config_name: str) -> ClientConfig | None:
     def get_configs(self) -> list[ClientConfig]:
         return sorted(self.all.values(), key=lambda config: config.name)
 
-    def match_view(self, view: sublime.View) -> Generator[ClientConfig, None, None]:
+    def match_view(self, view: sublime.View, workspace_folders: list[WorkspaceFolder]) -> Generator[ClientConfig]:
         """
         Yields matching configuration.
 
@@ -63,7 +64,7 @@ def match_view(self, view: sublime.View) -> Generator[ClientConfig, None, None]:
                 return
             scheme = parse_uri(uri)[0]
             for config in self.all.values():
-                if config.enabled and config.match_view(view, scheme):
+                if config.enabled and config.match_view(view, scheme, self._window, workspace_folders):
                     yield config
         except (IndexError, RuntimeError):
             pass
diff --git a/plugin/core/constants.py b/plugin/core/constants.py
index 3cfcc39dd..81aef4764 100644
--- a/plugin/core/constants.py
+++ b/plugin/core/constants.py
@@ -31,6 +31,7 @@
 ST_PACKAGES_PATH = sublime.packages_path()
 ST_PLATFORM = sublime.platform()
 ST_VERSION = int(sublime.version())
+# TODO: Convert to `Path` once `AbstractPlugin` is removed.
 ST_STORAGE_PATH = join(dirname(ST_CACHE_PATH), "Package Storage")
 """
 The "Package Storage" is a way to store server data without influencing the
diff --git a/plugin/core/registry.py b/plugin/core/registry.py
index 178ee5509..0a1c49f4b 100644
--- a/plugin/core/registry.py
+++ b/plugin/core/registry.py
@@ -60,11 +60,19 @@ class LspWindowCommand(sublime_plugin.WindowCommand):
 
     # When this is defined in a derived class, the command is enabled only if there exists a session with the given
     # capability attached to a view in the window.
-    capability = ''
+    capability: str = ''
 
     # When this is defined in a derived class, the command is enabled only if there exists a session with the given
     # name attached to a view in the window.
-    session_name = ''
+    session_name: str = ''
+
+    def __init__(self, window: sublime.Window) -> None:
+        super().__init__(window)
+        if not self.session_name:
+            # Auto-detect session_name based on package name. In case of the LSP package use empty string.
+            package_name = self.__module__.split('.')[0]
+            if package_name != 'LSP':
+                self.session_name = package_name
 
     def is_enabled(self) -> bool:
         return self.session() is not None
@@ -112,11 +120,19 @@ class LspTextCommand(sublime_plugin.TextCommand):
 
     # When this is defined in a derived class, the command is enabled only if there exists a session with the given
     # capability attached to the active view.
-    capability = ''
+    capability: str = ''
 
     # When this is defined in a derived class, the command is enabled only if there exists a session with the given
-    # name attached to the active view.
-    session_name = ''
+    # name attached to the active view. By default it will default to the name of the pacakge it's defined in.
+    session_name: str = ''
+
+    def __init__(self, view: sublime.View) -> None:
+        super().__init__(view)
+        if not self.session_name:
+            # Auto-detect session_name based on package name. In case of the LSP package use empty string.
+            package_name = self.__module__.split('.')[0]
+            if package_name != 'LSP':
+                self.session_name = package_name
 
     def is_enabled(self, event: dict | None = None, point: int | None = None) -> bool:
         if self.capability:
@@ -247,7 +263,7 @@ def _run_async(self, session_name: str) -> None:
                 debug(f'No listener for view {self.view}')
                 return
             scheme, _ = parse_uri(uri_from_view(self.view))
-            is_applicable = config.match_view(self.view, scheme)
+            is_applicable = config.match_view(self.view, scheme, wm.window, wm.workspace_folders)
             if session := wm.get_session(session_name, self.view.file_name() or ''):
                 session_view = session.session_view_for_view_async(self.view)
                 if is_applicable and not session_view:
diff --git a/plugin/core/sessions.py b/plugin/core/sessions.py
index de5e66269..69c0a87d4 100644
--- a/plugin/core/sessions.py
+++ b/plugin/core/sessions.py
@@ -73,6 +73,7 @@
 from ...protocol import WorkspaceFullDocumentDiagnosticReport
 from ..api import AbstractPlugin
 from ..api import APIHandler
+from ..api import LspPlugin
 from ..api import notification_handler
 from ..api import request_handler
 from ..diagnostics import DiagnosticsIdentifier
@@ -102,6 +103,8 @@
 from .progress import WindowProgressReporter
 from .promise import PackagedTask
 from .promise import Promise
+from .protocol import ClientNotification
+from .protocol import ClientRequest
 from .protocol import Error
 from .protocol import JSONRPCMessage
 from .protocol import Notification
@@ -109,6 +112,8 @@
 from .protocol import ResolvedCodeLens
 from .protocol import Response
 from .protocol import ResponseError
+from .protocol import ServerNotification
+from .protocol import ServerResponse
 from .settings import globalprefs
 from .settings import userprefs
 from .transports import TransportCallbacks
@@ -127,7 +132,6 @@
 from .url import normalize_uri
 from .url import parse_uri
 from .version import __version__
-from .views import extract_variables
 from .views import get_uri_and_range_from_location
 from .views import kind_contains_other_kind
 from .views import MarkdownLangMap
@@ -148,7 +152,6 @@
 from typing import TYPE_CHECKING
 from typing import TypeVar
 from typing import Union
-from typing_extensions import deprecated
 from typing_extensions import TypeAlias
 from typing_extensions import TypeGuard
 from weakref import WeakSet
@@ -960,7 +963,8 @@ def check_applicable(self, sb: SessionBufferProtocol, *, suppress_requests: bool
 class Session(APIHandler, TransportCallbacks):
 
     def __init__(self, manager: Manager, logger: Logger, workspace_folders: list[WorkspaceFolder],
-                 config: ClientConfig, plugin_class: type[AbstractPlugin] | None) -> None:
+                 config: ClientConfig, plugin_class: type[AbstractPlugin | LspPlugin] | None,
+    ) -> None:
         self.transport: TransportWrapper | None = None
         self.working_directory: str | None = None
         self.request_id = 0  # Our request IDs are always integers.
@@ -980,6 +984,7 @@ def __init__(self, manager: Manager, logger: Logger, workspace_folders: list[Wor
         self._init_callback: InitCallback | None = None
         self._initialize_error: tuple[int, Exception | None] | None = None
         self._views_opened = 0
+        self._variables: dict[str, str] = {}
         self._workspace_folders = workspace_folders
         self._session_views: WeakSet[SessionViewProtocol] = WeakSet()
         self._session_buffers: WeakSet[SessionBufferProtocol] = WeakSet()
@@ -988,7 +993,7 @@ def __init__(self, manager: Manager, logger: Logger, workspace_folders: list[Wor
         self._static_file_watchers: list[FileWatcher] = []
         self._dynamic_file_watchers: dict[str, list[FileWatcher]] = {}
         self._plugin_class = plugin_class
-        self._plugin: AbstractPlugin | None = None
+        self._plugin: AbstractPlugin | LspPlugin | None = None
         self._status_messages: dict[str, str] = {}
         self._semantic_tokens_map = get_semantic_tokens_map(config.semantic_tokens)
         self._is_executing_refactoring_command = False
@@ -1017,7 +1022,7 @@ def uses_plugin(self) -> bool:
         return self._plugin is not None
 
     @property
-    def plugin(self) -> AbstractPlugin | None:
+    def plugin(self) -> AbstractPlugin | LspPlugin | None:
         return self._plugin
 
     # --- session view management --------------------------------------------------------------------------------------
@@ -1057,18 +1062,6 @@ def _redraw_config_status_async(self) -> None:
         for sv in self.session_views_async():
             self.config.set_view_status(sv.view, self.config_status_message)
 
-    @deprecated("Use set_config_status_async(message) instead")
-    def set_window_status_async(self, key: str, message: str) -> None:
-        self._status_messages[key] = message
-        for sv in self.session_views_async():
-            sv.view.set_status(key, message)
-
-    @deprecated("Use set_config_status_async('') instead")
-    def erase_window_status_async(self, key: str) -> None:
-        self._status_messages.pop(key, None)
-        for sv in self.session_views_async():
-            sv.view.erase_status(key)
-
     # --- session buffer management ------------------------------------------------------------------------------------
 
     def register_session_buffer_async(self, sb: SessionBufferProtocol) -> None:
@@ -1124,9 +1117,6 @@ def compare_by_string(sb: SessionBufferProtocol | None) -> bool:
     def can_handle(self, view: sublime.View, scheme: str, capability: str | None, inside_workspace: bool) -> bool:
         if self.state != ClientStates.READY:
             return False
-        if self._plugin and self._plugin.should_ignore(view):  # TODO: remove after next release
-            debug(view, "ignored by plugin", self._plugin.__class__.__name__)
-            return False
         if scheme == "file":
             file_name = view.file_name()
             if not file_name:
@@ -1134,7 +1124,7 @@ def can_handle(self, view: sublime.View, scheme: str, capability: str | None, in
                 return False
             if not self.handles_path(file_name, inside_workspace):
                 return False
-        if self.config.match_view(view, scheme):
+        if self.config.match_view(view, scheme, self.window, self._workspace_folders):
             # If there's no capability requirement then this session can handle the view
             if capability is None:
                 return True
@@ -1242,7 +1232,8 @@ def initialize_async(
     ) -> None:
         self.transport = transport
         self.working_directory = working_directory
-        params = get_initialize_params(variables, self._workspace_folders, self.config)
+        self._variables = variables
+        params = get_initialize_params(self._variables, self._workspace_folders, self.config)
         self._init_callback = init_callback
         self.send_request_async(
             Request.initialize(params), self._handle_initialize_success, self._handle_initialize_error)
@@ -1255,11 +1246,16 @@ def _handle_initialize_success(self, result: InitializeResult) -> None:
         if diagnostic_options := capabilities.get('diagnosticProvider'):
             self.diagnostics.register_provider(diagnostic_options.get('id'), diagnostic_options)
         self.state = ClientStates.READY
-        if self._plugin_class is not None:
-            self._plugin = self._plugin_class(weakref.ref(self))
+        if self._plugin_class:
             # We've missed calling the "on_server_response_async" API as plugin was not created yet.
             # Handle it now and use fake request ID since it shouldn't matter.
-            self._plugin.on_server_response_async('initialize', Response(-1, result))
+            if issubclass(self._plugin_class, LspPlugin):
+                self._plugin = self._plugin_class(weakref.ref(self))
+                server_response: ServerResponse = {'method': 'initialize', 'result': result}
+                self._plugin.on_server_response_async(server_response)
+            else:
+                self._plugin = self._plugin_class(weakref.ref(self))
+                self._plugin.on_server_response_async('initialize', Response[InitializeResult](-1, result))
         self.send_notification(Notification.initialized())
         self._maybe_send_did_change_configuration()
         if execute_commands := self.get_capability('executeCommandProvider.commands'):
@@ -1311,29 +1307,26 @@ def _supports_workspace_folders(self) -> bool:
 
     def _maybe_send_did_change_configuration(self) -> None:
         if self.config.settings:
-            if self._plugin:
+            if isinstance(self._plugin, AbstractPlugin):
                 self._plugin.on_settings_changed(self.config.settings)
-            variables = self._template_variables()
-            resolved = self.config.settings.get_resolved(variables)
+            resolved = self.config.settings.get_resolved(self._variables)
             self.send_notification(Notification("workspace/didChangeConfiguration", {"settings": resolved}))
 
-    def _template_variables(self) -> dict[str, str]:
-        variables = extract_variables(self.window)
-        if self._plugin_class is not None:
-            if extra_vars := self._plugin_class.additional_variables():
-                variables.update(extra_vars)
-        return variables
-
     def execute_command(
         self, command: ExecuteCommandParams, *, progress: bool = False, view: sublime.View | None = None,
         is_refactoring: bool = False,
     ) -> Promise[R | Error | None]:  # pyright: ignore[reportInvalidTypeVarUse]
         """Run a command from any thread. Your .then() continuations will run in Sublime's worker thread."""
         if self._plugin:
-            task: PackagedTask[R | Error | None] = Promise.packaged_task()
-            promise, resolve = task
-            if self._plugin.on_pre_server_command(command, lambda: resolve(None)):
-                return promise
+            if isinstance(self._plugin, LspPlugin):
+                if promise := self._plugin.on_execute_command(command):
+                    return promise.then(lambda _: None)
+            else:
+                task: PackagedTask[R | Error | None] = Promise.packaged_task()
+                promise, resolve = task
+                if self._plugin.on_pre_server_command(command, lambda: resolve(None)):
+                    return promise
+                resolve(None)
         command_name = command['command']
         # Handle VSCode-specific command for triggering AC/sighelp
         if command_name == "editor.action.triggerSuggest" and view:
@@ -1424,7 +1417,17 @@ def try_open_uri_async(
             return Promise.resolve(view)
         # There is no pre-existing session-buffer, so we have to go through AbstractPlugin.on_open_uri_async.
         if self._plugin:
-            return self._open_uri_with_plugin_async(self._plugin, uri, r, flags, group)
+            if isinstance(self._plugin, LspPlugin):
+                def on_sheet_opened(sheet: sublime.Sheet) -> sublime.View | None:
+                    if view := sheet.view():
+                        view.settings().set('lsp_uri', uri)  # Preserve original URI given by the language server
+                        return view
+                    return None
+
+                if promise := self._plugin.on_open_uri_async(uri):
+                    return promise.then(on_sheet_opened)
+            else:
+                return self._open_uri_with_plugin_async(self._plugin, uri, r, flags, group)
         return None
 
     def open_uri_async(
@@ -1793,7 +1796,7 @@ def on_workspace_configuration(self, params: ConfigurationParams) -> Promise[lis
                 items.append(self._plugin.on_workspace_configuration(requested_item, configuration))
             else:
                 items.append(configuration)
-        return Promise.resolve(sublime.expand_variables(items, self._template_variables()))
+        return Promise.resolve(sublime.expand_variables(items, self._variables))
 
     @request_handler('workspace/applyEdit')
     def on_workspace_apply_edit(self, params: ApplyWorkspaceEditParams) -> Promise[ApplyWorkspaceEditResult]:
@@ -2144,8 +2147,12 @@ def send_request_async(
         on_error = on_error or (lambda _: None)
         self._response_handlers[request_id] = (request, on_result, on_error)
         self._invoke_views(request, "on_request_started_async", request_id, request)
-        if self._plugin:
+        if self._plugin and isinstance(self._plugin, AbstractPlugin):
             self._plugin.on_pre_send_request_async(request_id, request)
+        elif self._plugin:
+            client_request = cast('ClientRequest', cast('object', {'method': request.method, 'params': request.params}))
+            self._plugin.on_pre_send_request_async(client_request, request.view)
+            request.params = cast('P', client_request['params'])
         self._logger.outgoing_request(request_id, request.method, request.params)
         self.send_payload(request.to_payload(request_id))
         return request_id
@@ -2180,8 +2187,13 @@ def cancel_request_async(self, request_id: int) -> None:
             self._response_handlers[request_id] = (request, lambda *args: None, lambda *args: None)
 
     def send_notification(self, notification: Notification[P]) -> None:
-        if self._plugin:
+        if self._plugin and isinstance(self._plugin, AbstractPlugin):
             self._plugin.on_pre_send_notification_async(notification)
+        elif self._plugin:
+            client_notification = cast('ClientNotification',
+                                       cast('object', {'method': notification.method, 'params': notification.params}))
+            self._plugin.on_pre_send_notification_async(client_notification)
+            notification.params = cast('P', client_notification['params'])
         self._logger.outgoing_notification(notification.method, notification.params)
         self.send_payload(notification.to_payload())
 
@@ -2223,8 +2235,12 @@ def deduce_payload(
             else:
                 res = (handler, result, None, "notification", method)
                 self._logger.incoming_notification(method, result, res[0] is None)
-                if self._plugin:
+                if self._plugin and isinstance(self._plugin, AbstractPlugin):
                     self._plugin.on_server_notification_async(Notification(method, result))
+                elif self._plugin:
+                    server_notification = cast('ServerNotification',
+                                               cast('object', {'method': method, 'result': result}))
+                    self._plugin.on_server_notification_async(server_notification)
                 return res
         elif "id" in payload:
             response_id = payload["id"]
@@ -2234,8 +2250,14 @@ def deduce_payload(
             handler, method, result, is_error = self.response_handler(response_id, payload)
             self._logger.incoming_response(response_id, result, is_error)
             response = Response(response_id, result)
-            if self._plugin and not is_error:
-                self._plugin.on_server_response_async(method, response)  # type: ignore
+            if not is_error and self._plugin:
+                if isinstance(self._plugin, AbstractPlugin):
+                    self._plugin.on_server_response_async(cast('str', method), response)
+                else:
+                    server_response = cast('ServerResponse',
+                                           cast('object', {'method': method, 'result': response.result}))
+                    self._plugin.on_server_response_async(server_response)
+                    response.result = server_response['result']
             return handler, response.result, None, None, None
         else:
             debug("Unknown payload type: ", payload)  # pyright: ignore[reportUnreachable]
diff --git a/plugin/core/transports.py b/plugin/core/transports.py
index bccfd3ebb..40b11c416 100644
--- a/plugin/core/transports.py
+++ b/plugin/core/transports.py
@@ -257,6 +257,10 @@ def read(self) -> JSONRPCMessage | None:
     def write(self, payload: JSONRPCMessage) -> None:
         raise NotImplementedError
 
+    @abstractmethod
+    def write_bytes(self, payload: bytes) -> None:
+        raise NotImplementedError
+
     @abstractmethod
     def close(self) -> None:
         raise NotImplementedError
@@ -300,6 +304,11 @@ def write(self, payload: JSONRPCMessage) -> None:
         self._writer.writelines((f"Content-Length: {len(body)}\r\n\r\n".encode("ascii"), body))
         self._writer.flush()
 
+    @override
+    def write_bytes(self, payload: bytes) -> None:
+        self._writer.write(payload)
+        self._writer.flush()
+
     @override
     def close(self) -> None:
         self._writer.close()
@@ -349,7 +358,7 @@ def __init__(
         self._error_reader = error_reader
         self._reader_thread = threading.Thread(target=self._read_loop)
         self._writer_thread = threading.Thread(target=self._write_loop)
-        self._send_queue: Queue[JSONRPCMessage | None] = Queue(0)
+        self._send_queue: Queue[JSONRPCMessage | bytes | None] = Queue(0)
         self._reader_thread.start()
         self._writer_thread.start()
 
@@ -360,6 +369,9 @@ def process_args(self) -> Any:
     def send(self, payload: JSONRPCMessage) -> None:
         self._send_queue.put_nowait(payload)
 
+    def send_bytes(self, payload: bytes) -> None:
+        self._send_queue.put_nowait(payload)
+
     def close(self) -> None:
         if not self._closed:
             self._closed = True
@@ -431,7 +443,10 @@ def _write_loop(self) -> None:
             while self._transport:
                 if (d := self._send_queue.get()) is None:
                     break
-                self._transport.write(d)
+                if isinstance(d, bytes):
+                    self._transport.write_bytes(d)
+                else:
+                    self._transport.write(d)
         except (BrokenPipeError, AttributeError):
             pass
         except Exception as ex:
diff --git a/plugin/core/types.py b/plugin/core/types.py
index a1b11f9bb..c6268fd54 100644
--- a/plugin/core/types.py
+++ b/plugin/core/types.py
@@ -46,6 +46,7 @@
 
 if TYPE_CHECKING:
     from .file_watcher import FileWatcherEventType
+    from .workspace import WorkspaceFolder
 
 FEATURES_TIMEOUT = 300  # milliseconds
 
@@ -1030,7 +1031,9 @@ def erase_view_status(self, view: sublime.View) -> None:
         """
         self._view_status_handler.on_view_status_changed(self.name, view, None)
 
-    def match_view(self, view: sublime.View, scheme: str) -> bool:
+    def match_view(
+        self, view: sublime.View, scheme: str, window: sublime.Window, workspace_folders: list[WorkspaceFolder]
+    ) -> bool:
         """
         Return `True` if this server should be active for the given view.
 
@@ -1040,9 +1043,14 @@ def match_view(self, view: sublime.View, scheme: str) -> bool:
         :param view: The view to test.
         :param scheme: The URI scheme of the view's resource (e.g. `"file"`).
         """
+        from ..api import AbstractPlugin
         from ..api import get_plugin
+        from ..api import PluginContext
         if plugin := get_plugin(self.name):
-            return plugin.is_applicable(view, self)
+            if issubclass(plugin, AbstractPlugin):
+                return plugin.is_applicable(view, self)
+            plugin_context = PluginContext(self, view, window, workspace_folders)
+            return plugin.is_applicable(plugin_context)
         if (syntax := view.syntax()) and (selector := self.selector.strip()):
             return scheme in self.schemes and sublime.score_selector(syntax.scope, selector) > 0
         return False
diff --git a/plugin/core/windows.py b/plugin/core/windows.py
index e0d265d44..b9eeb13a3 100644
--- a/plugin/core/windows.py
+++ b/plugin/core/windows.py
@@ -8,7 +8,12 @@
 from ...protocol import ShowMessageParams
 from ...protocol import ShowMessageRequestParams
 from ...third_party import WebsocketServer  # type: ignore
+from ..api import AbstractPlugin
 from ..api import get_plugin
+from ..api import LspPlugin
+from ..api import PluginContext
+from ..api import PluginStartError
+from .collections import DottedDict
 from .configurations import RETRY_COUNT_TIMEDELTA
 from .configurations import RETRY_MAX_COUNT
 from .configurations import WindowConfigChangeListener
@@ -43,6 +48,7 @@
 from .views import make_link
 from .workspace import ProjectFolders
 from .workspace import sorted_workspace_folders
+from .workspace import WorkspaceFolder
 from collections import deque
 from datetime import datetime
 from subprocess import CalledProcessError
@@ -104,6 +110,10 @@ def __init__(self, window: sublime.Window, workspace: ProjectFolders, config_man
     def window(self) -> sublime.Window:
         return self._window
 
+    @property
+    def workspace_folders(self) -> list[WorkspaceFolder]:
+        return self._workspace.get_workspace_folders()
+
     def get_and_clear_server_log(self) -> list[tuple[str, str]]:
         log = self._server_log
         self._server_log = []
@@ -220,7 +230,7 @@ def _find_session(self, config_name: str, file_path: str) -> Session | None:
         return None
 
     def _needed_config(self, view: sublime.View) -> ClientConfig | None:
-        configs = self._config_manager.match_view(view)
+        configs = self._config_manager.match_view(view, self._workspace.get_workspace_folders())
         handled = False
         file_name = view.file_name()
         inside = self._workspace.contains(view)
@@ -232,9 +242,11 @@ def _needed_config(self, view: sublime.View) -> ClientConfig | None:
                     break
             if not handled:
                 if plugin := get_plugin(config.name):
-                    if plugin.should_ignore(view):  # TODO: remove after next release
-                        debug(view, "ignored by plugin", plugin.__name__)
-                    elif plugin.is_applicable(view, config):
+                    plugin_context = PluginContext(config, view, self._window, self._workspace.get_workspace_folders())
+                    if issubclass(plugin, AbstractPlugin):
+                        if plugin.is_applicable(view, config):
+                            return config
+                    elif plugin.is_applicable(plugin_context):
                         return config
                 else:
                     return config
@@ -245,41 +257,44 @@ def start_async(self, config: ClientConfig, initiating_view: sublime.View) -> No
         config.set_view_status_handler(self)
         file_path = initiating_view.file_name() or ''
         if not self._can_start_config(config.name, file_path):
-            # debug('Already starting on this window:', config.name)
             return
         try:
             workspace_folders = sorted_workspace_folders(self._workspace.folders, file_path)
             plugin_class = get_plugin(config.name)
             variables = extract_variables(self._window)
-            cwd: str | None = None
-            if plugin_class is not None:
-                if plugin_class.needs_update_or_installation():
+            cwd = workspace_folders[0].path if workspace_folders else None
+            plugin_context = PluginContext(config, initiating_view, self._window, workspace_folders)
+            if plugin_class:
+                if issubclass(plugin_class, LspPlugin):
                     config.set_view_status(initiating_view, "installing...")
-                    plugin_class.install_or_update()
-                additional_variables = plugin_class.additional_variables()
+                    plugin_class.install_async(plugin_context)
+                    additional_variables = plugin_class.additional_variables(plugin_context)
+                else:
+                    if plugin_class.needs_update_or_installation():
+                        config.set_view_status(initiating_view, "installing...")
+                        plugin_class.install_or_update()
+                    additional_variables = plugin_class.additional_variables()
                 if isinstance(additional_variables, dict):
                     variables.update(additional_variables)
-                cannot_start_reason = plugin_class.can_start(self._window, initiating_view, workspace_folders, config)
-                if cannot_start_reason:
-                    config.erase_view_status(initiating_view)
-                    message = f"cannot start {config.name}: {cannot_start_reason}"
-                    self._config_manager.disable_config(config.name, only_for_session=True)
-                    # Continue with handling pending listeners
-                    self._new_session = None
-                    sublime.set_timeout_async(self._dequeue_listener_async)
-                    self._window.status_message(message)
-                    return
-                cwd = plugin_class.on_pre_start(self._window, initiating_view, workspace_folders, config)
+                if issubclass(plugin_class, AbstractPlugin):
+                    cannot_start_reason = plugin_class.can_start(
+                        self._window, initiating_view, workspace_folders, config)
+                    if cannot_start_reason:
+                        raise PluginStartError(cannot_start_reason)
+                if issubclass(plugin_class, LspPlugin):
+                    config.command = plugin_class.command(plugin_context)
+                    config.initialization_options = DottedDict(plugin_class.initialization_options(plugin_context))
+                    cwd = plugin_class.working_directory(plugin_context)
+                elif plugin_cwd := plugin_class.on_pre_start(self._window, initiating_view, workspace_folders, config):
+                    cwd = plugin_cwd
             config.set_view_status(initiating_view, "starting...")
             session = Session(self, self._create_logger(config.name), workspace_folders, config, plugin_class)
-            if cwd:
-                transport_cwd: str | None = cwd
-            else:
-                transport_cwd = workspace_folders[0].path if workspace_folders else None
-            transport = config.create_transport_config().start(
-                config.command, config.env, transport_cwd, variables, session)
+            transport = config.create_transport_config().start(config.command, config.env, cwd, variables, session)
             if plugin_class:
-                plugin_class.on_post_start(self._window, initiating_view, workspace_folders, config)
+                if issubclass(plugin_class, AbstractPlugin):
+                    plugin_class.on_post_start(self._window, initiating_view, workspace_folders, config)
+                else:
+                    plugin_class.on_before_initialize(plugin_context, transport)
             config.set_view_status(initiating_view, "initialize")
             session.initialize_async(
                 variables=variables,
@@ -288,6 +303,14 @@ def start_async(self, config: ClientConfig, initiating_view: sublime.View) -> No
                 init_callback=functools.partial(self._on_post_session_initialize, initiating_view)
             )
             self._new_session = session
+        except PluginStartError as ex:
+            config.erase_view_status(initiating_view)
+            message = f"cannot start {config.name}: {ex!s}"
+            self._config_manager.disable_config(config.name, only_for_session=True)
+            # Continue with handling pending listeners
+            self._new_session = None
+            sublime.set_timeout_async(self._dequeue_listener_async)
+            self._window.status_message(message)
         except Exception as e:
             message = (f'Failed to start {config.name} - disabling for this window for the duration of the current '
                         'session.\nRe-enable by running "LSP: Enable Language Server In Project" from the Command '
diff --git a/plugin/tooling.py b/plugin/tooling.py
index e4bb4446a..168ac3bef 100644
--- a/plugin/tooling.py
+++ b/plugin/tooling.py
@@ -1,6 +1,10 @@
 from __future__ import annotations
 
+from .api import AbstractPlugin
 from .api import get_plugin
+from .api import PluginContext
+from .api import PluginStartError
+from .core.collections import DottedDict
 from .core.css import css
 from .core.logging import debug
 from .core.registry import windows
@@ -504,16 +508,27 @@ def __init__(
             workspace = ProjectFolders(window)
             workspace_folders = sorted_workspace_folders(workspace.folders, initiating_view.file_name() or '')
             cwd = None
-            if plugin_class is not None:
-                if plugin_class.needs_update_or_installation():
-                    plugin_class.install_or_update()
-                additional_variables = plugin_class.additional_variables()
+            if plugin_class:
+                plugin_context = PluginContext(config, initiating_view, window, workspace_folders)
+                if issubclass(plugin_class, AbstractPlugin):
+                    if plugin_class.needs_update_or_installation():
+                        plugin_class.install_or_update()
+                    additional_variables = plugin_class.additional_variables()
+                else:
+                    plugin_class.install_async(plugin_context)
+                    additional_variables = plugin_class.additional_variables(plugin_context)
                 if isinstance(additional_variables, dict):
                     variables.update(additional_variables)
-                cannot_start_reason = plugin_class.can_start(window, initiating_view, workspace_folders, config)
-                if cannot_start_reason:
-                    raise Exception(f'Plugin.can_start() prevented the start due to: {cannot_start_reason}')
-                cwd = plugin_class.on_pre_start(window, initiating_view, workspace_folders, config)
+                if issubclass(plugin_class, AbstractPlugin):
+                    reason = plugin_class.can_start(window, initiating_view, workspace_folders, config)
+                    if reason:
+                        raise PluginStartError(f'Plugin.can_start() prevented the start due to: {reason}')
+                if issubclass(plugin_class, AbstractPlugin):
+                    cwd = plugin_class.on_pre_start(window, initiating_view, workspace_folders, config)
+                else:
+                    config.command = plugin_class.command(plugin_context)
+                    config.initialization_options = DottedDict(plugin_class.initialization_options(plugin_context))
+                    cwd = plugin_class.working_directory(plugin_context)
             if not cwd and workspace_folders:
                 cwd = workspace_folders[0].path
             transport_config = config.create_transport_config()
diff --git a/tests/setup.py b/tests/setup.py
index 0dcc86659..3742280a5 100644
--- a/tests/setup.py
+++ b/tests/setup.py
@@ -117,7 +117,7 @@ def setUp(self) -> Generator:
         if not open_view:
             self.__class__.view = window.open_file(filename)
             yield {"condition": lambda: not self.view.is_loading(), "timeout": TIMEOUT_TIME}
-            self.assertTrue(self.wm.get_config_manager().match_view(self.view))
+            self.assertTrue(self.wm.get_config_manager().match_view(self.view, self.wm.workspace_folders))
         self.init_view_settings()
         yield self.ensure_document_listener_created
         params = yield from self.await_message("textDocument/didOpen")
diff --git a/tests/test_configurations.py b/tests/test_configurations.py
index 569cf9dc9..b7c07e44a 100644
--- a/tests/test_configurations.py
+++ b/tests/test_configurations.py
@@ -33,7 +33,7 @@ def test_no_configs(self) -> None:
         self.assertIsNotNone(self.view)
         self.assertIsNotNone(self.window)
         manager = WindowConfigManager(self.window, {})
-        self.assertEqual(list(manager.match_view(self.view)), [])
+        self.assertEqual(list(manager.match_view(self.view, [])), [])
 
     def test_with_single_config(self) -> None:
         self.assertIsNotNone(self.view)
@@ -46,7 +46,7 @@ def test_with_single_config(self) -> None:
             hidden=False
         ))
         self.view.settings().set("lsp_uri", "file:///foo/bar.txt")
-        self.assertEqual(list(manager.match_view(self.view)), [TEST_CONFIG])
+        self.assertEqual(list(manager.match_view(self.view, [])), [TEST_CONFIG])
 
     def test_applies_project_settings(self) -> None:
         self.window.project_data = MagicMock(return_value={
@@ -66,7 +66,7 @@ def test_applies_project_settings(self) -> None:
             hidden=False
         ))
         self.view.settings().set("lsp_uri", "file:///foo/bar.txt")
-        configs = list(manager.match_view(self.view))
+        configs = list(manager.match_view(self.view, []))
         self.assertEqual(len(configs), 1)
         config = configs[0]
         self.assertEqual(DISABLED_CONFIG.name, config.name)
@@ -86,4 +86,4 @@ def test_disables_temporarily(self) -> None:
         manager = WindowConfigManager(self.window, {DISABLED_CONFIG.name: DISABLED_CONFIG})
         # disables config in-memory
         manager.disable_config(DISABLED_CONFIG.name, only_for_session=True)
-        self.assertFalse(any(manager.match_view(self.view)))
+        self.assertFalse(any(manager.match_view(self.view, [])))
diff --git a/tests/test_documents.py b/tests/test_documents.py
index 018b3cb03..de67789d1 100644
--- a/tests/test_documents.py
+++ b/tests/test_documents.py
@@ -67,7 +67,7 @@ def test_sends_did_open_to_multiple_sessions(self) -> Generator:
         yield from close_test_view(open_view)
         self.view = self.window.open_file(filename)
         yield {"condition": lambda: not self.view.is_loading(), "timeout": TIMEOUT_TIME}
-        self.assertTrue(self.wm.get_config_manager().match_view(self.view))
+        self.assertTrue(self.wm.get_config_manager().match_view(self.view, self.wm.workspace_folders))
         # self.init_view_settings()
         yield {"condition": self.ensure_document_listener_created, "timeout": TIMEOUT_TIME}
         yield {
diff --git a/tests/test_server_requests.py b/tests/test_server_requests.py
index 98f525c7d..ad5fd3237 100644
--- a/tests/test_server_requests.py
+++ b/tests/test_server_requests.py
@@ -49,14 +49,7 @@ def test_m_workspace_configuration(self) -> Generator:
         self.session.config.settings.set("foo.a", 1)
         self.session.config.settings.set("foo.b", None)
         self.session.config.settings.set("foo.c", ["asdf ${hello} ${world}"])
-
-        class TempPlugin:
-
-            @classmethod
-            def additional_variables(cls) -> dict[str, str] | None:
-                return {"hello": "X", "world": "Y"}
-
-        self.session._plugin_class = TempPlugin  # type: ignore
+        self.session._variables.update({"hello": "X", "world": "Y"})
         method = "workspace/configuration"
         params = {"items": [{"section": "foo"}]}
         expected_output = [{"bar": "X", "baz": "Y", "a": 1, "b": None, "c": ["asdf X Y"]}]