feat: add support for agent context history

deepgram/__init__.py

@@ -340,6 +340,7 @@
     InjectUserMessageOptions,
     FunctionCallResponse,
     AgentKeepAlive,
+    Flags,
     # sub level
     Listen,
     Speak,
@@ -355,6 +356,10 @@
     Output,
     Audio,
     Endpoint,
+    Context,
+    HistoryConversationMessage,
+    HistoryFunctionCallsMessage,
+    FunctionCallHistory,
 )
 
 # utilities

deepgram/client.py

@@ -354,6 +354,7 @@
     InjectUserMessageOptions,
     FunctionCallResponse,
     AgentKeepAlive,
+    Flags,
     # sub level
     Listen,
     Speak,
@@ -369,6 +370,10 @@
     Output,
     Audio,
     Endpoint,
+    Context,
+    HistoryConversationMessage,
+    HistoryFunctionCallsMessage,
+    FunctionCallHistory,
 )
 
 

deepgram/clients/__init__.py

@@ -363,6 +363,7 @@
     InjectUserMessageOptions,
     FunctionCallResponse,
     AgentKeepAlive,
+    Flags,
     # sub level
     Listen,
     Speak,
@@ -378,4 +379,8 @@
     Output,
     Audio,
     Endpoint,
+    Context,
+    HistoryConversationMessage,
+    HistoryFunctionCallsMessage,
+    FunctionCallHistory,
 )

deepgram/clients/agent/__init__.py

@@ -38,6 +38,7 @@
     InjectUserMessageOptions,
     FunctionCallResponse,
     AgentKeepAlive,
+    Flags,
     # sub level
     Listen,
     Speak,
@@ -53,4 +54,8 @@
     Output,
     Audio,
     Endpoint,
+    Context,
+    HistoryConversationMessage,
+    HistoryFunctionCallsMessage,
+    FunctionCallHistory,
 )

deepgram/clients/agent/client.py

@@ -37,6 +37,7 @@
     InjectUserMessageOptions as LatestInjectUserMessageOptions,
     FunctionCallResponse as LatestFunctionCallResponse,
     AgentKeepAlive as LatestAgentKeepAlive,
+    Flags as LatestFlags,
     # sub level
     Listen as LatestListen,
     Speak as LatestSpeak,
@@ -52,6 +53,10 @@
     Output as LatestOutput,
     Audio as LatestAudio,
     Endpoint as LatestEndpoint,
+    Context as LatestContext,
+    HistoryConversationMessage as LatestHistoryConversationMessage,
+    HistoryFunctionCallsMessage as LatestHistoryFunctionCallsMessage,
+    FunctionCallHistory as LatestFunctionCallHistory,
 )
 
 
@@ -85,6 +90,7 @@
 InjectUserMessageOptions = LatestInjectUserMessageOptions
 FunctionCallResponse = LatestFunctionCallResponse
 AgentKeepAlive = LatestAgentKeepAlive
+Flags = LatestFlags
 
 Listen = LatestListen
 Speak = LatestSpeak
@@ -100,3 +106,7 @@
 Output = LatestOutput
 Audio = LatestAudio
 Endpoint = LatestEndpoint
+Context = LatestContext
+HistoryConversationMessage = LatestHistoryConversationMessage
+HistoryFunctionCallsMessage = LatestHistoryFunctionCallsMessage
+FunctionCallHistory = LatestFunctionCallHistory

deepgram/clients/agent/enums.py

@@ -24,6 +24,7 @@ class AgentWebSocketEvents(StrEnum):
     FunctionCallRequest: str = "FunctionCallRequest"
     AgentStartedSpeaking: str = "AgentStartedSpeaking"
     AgentAudioDone: str = "AgentAudioDone"
+    History: str = "History"
     Error: str = "Error"
     Unhandled: str = "Unhandled"
 

deepgram/clients/agent/v1/__init__.py

@@ -42,6 +42,7 @@
     InjectUserMessageOptions,
     FunctionCallResponse,
     AgentKeepAlive,
+    Flags,
     # sub level
     Listen,
     Speak,
@@ -57,4 +58,8 @@
     Output,
     Audio,
     Endpoint,
+    Context,
+    HistoryConversationMessage,
+    HistoryFunctionCallsMessage,
+    FunctionCallHistory,
 )

deepgram/clients/agent/v1/websocket/__init__.py

@@ -33,6 +33,7 @@
     InjectUserMessageOptions,
     FunctionCallResponse,
     AgentKeepAlive,
