from __future__ import annotations from datetime import datetime from enum import Enum import re from pydantic import BaseModel, Field class ServiceName(str, Enum): PRODUCT = "product" ADMIN = "admin" class ToolLifecycleStatus(str, Enum): DRAFT = "draft" GENERATED = "generated" VALIDATED = "validated" APPROVED = "approved" ACTIVE = "active" FAILED = "failed" ARCHIVED = "archived" class ToolLifecycleStageContract(BaseModel): code: ToolLifecycleStatus label: str description: str order: int = Field(ge=1) terminal: bool = False TOOL_LIFECYCLE_STAGES: tuple[ToolLifecycleStageContract, ...] = ( ToolLifecycleStageContract( code=ToolLifecycleStatus.DRAFT, label="Draft", description="Estado inicial de uma tool ainda em definicao.", order=1, terminal=False, ), ToolLifecycleStageContract( code=ToolLifecycleStatus.GENERATED, label="Generated", description="Implementacao gerada e pronta para analise tecnica.", order=2, terminal=False, ), ToolLifecycleStageContract( code=ToolLifecycleStatus.VALIDATED, label="Validated", description="Tool validada automaticamente com verificacoes basicas.", order=3, terminal=False, ), ToolLifecycleStageContract( code=ToolLifecycleStatus.APPROVED, label="Approved", description="Versao revisada e aprovada para publicacao controlada.", order=4, terminal=False, ), ToolLifecycleStageContract( code=ToolLifecycleStatus.ACTIVE, label="Active", description="Tool publicada e apta a abastecer o runtime de produto.", order=5, terminal=False, ), ToolLifecycleStageContract( code=ToolLifecycleStatus.FAILED, label="Failed", description="Falha registrada na geracao, validacao ou ativacao.", order=6, terminal=True, ), ToolLifecycleStageContract( code=ToolLifecycleStatus.ARCHIVED, label="Archived", description="Versao retirada de circulacao e mantida apenas para historico.", order=7, terminal=True, ), ) TOOL_LIFECYCLE_STATUS_SEQUENCE: tuple[ToolLifecycleStatus, ...] = tuple( stage.code for stage in TOOL_LIFECYCLE_STAGES ) _TOOL_LIFECYCLE_STAGE_BY_STATUS = { stage.code: stage for stage in TOOL_LIFECYCLE_STAGES } def get_tool_lifecycle_stage( status: ToolLifecycleStatus | str, ) -> ToolLifecycleStageContract: normalized_status = ( status if isinstance(status, ToolLifecycleStatus) else ToolLifecycleStatus(str(status or "").strip().lower()) ) return _TOOL_LIFECYCLE_STAGE_BY_STATUS[normalized_status] GENERATED_TOOLS_PACKAGE = "generated_tools" GENERATED_TOOL_ENTRYPOINT = "run" _GENERATED_TOOL_NAME_PATTERN = re.compile(r"^[a-z][a-z0-9_]{2,63}$") def _normalize_generated_tool_name(tool_name: str) -> str: normalized = str(tool_name or "").strip().lower() if not _GENERATED_TOOL_NAME_PATTERN.match(normalized): raise ValueError("tool_name must use lowercase snake_case to build the generated module path.") return normalized def build_generated_tool_module_name(tool_name: str) -> str: normalized = _normalize_generated_tool_name(tool_name) return f"{GENERATED_TOOLS_PACKAGE}.{normalized}" def build_generated_tool_module_path(tool_name: str) -> str: normalized = _normalize_generated_tool_name(tool_name) return f"{GENERATED_TOOLS_PACKAGE}/{normalized}.py" class ToolParameterType(str, Enum): STRING = "string" INTEGER = "integer" NUMBER = "number" BOOLEAN = "boolean" OBJECT = "object" ARRAY = "array" class ToolParameterContract(BaseModel): name: str parameter_type: ToolParameterType description: str required: bool = True class PublishedToolContract(BaseModel): tool_name: str display_name: str description: str version: int = Field(ge=1) status: ToolLifecycleStatus parameters: tuple[ToolParameterContract, ...] = () implementation_module: str implementation_callable: str checksum: str | None = None published_at: datetime | None = None published_by: str | None = None class ToolPublicationEnvelope(BaseModel): source_service: ServiceName = ServiceName.ADMIN target_service: ServiceName = ServiceName.PRODUCT publication_id: str published_tool: PublishedToolContract emitted_at: datetime