from __future__ import annotations from typing import Any, Optional, List from datetime import datetime, timezone from pydantic import BaseModel, Field, ConfigDict # ---------- CloudEvent (Pydantic v2 model) ---------- class CloudEvent(BaseModel): specversion: str = "1.0" id: str type: str source: str time: datetime datacontenttype: str = "application/json" data: Optional[Any] = None @staticmethod def now_utc() -> datetime: return datetime.now(timezone.utc) @classmethod def wrap(cls, *, event_id: str, event_type: str, source: str, data: Any) -> "CloudEvent": return cls( id=event_id, type=event_type or ("NullOrEmpty" if data is None else type(data).__name__), source=source, time=cls.now_utc(), data=data, ) # ---------- Permissive ancillary types ---------- class FunctionState(BaseModel): IsFunctionCall: bool = False IsFunctionCallResponse: bool = False IsFunctionCallError: bool = False IsFunctionCallStatus: bool = False IsFunctionStillRunning: bool = False class FunctionCallData(BaseModel): # Accept arbitrary keys so we mirror your .NET payloads without schema fights model_config = ConfigDict(extra="allow") class UserInfo(BaseModel): model_config = ConfigDict(extra="allow") # ---------- LLMServiceObj (field names match C# exactly) ---------- class LLMServiceObj(BaseModel): # strings SessionId: str = "" JsonFunction: str = "" LlmMessage: str = "" ResultMessage: str = "" UserInput: str = "" RequestSessionId: str = "" FunctionName: str = "" TimeZone: str = "" LLMRunnerType: str = "TurboLLM" SourceLlm: str = "" DestinationLlm: str = "" MessageID: str = "" LlmSessionStartName: str = "" SwapFunctionName: str = "" ChatAgentLocation: str = "" ToolsDefinitionId: Optional[str] = None JsonToolsBuilderSpec: Optional[str] = None # ints / bools TokensUsed: int = 0 IsUserLoggedIn: bool = False IsFuncAck: bool = False IsProcessed: bool = False IsSystemLlm: bool = False Timeout: Optional[int] = None # complex FunctionCallId: str = "" FunctionCallData: FunctionCallData = Field(default_factory=FunctionCallData) UserInfo: UserInfo = Field(default_factory=UserInfo) StartTimeUTC: datetime = Field(default_factory=lambda: datetime.now(timezone.utc)) # stacks (arrays; C# Stack will deserialize fine) LlmStack: List[str] = Field(default_factory=list) FunctionCallIdStack: List[str] = Field(default_factory=list) FunctionNameStack: List[str] = Field(default_factory=list) IsProcessedStack: List[bool] = Field(default_factory=list) MessageIDStack: List[str] = Field(default_factory=list) # function state flags (since your C# exposes these via methods/flags) IsFunctionCall: bool = False IsFunctionCallResponse: bool = False IsFunctionCallError: bool = False IsFunctionCallStatus: bool = False IsFunctionStillRunning: bool = False # ---------- ResultObj ---------- class ResultObj(BaseModel): Message: str = "" Success: bool = False Data: Optional[Any] = None