+    Flags,
     # sub level
     Listen,
     Speak,
@@ -48,4 +49,8 @@
     Output,
     Audio,
     Endpoint,
+    Context,
+    HistoryConversationMessage,
+    HistoryFunctionCallsMessage,
+    FunctionCallHistory,
 )

deepgram/clients/agent/v1/websocket/async_client.py

@@ -29,6 +29,10 @@
     ErrorResponse,
     UnhandledResponse,
 )
+from .options import (
+    HistoryConversationMessage,
+    HistoryFunctionCallsMessage,
+)
 from .options import (
     SettingsOptions,
     UpdatePromptOptions,
@@ -478,6 +482,28 @@ async def _process_text(self, message: str) -> None:
                         injection_refused=injection_refused_result,
                         **dict(cast(Dict[Any, Any], self._kwargs)),
                     )
+                case AgentWebSocketEvents.History:
+                    # Determine if this is conversation history or function call history
+                    history_result: Union[HistoryConversationMessage, HistoryFunctionCallsMessage, Dict[Any, Any]]
+
+                    if "role" in data and "content" in data:
+                        # This is conversation history
+                        history_result = HistoryConversationMessage.from_json(message)
+                        self._logger.verbose("HistoryConversationMessage: %s", history_result)
+                    elif "function_calls" in data:
+                        # This is function call history
+                        history_result = HistoryFunctionCallsMessage.from_json(message)
+                        self._logger.verbose("HistoryFunctionCallsMessage: %s", history_result)
+                    else:
+                        # Fallback for unknown History format
+                        history_result = data
+                        self._logger.verbose("History (unknown format): %s", history_result)
+
+                    await self._emit(
+                        AgentWebSocketEvents(AgentWebSocketEvents.History),
+                        history=history_result,
+                        **dict(cast(Dict[Any, Any], self._kwargs)),
+                    )
                 case AgentWebSocketEvents.Close:
                     close_result: CloseResponse = CloseResponse.from_json(message)
                     self._logger.verbose("CloseResponse: %s", close_result)

deepgram/clients/agent/v1/websocket/client.py

@@ -29,6 +29,10 @@
     ErrorResponse,
     UnhandledResponse,
 )
+from .options import (
+    HistoryConversationMessage,
+    HistoryFunctionCallsMessage,
+)
 from .options import (
     SettingsOptions,
     UpdatePromptOptions,
@@ -473,6 +477,28 @@ def _process_text(self, message: str) -> None:
                         injection_refused=injection_refused_result,
                         **dict(cast(Dict[Any, Any], self._kwargs)),
                     )
+                case AgentWebSocketEvents.History:
+                    # Determine if this is conversation history or function call history
+                    history_result: Union[HistoryConversationMessage, HistoryFunctionCallsMessage, Dict[Any, Any]]
+
+                    if "role" in data and "content" in data:
+                        # This is conversation history
+                        history_result = HistoryConversationMessage.from_json(message)
+                        self._logger.verbose("HistoryConversationMessage: %s", history_result)
+                    elif "function_calls" in data:
+                        # This is function call history
+                        history_result = HistoryFunctionCallsMessage.from_json(message)
+                        self._logger.verbose("HistoryFunctionCallsMessage: %s", history_result)
+                    else:
+                        # Fallback for unknown History format
+                        history_result = data
+                        self._logger.verbose("History (unknown format): %s", history_result)
+
+                    self._emit(
+                        AgentWebSocketEvents(AgentWebSocketEvents.History),
+                        history=history_result,
+                        **dict(cast(Dict[Any, Any], self._kwargs)),
+                    )
                 case AgentWebSocketEvents.Close:
                     close_result: CloseResponse = CloseResponse.from_json(message)
                     self._logger.verbose("CloseResponse: %s", close_result)

deepgram/clients/agent/v1/websocket/options.py

@@ -244,6 +244,94 @@ def __getitem__(self, key):
         return _dict[key]
 
 
