from collections.abc import Generator from typing import Any from pydantic import BaseModel from configs import dify_config # from core.plugin.entities.plugin import GenericProviderID, ToolProviderID from core.plugin.entities.plugin_daemon import CredentialType, PluginBasicBooleanResponse, PluginToolProviderEntity from core.plugin.impl.base import BasePluginClient from core.plugin.utils.chunk_merger import merge_blob_chunks from core.schemas.resolver import resolve_dify_schema_refs from core.tools.entities.tool_entities import ToolInvokeMessage, ToolParameter from models.provider_ids import GenericProviderID, ToolProviderID # The daemon's ``management/tools`` (ListTools) accepts ``page >= 1`` and # ``page_size <= 256`` and returns one tenant-filtered page with no total, so # the complete inventory is read page by page until a short or empty page. _TOOL_PROVIDER_PAGE_SIZE = 256 # 40 pages = 10,240 tool providers for one workspace -- far beyond any real # tenant. Reaching it means the daemon is not paging, so fail explicitly rather # than report a truncated inventory as complete. _MAX_TOOL_PROVIDER_PAGES = 40 class PluginToolManager(BasePluginClient): def fetch_tool_providers(self, tenant_id: str) -> list[PluginToolProviderEntity]: """ Fetch every tool provider installed for the given tenant. Pages through the daemon until a short or empty page. Providers are deduplicated on their stable ``plugin_id/provider`` id, within and across pages, because the daemon's page order is not guaranteed. A failed page fails the whole fetch; a page that adds nothing new (the daemon ignoring ``page``) or exceeding ``_MAX_TOOL_PROVIDER_PAGES`` raises ``ValueError`` -- a partial inventory is never returned as complete. """ providers: list[PluginToolProviderEntity] = [] seen: set[str] = set() for page in range(1, _MAX_TOOL_PROVIDER_PAGES + 1): page_providers = self._fetch_tool_provider_page(tenant_id, page) added = 0 for provider in page_providers: identity = provider.declaration.identity.name if identity in seen: continue seen.add(identity) providers.append(provider) added += 1 if page_providers and not added: raise ValueError(f"plugin daemon repeated tool provider page {page}; inventory is incomplete") if len(page_providers) < _TOOL_PROVIDER_PAGE_SIZE: return providers raise ValueError( f"plugin daemon returned more than {_MAX_TOOL_PROVIDER_PAGES} full tool provider pages; " "inventory is incomplete" ) def _fetch_tool_provider_page(self, tenant_id: str, page: int) -> list[PluginToolProviderEntity]: """Fetch one ``management/tools`` page with stable ``plugin_id/provider`` names.""" def transformer(json_response: dict[str, Any]): for provider in json_response.get("data", []): declaration = provider.get("declaration", {}) or {} provider_name = declaration.get("identity", {}).get("name") for tool in declaration.get("tools", []): tool["identity"]["provider"] = provider_name # resolve refs if tool.get("output_schema"): tool["output_schema"] = resolve_dify_schema_refs(tool["output_schema"]) return json_response response = self._request_with_plugin_daemon_response( "GET", f"plugin/{tenant_id}/management/tools", list[PluginToolProviderEntity], params={"page": page, "page_size": _TOOL_PROVIDER_PAGE_SIZE}, transformer=transformer, ) for provider in response: provider.declaration.identity.name = f"{provider.plugin_id}/{provider.declaration.identity.name}" # override the provider name for each tool to plugin_id/provider_name for tool in provider.declaration.tools: tool.identity.provider = provider.declaration.identity.name return response def fetch_tool_provider(self, tenant_id: str, provider: str) -> PluginToolProviderEntity: """ Fetch tool provider for the given tenant and plugin. """ tool_provider_id = ToolProviderID(provider) def transformer(json_response: dict[str, Any]): data = json_response.get("data") if data: for tool in data.get("declaration", {}).get("tools", []): tool["identity"]["provider"] = tool_provider_id.provider_name # resolve refs if tool.get("output_schema"): tool["output_schema"] = resolve_dify_schema_refs(tool["output_schema"]) return json_response response = self._request_with_plugin_daemon_response( "GET", f"plugin/{tenant_id}/management/tool", PluginToolProviderEntity, params={"provider": tool_provider_id.provider_name, "plugin_id": tool_provider_id.plugin_id}, transformer=transformer, ) response.declaration.identity.name = f"{response.plugin_id}/{response.declaration.identity.name}" # override the provider name for each tool to plugin_id/provider_name for tool in response.declaration.tools: tool.identity.provider = response.declaration.identity.name return response def invoke( self, tenant_id: str, user_id: str, tool_provider: str, tool_name: str, credentials: dict[str, Any], credential_type: CredentialType, tool_parameters: dict[str, Any], conversation_id: str | None = None, app_id: str | None = None, message_id: str | None = None, ) -> Generator[ToolInvokeMessage, None, None]: """ Invoke the tool with the given tenant, user, plugin, provider, name, credentials and parameters. """ tool_provider_id = GenericProviderID(tool_provider) response = self._request_with_plugin_daemon_response_stream( "POST", f"plugin/{tenant_id}/dispatch/tool/invoke", ToolInvokeMessage, data={ "user_id": user_id, "conversation_id": conversation_id, "app_id": app_id, "message_id": message_id, "data": { "provider": tool_provider_id.provider_name, "tool": tool_name, "credentials": credentials, "credential_type": credential_type, "tool_parameters": tool_parameters, }, }, headers={ "X-Plugin-ID": tool_provider_id.plugin_id, "Content-Type": "application/json", }, ) return merge_blob_chunks(response, max_file_size=dify_config.PLUGIN_MAX_FILE_SIZE) def validate_provider_credentials( self, tenant_id: str, user_id: str, provider: str, credentials: dict[str, Any] ) -> bool: """ validate the credentials of the provider """ tool_provider_id = GenericProviderID(provider) response = self._request_with_plugin_daemon_response_stream( "POST", f"plugin/{tenant_id}/dispatch/tool/validate_credentials", PluginBasicBooleanResponse, data={ "user_id": user_id, "data": { "provider": tool_provider_id.provider_name, "credentials": credentials, }, }, headers={ "X-Plugin-ID": tool_provider_id.plugin_id, "Content-Type": "application/json", }, ) for resp in response: return resp.result return False def validate_datasource_credentials( self, tenant_id: str, user_id: str, provider: str, credentials: dict[str, Any] ) -> bool: """ validate the credentials of the datasource """ tool_provider_id = GenericProviderID(provider) response = self._request_with_plugin_daemon_response_stream( "POST", f"plugin/{tenant_id}/dispatch/datasource/validate_credentials", PluginBasicBooleanResponse, data={ "user_id": user_id, "data": { "provider": tool_provider_id.provider_name, "credentials": credentials, }, }, headers={ "X-Plugin-ID": tool_provider_id.plugin_id, "Content-Type": "application/json", }, ) for resp in response: return resp.result return False def get_runtime_parameters( self, tenant_id: str, user_id: str, provider: str, credentials: dict[str, Any], tool: str, conversation_id: str | None = None, app_id: str | None = None, message_id: str | None = None, ) -> list[ToolParameter]: """ get the runtime parameters of the tool """ tool_provider_id = GenericProviderID(provider) class RuntimeParametersResponse(BaseModel): parameters: list[ToolParameter] response = self._request_with_plugin_daemon_response_stream( "POST", f"plugin/{tenant_id}/dispatch/tool/get_runtime_parameters", RuntimeParametersResponse, data={ "user_id": user_id, "conversation_id": conversation_id, "app_id": app_id, "message_id": message_id, "data": { "provider": tool_provider_id.provider_name, "tool": tool, "credentials": credentials, }, }, headers={ "X-Plugin-ID": tool_provider_id.plugin_id, "Content-Type": "application/json", }, ) for resp in response: return resp.parameters return []