add create call shortcut

Author: tschellenbach

agents-core/vision_agents/core/agents/agents.py

@@ -5,11 +5,12 @@
 import time
 import uuid
 from dataclasses import asdict
-from typing import TYPE_CHECKING, Any, Dict, List, Optional, TypeGuard
+from typing import TYPE_CHECKING, Any, Dict, List, Optional, TypeGuard, Coroutine
 from uuid import uuid4
 
 import getstream.models
 from aiortc import VideoStreamTrack
+from getstream.video.async_call import Call
 from getstream.video.rtc import Call
 
 from getstream.video.rtc.pb.stream.video.sfu.models.models_pb2 import TrackType
@@ -693,6 +694,13 @@ async def create_user(self) -> None:
 
         return None
 
+    async def create_call(self, call_type: str, call_id: str) -> Call:
+        """Shortcut for creating a call/room etc."""
+        call = self.edge.client.video.call(call_type, call_id)
+        response = await call.get_or_create(data={"created_by_id": self.agent_user.id})
+
+        return call
+
     def _on_vad_audio(self, event: VADAudioEvent):
         self.logger.debug(f"Vad audio event {self._truncate_for_logging(event)}")
 

examples/other_examples/openai_realtime_webrtc/openai_realtime_example.py

@@ -42,9 +42,7 @@ async def join_call(agent: Agent, call_type: str, call_id: str, **kwargs) -> Non
     # ensure the agent user is created
     await agent.create_user()
     # Create a call
-    call = agent.edge.client.video.call(call_type, call_id)
-    # Ensure the call exists server-side before joining
-    await call.get_or_create(data={"created_by_id": agent.agent_user.id})
+    call = await agent.create_call(call_type, call_id)
 
     logger.info("🤖 Starting OpenAI Realtime Agent...")