+# History and Context classes for Function Call Context / History feature
+
+@dataclass
+class Flags(BaseResponse):
+    """
+    This class defines configuration flags for the agent settings.
+    """
+
+    history: bool = field(default=True)
+
+
+@dataclass
+class HistoryConversationMessage(BaseResponse):
+    """
+    This class defines a conversation text message as part of the conversation history.
+    """
+
+    type: str = field(default="History")
+    role: str = field(default="")  # "user" or "assistant"
+    content: str = field(default="")
+
+
+@dataclass
+class FunctionCallHistory(BaseResponse):
+    """
+    This class defines a single function call in the history.
+    """
+
+    id: str = field(default="")
+    name: str = field(default="")
+    client_side: bool = field(default=False)
+    arguments: str = field(default="")
+    response: str = field(default="")
+
+
+@dataclass
+class HistoryFunctionCallsMessage(BaseResponse):
+    """
+    This class defines function call messages as part of the conversation history.
+    """
+
+    type: str = field(default="History")
+    function_calls: List[FunctionCallHistory] = field(default_factory=list)
+
+    def __post_init__(self):
+        """Convert dict function_calls to FunctionCallHistory objects; normalize None."""
+        if not self.function_calls:
+            self.function_calls = []
+            return
+
+        # Convert dicts and filter out any None values for safety
+        converted_calls = [
+            FunctionCallHistory.from_dict(call) if isinstance(call, dict) else call
+            for call in self.function_calls
+            if call is not None
+        ]
+        self.function_calls = [call for call in converted_calls if call is not None]
+
+
+@dataclass
+class Context(BaseResponse):
+    """
+    This class defines the conversation context including the history of messages and function calls.
+    """
+
+    messages: List[Union[HistoryConversationMessage, HistoryFunctionCallsMessage]] = field(default_factory=list)
+
+    def __post_init__(self):
+        """Convert dict messages to appropriate message objects; normalize None."""
+        if not self.messages:
+            self.messages = []
+            return
+
+        # Convert dicts to appropriate message objects and filter out None values
+        converted_messages = []
+        for message in self.messages:
+            if message is None:
+                continue
+            if isinstance(message, dict):
+                if "function_calls" in message:
+                    converted_messages.append(HistoryFunctionCallsMessage.from_dict(message))
+                else:
+                    converted_messages.append(HistoryConversationMessage.from_dict(message))
+            else:
+                converted_messages.append(message)
+        self.messages = converted_messages
+
+
 @dataclass
 class Agent(BaseResponse):
     """
@@ -271,10 +359,12 @@ class Agent(BaseResponse):
     greeting: Optional[str] = field(
         default=None, metadata=dataclass_config(exclude=lambda f: f is None)
     )
-
+    context: Optional[Context] = field(
+        default=None, metadata=dataclass_config(exclude=lambda f: f is None)
+    )
 
     def __post_init__(self):
-        """Handle conversion of dict/list data to proper Speak objects"""
+        """Handle conversion of dict/list data to proper Speak objects and Context objects"""
         # Handle speak conversion (OneOf pattern)
         if isinstance(self.speak, list):
             self.speak = [
@@ -284,6 +374,10 @@ def __post_init__(self):
         elif isinstance(self.speak, dict):
             self.speak = Speak.from_dict(self.speak)
 
+        # Handle context conversion
+        if isinstance(self.context, dict):
+            self.context = Context.from_dict(self.context)
+
     def __getitem__(self, key):
         _dict = self.to_dict()
         if "listen" in _dict and isinstance(_dict["listen"], dict):
@@ -295,6 +389,8 @@ def __getitem__(self, key):
                 _dict["speak"] = [Speak.from_dict(item) for item in _dict["speak"]]
             elif isinstance(_dict["speak"], dict):
                 _dict["speak"] = Speak.from_dict(_dict["speak"])
+        if "context" in _dict and isinstance(_dict["context"], dict):
+            _dict["context"] = Context.from_dict(_dict["context"])
         return _dict[key]
 
 
@@ -356,13 +452,19 @@ class SettingsOptions(BaseResponse):
     mip_opt_out: Optional[bool] = field(
         default=False, metadata=dataclass_config(exclude=lambda f: f is None)
     )
+    flags: Optional[Flags] = field(
+        default=None, metadata=dataclass_config(exclude=lambda f: f is None)
+    )
 
     def __getitem__(self, key):
         _dict = self.to_dict()
         if "audio" in _dict and isinstance(_dict["audio"], dict):
             _dict["audio"] = Audio.from_dict(_dict["audio"])
         if "agent" in _dict and isinstance(_dict["agent"], dict):
             _dict["agent"] = Agent.from_dict(_dict["agent"])
+        if "flags" in _dict and isinstance(_dict["flags"], dict):
+            _dict["flags"] = Flags.from_dict(_dict["flags"])
+        return _dict[key]
 
     def check(self):
         """