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

from datetime import datetime
from typing import TYPE_CHECKING

from pydantic import BaseModel, ConfigDict, Field

from core_domain import (
    AgentDefinitionContract,
    AgentMemoryPolicyContract,
    AgentModelConfigContract,
    AgentRunContract,
    AgentRunStatus,
    AgentSkillRefContract,
    AgentStatus,
    AgentToolRefContract,
    AgentVersionContract,
    AgentVersionStatus,
)
from core_shared import JSONValue

if TYPE_CHECKING:
    from app.db.models import AgentDefinition, AgentRun, AgentVersion


class AgentCreateRequest(BaseModel):
    tenant_id: str
    code: str
    name: str
    description: str | None = None
    agent_type: str = "assistant"
    owner_user_id: str | None = None
    metadata_json: dict[str, JSONValue] = Field(default_factory=dict)


class AgentStatusUpdateRequest(BaseModel):
    tenant_id: str
    status: AgentStatus


class AgentResponse(AgentDefinitionContract):
    @classmethod
    def from_entity(cls, entity: "AgentDefinition") -> "AgentResponse":
        return cls.model_validate(entity, from_attributes=True)


class AgentVersionCreateRequest(BaseModel):
    model_config = ConfigDict(populate_by_name=True)

    tenant_id: str
    agent_id: str
    status: AgentVersionStatus = "draft"
    role: str = "assistant"
    goal: str | None = None
    system_prompt: str
    model_config_data: AgentModelConfigContract = Field(
        default_factory=AgentModelConfigContract,
        alias="model_config",
    )
    memory_policy: AgentMemoryPolicyContract = Field(default_factory=AgentMemoryPolicyContract)
    tool_refs: list[AgentToolRefContract] = Field(default_factory=list)
    skill_refs: list[AgentSkillRefContract] = Field(default_factory=list)


class AgentVersionResponse(AgentVersionContract):
    @classmethod
    def from_entity(cls, entity: "AgentVersion") -> "AgentVersionResponse":
        return cls.model_validate(entity, from_attributes=True)


class AgentRunCreateRequest(BaseModel):
    tenant_id: str
    agent_id: str
    agent_version_id: str | None = None
    session_id: str | None = None
    input_text: str | None = None
    input_json: dict[str, JSONValue] | None = None


class AgentRunStatusUpdateRequest(BaseModel):
    tenant_id: str
    status: AgentRunStatus
    worker_key: str | None = None
    output_text: str | None = None
    output_json: dict[str, JSONValue] | None = None
    error_code: str | None = None
    error_message: str | None = None


class AgentRunResponse(AgentRunContract):
    @classmethod
    def from_entity(cls, entity: "AgentRun") -> "AgentRunResponse":
        return cls.model_validate(entity, from_attributes=True)


class AgentHealthResponse(BaseModel):
    service: str
    status: str
    database: str
    checked_time: datetime