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

from core_shared import JSONValue
from core_shared.secrets import EncryptedSecret, SecretCipher

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


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

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

    def list_tool_definitions(self) -> list[ToolDefinition]:
        return self.tool_definition_repository.list_all()

    def create_tool_version(self, payload: ToolVersionCreateRequest) -> ToolVersion:
        return self.tool_version_repository.create(
            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, tool_id: str) -> list[ToolVersion]:
        return self.tool_version_repository.list_by_tool(tool_id=tool_id)

    def create_tool_binding(self, payload: ToolBindingCreateRequest) -> ToolBinding:
        if payload.credential_id is not None:
            credential = self.tool_credential_repository.get_by_id(
                credential_id=payload.credential_id)
            if credential is None:
                raise ValueError(f"tool credential not found: {payload.credential_id}")
        return self.tool_binding_repository.create(
            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, app_id: str | None = None) -> list[ToolBinding]:
        return self.tool_binding_repository.list_by_scope(app_id=app_id)

    def create_tool_credential(self, payload: ToolCredentialCreateRequest) -> ToolCredential:
        encrypted = self.secret_cipher.encrypt_json(payload.secret_json)
        return self.tool_credential_repository.create(
            name=payload.name,
            credential_type=payload.credential_type,
            encrypted_payload_text=encrypted.ciphertext,
            secret_fingerprint=encrypted.fingerprint,
            encryption_algorithm=encrypted.algorithm,
            metadata_json=payload.metadata_json)

    def list_tool_credentials(self) -> list[ToolCredential]:
        return self.tool_credential_repository.list_all()

    def reveal_tool_credential(
        self,
        *,
        credential_id: str) -> tuple[ToolCredential, dict[str, JSONValue]] | None:
        credential = self.tool_credential_repository.get_by_id(
            credential_id=credential_id)
        if credential is None:
            return None
        secret_json = self.secret_cipher.decrypt_json(
            EncryptedSecret(
                ciphertext=credential.encrypted_payload_text,
                fingerprint=credential.secret_fingerprint,
                algorithm=credential.encryption_algorithm)
        )
        return credential, secret_json

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

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

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

        return binding, tool_version, tool_definition