dify/api/services/plugin/plugin_parameter_gateway.py

"""Resolve plugin providers and dispatch dynamic option requests to their package."""

from collections.abc import Mapping, Sequence
from typing import Any

from core.plugin.entities.parameters import PluginParameterOption
from core.plugin.impl.dynamic_select import DynamicSelectClient
from core.tools.plugin_tool.provider import PluginToolProviderController
from core.tools.tool_manager import ToolManager
from models.provider_ids import TriggerProviderID
from services.trigger.plugin_provider import PluginTriggerProvider
from services.trigger.provider_loader import PluginTriggerProviderLoader


class DaemonPluginParameterGateway:
    def __init__(self, *, providers: PluginTriggerProviderLoader) -> None:
        self._providers = providers

    def get_tool_provider(self, *, tenant_id: str, provider: str) -> PluginToolProviderController:
        return ToolManager.get_plugin_provider(provider=provider, tenant_id=tenant_id)

    def get_trigger_provider(self, *, tenant_id: str, provider_id: TriggerProviderID) -> PluginTriggerProvider:
        return self._providers.get_trigger_provider(tenant_id=tenant_id, provider_id=provider_id)

    def fetch_options(
        self,
        *,
        tenant_id: str,
        user_id: str,
        plugin_id: str,
        provider: str,
        action: str,
        parameter: str,
        credentials: Mapping[str, Any],
        credential_type: str,
        plugin_unique_identifier: str,
    ) -> Sequence[PluginParameterOption]:
        return (
            DynamicSelectClient()
            .fetch_dynamic_select_options(
                tenant_id=tenant_id,
                user_id=user_id,
                plugin_id=plugin_id,
                provider=provider,
                action=action,
                credentials=credentials,
                credential_type=credential_type,
                parameter=parameter,
                plugin_unique_identifier=plugin_unique_identifier,
            )
            .options
        )