feat: add support for custom REST handlers without UI in openapi.json

splunk_add_on_ucc_framework/commands/build.py

@@ -471,6 +471,7 @@ def generate(
                 "please unify them to build the add-on."
             )
             sys.exit(1)
+        global_config.parse_user_defined_handlers()
         scheme = global_config_builder_schema.GlobalConfigBuilderSchema(global_config)
         utils.recursive_overwrite(
             os.path.join(internal_root_dir, "package"),

splunk_add_on_ucc_framework/commands/openapi_generator/oas.py

@@ -70,7 +70,7 @@ class EncodingObject(Init):
 class MediaTypeObject(
     Init
 ):  # https://spec.openapis.org/oas/latest.html#media-type-object
-    schema: Optional[Union[SchemaObject, Dict[str, str]]] = None
+    schema: Optional[Union[SchemaObject, Dict[str, Any]]] = None
     example: Optional[Any] = None
     examples: Optional[Dict[str, ExampleObject]] = None
     encoding: Optional[Dict[str, EncodingObject]] = None
@@ -139,7 +139,7 @@ class OperationObject(
     description: Optional[str] = None
     externalDocs: Optional[ExternalDocumentationObject] = None
     operationId: Optional[str] = None
-    parameters: Optional[ParameterObject] = None
+    parameters: Optional[List[Union[ParameterObject, Dict[str, Any]]]] = None
     requestBody: Optional[RequestBodyObject] = None
     callbacks: Optional[CallbackObjects] = None
     deprecated: Optional[bool] = False

splunk_add_on_ucc_framework/commands/openapi_generator/ucc_to_oas.py

@@ -408,6 +408,20 @@ def __add_paths(
     return open_api_object
 
 
+def __add_user_defined_paths(
+    open_api_object: OpenAPIObject,
+    global_config: global_config_lib.GlobalConfig,
+) -> OpenAPIObject:
+    """
+    Adds user defined paths (globalConfig["options"]["restHandlers"]) to the OpenAPI object.
+    """
+    if open_api_object.paths is None:
+        open_api_object.paths = {}
+
+    open_api_object.paths.update(global_config.user_defined_handlers.oas_paths)
+    return open_api_object
+
+
 def transform(
     global_config: global_config_lib.GlobalConfig,
     app_manifest: app_manifest_lib.AppManifest,
@@ -427,4 +441,5 @@ def transform(
     open_api_object.security = [{"BasicAuth": []}]
     open_api_object = __add_schemas_object(open_api_object, global_config_dot_notation)
     open_api_object = __add_paths(open_api_object, global_config_dot_notation)
+    open_api_object = __add_user_defined_paths(open_api_object, global_config)
     return open_api_object

splunk_add_on_ucc_framework/commands/rest_builder/user_defined_rest_handlers.py

@@ -0,0 +1,314 @@
+#
+# Copyright 2025 Splunk Inc.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+from copy import deepcopy
+from dataclasses import dataclass
+from typing import Dict, Any, Union, Iterable, Optional, Set, List
+
+from splunk_add_on_ucc_framework.commands.openapi_generator import oas
+
+
+_EAI_OUTPUT_MODE = {
+    "name": "output_mode",
+    "in": "query",
+    "required": True,
+    "description": "Output mode",
+    "schema": {
+        "type": "string",
+        "enum": ["json"],
+        "default": "json",
+    },
+}
+EAI_DEFAULT_PARAMETERS = [_EAI_OUTPUT_MODE]
+EAI_DEFAULT_PARAMETERS_SPECIFIED = [
+    _EAI_OUTPUT_MODE,
+    {
+        "name": "name",
+        "in": "path",
+        "required": True,
+        "description": "The name of the item to operate on",
+        "schema": {"type": "string"},
+    },
+]
+
+
+def _eai_response_schema(schema: Any) -> oas.MediaTypeObject:
+    return oas.MediaTypeObject(
+        schema={
+            "type": "object",
+            "properties": {
+                "entry": {
+                    "type": "array",
+                    "items": {
+                        "type": "object",
+                        "properties": {
+                            "name": {"type": "string"},
+                            "content": schema,
+                        },
+                    },
+                }
+            },
+        }
+    )
+
+
+@dataclass
+class RestHandlerConfig:
+    """
+    Represents a REST handler configuration. See schema.json.
+    """
+
+    name: str
+    endpoint: str
+    handlerType: str
+    registerHandler: Optional[Dict[str, Any]] = None
+    requestParameters: Optional[Dict[str, Dict[str, Any]]] = None
+    responseParameters: Optional[Dict[str, Dict[str, Any]]] = None
+
+    @property
+    def request_parameters(self) -> Dict[str, Dict[str, Any]]:
+        return self.requestParameters or {}
+
+    @property
+    def response_parameters(self) -> Dict[str, Dict[str, Any]]:
+        return self.responseParameters or {}
+
+    @property
+    def supported_actions(self) -> Set[str]:
+        actions = set((self.registerHandler or {}).get("actions", []))
+        actions.update(self.request_parameters.keys())
+        actions.update(self.response_parameters.keys())
+        return actions
+
+    def _eai_params_to_schema_object(
+        self, params: Optional[Dict[str, Any]]
+    ) -> Optional[Dict[str, Any]]:
+        if not params:
+            return None
+
+        obj: Dict[str, Any] = {
+            "type": "object",
+            "properties": {},
+        }
+
+        required = []
+
+        for name, param in params.items():
+            obj["properties"][name] = param["schema"]
+
+            if param.get("required", False):
+                required.append(name)
+
+        if required:
+            obj["required"] = required
+
+        return obj
+
+    def _oas_object_eai_list_or_remove(
+        self, description: str, action: str
+    ) -> Optional[oas.OperationObject]:
+        if action not in self.supported_actions:
+            return None
+
+        op_obj: Dict[str, Any] = {
+            "description": description,
+            "responses": {
+                "200": oas.ResponseObject(description=description),
+            },
+        }
+
+        if self.request_parameters.get(action):
+            op_obj["parameters"] = [
+                {
+                    "name": key,
+                    "in": "query",
+                    "required": item.get("required", False),
+                    "schema": item["schema"],
+                }
+                for key, item in self.request_parameters[action].items()
+            ]
+
+        if self.response_parameters.get(action):
+            op_obj["responses"]["200"].content = {
+                "application/json": _eai_response_schema(
+                    self._eai_params_to_schema_object(self.response_parameters[action])
+                )
+            }
+
+        return oas.OperationObject(**op_obj)
+
+    def _oas_object_eai_create_or_edit(
+        self, description: str, action: str
+    ) -> Optional[oas.OperationObject]:
+        if action not in self.supported_actions:
+            return None
+
+        request_parameters = deepcopy(self.request_parameters[action])
+
+        if action == "create":
+            request_parameters["name"] = {
+                "schema": {"type": "string"},
+                "required": True,
+            }
+
+        op_obj: Dict[str, Any] = {
+            "description": description,
+            "responses": {
+                "200": oas.ResponseObject(description=description),
+            },
+        }
+
+        if request_parameters:
+            op_obj["requestBody"] = oas.RequestBodyObject(
+                content={
+                    "application/x-www-form-urlencoded": {
+                        "schema": self._eai_params_to_schema_object(request_parameters),
+                    },
+                }
+            )
+
+        if self.response_parameters.get(action):
+            op_obj["responses"]["200"].content = {
+                "application/json": _eai_response_schema(
+                    self._eai_params_to_schema_object(self.response_parameters[action])
+                )
+            }
+
+        return oas.OperationObject(**op_obj)
+
+    def _oas_object_eai_list_all(self) -> Optional[oas.OperationObject]:
+        return self._oas_object_eai_list_or_remove(
+            f"Get list of items for {self.name}", "list"
+        )
+
+    def _oas_object_eai_list_one(self) -> Optional[oas.OperationObject]:
+        return self._oas_object_eai_list_or_remove(
+            f"Get {self.name} item details", "list"
+        )
+
+    def _oas_object_eai_create(self) -> Optional[oas.OperationObject]:
+        return self._oas_object_eai_create_or_edit(
+            f"Create item in {self.name}", "create"
+        )
+
+    def _oas_object_eai_edit(self) -> Optional[oas.OperationObject]:
+        return self._oas_object_eai_create_or_edit(f"Update {self.name} item", "edit")
+
+    def _oas_object_eai_remove(self) -> Optional[oas.OperationObject]:
+        return self._oas_object_eai_list_or_remove(f"Delete {self.name} item", "remove")
+
+    def _oas_objects_eai_normal(self) -> Dict[str, oas.PathItemObject]:
+        endpoint = self.endpoint.strip("/")
+
+        obj: Dict[str, Any] = {}
+        list_all = self._oas_object_eai_list_all()
+
+        if list_all:
+            obj["get"] = list_all
+
+        create = self._oas_object_eai_create()
+
+        if create:
+            obj["post"] = create
+
+        if obj:
+            obj["parameters"] = EAI_DEFAULT_PARAMETERS
+            return {f"/{endpoint}": oas.PathItemObject(**obj)}
+
+        return {}
+
+    def _oas_objects_eai_specified(self) -> Dict[str, oas.PathItemObject]:
+        endpoint = self.endpoint.strip("/")
+
+        obj_specified: Dict[str, Any] = {}
+
+        list_one = self._oas_object_eai_list_one()
+
+        if list_one:
+            obj_specified["get"] = list_one
+
+        edit = self._oas_object_eai_edit()
+
+        if edit:
+            obj_specified["post"] = edit
+
+        remove = self._oas_object_eai_remove()
+
+        if remove:
+            obj_specified["delete"] = remove
+
+        if obj_specified:
+            obj_specified["parameters"] = EAI_DEFAULT_PARAMETERS_SPECIFIED
+            return {f"/{endpoint}/{{name}}": oas.PathItemObject(**obj_specified)}
+
+        return {}
+
+    def _oas_objects_eai(self) -> Dict[str, oas.PathItemObject]:
+        obj_dict: Dict[str, oas.PathItemObject] = {}
+
+        obj_dict.update(self._oas_objects_eai_normal())
+        obj_dict.update(self._oas_objects_eai_specified())
+
+        return obj_dict
+
+    @property
+    def oas_paths(self) -> Dict[str, oas.PathItemObject]:
+        if self.handlerType == "EAI":
+            return self._oas_objects_eai()
+        else:
+            raise ValueError(f"Unsupported handler type: {self.handlerType}")
+
+
+class UserDefinedRestHandlers:
+    """
+    Represents a logic for dealing with user-defined REST handlers
+    """
+
+    def __init__(self) -> None:
+        self._definitions: List[RestHandlerConfig] = []
+        self._names: Set[str] = set()
+        self._endpoints: Set[str] = set()
+
+    def add_definitions(
+        self, definitions: Iterable[Union[Dict[str, Any], RestHandlerConfig]]
+    ) -> None:
+        for definition in definitions:
+            self.add_definition(definition)
+
+    def add_definition(
+        self, definition: Union[Dict[str, Any], RestHandlerConfig]
+    ) -> None:
+        if not isinstance(definition, RestHandlerConfig):
+            definition = RestHandlerConfig(**definition)
+
+        if definition.name in self._names:
+            raise ValueError(f"Duplicate REST handler name: {definition.name}")
+
+        if definition.endpoint in self._endpoints:
+            raise ValueError(f"Duplicate REST handler endpoint: {definition.endpoint}")
+
+        self._names.add(definition.name)
+        self._endpoints.add(definition.endpoint)
+
+        self._definitions.append(definition)
+
+    @property
+    def oas_paths(self) -> Dict[str, oas.PathItemObject]:
+        paths = {}
+
+        for definition in self._definitions:
+            paths.update(definition.oas_paths)
+
+        return paths

splunk_add_on_ucc_framework/global_config.py

@@ -21,6 +21,9 @@
 import yaml
 
 from splunk_add_on_ucc_framework import utils
+from splunk_add_on_ucc_framework.commands.rest_builder.user_defined_rest_handlers import (
+    UserDefinedRestHandlers,
+)
 from splunk_add_on_ucc_framework.entity import expand_entity
 from splunk_add_on_ucc_framework.tabs import resolve_tab, LoggingTab
 
@@ -76,6 +79,12 @@ def __init__(self, global_config_path: str) -> None:
             else json.loads(config_raw)
         )
         self._original_path = global_config_path
+        self.user_defined_handlers = UserDefinedRestHandlers()
+
+    def parse_user_defined_handlers(self) -> None:
+        """Parse user-defined REST handlers from globalConfig["options"]["restHandlers"]"""
+        rest_handlers = self._content.get("options", {}).get("restHandlers", [])
+        self.user_defined_handlers.add_definitions(rest_handlers)
 
     def dump(self, path: str) -> None:
         if self._is_global_config_yaml: