opeani realtime websocket integration #51
Conversation
Summary by CodeRabbit
WalkthroughThe changes introduce a new Changes
Sequence Diagram(s)sequenceDiagram
participant Client as Client Application
participant Wrapper as OpenAIRealtimeWebsocketWrapper
participant WS as WebSocket Server
Client->>Wrapper: send(message)
Wrapper->>WS: send(message)
Wrapper-->>Client: log outbound event
WS->>Wrapper: message
Wrapper-->>Client: log inbound event
Wrapper->>Client: invoke event callbacks
Client->>Wrapper: close()
Wrapper->>WS: close()
Wrapper-->>Client: log connection closed
Suggested labels
Suggested reviewers
Poem
✨ Finishing Touches
🧪 Generate unit tests
🪧 TipsChatThere are 3 ways to chat with CodeRabbit:
SupportNeed help? Create a ticket on our support page for assistance with any issues or questions. Note: Be mindful of the bot's finite context window. It's strongly recommended to break down tasks such as reading entire modules into smaller chunks. For a focused discussion, use review comments to chat about specific files and their changes, instead of using the PR comments. CodeRabbit Commands (Invoked using PR comments)
Other keywords and placeholders
CodeRabbit Configuration File (
|
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 5
🔭 Outside diff range comments (1)
maxim/logger/openai/realtime/websocket/wrapper.py (1)
341-354: Fix context manager implementation for async compatibility.The class inherits from an async
ClientConnectionbut implements synchronous context manager methods. This creates a mismatch - theclose()method is async and should be awaited, but__exit__is synchronous.Either:
- Implement async context manager methods (
__aenter__and__aexit__)- Remove the context manager methods and rely on the parent class implementation
- Document that only async context usage is supported
- def __enter__(self): - """Context manager entry.""" - return self - - def __exit__(self, exc_type, exc_val, exc_tb): - """Context manager exit.""" - try: - # Note: In async context, close() should be awaited, but this is sync context - self._cleanup() - except Exception as e: - scribe().warning( - f"[MaximSDK][OpenAIRealtimeWebsocketWrapper] Error in context manager exit: {e}" - ) + async def __aenter__(self): + """Async context manager entry.""" + return self + + async def __aexit__(self, exc_type, exc_val, exc_tb): + """Async context manager exit.""" + try: + await self.close() + except Exception as e: + scribe().warning( + f"[MaximSDK][OpenAIRealtimeWebsocketWrapper] Error in context manager exit: {e}" + )
📜 Review details
Configuration used: CodeRabbit UI
Review profile: ASSERTIVE
Plan: Pro
📒 Files selected for processing (4)
README.md(1 hunks)maxim/logger/livekit/agent_session.py(1 hunks)maxim/logger/openai/realtime/websocket/wrapper.py(1 hunks)pyproject.toml(1 hunks)
🧰 Additional context used
🧠 Learnings (1)
pyproject.toml (3)
Learnt from: danpiths
PR: maximhq/maxim-py#15
File: maxim/models/dataset.py:94-95
Timestamp: 2025-06-04T14:52:14.757Z
Learning: The maxim-py project maintains backward compatibility with Python 3.9 and onwards. The team prefers to keep older typing syntax (Dict/List) instead of modern built-in types (dict/list) for consistency, even though the modern syntax is supported in their minimum Python version.
Learnt from: danpiths
PR: maximhq/maxim-py#15
File: maxim/models/dataset.py:94-95
Timestamp: 2025-06-04T14:52:14.757Z
Learning: The maxim-py project maintains backward compatibility with Python 3.9 and onwards, so the team prefers to keep older typing syntax even when modern alternatives are available.
Learnt from: SamstyleGhost
PR: maximhq/maxim-py#47
File: pyproject.toml:63-64
Timestamp: 2025-07-15T17:47:52.150Z
Learning: In the maxim-py repository, all integrations (like fireworks-ai, groq, anthropic, etc.) are intentionally kept as dev dependencies rather than optional extras. This is because maxim-py is typically installed as an observability/logging layer on top of existing AI SDK installations that users already have, rather than users installing maxim-py first and then adding AI SDKs as optional extras.
🪛 LanguageTool
README.md
[grammar] ~67-~67: Use proper capitalization
Context: ...t for OpenAI realtime API tracing using Websockets ### 3.9.10 - fix: removes signal hand...
(QB_NEW_EN_OTHER_ERROR_IDS_6)
🪛 Ruff (0.12.2)
maxim/logger/openai/realtime/websocket/wrapper.py
2-2: threading imported but unused
Remove unused import: threading
(F401)
6-6: typing.Dict is deprecated, use dict instead
(UP035)
6-6: typing.List is deprecated, use list instead
(UP035)
7-7: queue.Queue imported but unused
Remove unused import: queue.Queue
(F401)
16-16: Missing return type annotation for special method __init__
Add return type annotation: None
(ANN204)
19-19: Dynamically typed expressions (typing.Any) are disallowed in data
(ANN401)
29-29: Use dict instead of Dict for type annotation
Replace with dict
(UP006)
40-40: Missing return type annotation for public function connect_with_maxim_wrapper
Add return type annotation: None
(ANN201)
53-53: Missing return type annotation for special method __init__
Add return type annotation: None
(ANN204)
54-54: Missing type annotation for *args
(ANN002)
54-54: Missing type annotation for **kwargs
(ANN003)
54-54: Trailing comma missing
Add trailing comma
(COM812)
61-61: Use list instead of List for type annotation
Replace with list
(UP006)
63-63: Trailing comma missing
Add trailing comma
(COM812)
74-74: Trailing comma missing
Add trailing comma
(COM812)
77-77: Missing return type annotation for public function send
(ANN201)
77-77: Missing type annotation for function argument message
(ANN001)
77-77: Missing type annotation for **kwargs
(ANN003)
87-87: f-string without any placeholders
Remove extraneous f prefix
(F541)
87-87: Trailing comma missing
Add trailing comma
(COM812)
90-90: Consider moving this statement to an else block
(TRY300)
94-94: Trailing comma missing
Add trailing comma
(COM812)
98-98: Missing return type annotation for public function recv
(ANN201)
98-98: Missing type annotation for **kwargs
(ANN003)
108-108: f-string without any placeholders
Remove extraneous f prefix
(F541)
108-108: Trailing comma missing
Add trailing comma
(COM812)
111-111: Consider moving this statement to an else block
(TRY300)
115-115: Trailing comma missing
Add trailing comma
(COM812)
119-119: Missing return type annotation for public function recv_streaming
(ANN201)
119-119: Missing type annotation for **kwargs
(ANN003)
127-127: f-string without any placeholders
Remove extraneous f prefix
(F541)
127-127: Trailing comma missing
Add trailing comma
(COM812)
130-130: Consider moving this statement to an else block
(TRY300)
134-134: Trailing comma missing
Add trailing comma
(COM812)
138-138: Missing return type annotation for public function close
(ANN201)
138-138: Missing type annotation for function argument code
(ANN001)
138-138: Missing type annotation for function argument reason
(ANN001)
138-138: Missing type annotation for **kwargs
(ANN003)
142-142: Trailing comma missing
Add trailing comma
(COM812)
151-151: Consider moving this statement to an else block
(TRY300)
155-155: Trailing comma missing
Add trailing comma
(COM812)
159-159: Missing return type annotation for public function ping
(ANN201)
159-159: Missing type annotation for function argument data
(ANN001)
159-159: Missing type annotation for **kwargs
(ANN003)
162-162: f-string without any placeholders
Remove extraneous f prefix
(F541)
167-167: Consider moving this statement to an else block
(TRY300)
167-167: Unnecessary assignment to result before return statement
Remove unnecessary assignment
(RET504)
171-171: Trailing comma missing
Add trailing comma
(COM812)
175-175: Missing return type annotation for public function pong
(ANN201)
175-175: Missing type annotation for function argument data
(ANN001)
175-175: Missing type annotation for **kwargs
(ANN003)
178-178: f-string without any placeholders
Remove extraneous f prefix
(F541)
183-183: Consider moving this statement to an else block
(TRY300)
183-183: Unnecessary assignment to result before return statement
Remove unnecessary assignment
(RET504)
187-187: Trailing comma missing
Add trailing comma
(COM812)
191-191: Missing return type annotation for public function add_event_callback
Add return type annotation: None
(ANN201)
195-195: Trailing comma missing
Add trailing comma
(COM812)
198-198: Missing return type annotation for public function remove_event_callback
Add return type annotation: None
(ANN201)
203-203: Trailing comma missing
Add trailing comma
(COM812)
206-206: Missing return type annotation for private function _process_inbound_message
Add return type annotation: None
(ANN202)
206-206: Dynamically typed expressions (typing.Any) are disallowed in message
(ANN401)
225-225: Trailing comma missing
Add trailing comma
(COM812)
232-232: Do not catch blind exception: Exception
(BLE001)
234-234: Trailing comma missing
Add trailing comma
(COM812)
237-237: Missing return type annotation for private function _process_outbound_message
Add return type annotation: None
(ANN202)
237-237: Dynamically typed expressions (typing.Any) are disallowed in message
(ANN401)
256-256: Trailing comma missing
Add trailing comma
(COM812)
263-263: Do not catch blind exception: Exception
(BLE001)
265-265: Trailing comma missing
Add trailing comma
(COM812)
268-268: Missing return type annotation for private function _process_event_async
Add return type annotation: None
(ANN202)
273-273: Do not catch blind exception: Exception
(BLE001)
275-275: Trailing comma missing
Add trailing comma
(COM812)
278-278: Missing return type annotation for private function _process_event_sync
Add return type annotation: None
(ANN202)
285-285: Do not catch blind exception: Exception
(BLE001)
287-287: Trailing comma missing
Add trailing comma
(COM812)
295-295: Do not catch blind exception: Exception
(BLE001)
297-297: Trailing comma missing
Add trailing comma
(COM812)
301-301: Trailing comma missing
Add trailing comma
(COM812)
304-304: Do not catch blind exception: Exception
(BLE001)
306-306: Trailing comma missing
Add trailing comma
(COM812)
309-309: Use dict instead of Dict for type annotation
Replace with dict
(UP006)
326-326: Missing return type annotation for private function _cleanup
Add return type annotation: None
(ANN202)
334-334: Trailing comma missing
Add trailing comma
(COM812)
336-336: Do not catch blind exception: Exception
(BLE001)
338-338: Trailing comma missing
Add trailing comma
(COM812)
341-341: Missing return type annotation for special method __enter__
(ANN204)
345-345: Missing return type annotation for special method __exit__
(ANN204)
345-345: Missing type annotation for function argument exc_type
(ANN001)
345-345: Missing type annotation for function argument exc_val
(ANN001)
345-345: Missing type annotation for function argument exc_tb
(ANN001)
350-350: Do not catch blind exception: Exception
(BLE001)
352-352: Trailing comma missing
Add trailing comma
(COM812)
🔇 Additional comments (4)
pyproject.toml (1)
7-7: Version bump looks good.The version increment from 3.9.10 to 3.9.11 appropriately reflects the addition of OpenAI realtime WebSocket support.
maxim/logger/livekit/agent_session.py (1)
200-201: Good defensive programming practice.Adding the None check prevents potential AttributeError when accessing
output.call_id. This improves the robustness of the event handling.maxim/logger/openai/realtime/websocket/wrapper.py (2)
40-42: Clarify the purpose of this empty function.The
connect_with_maxim_wrapperfunction is empty and lacks documentation. Please either:
- Implement the intended functionality
- Add a docstring explaining it's a placeholder for future implementation
- Remove it if it's not needed
232-235: Consider more specific exception handling in internal methods.While the main WebSocket methods appropriately log and re-raise exceptions, several internal methods catch broad
Exceptiontypes without re-raising. This could hide unexpected errors and make debugging difficult.Consider catching more specific exceptions or at least logging the full exception details (including stack trace) when catching broad exceptions in methods like
_process_event_sync,_process_event_async, etc.Also applies to: 263-266, 273-276, 285-288, 295-298, 304-307, 336-339, 350-353
⛔ Skipped due to learnings
Learnt from: SamstyleGhost PR: maximhq/maxim-py#46 File: maxim/logger/groq/helpers.py:0-0 Timestamp: 2025-07-12T08:10:50.682Z Learning: In the maxim-py repository, when implementing logging/instrumentation integrations (like the Groq SDK integration), user SamstyleGhost prefers catching broad Exception types rather than specific exceptions to ensure that logging failures never cause user operations to fail. The integration should be completely non-intrusive and maintain operational reliability over exception handling specificity.
| async def recv_streaming(self, **kwargs): | ||
| """Override recv_streaming to log inbound streaming messages.""" | ||
| try: | ||
| # Call the parent recv_streaming method | ||
| stream = await super().recv_streaming(**kwargs) | ||
|
|
||
| # TODO: Consider how to handle streaming message logging | ||
| scribe().debug( | ||
| f"[MaximSDK][OpenAIRealtimeWebsocketWrapper] Received streaming message" | ||
| ) | ||
|
|
||
| return stream | ||
|
|
||
| except Exception as e: | ||
| scribe().warning( | ||
| f"[MaximSDK][OpenAIRealtimeWebsocketWrapper] Failed to receive streaming message: {e}" | ||
| ) | ||
| raise |
There was a problem hiding this comment.
🧹 Nitpick (assertive)
Address the TODO for streaming message logging.
The recv_streaming method has a TODO comment about handling streaming message logging. Currently, the streaming data is not processed or logged in detail like regular messages.
Would you like me to help implement proper streaming message processing or create an issue to track this TODO?
🧰 Tools
🪛 Ruff (0.12.2)
119-119: Missing return type annotation for public function recv_streaming
(ANN201)
119-119: Missing type annotation for **kwargs
(ANN003)
127-127: f-string without any placeholders
Remove extraneous f prefix
(F541)
127-127: Trailing comma missing
Add trailing comma
(COM812)
130-130: Consider moving this statement to an else block
(TRY300)
134-134: Trailing comma missing
Add trailing comma
(COM812)
🤖 Prompt for AI Agents
In maxim/logger/openai/realtime/websocket/wrapper.py around lines 119 to 136,
the recv_streaming method has a TODO about logging streaming messages but
currently only logs a generic message without processing the stream data. To fix
this, modify the method to asynchronously iterate over the streaming messages
received from the parent recv_streaming call, log each message's content or
relevant details for debugging, and yield or return the messages appropriately
to preserve the streaming behavior.
| result = await super().send(message, **kwargs) | ||
|
|
||
| scribe().debug( | ||
| f"[MaximSDK][OpenAIRealtimeWebsocketWrapper] Sent message successfully" |
There was a problem hiding this comment.
🧹 Nitpick (assertive)
Remove unnecessary f-string prefixes.
Several log messages use f-string syntax without any placeholders. These should be regular strings.
For example:
- f"[MaximSDK][OpenAIRealtimeWebsocketWrapper] Sent message successfully"
+ "[MaximSDK][OpenAIRealtimeWebsocketWrapper] Sent message successfully"Apply similar changes to lines 108, 127, 162, and 178.
Also applies to: 108-108, 127-127, 162-162, 178-178
🧰 Tools
🪛 Ruff (0.12.2)
87-87: f-string without any placeholders
Remove extraneous f prefix
(F541)
87-87: Trailing comma missing
Add trailing comma
(COM812)
🤖 Prompt for AI Agents
In maxim/logger/openai/realtime/websocket/wrapper.py at lines 87, 108, 127, 162,
and 178, remove the unnecessary f-string prefixes from log messages that do not
contain any placeholders. Replace the f-strings with regular string literals to
simplify the code and improve readability.
|
|
||
| # Process event asynchronously | ||
| self._process_event_async(event) | ||
| self.events_processed += 1 |
There was a problem hiding this comment.
🧹 Nitpick (assertive)
Consider thread-safe counter and proper executor shutdown.
-
The
events_processedcounter is incremented from async methods that could potentially be called concurrently. Consider usingthreading.Lockorasyncio.Lockfor thread-safe increments. -
The executor shutdown with
wait=Falsemight leave event processing tasks incomplete. Consider usingwait=Trueor implementing a graceful shutdown mechanism.
+ self._event_counter_lock = asyncio.Lock()
self.events_processed = 0
# In _process_inbound_message and _process_outbound_message:
- self.events_processed += 1
+ async with self._event_counter_lock:
+ self.events_processed += 1
# In _cleanup:
- self.executor.shutdown(wait=False)
+ self.executor.shutdown(wait=True, cancel_futures=True)Also applies to: 261-261, 329-330
🤖 Prompt for AI Agents
In maxim/logger/openai/realtime/websocket/wrapper.py at lines 230, 261, and
329-330, the events_processed counter is incremented in async contexts without
thread safety, risking race conditions. Use an asyncio.Lock or threading.Lock to
protect increments of events_processed to ensure thread-safe updates.
Additionally, revise the executor shutdown calls to use wait=True or implement a
graceful shutdown to ensure all event processing tasks complete before shutdown.
| import json | ||
| import threading | ||
| import time | ||
| import uuid | ||
| from concurrent.futures import ThreadPoolExecutor | ||
| from typing import Any, Callable, Dict, List, Optional | ||
| from queue import Queue | ||
| from websockets.asyncio.client import ClientConnection | ||
| from ....logger import Logger | ||
| from .....scribe import scribe | ||
|
|
There was a problem hiding this comment.
🧹 Nitpick (assertive)
Remove unused imports.
The threading and Queue imports are not used in the code and should be removed.
import json
-import threading
import time
import uuid
from concurrent.futures import ThreadPoolExecutor
from typing import Any, Callable, Dict, List, Optional
-from queue import Queue
from websockets.asyncio.client import ClientConnection
from ....logger import Logger
from .....scribe import scribe📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| import json | |
| import threading | |
| import time | |
| import uuid | |
| from concurrent.futures import ThreadPoolExecutor | |
| from typing import Any, Callable, Dict, List, Optional | |
| from queue import Queue | |
| from websockets.asyncio.client import ClientConnection | |
| from ....logger import Logger | |
| from .....scribe import scribe | |
| import json | |
| import time | |
| import uuid | |
| from concurrent.futures import ThreadPoolExecutor | |
| from typing import Any, Callable, Dict, List, Optional | |
| from websockets.asyncio.client import ClientConnection | |
| from ....logger import Logger | |
| from .....scribe import scribe |
🧰 Tools
🪛 Ruff (0.12.2)
2-2: threading imported but unused
Remove unused import: threading
(F401)
6-6: typing.Dict is deprecated, use dict instead
(UP035)
6-6: typing.List is deprecated, use list instead
(UP035)
7-7: queue.Queue imported but unused
Remove unused import: queue.Queue
(F401)
🤖 Prompt for AI Agents
In maxim/logger/openai/realtime/websocket/wrapper.py at lines 1 to 11, the
imports for threading and Queue are not used anywhere in the code. Remove these
two import statements to clean up the code and avoid unnecessary imports.
|
|
||
| ### 3.9.11 | ||
|
|
||
| - feat: Adds support for OpenAI realtime API tracing using Websockets |
There was a problem hiding this comment.
🧹 Nitpick (assertive)
Use proper capitalization for "WebSockets".
-- feat: Adds support for OpenAI realtime API tracing using Websockets
+- feat: Adds support for OpenAI realtime API tracing using WebSockets📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| - feat: Adds support for OpenAI realtime API tracing using Websockets | |
| - feat: Adds support for OpenAI realtime API tracing using WebSockets |
🧰 Tools
🪛 LanguageTool
[grammar] ~67-~67: Use proper capitalization
Context: ...t for OpenAI realtime API tracing using Websockets ### 3.9.10 - fix: removes signal hand...
(QB_NEW_EN_OTHER_ERROR_IDS_6)
🤖 Prompt for AI Agents
In README.md at line 67, the term "Websockets" is not properly capitalized.
Change "Websockets" to "WebSockets" to use the correct capitalization.
Add support for OpenAI realtime API tracing using Websockets
TL;DR
Added WebSocket wrapper for OpenAI's realtime API to enable tracing and logging of WebSocket communications.
What changed?
OpenAIRealtimeWebsocketWrapperclass that extends the WebSockets ClientConnection interfacehandle_tool_call_executedto handle None values in function call outputsHow to test?
Why make this change?
This enhancement allows for comprehensive tracing and debugging of real-time communications with OpenAI's API using WebSockets. It provides visibility into the bidirectional message flow, which is essential for monitoring, debugging, and analyzing interactions with OpenAI's real-time services.