feat: simpler permissions

lnbits/core/extensions/__init__.py

@@ -1 +1,19 @@
 """Extension runtime contracts."""
+
+from .api import (
+    ExtensionAPI,
+    ExtensionAPIMethod,
+    extension_api_contract,
+    extension_api_method,
+    get_extension_api_method,
+    list_extension_api_methods,
+)
+
+__all__ = [
+    "ExtensionAPI",
+    "ExtensionAPIMethod",
+    "extension_api_contract",
+    "extension_api_method",
+    "get_extension_api_method",
+    "list_extension_api_methods",
+]

lnbits/core/extensions/api.py

@@ -0,0 +1,397 @@
+from __future__ import annotations
+
+import inspect
+import logging
+import secrets
+import time
+from collections.abc import Awaitable, Callable, Iterable
+from dataclasses import dataclass
+from functools import wraps
+from typing import Literal, TypeVar, get_type_hints
+
+from pydantic import BaseModel, Field
+
+logger = logging.getLogger("lnbits.extensions")
+
+_EXTENSION_API_METHOD_ATTR = "__lnbits_extension_api_method__"
+_RequestModel = TypeVar("_RequestModel", bound=BaseModel)
+_ResponseModel = TypeVar("_ResponseModel", bound=BaseModel)
+
+
+class EmptyRequest(BaseModel):
+    pass
+
+
+class KvGetRequest(BaseModel):
+    key: str = Field(..., min_length=1, max_length=512)
+
+
+class KvGetResponse(BaseModel):
+    value: str | None = None
+
+
+class KvSetRequest(BaseModel):
+    key: str = Field(..., min_length=1, max_length=512)
+    value: str = Field(..., max_length=65536)
+
+
+class KvSetResponse(BaseModel):
+    ok: bool = True
+
+
+class KvListRequest(BaseModel):
+    prefix: str = Field(..., min_length=1, max_length=512)
+
+
+class KvListResponse(BaseModel):
+    keys: list[str] = Field(default_factory=list)
+
+
+class CreateInvoiceRequest(BaseModel):
+    wallet_id: str = Field(..., min_length=1, max_length=128)
+    amount_sat: int = Field(..., gt=0)
+    memo: str = Field(..., max_length=512)
+    tag: str = Field(..., min_length=1, max_length=64)
+    extra: dict[str, str] = Field(default_factory=dict)
+
+
+class CreateInvoiceResponse(BaseModel):
+    payment_hash: str
+    payment_request: str
+    checking_id: str
+
+
+class WatchPaymentRequest(BaseModel):
+    payment_hash: str = Field(..., min_length=1, max_length=128)
+    callback_export: str = Field(..., min_length=1, max_length=128)
+
+
+class WatchPaymentResponse(BaseModel):
+    ok: bool = True
+
+
+class RandomIdRequest(BaseModel):
+    prefix: str = Field(..., min_length=1, max_length=32)
+
+
+class RandomIdResponse(BaseModel):
+    id: str
+
+
+class NowResponse(BaseModel):
+    timestamp: int
+
+
+class LogRequest(BaseModel):
+    level: Literal["debug", "info", "warning", "error"] = "info"
+    message: str = Field(..., min_length=1, max_length=2048)
+
+
+class LogResponse(BaseModel):
+    ok: bool = True
+
+
+@dataclass(frozen=True)
+class ExtensionAPIMethodExport:
+    method_id: str
+    namespace: str
+    name: str
+    host_name: str
+    sdk_name: str
+    description: str
+    required_permission: str | None = None
+    public_context: bool = False
+
+
+@dataclass(frozen=True)
+class ExtensionAPIMethod:
+    method_id: str
+    namespace: str
+    name: str
+    python_name: str
+    host_name: str
+    sdk_name: str
+    description: str
+    request_model: type[BaseModel]
+    response_model: type[BaseModel]
+    required_permission: str | None = None
+    public_context: bool = False
+
+    @property
+    def sdk_qualified_name(self) -> str:
+        return f"{self.namespace}.{self.sdk_name}"
+
+
+def extension_api_method(
+    *,
+    method_id: str,
+    namespace: str,
+    name: str,
+    host_name: str,
+    sdk_name: str,
+    description: str,
+    required_permission: str | None = None,
+    public_context: bool = False,
+) -> Callable[
+    [Callable[["ExtensionAPI", _RequestModel], Awaitable[_ResponseModel]]],
+    Callable[["ExtensionAPI", _RequestModel], Awaitable[_ResponseModel]],
+]:
+    export = ExtensionAPIMethodExport(
+        method_id=method_id,
+        namespace=namespace,
+        name=name,
+        host_name=host_name,
+        sdk_name=sdk_name,
+        description=description,
+        required_permission=required_permission,
+        public_context=public_context,
+    )
+
+    def decorator(
+        function: Callable[["ExtensionAPI", _RequestModel], Awaitable[_ResponseModel]],
+    ) -> Callable[["ExtensionAPI", _RequestModel], Awaitable[_ResponseModel]]:
+        @wraps(function)
+        async def wrapper(
+            self: "ExtensionAPI", request: _RequestModel
+        ) -> _ResponseModel:
+            self.require_permission(required_permission)
+            return await function(self, request)
+
+        setattr(wrapper, _EXTENSION_API_METHOD_ATTR, export)
+        return wrapper
+
+    return decorator
+
+
+class ExtensionAPI:
+    def __init__(
+        self,
+        extension_id: str,
+        permissions: Iterable[str],
+        *,
+        user_id: str | None = None,
+        wallet_id: str | None = None,
+    ) -> None:
+        self.extension_id = extension_id
+        self.permissions = set(permissions)
+        self.user_id = user_id
+        self.wallet_id = wallet_id
+
+    def require_permission(self, permission: str | None) -> None:
+        if permission and permission not in self.permissions:
+            raise PermissionError(
+                f"Extension '{self.extension_id}' is missing permission '{permission}'."
+            )
+
+    @extension_api_method(
+        method_id="storage.get",
+        namespace="storage",
+        name="Get storage value",
+        host_name="kv_get",
+        sdk_name="get",
+        description="Read one value from the extension storage namespace.",
+        required_permission="ext.storage.read_write",
+        public_context=True,
+    )
+    async def storage_get(self, request: KvGetRequest) -> KvGetResponse:
+        self._raise_unwired_runtime("storage_get")
+
+    @extension_api_method(
+        method_id="storage.set",
+        namespace="storage",
+        name="Set storage value",
+        host_name="kv_set",
+        sdk_name="set",
+        description="Write one value to the extension storage namespace.",
+        required_permission="ext.storage.read_write",
+    )
+    async def storage_set(self, request: KvSetRequest) -> KvSetResponse:
+        self._raise_unwired_runtime("storage_set")
+
+    @extension_api_method(
+        method_id="storage.list",
+        namespace="storage",
+        name="List storage keys",
+        host_name="kv_list",
+        sdk_name="list",
+        description="List keys under a prefix in the extension storage namespace.",
+        required_permission="ext.storage.read_write",
+        public_context=True,
+    )
+    async def storage_list(self, request: KvListRequest) -> KvListResponse:
+        self._raise_unwired_runtime("storage_list")
+
+    @extension_api_method(
+        method_id="wallet.create_invoice",
+        namespace="wallet",
+        name="Create invoice",
+        host_name="create_invoice",
+        sdk_name="createInvoice",
+        description="Create an incoming Lightning invoice for an allowed wallet.",
+        required_permission="wallet.create_invoice",
+        public_context=True,
+    )
+    async def wallet_create_invoice(
+        self, request: CreateInvoiceRequest
+    ) -> CreateInvoiceResponse:
+        self._raise_unwired_runtime("wallet_create_invoice")
+
+    @extension_api_method(
+        method_id="payments.watch",
+        namespace="payments",
+        name="Watch payment",
+        host_name="watch_payment",
+        sdk_name="watch",
+        description="Subscribe the extension to a payment state callback.",
+        required_permission="payments.watch",
+    )
+    async def payments_watch(self, request: WatchPaymentRequest) -> WatchPaymentResponse:
+        self._raise_unwired_runtime("payments_watch")
+
+    @extension_api_method(
+        method_id="system.random_id",
+        namespace="system",
+        name="Random ID",
+        host_name="random_id",
+        sdk_name="id",
+        description="Create a random extension-local identifier.",
+        public_context=True,
+    )
+    async def system_random_id(self, request: RandomIdRequest) -> RandomIdResponse:
+        return RandomIdResponse(
+            id=f"{request.prefix}_{secrets.token_urlsafe(12).replace('-', '_')}"
+        )
+
+    @extension_api_method(
+        method_id="system.now",
+        namespace="system",
+        name="Current timestamp",
+        host_name="now",
+        sdk_name="now",
+        description="Return the current Unix timestamp.",
+        public_context=True,
+    )
+    async def system_now(self, request: EmptyRequest) -> NowResponse:
+        return NowResponse(timestamp=int(time.time()))
+
+    @extension_api_method(
+        method_id="system.log",
+        namespace="system",
+        name="Log message",
+        host_name="log",
+        sdk_name="log",
+        description="Write a bounded message to the extension log.",
+        public_context=True,
+    )
+    async def system_log(self, request: LogRequest) -> LogResponse:
+        log = getattr(logger, request.level)
+        log("extension:%s %s", self.extension_id, request.message)
+        return LogResponse()
+
+    def _raise_unwired_runtime(self, method_name: str) -> None:
+        raise NotImplementedError(
+            f"ExtensionAPI.{method_name} must be wired to LNbits services before use."
+        )
+
+
+def list_extension_api_methods(
+    api_cls: type[ExtensionAPI] = ExtensionAPI,
+) -> list[ExtensionAPIMethod]:
+    methods: list[ExtensionAPIMethod] = []
+
+    for python_name, function in inspect.getmembers(api_cls, inspect.isfunction):
+        export = getattr(function, _EXTENSION_API_METHOD_ATTR, None)
+        if not export:
+            continue
+
+        request_model, response_model = _get_method_models(function)
+        methods.append(
+            ExtensionAPIMethod(
+                method_id=export.method_id,
+                namespace=export.namespace,
+                name=export.name,
+                python_name=python_name,
+                host_name=export.host_name,
+                sdk_name=export.sdk_name,
+                description=export.description,
+                request_model=request_model,
+                response_model=response_model,
+                required_permission=export.required_permission,
+                public_context=export.public_context,
+            )
+        )
+
+    return sorted(methods, key=lambda method: method.method_id)
+
+
+def get_extension_api_method(
+    method_id: str,
+    api_cls: type[ExtensionAPI] = ExtensionAPI,
+) -> ExtensionAPIMethod:
+    for method in list_extension_api_methods(api_cls):
+        if method.method_id == method_id:
+            return method
+    raise KeyError(f"Unknown extension API method '{method_id}'.")
+
+
+def extension_api_contract(
+    api_cls: type[ExtensionAPI] = ExtensionAPI,
+) -> dict[str, object]:
+    return {
+        "version": 1,
+        "methods": [
+            {
+                "id": method.method_id,
+                "namespace": method.namespace,
+                "name": method.name,
+                "python_name": method.python_name,
+                "host_name": method.host_name,
+                "sdk_name": method.sdk_name,
+                "sdk_qualified_name": method.sdk_qualified_name,
+                "description": method.description,
+                "required_permission": method.required_permission,
+                "public_context": method.public_context,
+                "request_schema": method.request_model.schema(
+                    ref_template="#/definitions/{model}"
+                ),
+                "response_schema": method.response_model.schema(
+                    ref_template="#/definitions/{model}"
+                ),
+            }
+            for method in list_extension_api_methods(api_cls)
+        ],
+    }
+
+
+def _get_method_models(
+    function: Callable[..., object],
+) -> tuple[type[BaseModel], type[BaseModel]]:
+    signature = inspect.signature(function)
+    request_parameters = [
+        parameter
+        for parameter in signature.parameters.values()
+        if parameter.name != "self"
+    ]
+    if len(request_parameters) != 1:
+        raise TypeError(
+            f"Extension API method '{function.__name__}' must accept one request model."
+        )
+
+    hints = get_type_hints(function)
+    request_model = hints.get(request_parameters[0].name)
+    response_model = hints.get("return")
+
+    if not _is_pydantic_model(request_model):
+        raise TypeError(
+            f"Extension API method '{function.__name__}' request must be a BaseModel."
+        )
+    if not _is_pydantic_model(response_model):
+        raise TypeError(
+            f"Extension API method '{function.__name__}' response must be a BaseModel."
+        )
+
+    return request_model, response_model
+
+
+def _is_pydantic_model(value: object) -> bool:
+    return isinstance(value, type) and issubclass(value, BaseModel)

