feat(agentkit): add GenericAvatar and session-aware avatar validation

Adds the GenericAvatar vendor wrapper and extends avatar validation
helpers for generic and RTC-backed avatars. Session-derived fields such
as agora_appid, agora_channel, and agora_token can now be validated
after AgentSession enrichment.

Author: digitallysavvy

src/agora_agent/agentkit/avatar_types.py

@@ -17,7 +17,21 @@ def is_anam_avatar(config: typing.Dict[str, typing.Any]) -> bool:
     return config.get("vendor") == "anam"
 
 
-def validate_avatar_config(config: typing.Dict[str, typing.Any]) -> None:
+def is_generic_avatar(config: typing.Dict[str, typing.Any]) -> bool:
+    return config.get("vendor") == "generic"
+
+
+def is_rtc_avatar(config: typing.Dict[str, typing.Any]) -> bool:
+    params = config.get("params", {})
+    return isinstance(params, dict) and bool(params.get("agora_uid")) and (
+        is_heygen_avatar(config) or is_live_avatar_avatar(config) or is_generic_avatar(config)
+    )
+
+
+def validate_avatar_config(
+    config: typing.Dict[str, typing.Any],
+    require_session_fields: bool = False,
+) -> None:
     """Validates avatar configuration at runtime.
 
     Parameters
@@ -45,6 +59,8 @@ def validate_avatar_config(config: typing.Dict[str, typing.Any]) -> None:
                 f"Invalid quality for {label}: {params.get('quality')}. "
                 f"Must be one of: {', '.join(valid_qualities)}"
             )
+        if require_session_fields and not params.get("agora_token"):
+            raise ValueError(f"{label} avatar requires agora_token after session enrichment")
     elif is_akool_avatar(config):
         params = config.get("params", {})
         if not params.get("api_key"):
@@ -53,6 +69,23 @@ def validate_avatar_config(config: typing.Dict[str, typing.Any]) -> None:
         params = config.get("params", {})
         if not params.get("api_key"):
             raise ValueError("Anam avatar requires api_key")
+    elif is_generic_avatar(config):
+        params = config.get("params", {})
+        if not params.get("api_key"):
+            raise ValueError("Generic avatar requires api_key")
+        if not params.get("api_base_url"):
+            raise ValueError("Generic avatar requires api_base_url")
+        if not params.get("avatar_id"):
+            raise ValueError("Generic avatar requires avatar_id")
+        if not params.get("agora_uid"):
+            raise ValueError("Generic avatar requires agora_uid")
+        if require_session_fields:
+            if not params.get("agora_token"):
+                raise ValueError("Generic avatar requires agora_token after session enrichment")
+            if not params.get("agora_appid"):
+                raise ValueError("Generic avatar requires agora_appid after session enrichment")
+            if not params.get("agora_channel"):
+                raise ValueError("Generic avatar requires agora_channel after session enrichment")
 
 
 def validate_tts_sample_rate(

src/agora_agent/agentkit/vendors/avatar.py

@@ -132,6 +132,48 @@ def to_config(self) -> Dict[str, Any]:
         return {"enable": enable, "vendor": "liveavatar", "params": params}
 
 
+class GenericAvatarOptions(BaseModel):
+    model_config = ConfigDict(extra="forbid")
+
+    api_key: str = Field(..., description="Generic avatar provider API key")
+    api_base_url: str = Field(..., description="Avatar provider API base URL")
+    avatar_id: str = Field(..., description="Avatar ID")
+    agora_uid: str = Field(..., description="Agora UID for the avatar video stream")
+    agora_appid: Optional[str] = Field(default=None, description="Agora App ID; filled by AgentSession when omitted")
+    agora_token: Optional[str] = Field(default=None, description="RTC token; generated by AgentSession when omitted")
+    agora_channel: Optional[str] = Field(default=None, description="Agora channel; filled by AgentSession when omitted")
+    enable: Optional[bool] = Field(default=None, description="Enable avatar (default: true)")
+    additional_params: Optional[Dict[str, Any]] = Field(default=None, description="Additional vendor-specific parameters")
+
+class GenericAvatar(BaseAvatar):
+    def __init__(self, **kwargs: Any):
+        self.options = GenericAvatarOptions(**kwargs)
+
+    @property
+    def required_sample_rate(self) -> int:
+        return 0
+
+    def to_config(self) -> Dict[str, Any]:
+        params: Dict[str, Any] = {
+            "api_key": self.options.api_key,
+            "api_base_url": self.options.api_base_url,
+            "avatar_id": self.options.avatar_id,
+            "agora_uid": self.options.agora_uid,
+        }
+
+        if self.options.agora_appid is not None:
+            params["agora_appid"] = self.options.agora_appid
+        if self.options.agora_token is not None:
+            params["agora_token"] = self.options.agora_token
+        if self.options.agora_channel is not None:
+            params["agora_channel"] = self.options.agora_channel
+        if self.options.additional_params is not None:
+            params = {**self.options.additional_params, **params}
+
+        enable = self.options.enable if self.options.enable is not None else True
+        return {"enable": enable, "vendor": "generic", "params": params}
+
+
 class AnamAvatarOptions(BaseModel):
     model_config = ConfigDict(extra="forbid")