SGLang Plugin System

DeviceMixin (shared device identity + operations)
├── SRTPlatform(DeviceMixin)           # + graph runner, KV pool, …
│   └── MySRTPlatform(SRTPlatform, MyDeviceMixin)   # OOT plugin
└── MMPlatform(DeviceMixin)            # + attention backend, VAE, … (future)
    └── MyMMPlatform(MMPlatform, MyDeviceMixin)      # OOT plugin

entry_points("sglang.srt.platforms")  → Enumerate ALL plugins by name (metadata only)
  │
  ├─ SGLANG_PLATFORM set (front-loading filter):
  │   ├─ Name not found in discovered → RuntimeError
  │   ├─ activate() returns non-None  → load that platform
  │   └─ activate() returns None      → RuntimeError (hardware unavailable)
  │
  └─ SGLANG_PLATFORM unset (auto-discover, activate all):
      ├─ 0 activated → fallback base SRTPlatform
      ├─ 1 activated → use it
      └─ N activated → RuntimeError (must set SGLANG_PLATFORM)

load_plugins()
  ├── _get_excluded_dists()                       → compute dists to skip (via SGLANG_PLATFORM)
  ├── load_plugins_by_group("sglang.srt.plugins",     → discover entry_points, filter by SGLANG_PLUGINS
  │     excluded_dists=...)                          skip plugins from unselected platform packages
  ├── for each plugin:                            → set _current_plugin_source context var
  │     func()                                      side effects (register hooks with source tracking)
  └── HookRegistry.apply_hooks()                  → monkey-patch targets

my_platform_plugin/
├── pyproject.toml
└── my_platform_plugin/
    ├── __init__.py    # activate() function
    ├── device.py      # MyDeviceMixin
    └── platform.py    # MySRTPlatform

[build-system]
requires = ["setuptools"]
build-backend = "setuptools.build_meta"

[project]
name = "my-platform-plugin"
version = "0.1.0"

[project.entry-points."sglang.srt.platforms"]
my_device = "my_platform_plugin:activate"

def activate():
    """Return fully-qualified class name to activate, or None to skip."""
    if _my_device_is_available():
        return "my_platform_plugin.platform.MySRTPlatform"
    return None

from sglang.srt.platforms.device_mixin import DeviceMixin, PlatformEnum

class MyDeviceMixin(DeviceMixin):
    _enum = PlatformEnum.OOT
    device_name = "my_device"
    device_type = "my_device"   # torch device type

    def set_device(self, device) -> None: ...
    def get_device_name(self, device_id=0) -> str: ...
    def get_device_total_memory(self, device_id=0) -> int: ...
    def get_current_memory_usage(self, device=None) -> float: ...
    def get_device_capability(self, device_id=0): ...
    def get_torch_distributed_backend_str(self) -> str: ...

from sglang.srt.platforms.interface import SRTPlatform
from my_platform_plugin.device import MyDeviceMixin

class MySRTPlatform(SRTPlatform, MyDeviceMixin):
    def get_default_attention_backend(self) -> str: ...
    def support_cuda_graph(self) -> bool: ...
    # ... override other methods as needed

pip install -e my_platform_plugin/
python -c "from sglang.srt.platforms import current_platform; print(current_platform)"

my_general_plugin/
├── pyproject.toml
└── my_general_plugin/
    └── __init__.py    # register() function

[build-system]
requires = ["setuptools"]
build-backend = "setuptools.build_meta"

[project]
name = "my-general-plugin"
version = "0.1.0"

[project.entry-points."sglang.srt.plugins"]
my_plugin = "my_general_plugin:register"

from sglang.srt.plugins.hook_registry import HookRegistry, HookType

def register():
    """Entry point called by load_plugins()."""
    HookRegistry.register(
        "sglang.srt.managers.scheduler.Scheduler.__init__",
        my_hook,
        HookType.AROUND,
    )

def my_hook(original_fn, self, *args, **kwargs):
    result = original_fn(self, *args, **kwargs)
    print(f"Scheduler initialized! gpu_id={self.gpu_id}")
    return result

pip install -e my_general_plugin/
sglang serve --model-path <model> [options]
# Look for "Scheduler initialized!" in logs

# --- Imperative API ---
from sglang.srt.plugins.hook_registry import HookRegistry, HookType

def my_timer(original_fn, *args, **kwargs):
    start = time.perf_counter()
    result = original_fn(*args, **kwargs)
    print(f"Elapsed: {time.perf_counter() - start:.3f}s")
    return result

HookRegistry.register(
    "sglang.srt.managers.scheduler.Scheduler.get_next_batch_to_run",
    my_timer,
    HookType.AROUND,
)

# --- Decorator API ---
from sglang.srt.plugins.hook_registry import plugin_hook, HookType

@plugin_hook(
    "sglang.srt.managers.scheduler.Scheduler.get_next_batch_to_run",
    type=HookType.AROUND,
)
def my_timer(original_fn, *args, **kwargs):
    start = time.perf_counter()
    result = original_fn(*args, **kwargs)
    print(f"Elapsed: {time.perf_counter() - start:.3f}s")
    return result

# --- Class replacement (REPLACE) ---
from sglang.srt.plugins.hook_registry import plugin_hook, HookType
from sglang.srt.managers.scheduler import Scheduler

@plugin_hook(
    "sglang.srt.managers.scheduler.Scheduler",
    type=HookType.REPLACE,
)
class MyScheduler(Scheduler):
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        print("Enhanced scheduler initialized!")