lnbits/core/extensions/capabilities/__init__.py

@@ -1,8 +0,0 @@
-from .catalog import get_extension_capability_registry
-from .registry import CapabilityDefinition, CapabilityRegistry
-
-__all__ = [
-    "CapabilityDefinition",
-    "CapabilityRegistry",
-    "get_extension_capability_registry",
-]

lnbits/core/extensions/capabilities/catalog.py

@@ -1,120 +0,0 @@
-from .models import (
-    CreateInvoiceRequest,
-    CreateInvoiceResponse,
-    EmptyRequest,
-    KvGetRequest,
-    KvGetResponse,
-    KvListRequest,
-    KvListResponse,
-    KvSetRequest,
-    KvSetResponse,
-    LogRequest,
-    LogResponse,
-    NowResponse,
-    RandomIdRequest,
-    RandomIdResponse,
-    WatchPaymentRequest,
-    WatchPaymentResponse,
-)
-from .registry import CapabilityDefinition, CapabilityRegistry
-
-
-def get_extension_capability_registry() -> CapabilityRegistry:
-    registry = CapabilityRegistry()
-
-    for capability in [
-        CapabilityDefinition(
-            id="storage.get",
-            namespace="storage",
-            name="Get storage value",
-            host_name="kv_get",
-            sdk_name="get",
-            description="Read one value from the extension storage namespace.",
-            request_model=KvGetRequest,
-            response_model=KvGetResponse,
-            required_permission="ext.storage.read_write",
-            public_context=True,
-        ),
-        CapabilityDefinition(
-            id="storage.set",
-            namespace="storage",
-            name="Set storage value",
-            host_name="kv_set",
-            sdk_name="set",
-            description="Write one value to the extension storage namespace.",
-            request_model=KvSetRequest,
-            response_model=KvSetResponse,
-            required_permission="ext.storage.read_write",
-        ),
-        CapabilityDefinition(
-            id="storage.list",
-            namespace="storage",
-            name="List storage keys",
-            host_name="kv_list",
-            sdk_name="list",
-            description="List keys under a prefix in the extension storage namespace.",
-            request_model=KvListRequest,
-            response_model=KvListResponse,
-            required_permission="ext.storage.read_write",
-            public_context=True,
-        ),
-        CapabilityDefinition(
-            id="wallet.create_invoice",
-            namespace="wallet",
-            name="Create invoice",
-            host_name="create_invoice",
-            sdk_name="createInvoice",
-            description="Create an incoming Lightning invoice for an allowed wallet.",
-            request_model=CreateInvoiceRequest,
-            response_model=CreateInvoiceResponse,
-            required_permission="wallet.create_invoice",
-            public_context=True,
-        ),
-        CapabilityDefinition(
-            id="payments.watch",
-            namespace="payments",
-            name="Watch payment",
-            host_name="watch_payment",
-            sdk_name="watch",
-            description="Subscribe the extension to a payment state callback.",
-            request_model=WatchPaymentRequest,
-            response_model=WatchPaymentResponse,
-            required_permission="payments.watch",
-        ),
-        CapabilityDefinition(
-            id="system.random_id",
-            namespace="system",
-            name="Random ID",
-            host_name="random_id",
-            sdk_name="id",
-            description="Create a random extension-local identifier.",
-            request_model=RandomIdRequest,
-            response_model=RandomIdResponse,
-            public_context=True,
-        ),
-        CapabilityDefinition(
-            id="system.now",
-            namespace="system",
-            name="Current timestamp",
-            host_name="now",
-            sdk_name="now",
-            description="Return the current Unix timestamp.",
-            request_model=EmptyRequest,
-            response_model=NowResponse,
-            public_context=True,
-        ),
-        CapabilityDefinition(
-            id="system.log",
-            namespace="system",
-            name="Log message",
-            host_name="log",
-            sdk_name="log",
-            description="Write a bounded message to the extension log.",
-            request_model=LogRequest,
-            response_model=LogResponse,
-            public_context=True,
-        ),
-    ]:
-        registry.register(capability)
-
-    return registry

