fix example with function calling and add tests

Author: maxkahan

README.md

@@ -95,6 +95,15 @@ call.get_or_create(
 )
 ```
 
+### Getting Response Data
+
+Many calls return a `StreamResponse` object, with the specific dataclass for the method call nested inside. You can access this via:
+
+```python
+response: StreamResponse[StartClosedCaptionsResponse] = call.start_closed_captions()
+response.data  # Gives the StartClosedCaptionsResponse model
+```
+
 ### App configuration
 
 ```python

examples/env.example

@@ -1,7 +1,7 @@
 # Stream API credentials
 STREAM_API_KEY=your_stream_api_key_here
 STREAM_API_SECRET=your_stream_api_secret_here
-STREAM_BASE_URL=stream_base_url_here
+EXAMPLE_BASE_URL=example_base_url_here
 
 # Deepgram API credentials
 DEEPGRAM_API_KEY=your_deepgram_api_key_here

examples/llm_audio_conversation/README.md

@@ -53,7 +53,7 @@ This example demonstrates how to build a real-time AI conversation bot using Str
    ```env
    STREAM_API_KEY=your_actual_stream_api_key
    STREAM_API_SECRET=your_actual_stream_api_secret
-   STREAM_BASE_URL=https://chat.stream-io-api.com/
+   STREAM_BASE_URL=the_stream_base_url
    DEEPGRAM_API_KEY=your_actual_deepgram_api_key
    ELEVENLABS_API_KEY=your_actual_elevenlabs_api_key
    OPENAI_API_KEY=your_actual_openai_api_key

examples/openai_realtime_speech_to_speech/main.py

@@ -5,7 +5,10 @@
 from dotenv import load_dotenv
 from examples.utils import create_user, open_browser
 from getstream import Stream
+from getstream.models import StartClosedCaptionsResponse
 from getstream.plugins.sts.openai_realtime import OpenAIRealtime
+from dataclasses import asdict
+import json
 
 
 logging.basicConfig(
@@ -54,33 +57,6 @@ async def main():
         voice="alloy",
     )
 
-    @sts_bot.on("connected")
-    async def _on_connected():
-        print("✅ CONNECTED EVENT RECEIVED")
-        logging.info("✅ Bot connected successfully")
-
-    @sts_bot.on("disconnected")
-    async def _on_disconnected():
-        print("❌ DISCONNECTED EVENT RECEIVED")
-        logging.info("❌ Bot disconnected")
-
-    @sts_bot.on("error")
-    async def _on_error(error):
-        print(f"💥 ERROR EVENT RECEIVED: {error}")
-        logging.error("💥 Bot error: %s", error)
-
-    @sts_bot.on("session.created")
-    @sts_bot.on("session.updated")
-    @sts_bot.on("conversation.item.created")
-    @sts_bot.on("response.created")
-    @sts_bot.on("response.done")
-    @sts_bot.on("call.session_participant_joined")
-    @sts_bot.on("call.session_participant_left")
-    async def _on_openai_event(event):
-        print(f"🔔 Event received: {event.type}")
-        print(f"   Event data: {event}")
-        logging.info("🔔 Event: %s", event.type)
-
     try:
         logging.info("Connecting to OpenAI Realtime...")
 
@@ -89,20 +65,66 @@ async def _on_openai_event(event):
             logging.error("❌ OPENAI_API_KEY not found in environment")
             return
 
-        await sts_bot.connect(call, agent_user_id=bot_user_id)
-        logging.info("🎧 Listening for responses... (Press Ctrl+C to stop)")
-        logging.info("💡 Try speaking in the browser to generate audio events!")
-
-        while sts_bot.is_connected:
-            await asyncio.sleep(1)
+        async with await sts_bot.connect(call, agent_user_id=bot_user_id) as connection:
+            tools = [
+                {
+                    "type": "function",
+                    "name": "start_closed_captions",
+                    "description": "start closed captions for the call",
+                    "parameters": {
+                        "type": "object",
+                        "properties": {},
+                        "required": [],
+                    },
+                }
+            ]
+
+            await sts_bot.update_session(
+                turn_detection={
+                    "type": "semantic_vad",
+                    "eagerness": "low",
+                    "create_response": True,
+                    "interrupt_response": True,
+                },
+                tools=tools,
+            )
+
+            logging.info("🎧 Listening for responses... (Press Ctrl+C to stop)")
+            logging.info("💡 Try speaking in the browser – ask it something like 'start closed captions' to trigger the function call.")
+
+            async def start_closed_captions() -> StartClosedCaptionsResponse:
+                """Helper that starts closed captions for the call."""
+                return call.start_closed_captions().data
+
+            async for event in connection:
+                logging.info("🔔 Event received: %s", event.type)
+
+                if (
+                    event.type == "response.done"
+                    and event.response.output is not None
+                    and len(event.response.output) > 0
+                    and event.response.output[0].type == "function_call"
+                ):
+                    tool_call_id = event.response.output[0].call_id
+                    
+                    if event.response.output[0].name == "start_closed_captions":
+                        logging.info("🛠  Assistant requested start_closed_captions()")
+
+                        result = await start_closed_captions()
+
+                        # Send the tool result back to the assistant
+                        await sts_bot.send_function_call_output(tool_call_id, result.to_json())
+
+                        logging.info("🛠  Replied to tool call with result: %s", result)
+
+            
 
     except KeyboardInterrupt:  # noqa: WPS420
         logging.info("\n⏹️  Stopping OpenAI Realtime Speech to Speech bot…")
     except Exception as e:  # noqa: BLE001
         logging.exception("❌ Error: %s", e)
     finally:
         logging.info("Cleaning up...")
-        await sts_bot.close()
         client.delete_users([user_id, bot_user_id])
         logging.info("Cleanup complete")
 

examples/utils.py

@@ -5,11 +5,14 @@
 This module contains common functions that can be reused across different examples.
 """
 
+import os
 import webbrowser
 from urllib.parse import urlencode
 from getstream.models import UserRequest
 from getstream.stream import Stream
+from dotenv import load_dotenv
 
+load_dotenv()
 
 def create_user(client: Stream, id: str, name: str) -> None:
     """
@@ -36,7 +39,7 @@ def open_browser(api_key: str, token: str, call_id: str) -> str:
     Returns:
         The URL that was opened
     """
-    base_url = "https://pronto.getstream.io/bare/join/"
+    base_url = f"{os.getenv('EXAMPLE_BASE_URL')}/join/"
     params = {"api_key": api_key, "token": token, "skip_lobby": "true"}
 
     url = f"{base_url}{call_id}?{urlencode(params)}"

getstream/plugins/sts/__init__.py

@@ -8,41 +8,28 @@
 
 
 class STS(AsyncIOEventEmitter, abc.ABC):
-    """Speech-to-Speech (full duplex) base class.
-
-    Implementations are expected to:
-    • establish an audio session (usually via Stream Video `Call.connect_openai`)
-    • emit high-level events coming from the AI agent (for example
-      ``conversation.updated`` or ``error``)
-    • optionally expose helper methods like ``update_session`` or
-      ``send_user_message``.
-
-    Events emitted by *all* STS implementations:
-        - *connected*: fired once the underlying websocket is ready
-        - *disconnected*: fired when the websocket is closed (graceful or error)
-        - *error*: emitted for any exception that bubbles up
-        - *<any other event type coming from the provider>*: forwarded verbatim
+    """Base class for Speech-to-Speech (STS) implementations.
+
+    This abstract base class provides the foundation for implementing real-time
+    speech-to-speech communication with AI agents. It handles event emission
+    and connection state management.
+
+    Key Features:
+    - Event-driven architecture using AsyncIOEventEmitter
+    - Connection state tracking
+    - Standardized event interface
+
+    Implementations should:
+    1. Establish and manage the audio session
+    2. Handle provider-specific authentication and setup
+    3. Emit appropriate events for state changes and interactions
+    4. Implement any provider-specific helper methods
     """
 
     def __init__(self):
         super().__init__()
         self._is_connected = False
 
-    # ---------------------------------------------------------------------
-    # Lifecycle helpers
-    # ---------------------------------------------------------------------
-    @abc.abstractmethod
-    async def connect(self, *args, **kwargs):  # pragma: no cover
-        """Establish the realtime connection (provider-specific)."""
-
-    @abc.abstractmethod
-    async def close(self):  # pragma: no cover
-        """Close the connection and release all resources."""
-
-    # Derived classes should set ``self._is_connected`` accordingly so that
-    # embedders can introspect the state.
-    # ---------------------------------------------------------------------
-
     @property
     def is_connected(self) -> bool:
         """Return True if the realtime session is currently active."""