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

from datetime import datetime
from typing import TYPE_CHECKING, Generic, TypeVar

from core_domain import (
    ToolBindingContract,
    ToolBindingDetailContract,
    ToolCredentialContract,
    ToolCredentialRevealContract,
    ToolDefinitionContract,
    ToolVersionContract,
)
from core_shared import JSONValue
from pydantic import BaseModel, Field

if TYPE_CHECKING:
    from app.db.models import ToolBinding, ToolCredential, ToolDefinition, ToolVersion

T = TypeVar("T")


class ApiErrorResponse(BaseModel):
    errorType: str
    message: str
    details: dict[str, JSONValue] = Field(default_factory=dict)


class ApiResponse(BaseModel, Generic[T]):
    success: bool = True
    data: T | None = None
    error: ApiErrorResponse | None = None
    requestId: str
    serverTime: datetime


class PageRequest(BaseModel):
    page: int = Field(default=1, ge=1)
    pageSize: int = Field(default=20, ge=1, le=200)
    keyword: str | None = None

    @property
    def offset(self) -> int:
        return (self.page - 1) * self.pageSize


class PageResult(BaseModel, Generic[T]):
    items: list[T]
    total: int
    page: int
    pageSize: int
    hasMore: bool

    @classmethod
    def from_items(
        cls,
        *,
        items: list[T],
        total: int,
        page: int,
        page_size: int) -> "PageResult[T]":
        return cls(
            items=items,
            total=total,
            page=page,
            pageSize=page_size,
            hasMore=page * page_size < total)


class ToolCreateRequest(BaseModel):
    plugin_id: str | None = None
    code: str | None = None
    name: str
    tool_type: str
    description: str | None = None


class ToolResponse(ToolDefinitionContract):

    @classmethod
    def from_entity(cls, entity: "ToolDefinition") -> "ToolResponse":
        return cls.model_validate(entity, from_attributes=True)


class ToolVersionCreateRequest(BaseModel):
    tool_id: str
    input_schema_json: dict[str, JSONValue] = Field(default_factory=dict)
    output_schema_json: dict[str, JSONValue] = Field(default_factory=dict)
    invoke_config_json: dict[str, JSONValue] = Field(default_factory=dict)
    timeout_ms: int | None = None
    retry_policy_json: dict[str, JSONValue] = Field(default_factory=dict)


class ToolVersionResponse(ToolVersionContract):

    @classmethod
    def from_entity(cls, entity: "ToolVersion") -> "ToolVersionResponse":
        return cls.model_validate(entity, from_attributes=True)


class ToolBindingCreateRequest(BaseModel):
    app_id: str
    tool_version_id: str
    credential_id: str | None = None
    binding_scope: str = "app"
    enabled: bool = True
    config_json: dict[str, JSONValue] = Field(default_factory=dict)


class ToolBindingResponse(ToolBindingContract):

    @classmethod
    def from_entity(cls, entity: "ToolBinding") -> "ToolBindingResponse":
        return cls.model_validate(entity, from_attributes=True)


class ToolBindingDetailResponse(ToolBindingDetailContract):
    binding: ToolBindingResponse
    tool_version: ToolVersionResponse
    tool_definition: ToolResponse


class ToolCredentialCreateRequest(BaseModel):
    name: str
    credential_type: str = "generic"
    secret_json: dict[str, JSONValue] = Field(default_factory=dict)
    metadata_json: dict[str, JSONValue] = Field(default_factory=dict)


class ToolCredentialResponse(ToolCredentialContract):

    @classmethod
    def from_entity(cls, entity: "ToolCredential") -> "ToolCredentialResponse":
        return cls.model_validate(entity, from_attributes=True)


class ToolCredentialRevealResponse(ToolCredentialRevealContract):
    credential: ToolCredentialResponse


class ToolDto(BaseModel):
    id: str
    pluginId: str | None = None
    name: str
    toolType: str
    description: str | None = None
    createdTime: datetime

    @classmethod
    def from_entity(cls, entity: "ToolDefinition") -> "ToolDto":
        return cls(
            id=entity.id,
            pluginId=entity.plugin_id,
            name=entity.name,
            toolType=entity.tool_type,
            description=entity.description,
            createdTime=entity.created_time)


class ToolCreateRequestDto(BaseModel):
    pluginId: str | None = None
    name: str
    toolType: str = "mcp"
    description: str | None = None


class ToolDeleteRequestDto(BaseModel):
    toolId: str


class ToolDetailRequestDto(BaseModel):
    toolId: str


class ToolUpdateRequestDto(BaseModel):
    toolId: str
    name: str | None = None
    toolType: str | None = None
    description: str | None = None
    pluginId: str | None = None


