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