Adding support for the HeyGen LiveAvatar API

env.example

@@ -84,6 +84,7 @@ GROQ_API_KEY=...
 
 # Heygen
 HEYGEN_API_KEY=...
+HEYGEN_LIVE_AVATAR_API_KEY=...
 
 # Hume
 HUME_API_KEY=...

examples/foundational/43a-heygen-video-service.py

@@ -25,7 +25,7 @@
 from pipecat.services.cartesia.tts import CartesiaTTSService
 from pipecat.services.deepgram.stt import DeepgramSTTService
 from pipecat.services.google.llm import GoogleLLMService
-from pipecat.services.heygen.api import AvatarQuality, NewSessionRequest
+from pipecat.services.heygen.client import ServiceType
 from pipecat.services.heygen.video import HeyGenVideoService
 from pipecat.transports.base_transport import BaseTransport, TransportParams
 from pipecat.transports.daily.transport import DailyParams, DailyTransport
@@ -73,11 +73,9 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
         llm = GoogleLLMService(api_key=os.getenv("GOOGLE_API_KEY"))
 
         heyGen = HeyGenVideoService(
-            api_key=os.getenv("HEYGEN_API_KEY"),
+            api_key=os.getenv("HEYGEN_LIVE_AVATAR_API_KEY"),
+            service_type=ServiceType.LIVE_AVATAR,
             session=session,
-            session_request=NewSessionRequest(
-                avatar_id="Shawn_Therapist_public", version="v2", quality=AvatarQuality.high
-            ),
         )
 
         messages = [

src/pipecat/services/heygen/api.py → src/pipecat/services/heygen/api_interactive_avatar.py

@@ -16,6 +16,8 @@
 from loguru import logger
 from pydantic import BaseModel, Field
 
+from pipecat.services.heygen.base_api import BaseAvatarApi, StandardSessionResponse
+
 
 class AvatarQuality(str, Enum):
     """Enum representing different avatar quality levels."""
@@ -136,7 +138,7 @@ def __init__(self, message: str, status: int, response_text: str) -> None:
         self.response_text = response_text
 
 
-class HeyGenApi:
+class HeyGenApi(BaseAvatarApi):
     """HeyGen Streaming API client."""
 
     BASE_URL = "https://api.heygen.com/v1"
@@ -193,16 +195,16 @@ async def _request(self, path: str, params: Dict[str, Any], expect_data: bool =
             logger.error(f"Network error while calling HeyGen API: {str(e)}")
             raise
 
-    async def new_session(self, request_data: NewSessionRequest) -> HeyGenSession:
-        """Create a new streaming session.
+    async def new_session(self, request_data: NewSessionRequest) -> StandardSessionResponse:
+        """Create a new streaming session and start it immediately.
 
         https://docs.heygen.com/reference/new-session
 
         Args:
             request_data: Session configuration parameters.
 
         Returns:
-            Session information, including ID and access token.
+            StandardSessionResponse: Standardized session information with HeyGen raw response.
         """
         params = {
             "quality": request_data.quality,
@@ -225,9 +227,21 @@ async def new_session(self, request_data: NewSessionRequest) -> HeyGenSession:
         session_info = await self._request("/streaming.new", params)
         print("heygen session info", session_info)
 
-        return HeyGenSession.model_validate(session_info)
+        heygen_session = HeyGenSession.model_validate(session_info)
+
+        await self._start_session(heygen_session.session_id)
+
+        # Convert to standardized response
+        return StandardSessionResponse(
+            session_id=heygen_session.session_id,
+            access_token=heygen_session.access_token,
+            livekit_url=heygen_session.url,
+            livekit_agent_token=heygen_session.livekit_agent_token,
+            ws_url=heygen_session.realtime_endpoint,
+            raw_response=heygen_session,
+        )
 
-    async def start_session(self, session_id: str) -> Any:
+    async def _start_session(self, session_id: str) -> Any:
         """Start the streaming session.
 
         https://docs.heygen.com/reference/start-session