class ToolVersionDto(BaseModel):
    id: str
    toolId: str
    versionNo: int
    inputSchema: dict[str, JSONValue] | None = None
    outputSchema: dict[str, JSONValue] | None = None
    invokeConfig: dict[str, JSONValue] | None = None
    timeoutMs: int | None = None
    retryPolicy: dict[str, JSONValue] | None = None
    createdTime: datetime

    @classmethod
    def from_entity(cls, entity: "ToolVersion") -> "ToolVersionDto":
        return cls(
            id=entity.id,
            toolId=entity.tool_id,
            versionNo=entity.version_no,
            inputSchema=entity.input_schema_json,
            outputSchema=entity.output_schema_json,
            invokeConfig=entity.invoke_config_json,
            timeoutMs=entity.timeout_ms,
            retryPolicy=entity.retry_policy_json,
            createdTime=entity.created_time)


class ToolVersionListRequestDto(PageRequest):
    toolId: str | None = None


class ToolVersionCreateRequestDto(BaseModel):
    toolId: str
    inputSchema: dict[str, JSONValue] = Field(default_factory=dict)
    outputSchema: dict[str, JSONValue] = Field(default_factory=dict)
    invokeConfig: dict[str, JSONValue] = Field(default_factory=dict)
    timeoutMs: int | None = None
    retryPolicy: dict[str, JSONValue] = Field(default_factory=dict)


class ToolVersionDeleteRequestDto(BaseModel):
    connectionId: str


class ToolVersionDetailRequestDto(BaseModel):
    connectionId: str


class ToolVersionUpdateRequestDto(BaseModel):
    connectionId: str
    inputSchema: dict[str, JSONValue] | None = None
    outputSchema: dict[str, JSONValue] | None = None
    invokeConfig: dict[str, JSONValue] | None = None
    timeoutMs: int | None = None
    retryPolicy: dict[str, JSONValue] | None = None


class ToolBindingDto(BaseModel):
    id: str
    appId: str
    toolVersionId: str
    credentialId: str | None = None
    bindingScope: str
    configJson: dict[str, JSONValue] | None = None
    createdTime: datetime

    @classmethod
    def from_entity(cls, entity: "ToolBinding") -> "ToolBindingDto":
        return cls(
            id=entity.id,
            appId=entity.app_id,
            toolVersionId=entity.tool_version_id,
            credentialId=entity.credential_id,
            bindingScope=entity.binding_scope,
            configJson=entity.config_json,
            createdTime=entity.created_time)


class ToolBindingListRequestDto(PageRequest):
    appId: str | None = None


class ToolBindingCreateRequestDto(BaseModel):
    appId: str
    toolVersionId: str
    credentialId: str | None = None
    bindingScope: str = "app"
    configJson: dict[str, JSONValue] = Field(default_factory=dict)


class ToolBindingDeleteRequestDto(BaseModel):
    bindingId: str


class ToolBindingDetailRequestDto(BaseModel):
    bindingId: str


class ToolBindingUpdateRequestDto(BaseModel):
    bindingId: str
    credentialId: str | None = None
    bindingScope: str | None = None
    configJson: dict[str, JSONValue] | None = None


class ToolCredentialDto(BaseModel):
    id: str
    name: str
    credentialType: str
    secretFingerprint: str
    encryptionAlgorithm: str
    metadataJson: dict[str, JSONValue] = Field(default_factory=dict)
    createdTime: datetime

    @classmethod
    def from_entity(cls, entity: "ToolCredential") -> "ToolCredentialDto":
        return cls(
            id=entity.id,
            name=entity.name,
            credentialType=entity.credential_type,
            secretFingerprint=entity.secret_fingerprint,
            encryptionAlgorithm=entity.encryption_algorithm,
            metadataJson=entity.metadata_json,
            createdTime=entity.created_time)


class ToolCredentialCreateRequestDto(BaseModel):
    name: str
    credentialType: str = "generic"
    secretJson: dict[str, JSONValue] = Field(default_factory=dict)
    metadataJson: dict[str, JSONValue] = Field(default_factory=dict)


class ToolCredentialDeleteRequestDto(BaseModel):
    credentialId: str


class ToolCredentialDetailRequestDto(BaseModel):
    credentialId: str


class ToolCredentialUpdateRequestDto(BaseModel):
    credentialId: str
    name: str | None = None
    metadataJson: dict[str, JSONValue] | None = None


class ToolCredentialRevealRequestDto(BaseModel):
    credentialId: str


class ToolCredentialRevealDto(BaseModel):
    credential: ToolCredentialDto
    secretJson: dict[str, JSONValue] = Field(default_factory=dict)


class McpToolDto(BaseModel):
    name: str
    description: str | None = None
    inputSchema: dict[str, JSONValue] | None = None


class McpConnectRequestDto(BaseModel):
    name: str | None = None
    config: dict[str, JSONValue]


class McpConnectData(BaseModel):
    tool: ToolDto
    connection: ToolVersionDto
    discoveredTools: list[McpToolDto] = Field(default_factory=list)


class DeleteData(BaseModel):
    deleted: bool
    toolId: str | None = None
    connectionId: str | None = None
    bindingId: str | None = None
    credentialId: str | None = None