import json import logging import mimetypes import time from collections.abc import Generator, Mapping from os import listdir, path from threading import Lock from typing import TYPE_CHECKING, Any, Literal, Protocol, cast import sqlalchemy as sa from pydantic import TypeAdapter from sqlalchemy import select from sqlalchemy.orm import Session from typing_extensions import TypedDict from yarl import URL import contexts from configs import dify_config from core.entities import PluginCredentialType from core.helper.provider_cache import ToolProviderCredentialsCache from core.plugin.impl.tool import PluginToolManager from core.tools.__base.tool_provider import ToolProviderController from core.tools.__base.tool_runtime import ToolRuntime from core.tools.mcp_tool.provider import MCPToolProviderController from core.tools.mcp_tool.tool import MCPTool from core.tools.plugin_tool.provider import PluginToolProviderController from core.tools.plugin_tool.tool import PluginTool from core.tools.utils.uuid_utils import is_valid_uuid from core.tools.workflow_as_tool.provider import WorkflowToolProviderController from extensions.ext_database import db from graphon.runtime import VariablePool from models.provider_ids import ToolProviderID from services.tools.mcp_tools_manage_service import MCPToolManageService if TYPE_CHECKING: pass from core.agent.entities import AgentToolEntity from core.app.entities.app_invoke_entities import InvokeFrom from core.helper.module_import_helper import load_single_subclass_from_source from core.helper.position_helper import is_filtered from core.tools.__base.tool import Tool from core.tools.builtin_tool.provider import BuiltinToolProviderController from core.tools.builtin_tool.providers._positions import BuiltinToolProviderSort from core.tools.builtin_tool.tool import BuiltinTool from core.tools.custom_tool.provider import ApiToolProviderController from core.tools.custom_tool.tool import ApiTool from core.tools.entities.api_entities import ToolProviderApiEntity, ToolProviderTypeApiLiteral from core.tools.entities.common_entities import I18nObject from core.tools.entities.tool_entities import ( ApiProviderAuthType, EmojiIconDict, ToolInvokeFrom, ToolParameter, ToolProviderType, emoji_icon_adapter, ) from core.tools.errors import ToolProviderNotFoundError from core.tools.tool_label_manager import ToolLabelManager from core.tools.utils.configuration import ToolParameterConfigurationManager from core.tools.utils.encryption import create_provider_encrypter, create_tool_provider_encrypter from core.tools.workflow_as_tool.tool import WorkflowTool from graphon.model_runtime.utils.encoders import jsonable_encoder from models.tools import ApiToolProvider, BuiltinToolProvider, WorkflowToolProvider from services.tools.tools_transform_service import ToolTransformService if TYPE_CHECKING: pass logger = logging.getLogger(__name__) class ApiProviderControllerItem(TypedDict): provider: ApiToolProvider controller: ApiToolProviderController _credentials_adapter: TypeAdapter[dict[str, Any]] = TypeAdapter(dict[str, Any]) class WorkflowToolRuntimeSpec(Protocol): @property def provider_type(self) -> ToolProviderType: ... @property def provider_id(self) -> str: ... @property def tool_name(self) -> str: ... @property def tool_configurations(self) -> Mapping[str, Any]: ... @property def credential_id(self) -> str | None: ... class ToolManager: _builtin_provider_lock = Lock() _hardcoded_providers: dict[str, BuiltinToolProviderController] = {} _builtin_providers_loaded = False _builtin_tools_labels: dict[str, I18nObject | None] = {} @classmethod def get_hardcoded_provider(cls, provider: str) -> BuiltinToolProviderController: """ get the hardcoded provider """ if len(cls._hardcoded_providers) == 0: # init the builtin providers cls.load_hardcoded_providers_cache() return cls._hardcoded_providers[provider] @classmethod def get_builtin_provider( cls, provider: str, tenant_id: str ) -> BuiltinToolProviderController | PluginToolProviderController: """ get the builtin provider :param provider: the name of the provider :param tenant_id: the id of the tenant :return: the provider """ # split provider to if len(cls._hardcoded_providers) == 0: # init the builtin providers cls.load_hardcoded_providers_cache() if provider not in cls._hardcoded_providers: # get plugin provider plugin_provider = cls.get_plugin_provider(provider, tenant_id) if plugin_provider: return plugin_provider return cls._hardcoded_providers[provider] @classmethod def get_plugin_provider(cls, provider: str, tenant_id: str) -> PluginToolProviderController: """ get the plugin provider """ # check if context is set try: contexts.plugin_tool_providers.get() except LookupError: contexts.plugin_tool_providers.set({}) contexts.plugin_tool_providers_lock.set(Lock()) plugin_tool_providers = contexts.plugin_tool_providers.get() if provider in plugin_tool_providers: return plugin_tool_providers[provider] with contexts.plugin_tool_providers_lock.get(): # double check plugin_tool_providers = contexts.plugin_tool_providers.get() if provider in plugin_tool_providers: return plugin_tool_providers[provider] manager = PluginToolManager() provider_entity = manager.fetch_tool_provider(tenant_id, provider) if not provider_entity: raise ToolProviderNotFoundError(f"plugin provider {provider} not found") controller = PluginToolProviderController( entity=provider_entity.declaration, plugin_id=provider_entity.plugin_id, plugin_unique_identifier=provider_entity.plugin_unique_identifier, tenant_id=tenant_id, ) plugin_tool_providers[provider] = controller return controller @classmethod def get_tool_runtime( cls, provider_type: ToolProviderType, provider_id: str, tool_name: str, tenant_id: str, user_id: str | None = None, invoke_from: InvokeFrom = InvokeFrom.DEBUGGER, tool_invoke_from: ToolInvokeFrom = ToolInvokeFrom.AGENT, credential_id: str | None = None, ) -> BuiltinTool | PluginTool | ApiTool | WorkflowTool | MCPTool: """ get the tool runtime :param provider_type: the type of the provider :param provider_id: the id of the provider :param tool_name: the name of the tool :param tenant_id: the tenant id :param user_id: the caller id bound to runtime-scoped model/tool lookups :param invoke_from: invoke from :param tool_invoke_from: the tool invoke from :param credential_id: the credential id :return: the tool """ match provider_type: case ToolProviderType.BUILT_IN: provider_controller = cls.get_builtin_provider(provider_id, tenant_id) builtin_tool = provider_controller.get_tool(tool_name) if not builtin_tool: raise ToolProviderNotFoundError(f"builtin tool {tool_name} not found") if not provider_controller.need_credentials: return builtin_tool.fork_tool_runtime( runtime=ToolRuntime( tenant_id=tenant_id, user_id=user_id, credentials={}, invoke_from=invoke_from, tool_invoke_from=tool_invoke_from, ) ) builtin_provider = None if isinstance(provider_controller, PluginToolProviderController): provider_id_entity = ToolProviderID(provider_id) if is_valid_uuid(credential_id): try: builtin_provider_stmt = select(BuiltinToolProvider).where( BuiltinToolProvider.tenant_id == tenant_id, BuiltinToolProvider.id == credential_id, ) builtin_provider = db.session.scalar(builtin_provider_stmt) except Exception as e: builtin_provider = None logger.info("Error getting builtin provider %s:%s", credential_id, e, exc_info=True) if builtin_provider is None: raise ToolProviderNotFoundError(f"provider has been deleted: {credential_id}") if builtin_provider is None: with Session(db.engine) as session: builtin_provider = session.scalar( sa.select(BuiltinToolProvider) .where( BuiltinToolProvider.tenant_id == tenant_id, (BuiltinToolProvider.provider == str(provider_id_entity)) | (BuiltinToolProvider.provider == provider_id_entity.provider_name), ) .order_by(BuiltinToolProvider.is_default.desc(), BuiltinToolProvider.created_at.asc()) ) if builtin_provider is None: raise ToolProviderNotFoundError(f"no default provider for {provider_id}") else: builtin_provider = db.session.scalar( select(BuiltinToolProvider) .where( BuiltinToolProvider.tenant_id == tenant_id, (BuiltinToolProvider.provider == provider_id) ) .order_by(BuiltinToolProvider.is_default.desc(), BuiltinToolProvider.created_at.asc()) .limit(1) ) if builtin_provider is None: raise ToolProviderNotFoundError(f"builtin provider {provider_id} not found") from core.helper.credential_utils import runtime_check_credential_policy_compliance runtime_check_credential_policy_compliance( credential_id=builtin_provider.id, provider=provider_id, credential_type=PluginCredentialType.TOOL, check_existence=False, ) encrypter, cache = create_provider_encrypter( tenant_id=tenant_id, config=[ x.to_basic_provider_config() for x in provider_controller.get_credentials_schema_by_type(builtin_provider.credential_type) ], cache=ToolProviderCredentialsCache( tenant_id=tenant_id, provider=provider_id, credential_id=builtin_provider.id ), ) decrypted_credentials: Mapping[str, Any] = encrypter.decrypt(builtin_provider.credentials) if builtin_provider.expires_at != -1 and (builtin_provider.expires_at - 60) < int(time.time()): # TODO: circular import from core.plugin.impl.oauth import OAuthHandler from services.tools.builtin_tools_manage_service import BuiltinToolManageService tool_provider = ToolProviderID(provider_id) provider_name = tool_provider.provider_name redirect_uri = f"{dify_config.CONSOLE_API_URL}/console/api/oauth/plugin/{provider_id}/tool/callback" system_credentials = BuiltinToolManageService.get_oauth_client(tenant_id, provider_id) oauth_handler = OAuthHandler() refreshed_credentials = oauth_handler.refresh_credentials( tenant_id=tenant_id, user_id=builtin_provider.user_id, plugin_id=tool_provider.plugin_id, provider=provider_name, redirect_uri=redirect_uri, system_credentials=system_credentials or {}, credentials=decrypted_credentials, ) # update the credentials builtin_provider.encrypted_credentials = json.dumps( encrypter.encrypt(refreshed_credentials.credentials) ) builtin_provider.expires_at = refreshed_credentials.expires_at db.session.commit() decrypted_credentials = refreshed_credentials.credentials cache.delete() return builtin_tool.fork_tool_runtime( runtime=ToolRuntime( tenant_id=tenant_id, user_id=user_id, credentials=dict(decrypted_credentials), credential_type=builtin_provider.credential_type, runtime_parameters={}, invoke_from=invoke_from, tool_invoke_from=tool_invoke_from, ) ) case ToolProviderType.API: api_provider, credentials = cls.get_api_provider_controller(tenant_id, provider_id) encrypter, _ = create_tool_provider_encrypter( tenant_id=tenant_id, controller=api_provider, ) return api_provider.get_tool(tool_name).fork_tool_runtime( runtime=ToolRuntime( tenant_id=tenant_id, user_id=user_id, credentials=dict(encrypter.decrypt(credentials)), invoke_from=invoke_from, tool_invoke_from=tool_invoke_from, ) ) case ToolProviderType.WORKFLOW: workflow_provider_stmt = select(WorkflowToolProvider).where( WorkflowToolProvider.tenant_id == tenant_id, WorkflowToolProvider.id == provider_id ) with Session(db.engine, expire_on_commit=False) as session, session.begin(): workflow_provider = session.scalar(workflow_provider_stmt) if workflow_provider is None: raise ToolProviderNotFoundError(f"workflow provider {provider_id} not found") controller = ToolTransformService.workflow_provider_to_controller(db_provider=workflow_provider) controller_tools: list[WorkflowTool] = controller.get_tools(tenant_id=workflow_provider.tenant_id) if controller_tools is None or len(controller_tools) == 0: raise ToolProviderNotFoundError(f"workflow provider {provider_id} not found") return controller.get_tools(tenant_id=workflow_provider.tenant_id)[0].fork_tool_runtime( runtime=ToolRuntime( tenant_id=tenant_id, user_id=user_id, credentials={}, invoke_from=invoke_from, tool_invoke_from=tool_invoke_from, ) ) case ToolProviderType.APP: raise NotImplementedError("app provider not implemented") case ToolProviderType.PLUGIN: plugin_tool = cls.get_plugin_provider(provider_id, tenant_id).get_tool(tool_name) runtime = getattr(plugin_tool, "runtime", None) if runtime is not None: runtime.user_id = user_id runtime.invoke_from = invoke_from runtime.tool_invoke_from = tool_invoke_from return plugin_tool case ToolProviderType.MCP: mcp_tool = cls.get_mcp_provider_controller(tenant_id, provider_id).get_tool(tool_name) runtime = getattr(mcp_tool, "runtime", None) if runtime is not None: runtime.user_id = user_id runtime.invoke_from = invoke_from runtime.tool_invoke_from = tool_invoke_from return mcp_tool case ToolProviderType.DATASET_RETRIEVAL: raise ToolProviderNotFoundError(f"provider type {provider_type.value} not found") case _: raise ToolProviderNotFoundError(f"provider type {provider_type} not found") @classmethod def get_agent_tool_runtime( cls, tenant_id: str, app_id: str, agent_tool: AgentToolEntity, user_id: str | None = None, invoke_from: InvokeFrom = InvokeFrom.DEBUGGER, variable_pool: "VariablePool | None" = None, ) -> Tool: """ get the agent tool runtime """ tool_entity = cls.get_tool_runtime( provider_type=agent_tool.provider_type, provider_id=agent_tool.provider_id, tool_name=agent_tool.tool_name, tenant_id=tenant_id, user_id=user_id, invoke_from=invoke_from, tool_invoke_from=ToolInvokeFrom.AGENT, credential_id=agent_tool.credential_id, ) runtime_parameters = {} parameters = tool_entity.get_merged_runtime_parameters() runtime_parameters = cls._convert_tool_parameters_type( parameters, variable_pool, agent_tool.tool_parameters, typ="agent" ) # decrypt runtime parameters encryption_manager = ToolParameterConfigurationManager( tenant_id=tenant_id, tool_runtime=tool_entity, provider_name=agent_tool.provider_id, provider_type=agent_tool.provider_type, identity_id=f"AGENT.{app_id}", ) runtime_parameters = encryption_manager.decrypt_tool_parameters(runtime_parameters) if tool_entity.runtime is None or tool_entity.runtime.runtime_parameters is None: raise ValueError("runtime not found or runtime parameters not found") tool_entity.runtime.runtime_parameters.update(runtime_parameters) return tool_entity @classmethod def get_workflow_tool_runtime( cls, tenant_id: str, app_id: str, node_id: str, workflow_tool: WorkflowToolRuntimeSpec, user_id: str | None = None, invoke_from: InvokeFrom = InvokeFrom.DEBUGGER, variable_pool: "VariablePool | None" = None, ) -> Tool: """ get the workflow tool runtime """ tool_runtime = cls.get_tool_runtime( provider_type=workflow_tool.provider_type, provider_id=workflow_tool.provider_id, tool_name=workflow_tool.tool_name, tenant_id=tenant_id, user_id=user_id, invoke_from=invoke_from, tool_invoke_from=ToolInvokeFrom.WORKFLOW, credential_id=workflow_tool.credential_id, ) parameters = tool_runtime.get_merged_runtime_parameters() runtime_parameters = cls._convert_tool_parameters_type( parameters, variable_pool, workflow_tool.tool_configurations, typ="workflow" ) # decrypt runtime parameters encryption_manager = ToolParameterConfigurationManager( tenant_id=tenant_id, tool_runtime=tool_runtime, provider_name=workflow_tool.provider_id, provider_type=workflow_tool.provider_type, identity_id=f"WORKFLOW.{app_id}.{node_id}", ) if runtime_parameters: runtime_parameters = encryption_manager.decrypt_tool_parameters(runtime_parameters) tool_runtime.runtime.runtime_parameters.update(runtime_parameters) return tool_runtime @classmethod def get_tool_runtime_from_plugin( cls, tool_type: ToolProviderType, tenant_id: str, provider: str, tool_name: str, tool_parameters: dict[str, Any], user_id: str | None = None, credential_id: str | None = None, ) -> Tool: """ get tool runtime from plugin """ tool_entity = cls.get_tool_runtime( provider_type=tool_type, provider_id=provider, tool_name=tool_name, tenant_id=tenant_id, user_id=user_id, invoke_from=InvokeFrom.SERVICE_API, tool_invoke_from=ToolInvokeFrom.PLUGIN, credential_id=credential_id, ) runtime_parameters = {} parameters = tool_entity.get_merged_runtime_parameters() for parameter in parameters: if parameter.form == ToolParameter.ToolParameterForm.FORM: # save tool parameter to tool entity memory value = parameter.init_frontend_parameter(tool_parameters.get(parameter.name)) runtime_parameters[parameter.name] = value tool_entity.runtime.runtime_parameters.update(runtime_parameters) return tool_entity @classmethod def get_hardcoded_provider_icon(cls, provider: str) -> tuple[str, str]: """ get the absolute path of the icon of the hardcoded provider :param provider: the name of the provider :return: the absolute path of the icon, the mime type of the icon """ # get provider provider_controller = cls.get_hardcoded_provider(provider) absolute_path = path.join( path.dirname(path.realpath(__file__)), "builtin_tool", "providers", provider, "_assets", provider_controller.entity.identity.icon, ) # check if the icon exists if not path.exists(absolute_path): raise ToolProviderNotFoundError(f"builtin provider {provider} icon not found") # get the mime type mime_type, _ = mimetypes.guess_type(absolute_path) mime_type = mime_type or "application/octet-stream" return absolute_path, mime_type @classmethod def list_hardcoded_providers(cls): # use cache first if cls._builtin_providers_loaded: yield from list(cls._hardcoded_providers.values()) return with cls._builtin_provider_lock: if cls._builtin_providers_loaded: yield from list(cls._hardcoded_providers.values()) return yield from cls._list_hardcoded_providers() @classmethod def list_plugin_providers(cls, tenant_id: str) -> list[PluginToolProviderController]: """ list all the plugin providers """ manager = PluginToolManager() provider_entities = manager.fetch_tool_providers(tenant_id) return [ PluginToolProviderController( entity=provider.declaration, plugin_id=provider.plugin_id, plugin_unique_identifier=provider.plugin_unique_identifier, tenant_id=tenant_id, ) for provider in provider_entities ] @classmethod def list_builtin_providers( cls, tenant_id: str ) -> Generator[BuiltinToolProviderController | PluginToolProviderController, None, None]: """ list all the builtin providers """ yield from cls.list_hardcoded_providers() # get plugin providers yield from cls.list_plugin_providers(tenant_id) @classmethod def _list_hardcoded_providers(cls) -> Generator[BuiltinToolProviderController, None, None]: """ list all the builtin providers """ for provider_path in listdir(path.join(path.dirname(path.realpath(__file__)), "builtin_tool", "providers")): if provider_path.startswith("__"): continue if path.isdir(path.join(path.dirname(path.realpath(__file__)), "builtin_tool", "providers", provider_path)): if provider_path.startswith("__"): continue # init provider try: provider_class = load_single_subclass_from_source( module_name=f"core.tools.builtin_tool.providers.{provider_path}.{provider_path}", script_path=path.join( path.dirname(path.realpath(__file__)), "builtin_tool", "providers", provider_path, f"{provider_path}.py", ), parent_type=BuiltinToolProviderController, ) provider: BuiltinToolProviderController = provider_class() cls._hardcoded_providers[provider.entity.identity.name] = provider for tool in provider.get_tools(): cls._builtin_tools_labels[tool.entity.identity.name] = tool.entity.identity.label yield provider except Exception: logger.exception("load builtin provider %s", provider_path) continue # set builtin providers loaded cls._builtin_providers_loaded = True @classmethod def load_hardcoded_providers_cache(cls): for _ in cls.list_hardcoded_providers(): pass @classmethod def clear_hardcoded_providers_cache(cls): cls._hardcoded_providers = {} cls._builtin_providers_loaded = False @classmethod def get_tool_label(cls, tool_name: str) -> I18nObject | None: """ get the tool label :param tool_name: the name of the tool :return: the label of the tool """ if len(cls._builtin_tools_labels) == 0: # init the builtin providers cls.load_hardcoded_providers_cache() if tool_name not in cls._builtin_tools_labels: return None return cls._builtin_tools_labels[tool_name] @classmethod def list_default_builtin_providers(cls, tenant_id: str) -> list[BuiltinToolProvider]: """ list all the builtin providers """ # according to multi credentials, select the one with is_default=True first, then created_at oldest # for compatibility with old version if dify_config.SQLALCHEMY_DATABASE_URI_SCHEME == "postgresql": # PostgreSQL: Use DISTINCT ON sql = """ SELECT DISTINCT ON (tenant_id, provider) id FROM tool_builtin_providers WHERE tenant_id = :tenant_id ORDER BY tenant_id, provider, is_default DESC, created_at DESC """ else: # MySQL: Use window function to achieve same result sql = """ SELECT id FROM ( SELECT id, ROW_NUMBER() OVER ( PARTITION BY tenant_id, provider ORDER BY is_default DESC, created_at DESC ) as rn FROM tool_builtin_providers WHERE tenant_id = :tenant_id ) ranked WHERE rn = 1 """ with Session(db.engine, autoflush=False) as session: ids = [row.id for row in session.execute(sa.text(sql), {"tenant_id": tenant_id}).all()] return list(session.scalars(select(BuiltinToolProvider).where(BuiltinToolProvider.id.in_(ids)))) @classmethod def list_providers_from_api( cls, user_id: str, tenant_id: str, typ: ToolProviderTypeApiLiteral ) -> list[ToolProviderApiEntity]: result_providers: dict[str, ToolProviderApiEntity] = {} filters = [] if not typ: filters.extend(["builtin", "api", "workflow", "mcp"]) else: filters.append(typ) # Use a single session for all database operations to reduce connection overhead with Session(db.engine) as session: if "builtin" in filters: builtin_providers = list(cls.list_builtin_providers(tenant_id)) # key: provider name, value: provider db_builtin_providers = { str(ToolProviderID(provider.provider)): provider for provider in cls.list_default_builtin_providers(tenant_id) } # append builtin providers for provider in builtin_providers: # handle include, exclude if is_filtered( include_set=dify_config.POSITION_TOOL_INCLUDES_SET, exclude_set=dify_config.POSITION_TOOL_EXCLUDES_SET, data=provider, name_func=lambda x: x.entity.identity.name, ): continue user_provider = ToolTransformService.builtin_provider_to_user_provider( provider_controller=provider, db_provider=db_builtin_providers.get(provider.entity.identity.name), decrypt_credentials=False, ) if isinstance(provider, PluginToolProviderController): result_providers[f"plugin_provider.{user_provider.name}"] = user_provider else: result_providers[f"builtin_provider.{user_provider.name}"] = user_provider # get db api providers if "api" in filters: db_api_providers = session.scalars( select(ApiToolProvider).where(ApiToolProvider.tenant_id == tenant_id) ).all() # Batch create controllers api_provider_controllers: list[ApiProviderControllerItem] = [] for api_provider in db_api_providers: try: controller = ToolTransformService.api_provider_to_controller(api_provider) api_provider_controllers.append({"provider": api_provider, "controller": controller}) except Exception: # Skip invalid providers but continue processing others logger.warning("Failed to create controller for API provider %s", api_provider.id) # Batch get labels for all API providers if api_provider_controllers: controllers = cast( list[ToolProviderController], [item["controller"] for item in api_provider_controllers] ) labels = ToolLabelManager.get_tools_labels(controllers) for item in api_provider_controllers: provider_controller = item["controller"] db_provider = item["provider"] provider_labels = labels.get(provider_controller.provider_id, []) user_provider = ToolTransformService.api_provider_to_user_provider( provider_controller=provider_controller, db_provider=db_provider, decrypt_credentials=False, labels=provider_labels, ) result_providers[f"api_provider.{user_provider.name}"] = user_provider if "workflow" in filters: # get workflow providers workflow_providers = session.scalars( select(WorkflowToolProvider).where(WorkflowToolProvider.tenant_id == tenant_id) ).all() workflow_provider_controllers: list[WorkflowToolProviderController] = [] for workflow_provider in workflow_providers: try: workflow_controller: WorkflowToolProviderController = ( ToolTransformService.workflow_provider_to_controller(db_provider=workflow_provider) ) workflow_provider_controllers.append(workflow_controller) except Exception: # app has been deleted logger.exception("Failed to transform workflow provider %s to controller", workflow_provider.id) continue # Batch get labels for workflow providers if workflow_provider_controllers: workflow_controllers: list[ToolProviderController] = [ cast(ToolProviderController, controller) for controller in workflow_provider_controllers ] labels = ToolLabelManager.get_tools_labels(workflow_controllers) for workflow_provider_controller in workflow_provider_controllers: provider_labels = labels.get(workflow_provider_controller.provider_id, []) user_provider = ToolTransformService.workflow_provider_to_user_provider( provider_controller=workflow_provider_controller, labels=provider_labels, ) result_providers[f"workflow_provider.{user_provider.name}"] = user_provider if "mcp" in filters: mcp_service = MCPToolManageService(session=session) mcp_providers = mcp_service.list_providers(tenant_id=tenant_id, for_list=True) for mcp_provider in mcp_providers: result_providers[f"mcp_provider.{mcp_provider.name}"] = mcp_provider return BuiltinToolProviderSort.sort(list(result_providers.values())) @classmethod def get_api_provider_controller( cls, tenant_id: str, provider_id: str ) -> tuple[ApiToolProviderController, dict[str, Any]]: """ get the api provider :param tenant_id: the id of the tenant :param provider_id: the id of the provider :return: the provider controller, the credentials """ provider: ApiToolProvider | None = db.session.scalar( select(ApiToolProvider) .where( ApiToolProvider.id == provider_id, ApiToolProvider.tenant_id == tenant_id, ) .limit(1) ) if provider is None: raise ToolProviderNotFoundError(f"api provider {provider_id} not found") auth_type = ApiProviderAuthType.NONE provider_auth_type = provider.credentials.get("auth_type") if provider_auth_type in ("api_key_header", "api_key"): # backward compatibility auth_type = ApiProviderAuthType.API_KEY_HEADER elif provider_auth_type == "api_key_query": auth_type = ApiProviderAuthType.API_KEY_QUERY controller = ApiToolProviderController.from_db( provider, auth_type, ) controller.load_bundled_tools(provider.tools) return controller, provider.credentials @classmethod def get_mcp_provider_controller(cls, tenant_id: str, provider_id: str) -> MCPToolProviderController: """ get the api provider :param tenant_id: the id of the tenant :param provider_id: the id of the provider :return: the provider controller, the credentials """ with Session(db.engine) as session: mcp_service = MCPToolManageService(session=session) try: provider = mcp_service.get_provider(server_identifier=provider_id, tenant_id=tenant_id) except ValueError: raise ToolProviderNotFoundError(f"mcp provider {provider_id} not found") controller = MCPToolProviderController.from_db(provider) return controller @classmethod def user_get_api_provider(cls, provider: str, tenant_id: str, mask: bool = True): """ get api provider """ provider_name = provider provider_obj: ApiToolProvider | None = db.session.scalar( select(ApiToolProvider) .where( ApiToolProvider.tenant_id == tenant_id, ApiToolProvider.name == provider, ) .limit(1) ) if provider_obj is None: raise ValueError(f"you have not added provider {provider_name}") try: credentials = _credentials_adapter.validate_json(provider_obj.credentials_str) or {} except Exception: credentials = {} # package tool provider controller auth_type = ApiProviderAuthType.NONE credentials_auth_type = credentials.get("auth_type") if credentials_auth_type in ("api_key_header", "api_key"): # backward compatibility auth_type = ApiProviderAuthType.API_KEY_HEADER elif credentials_auth_type == "api_key_query": auth_type = ApiProviderAuthType.API_KEY_QUERY controller = ApiToolProviderController.from_db( provider_obj, auth_type, ) # init tool configuration encrypter, _ = create_tool_provider_encrypter( tenant_id=tenant_id, controller=controller, ) if mask: masked_credentials = encrypter.mask_plugin_credentials(encrypter.decrypt(credentials)) else: masked_credentials = encrypter.decrypt(credentials) try: icon = emoji_icon_adapter.validate_json(provider_obj.icon) except Exception: icon = {"background": "#252525", "content": "\ud83d\ude01"} # add tool labels labels = ToolLabelManager.get_tool_labels(controller) return cast( dict, jsonable_encoder( { "schema_type": provider_obj.schema_type, "schema": provider_obj.schema, "tools": provider_obj.tools, "icon": icon, "description": provider_obj.description, "credentials": masked_credentials, "privacy_policy": provider_obj.privacy_policy, "custom_disclaimer": provider_obj.custom_disclaimer, "labels": labels, } ), ) @classmethod def generate_builtin_tool_icon_url(cls, provider_id: str) -> str: return str( URL(dify_config.CONSOLE_API_URL or "/") / "console" / "api" / "workspaces" / "current" / "tool-provider" / "builtin" / provider_id / "icon" ) @classmethod def generate_plugin_tool_icon_url(cls, tenant_id: str, filename: str) -> str: return str( URL(dify_config.CONSOLE_API_URL or "/") / "console" / "api" / "workspaces" / "current" / "plugin" / "icon" % {"tenant_id": tenant_id, "filename": filename} ) @classmethod def generate_workflow_tool_icon_url(cls, tenant_id: str, provider_id: str) -> EmojiIconDict: try: workflow_provider: WorkflowToolProvider | None = db.session.scalar( select(WorkflowToolProvider) .where(WorkflowToolProvider.tenant_id == tenant_id, WorkflowToolProvider.id == provider_id) .limit(1) ) if workflow_provider is None: raise ToolProviderNotFoundError(f"workflow provider {provider_id} not found") icon = emoji_icon_adapter.validate_json(workflow_provider.icon) return icon except Exception: return {"background": "#252525", "content": "\ud83d\ude01"} @classmethod def generate_api_tool_icon_url(cls, tenant_id: str, provider_id: str) -> EmojiIconDict: try: api_provider: ApiToolProvider | None = db.session.scalar( select(ApiToolProvider) .where(ApiToolProvider.tenant_id == tenant_id, ApiToolProvider.id == provider_id) .limit(1) ) if api_provider is None: raise ToolProviderNotFoundError(f"api provider {provider_id} not found") icon = emoji_icon_adapter.validate_json(api_provider.icon) return icon except Exception: return {"background": "#252525", "content": "\ud83d\ude01"} @classmethod def generate_mcp_tool_icon_url(cls, tenant_id: str, provider_id: str) -> EmojiIconDict | str: try: with Session(db.engine) as session: mcp_service = MCPToolManageService(session=session) try: mcp_provider = mcp_service.get_provider_entity( provider_id=provider_id, tenant_id=tenant_id, by_server_id=True ) return cast(EmojiIconDict | str, mcp_provider.provider_icon) except ValueError: raise ToolProviderNotFoundError(f"mcp provider {provider_id} not found") except Exception: return {"background": "#252525", "content": "\ud83d\ude01"} @classmethod def get_tool_icon( cls, tenant_id: str, provider_type: ToolProviderType, provider_id: str, ) -> str | EmojiIconDict: """ get the tool icon :param tenant_id: the id of the tenant :param provider_type: the type of the provider :param provider_id: the id of the provider :return: """ match provider_type: case ToolProviderType.BUILT_IN: provider = ToolManager.get_builtin_provider(provider_id, tenant_id) if isinstance(provider, PluginToolProviderController): try: return cls.generate_plugin_tool_icon_url(tenant_id, provider.entity.identity.icon) except Exception: return {"background": "#252525", "content": "\ud83d\ude01"} return cls.generate_builtin_tool_icon_url(provider_id) case ToolProviderType.API: return cls.generate_api_tool_icon_url(tenant_id, provider_id) case ToolProviderType.WORKFLOW: return cls.generate_workflow_tool_icon_url(tenant_id, provider_id) case ToolProviderType.PLUGIN: provider = ToolManager.get_plugin_provider(provider_id, tenant_id) try: return cls.generate_plugin_tool_icon_url(tenant_id, provider.entity.identity.icon) except Exception: return {"background": "#252525", "content": "\ud83d\ude01"} case ToolProviderType.MCP: return cls.generate_mcp_tool_icon_url(tenant_id, provider_id) case ToolProviderType.APP | ToolProviderType.DATASET_RETRIEVAL: raise ValueError(f"provider type {provider_type} not found") case _: raise ValueError(f"provider type {provider_type} not found") @classmethod def _convert_tool_parameters_type( cls, parameters: list[ToolParameter], variable_pool: "VariablePool | None", tool_configurations: Mapping[str, Any], typ: Literal["agent", "workflow", "tool"] = "workflow", ) -> dict[str, Any]: """ Convert tool parameters type """ from graphon.nodes.tool.entities import ToolNodeData from graphon.nodes.tool.exc import ToolParameterError runtime_parameters = {} for parameter in parameters: if ( parameter.type in { ToolParameter.ToolParameterType.SYSTEM_FILES, ToolParameter.ToolParameterType.FILE, ToolParameter.ToolParameterType.FILES, } and parameter.required and typ == "agent" ): raise ValueError(f"file type parameter {parameter.name} not supported in agent") # save tool parameter to tool entity memory if parameter.form == ToolParameter.ToolParameterForm.FORM: if variable_pool: config = tool_configurations.get(parameter.name, {}) selector_value = cls._extract_runtime_selector_value(parameter, config) if selector_value is not None: # Selector parameters carry structured dictionaries, not scalar ToolInput values. runtime_parameters[parameter.name] = selector_value continue if not (config and isinstance(config, dict) and config.get("value") is not None): continue tool_input = ToolNodeData.ToolInput.model_validate(tool_configurations.get(parameter.name, {})) if tool_input.type == "variable": variable_selector = tool_input.value if not isinstance(variable_selector, list) or not all( isinstance(selector_part, str) for selector_part in variable_selector ): raise ToolParameterError("Variable tool input must be a variable selector") variable = variable_pool.get(variable_selector) if variable is None: raise ToolParameterError(f"Variable {tool_input.value} does not exist") parameter_value = variable.value elif tool_input.type == "constant": parameter_value = tool_input.value elif tool_input.type == "mixed": segment_group = variable_pool.convert_template(str(tool_input.value)) parameter_value = segment_group.text else: raise ToolParameterError(f"Unknown tool input type '{tool_input.type}'") runtime_parameters[parameter.name] = parameter_value else: value = parameter.init_frontend_parameter(tool_configurations.get(parameter.name)) runtime_parameters[parameter.name] = value return runtime_parameters @classmethod def _extract_runtime_selector_value(cls, parameter: ToolParameter, config: Any) -> dict[str, Any] | None: if parameter.type not in { ToolParameter.ToolParameterType.MODEL_SELECTOR, ToolParameter.ToolParameterType.APP_SELECTOR, }: return None if not isinstance(config, dict): return None input_value = config.get("value") if isinstance(input_value, dict) and cls._is_selector_value(parameter, input_value): return cast("dict[str, Any]", parameter.init_frontend_parameter(input_value)) if cls._is_selector_value(parameter, config): selector_value = dict(config) selector_value.pop("type", None) selector_value.pop("value", None) return cast("dict[str, Any]", parameter.init_frontend_parameter(selector_value)) return None @classmethod def _is_selector_value(cls, parameter: ToolParameter, value: Mapping[str, Any]) -> bool: if parameter.type == ToolParameter.ToolParameterType.MODEL_SELECTOR: return ( isinstance(value.get("provider"), str) and isinstance(value.get("model"), str) and isinstance(value.get("model_type"), str) ) if parameter.type == ToolParameter.ToolParameterType.APP_SELECTOR: return isinstance(value.get("app_id"), str) return False ToolManager.load_hardcoded_providers_cache()