dify/api/extensions/application_services/tools.py

"""Compose provider-scoped tool credential resolution for workflow execution."""

from collections.abc import Mapping

from configs import dify_config
from core.db.session_factory import get_session_maker
from core.plugin.entities.plugin_daemon import CredentialType
from core.tools.builtin_tool.provider import BuiltinToolProviderController
from core.tools.plugin_tool.provider import PluginToolProviderController
from extensions.ext_redis import redis_client
from models.provider_ids import ToolProviderID
from repositories.credentials.query_repository import CredentialQueryRepository
from services.credentials.redis_refresh_lock import RedisCredentialRefreshLock
from services.credentials.tool_runtime import ToolCredentialService
from services.credentials.tool_runtime_adapters import ProviderToolCredentialRuntime


def resolve_tool_credentials(
    *, controller: BuiltinToolProviderController, provider_id: str, workspace_id: str, credential_id: str | None
) -> tuple[Mapping[str, object], CredentialType, str]:
    if isinstance(controller, PluginToolProviderController):
        provider = ToolProviderID(provider_id)
        providers = (str(provider), provider.provider_name)
    else:
        providers = (provider_id,)
    runtime = ProviderToolCredentialRuntime(
        controller=controller, provider_id=provider_id, console_api_url=dify_config.CONSOLE_API_URL
    )
    service = ToolCredentialService(
        credentials=CredentialQueryRepository(session_factory=get_session_maker()),
        runtime=runtime,
        refresh_lock=RedisCredentialRefreshLock(client=redis_client),
        providers=providers,
    )
    return runtime.resolve(service=service, workspace_id=workspace_id, credential_id=credential_id)