agent-platform/services/tool-service/app/application/services.py

from app.db.models import ToolBinding, ToolDefinition, ToolVersion
from app.domain.repositories import ToolBindingRepository, ToolDefinitionRepository, ToolVersionRepository
from app.schemas.tool import (
    ToolBindingCreateRequest,
    ToolCreateRequest,
    ToolVersionCreateRequest,
)


class ToolApplicationService:
    def __init__(
        self,
        tool_definition_repository: ToolDefinitionRepository,
        tool_version_repository: ToolVersionRepository,
        tool_binding_repository: ToolBindingRepository,
    ) -> None:
        self.tool_definition_repository = tool_definition_repository
        self.tool_version_repository = tool_version_repository
        self.tool_binding_repository = tool_binding_repository

    def create_tool_definition(self, payload: ToolCreateRequest) -> ToolDefinition:
        return self.tool_definition_repository.create(
            tenant_id=payload.tenant_id,
            plugin_id=payload.plugin_id,
            code=payload.code,
            name=payload.name,
            tool_type=payload.tool_type,
            description=payload.description,
        )

    def list_tool_definitions(self, tenant_id: str) -> list[ToolDefinition]:
        return self.tool_definition_repository.list_by_tenant(tenant_id)

    def create_tool_version(self, payload: ToolVersionCreateRequest) -> ToolVersion:
        return self.tool_version_repository.create(
            tenant_id=payload.tenant_id,
            tool_id=payload.tool_id,
            input_schema_json=payload.input_schema_json,
            output_schema_json=payload.output_schema_json,
            invoke_config_json=payload.invoke_config_json,
            timeout_ms=payload.timeout_ms,
            retry_policy_json=payload.retry_policy_json,
        )

    def list_tool_versions(self, tenant_id: str, tool_id: str) -> list[ToolVersion]:
        return self.tool_version_repository.list_by_tool(tenant_id=tenant_id, tool_id=tool_id)

    def create_tool_binding(self, payload: ToolBindingCreateRequest) -> ToolBinding:
        return self.tool_binding_repository.create(
            tenant_id=payload.tenant_id,
            app_id=payload.app_id,
            tool_version_id=payload.tool_version_id,
            credential_id=payload.credential_id,
            binding_scope=payload.binding_scope,
            enabled=payload.enabled,
            config_json=payload.config_json,
        )

    def list_tool_bindings(self, tenant_id: str, app_id: str | None = None) -> list[ToolBinding]:
        return self.tool_binding_repository.list_by_scope(tenant_id=tenant_id, app_id=app_id)

    def get_tool_binding_detail(
        self,
        *,
        tenant_id: str,
        binding_id: str,
    ) -> tuple[ToolBinding, ToolVersion, ToolDefinition] | None:
        binding = self.tool_binding_repository.get_by_id(tenant_id=tenant_id, binding_id=binding_id)
        if binding is None:
            return None

        tool_version = self.tool_version_repository.get_by_id(
            tenant_id=tenant_id,
            tool_version_id=binding.tool_version_id,
        )
        if tool_version is None:
            return None

        tool_definition = self.tool_definition_repository.get_by_id(
            tenant_id=tenant_id,
            tool_id=tool_version.tool_id,
        )
        if tool_definition is None:
            return None

        return binding, tool_version, tool_definition