lnbits/core/extensions/capabilities/models.py

@@ -1,76 +0,0 @@
-from typing import Literal
-
-from pydantic import BaseModel, Field
-
-
-class EmptyRequest(BaseModel):
-    pass
-
-
-class KvGetRequest(BaseModel):
-    key: str = Field(..., min_length=1, max_length=512)
-
-
-class KvGetResponse(BaseModel):
-    value: str | None = None
-
-
-class KvSetRequest(BaseModel):
-    key: str = Field(..., min_length=1, max_length=512)
-    value: str = Field(..., max_length=65536)
-
-
-class KvSetResponse(BaseModel):
-    ok: bool = True
-
-
-class KvListRequest(BaseModel):
-    prefix: str = Field(..., min_length=1, max_length=512)
-
-
-class KvListResponse(BaseModel):
-    keys: list[str] = []
-
-
-class CreateInvoiceRequest(BaseModel):
-    wallet_id: str = Field(..., min_length=1, max_length=128)
-    amount_sat: int = Field(..., gt=0)
-    memo: str = Field(..., max_length=512)
-    tag: str = Field(..., min_length=1, max_length=64)
-    extra: dict[str, str] = Field(default_factory=dict)
-
-
-class CreateInvoiceResponse(BaseModel):
-    payment_hash: str
-    payment_request: str
-    checking_id: str
-
-
-class WatchPaymentRequest(BaseModel):
-    payment_hash: str = Field(..., min_length=1, max_length=128)
-    callback_export: str = Field(..., min_length=1, max_length=128)
-
-
-class WatchPaymentResponse(BaseModel):
-    ok: bool = True
-
-
-class RandomIdRequest(BaseModel):
-    prefix: str = Field(..., min_length=1, max_length=32)
-
-
-class RandomIdResponse(BaseModel):
-    id: str
-
-
-class NowResponse(BaseModel):
-    timestamp: int
-
-
-class LogRequest(BaseModel):
-    level: Literal["debug", "info", "warning", "error"] = "info"
-    message: str = Field(..., min_length=1, max_length=2048)
-
-
-class LogResponse(BaseModel):
-    ok: bool = True

