dify/api/core/plugin/impl/endpoint.py

from typing import Any

from core.plugin.entities.endpoint import EndpointEntityWithInstance
from core.plugin.impl.base import BasePluginClient
from core.plugin.impl.exc import PluginDaemonInternalServerError


class PluginEndpointClient(BasePluginClient):
    def create_endpoint(
        self,
        tenant_id: str,
        user_id: str,
        plugin_unique_identifier: str,
        name: str,
        settings: dict[str, Any],
    ) -> bool:
        """
        Create an endpoint for the given plugin.

        Errors will be raised if any error occurs.
        """
        return self._request_with_plugin_daemon_response(
            "POST",
            f"plugin/{tenant_id}/endpoint/setup",
            bool,
            headers={
                "Content-Type": "application/json",
            },
            data={
                "user_id": user_id,
                "plugin_unique_identifier": plugin_unique_identifier,
                "settings": settings,
                "name": name,
            },
        )

    def list_endpoints(self, tenant_id: str, user_id: str, page: int, page_size: int):
        """
        List all endpoints for the given tenant.

        The daemon list route binds only tenant and pagination fields; user_id is
        retained in this client signature for consistency with endpoint services.
        """
        return self._request_with_plugin_daemon_response(
            "GET",
            f"plugin/{tenant_id}/endpoint/list",
            list[EndpointEntityWithInstance],
            params={"page": page, "page_size": page_size},
        )

    def list_endpoints_for_single_plugin(self, tenant_id: str, user_id: str, plugin_id: str, page: int, page_size: int):
        """
        List all endpoints for the given tenant and plugin.

        The daemon list route binds tenant, plugin and pagination fields; user_id
        is retained in this client signature for consistency with endpoint services.
        """
        return self._request_with_plugin_daemon_response(
            "GET",
            f"plugin/{tenant_id}/endpoint/list/plugin",
            list[EndpointEntityWithInstance],
            params={"plugin_id": plugin_id, "page": page, "page_size": page_size},
        )

    def update_endpoint(
        self, tenant_id: str, user_id: str, endpoint_id: str, name: str, settings: dict[str, Any]
    ) -> bool:
        """
        Update the settings of the given endpoint.
        """
        return self._request_with_plugin_daemon_response(
            "POST",
            f"plugin/{tenant_id}/endpoint/update",
            bool,
            data={
                "user_id": user_id,
                "endpoint_id": endpoint_id,
                "name": name,
                "settings": settings,
            },
            headers={
                "Content-Type": "application/json",
            },
        )

    def delete_endpoint(self, tenant_id: str, user_id: str, endpoint_id: str):
        """
        Delete the given endpoint.

        This operation is idempotent: if the endpoint is already deleted (record not found),
        it will return True instead of raising an error.
        """
        try:
            return self._request_with_plugin_daemon_response(
                "POST",
                f"plugin/{tenant_id}/endpoint/remove",
                bool,
                data={
                    "endpoint_id": endpoint_id,
                },
                headers={
                    "Content-Type": "application/json",
                },
            )
        except PluginDaemonInternalServerError as e:
            # Make delete idempotent: if record is not found, consider it a success
            if "record not found" in str(e.description).lower():
                return True
            raise

    def enable_endpoint(self, tenant_id: str, user_id: str, endpoint_id: str):
        """
        Enable the given endpoint.
        """
        return self._request_with_plugin_daemon_response(
            "POST",
            f"plugin/{tenant_id}/endpoint/enable",
            bool,
            data={
                "endpoint_id": endpoint_id,
            },
            headers={
                "Content-Type": "application/json",
            },
        )

    def disable_endpoint(self, tenant_id: str, user_id: str, endpoint_id: str):
        """
        Disable the given endpoint.
        """
        return self._request_with_plugin_daemon_response(
            "POST",
            f"plugin/{tenant_id}/endpoint/disable",
            bool,
            data={
                "endpoint_id": endpoint_id,
            },
            headers={
                "Content-Type": "application/json",
            },
        )