lnbits/core/extensions/capabilities/registry.py

@@ -1,76 +0,0 @@
-from __future__ import annotations
-
-from dataclasses import dataclass
-from typing import Any
-
-from pydantic import BaseModel
-
-
-@dataclass(frozen=True)
-class CapabilityDefinition:
-    id: str
-    namespace: str
-    name: str
-    host_name: str
-    sdk_name: str
-    description: str
-    request_model: type[BaseModel]
-    response_model: type[BaseModel]
-    required_permission: str | None = None
-    public_context: bool = False
-
-    @property
-    def sdk_qualified_name(self) -> str:
-        return f"{self.namespace}.{self.sdk_name}"
-
-    def request_schema(self) -> dict[str, Any]:
-        return self.request_model.schema(ref_template="#/definitions/{model}")
-
-    def response_schema(self) -> dict[str, Any]:
-        return self.response_model.schema(ref_template="#/definitions/{model}")
-
-    def to_codegen_contract(self) -> dict[str, Any]:
-        return {
-            "id": self.id,
-            "namespace": self.namespace,
-            "name": self.name,
-            "host_name": self.host_name,
-            "sdk_name": self.sdk_name,
-            "sdk_qualified_name": self.sdk_qualified_name,
-            "description": self.description,
-            "required_permission": self.required_permission,
-            "public_context": self.public_context,
-            "request_schema": self.request_schema(),
-            "response_schema": self.response_schema(),
-        }
-
-
-class CapabilityRegistry:
-    def __init__(self) -> None:
-        self._capabilities: dict[str, CapabilityDefinition] = {}
-
-    def register(self, capability: CapabilityDefinition) -> None:
-        if capability.id in self._capabilities:
-            raise ValueError(f"Capability '{capability.id}' already registered.")
-        self._capabilities[capability.id] = capability
-
-    def get(self, capability_id: str) -> CapabilityDefinition | None:
-        return self._capabilities.get(capability_id)
-
-    def require(self, capability_id: str) -> CapabilityDefinition:
-        capability = self.get(capability_id)
-        if not capability:
-            raise KeyError(f"Unknown capability '{capability_id}'.")
-        return capability
-
-    def all(self) -> list[CapabilityDefinition]:
-        return sorted(self._capabilities.values(), key=lambda c: c.id)
-
-    def by_permission(self, permission_id: str) -> list[CapabilityDefinition]:
-        return [c for c in self.all() if c.required_permission == permission_id]
-
-    def to_codegen_contract(self) -> dict[str, Any]:
-        return {
-            "version": 1,
-            "capabilities": [c.to_codegen_contract() for c in self.all()],
-        }

lnbits/core/extensions/codegen/__init__.py

@@ -0,0 +1 @@
+"""SDK generators for extension API contracts."""

lnbits/core/extensions/codegen/typescript.py

@@ -0,0 +1,290 @@
+from __future__ import annotations
+
+import argparse
+from collections import defaultdict
+from collections.abc import Sequence
+from pathlib import Path
+from types import UnionType
+from typing import Any, Literal, Union, get_args, get_origin
+
+from pydantic import BaseModel
+
+from lnbits.core.extensions import (
+    ExtensionAPI,
+    ExtensionAPIMethod,
+    get_extension_api_method,
+    list_extension_api_methods,
+)
+
+
+def generate_typescript_sdk(
+    api_cls: type[ExtensionAPI] | None = None,
+    method_ids: Sequence[str] | None = None,
+) -> str:
+    api_cls = api_cls or ExtensionAPI
+    methods = _select_methods(api_cls, method_ids)
+    models = _collect_models(methods)
+
+    lines = [
+        "/* Generated by LNbits ExtensionAPI codegen. */",
+        "/* Do not edit by hand. */",
+        "",
+        "export type MaybePromise<T> = T | Promise<T>",
+        "",
+    ]
+
+    for model in models:
+        lines.extend(_render_model_type(model))
+        lines.append("")
+
+    lines.extend(_render_method_metadata(methods))
+    lines.append("")
+    lines.extend(_render_host_type(methods))
+    lines.append("")
+    lines.extend(_render_sdk_type(methods))
+    lines.append("")
+
+    lines.extend(_render_create_sdk(methods))
+
+    return "\n".join(lines).rstrip() + "\n"
+
+
+def write_typescript_sdk(
+    path: str | Path,
+    api_cls: type[ExtensionAPI] | None = None,
+    method_ids: Sequence[str] | None = None,
+) -> None:
+    Path(path).write_text(generate_typescript_sdk(api_cls, method_ids), encoding="utf-8")
+
+
+def _select_methods(
+    api_cls: type[ExtensionAPI], method_ids: Sequence[str] | None
+) -> list[ExtensionAPIMethod]:
+    if not method_ids:
+        return list_extension_api_methods(api_cls)
+    return [get_extension_api_method(method_id, api_cls) for method_id in method_ids]
+
+
+def _collect_models(methods: Sequence[ExtensionAPIMethod]) -> list[type[BaseModel]]:
+    models: dict[str, type[BaseModel]] = {}
+    for method in methods:
+        models[method.request_model.__name__] = method.request_model
+        models[method.response_model.__name__] = method.response_model
+    return [models[name] for name in sorted(models)]
+
+
+def _render_model_type(model: type[BaseModel]) -> list[str]:
+    name = _model_name(model)
+    fields = model.__fields__
+    if not fields:
+        return [f"export type {name} = Record<string, never>"]
+
+    lines = [f"export type {name} = {{"]
+    for field_name, field in fields.items():
+        optional = "?" if not field.required else ""
+        ts_type = _python_type_to_ts(field.outer_type_, field.allow_none)
+        lines.append(f"  {_camel(field_name)}{optional}: {ts_type}")
+    lines.append("}")
+    return lines
+
+
+def _python_type_to_ts(type_: Any, allow_none: bool = False) -> str:
+    origin = get_origin(type_)
+    args = get_args(type_)
+
+    if origin in (UnionType, Union):
+        ts = " | ".join(
+            _python_type_to_ts(arg) for arg in args if arg is not type(None)
+        )
+        if type(None) in args:
+            ts = f"{ts} | null"
+        return ts
+
+    if origin is Literal:
+        return " | ".join(_literal_to_ts(arg) for arg in args)
+
+    if origin in (list, Sequence):
+        item_type = _python_type_to_ts(args[0]) if args else "unknown"
+        ts = f"{item_type}[]"
+    elif origin is dict:
+        key_type = _python_type_to_ts(args[0]) if args else "string"
+        value_type = _python_type_to_ts(args[1]) if len(args) > 1 else "unknown"
+        ts = (
+            f"Record<{key_type}, {value_type}>"
+            if key_type == "string"
+            else f"{{ [key: string]: {value_type} }}"
+        )
+    elif _is_subclass(type_, str):
+        ts = "string"
+    elif _is_subclass(type_, bool):
+        ts = "boolean"
+    elif _is_subclass(type_, int) or _is_subclass(type_, float):
+        ts = "number"
+    else:
+        ts = "unknown"
+
+    return f"{ts} | null" if allow_none else ts
+
+
+def _literal_to_ts(value: Any) -> str:
+    if isinstance(value, str):
+        return f'"{value}"'
+    if isinstance(value, bool):
+        return "true" if value else "false"
+    if value is None:
+        return "null"
+    return str(value)
+
+
+def _is_subclass(type_: Any, class_: type) -> bool:
+    try:
+        return isinstance(type_, type) and issubclass(type_, class_)
+    except TypeError:
+        return False
+
+
+def _render_method_metadata(
+    methods: Sequence[ExtensionAPIMethod],
+) -> list[str]:
+    lines = ["export const extensionApiMethods = ["]
+    for method in methods:
+        permission = (
+            f'"{method.required_permission}"'
+            if method.required_permission
+            else "null"
+        )
+        lines.extend(
+            [
+                "  {",
+                f'    id: "{method.method_id}",',
+                f'    namespace: "{method.namespace}",',
+                f'    sdkName: "{method.sdk_name}",',
+                f'    pythonName: "{method.python_name}",',
+                f'    hostName: "{method.host_name}",',
+                f'    hostJsName: "{_camel(method.host_name)}",',
+                f"    requiredPermission: {permission},",
+                f"    publicContext: {_bool(method.public_context)},",
+                "  },",
+            ]
+        )
+    lines.append("] as const")
+    return lines
+
+
+def _render_host_type(methods: Sequence[ExtensionAPIMethod]) -> list[str]:
+    lines = ["export type ExtensionHost = {"]
+    for method in sorted(methods, key=lambda item: item.host_name):
+        request = _model_name(method.request_model)
+        response = _model_name(method.response_model)
+        if _is_empty_model(method.request_model):
+            lines.append(f"  {_camel(method.host_name)}(): MaybePromise<{response}>")
+        else:
+            lines.append(
+                f"  {_camel(method.host_name)}"
+                f"(input: {request}): MaybePromise<{response}>"
+            )
+    lines.append("}")
+    return lines
+
+
+def _render_sdk_type(methods: Sequence[ExtensionAPIMethod]) -> list[str]:
+    namespaces = _methods_by_namespace(methods)
+    lines = ["export type ExtensionSdk = {"]
+    for namespace, namespace_methods in namespaces.items():
+        lines.append(f"  {namespace}: {{")
+        for method in namespace_methods:
+            request = _model_name(method.request_model)
+            response = _model_name(method.response_model)
+            if _is_empty_model(method.request_model):
+                lines.append(f"    {method.sdk_name}(): Promise<{response}>")
+            else:
+                lines.append(
+                    f"    {method.sdk_name}(input: {request}): Promise<{response}>"
+                )
+        lines.append("  }")
+    lines.append("}")
+    return lines
+
+
+def _render_create_sdk(methods: Sequence[ExtensionAPIMethod]) -> list[str]:
+    namespaces = _methods_by_namespace(methods)
+    lines = [
+        "export function createExtensionSdk(",
+        "  host: ExtensionHost",
+        "): ExtensionSdk {",
+        "  return {",
+    ]
+    for namespace, namespace_methods in namespaces.items():
+        lines.append(f"    {namespace}: {{")
+        for method in namespace_methods:
+            host_name = _camel(method.host_name)
+            if _is_empty_model(method.request_model):
+                signature = f"{method.sdk_name}()"
+                host_call = f"host.{host_name}()"
+            else:
+                signature = f"{method.sdk_name}(input)"
+                host_call = f"host.{host_name}(input)"
+            lines.extend(
+                [
+                    f"      async {signature} {{",
+                    f"        return {host_call}",
+                    "      },",
+                ]
+            )
+        lines.append("    },")
+    lines.extend(["  }", "}"])
+    return lines
+
+
+def _methods_by_namespace(
+    methods: Sequence[ExtensionAPIMethod],
+) -> dict[str, list[ExtensionAPIMethod]]:
+    namespaces: dict[str, list[ExtensionAPIMethod]] = defaultdict(list)
+    for method in sorted(methods, key=lambda item: (item.namespace, item.sdk_name)):
+        namespaces[method.namespace].append(method)
+    return dict(sorted(namespaces.items()))
+
+
+def _model_name(model: type[BaseModel]) -> str:
+    return model.__name__
+
+
+def _camel(value: str) -> str:
+    head, *tail = value.split("_")
+    return head + "".join(part.capitalize() for part in tail)
+
+
+def _bool(value: bool) -> str:
+    return "true" if value else "false"
+
+
+def _is_empty_model(model: type[BaseModel]) -> bool:
+    return not model.__fields__
+
+
+def main(argv: Sequence[str] | None = None) -> int:
+    parser = argparse.ArgumentParser(
+        description="Generate a TypeScript SDK from the LNbits ExtensionAPI."
+    )
+    parser.add_argument(
+        "--method",
+        action="append",
+        dest="method_ids",
+        help="ExtensionAPI method id to include. Can be passed multiple times.",
+    )
+    parser.add_argument(
+        "--out",
+        help="Output file. If omitted, the generated SDK is printed to stdout.",
+    )
+    args = parser.parse_args(argv)
+
+    sdk = generate_typescript_sdk(method_ids=args.method_ids)
+    if args.out:
+        Path(args.out).write_text(sdk, encoding="utf-8")
+    else:
+        print(sdk, end="")
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())