From 25ebd8ca81c078489b5a1e11f7026f84d272673a Mon Sep 17 00:00:00 2001 From: abhinav7x94 Date: Sun, 16 Aug 2026 09:07:55 +0530 Subject: [PATCH] fix(openai-sdk-python): restore complete async middleware update --- .../src/supermemory_openai/middleware.py | 549 +++++++++++++----- .../tests/test_client_isolation.py | 150 +++++ 2 files changed, 545 insertions(+), 154 deletions(-) diff --git a/packages/openai-sdk-python/src/supermemory_openai/middleware.py b/packages/openai-sdk-python/src/supermemory_openai/middleware.py index c377c38d..01315369 100644 --- a/packages/openai-sdk-python/src/supermemory_openai/middleware.py +++ b/packages/openai-sdk-python/src/supermemory_openai/middleware.py @@ -439,164 +439,405 @@ class SupermemoryOpenAIWrapper: def _create_client_variant_facade(self, name: str) -> _ResourceFacade: """Wrap completions reached through a client response variant.""" client_resource = getattr(self._client, name) - chat_resource = client_resource.chaNkwX\][ۜ\]ȋ]ܙX]W٘XܞB -BY[ ] ]ܘ]ܙ\ۜHH[\S[Y\XJ\][ۜX\][ۗܙ\\J] \]ȋ]ܙX]W٘XܞJB -BY[ ]ܘ]ܙ\ۜHH[\S[Y\XJ]T[\S[Y\XJ\][ۜX\][ۗܙ\\JY[ \]ȋ]ܙX]W٘XܞJB -B -BY[ ] \][ۜ˝]X[Z[ܙ\ۜHH\][ۗܙ\\J\][ۜ\X[HX[Z[ܙX]W٘XܞB -BY[ ] ]X[Z[ܙ\ۜHH[\S[Y\XJ\][ۜX\][ۗܙ\\J] \X[HX[Z[ܙX]W٘XܞJB -BY[ ]X[Z[ܙ\ۜHH[\S[Y\XJ]T[\S[Y\XJ\][ۜX\][ۗܙ\\JY[ \X[HX[Z[ܙX]W٘XܞJB -B -B]\[‚Y\ۜWݘ\X[ܙX]\Y[[KNH O\[WN]\ˆ]]Y[ ] \][ۜJKܙX]K]]Y[ ] JK\][ۜ˘ܙX]K]]Y[ JK] \][ۜ˘ܙX]KBPSTSTUSӗUH -ܛX[\][ۜ˜]ȋ] ]ȋY[ ]ȋ\][ۜ˜X[Z[ȋ] X[Z[ȋY[ X[Z[ȋBYX[\[\][ۗܙX]JY[[K]H O[NY]OHܛX[]\Y[ ] \][ۜ˘ܙX]BY]OH\][ۜ˜]Ȏ]\Y[ ] \][ۜ˝]ܘ]ܙ\ۜKܙX]BY]OH] ]Ȏ]\Y[ ] ]ܘ]ܙ\ۜK\][ۜ˘ܙX]BY]OHY[ ]Ȏ]\Y[ ]ܘ]ܙ\ۜK] \][ۜ˘ܙX]BY]OH\][ۜ˜X[Z[Ȏ]\Y[ ] \][ۜ˝]X[Z[ܙ\ۜKܙX]BY]OH] X[Z[Ȏ]\Y[ ] ]X[Z[ܙ\ۜK\][ۜ˘ܙX]BY]OHY[ X[Z[Ȏ]\Y[ ]X[Z[ܙ\ۜK] \][ۜ˘ܙX]BZ\H\\[ۑ\܊[ۛۈ\][ۈ]]HB]\ ^\J]]\OUYJH\NYۛܙV[\Y YXܘ]ܗBY\\Y[[ܞW\W^J -H O[\]ܖӛۙKۙKۙWN]] X -˙[\ۋȔTTQSSԖWTWVH\ Z^HJNZY[Y\\Y[Y[\ۛX[[ZY]\J -H OۙN\WY[ ܚY[[ܙX]HHܙX]W[Y[ + chat_resource = client_resource.chat + completions_resource = chat_resource.completions + return _ResourceFacade( + client_resource, + chat=_ResourceFacade( + chat_resource, + completions=_ResourceFacade( + completions_resource, + create=self._create_completion_method( + completions_resource.create, + streaming_response=name == "with_streaming_response", + ), + ), + ), + ) + + def _create_completion_method( + self, + original_create: Any, + *, + streaming_response: bool = False, + ) -> Any: + """Wrap one completion create implementation with memory injection.""" + if self._is_async_client and streaming_response: -B\Έ\HHB\[YZW\ -Y\Y\Έ\[WK۝Z[\YΈ\[K[N[K\W^N -H O\[WN\˘\[ -۝Z[\YB]\ˆȜH\[H۝[Xܙ] ^۝Z[\YHK -Y\Y\B]] -\\Y[[ܞW[ZKZY]\K\\Y[[ܞK\\Y[[ܞH]\ݘ[YOS[ -K -K] -\\Y[[ܞW[ZKZY]\KY\[W\YWYXYZW\  -N[[N[HH]\\Y[[ܞJ\WY[ ZY]\W[ۜ[[ XHK -B[[؎[HH]\\Y[[ܞJ\WY[ ZY]\W[ۜ[[ XK -B\\\WY[ ] \][ۜ˘ܙX]H\ܚY[[ܙX]B\\[[K]\\WY[ ]\\[[؋] X\\OH] [X\\\\[[؋] \][ۜ˛X\\OH\][ۜ[X\\[[؋] \][ۜ˘ܙX]J[[H ]\Y\Y\VȜH\\۝[]]HWK -B\\\OHȝ[[ XBY\Y\HܚY[[ܙX]K[\˚\țY\Y\ȗB\\Y\Y\VȘ۝[HOHXܙ] ][[ X\\[ -[[ XH[Y\YJH܈Y\YH[Y\Y\B\WY[ ] \][ۜ˘ܙX]J[[H ]\Y\Y\VȜH\\۝[[ܘ\YWK -B\\\OHȝ[[ XBY\ܙ]ܘ\[W٘XYWܙXݙ\W\[WY[ + def create_streaming_with_memory( + **kwargs: Any, + ) -> _AsyncMemoryResponseContextManager: + return _AsyncMemoryResponseContextManager( + self, + original_create, + kwargs, + ) -H OۙN\WY[ ܚY[[ܙX]HHܙX]W[Y[ + return create_streaming_with_memory -B\Έ\HHB\[YZW\ -Y\Y\Έ\[WK۝Z[\YΈ\[K[N[K\W^N -H O\[WN\˘\[ -۝Z[\YB]\ˆȜH\[H۝[Xܙ] ^۝Z[\YHK -Y\Y\B]] -\\Y[[ܞW[ZKZY]\K\\Y[[ܞK\\Y[[ܞH]\ݘ[YOS[ -K -K] -\\Y[[ܞW[ZKZY]\KY\[W\YWYXYZW\  -N[[N[HH]\\Y[[ܞJ\WY[ ZY]\W[ۜ[[ XHK -B[[؎[HH]\\Y[[ܞJ[[KZY]\W[ۜ[[ XK -B[[؋] \][ۜ˘ܙX]J[[H ]\Y\Y\VȜH\\۝[]]HWK -B\\\OHȝ[[ XB\\ܚY[[ܙX]K[[OH B\\\WY[ ] \][ۜ˘ܙX]H\ܚY[[ܙX]BY\ܘ][X[Z[ܙ\ۜW٘XY\ܙ[XZ[Y[[ܞW]\J -H OۙN\WY[ HܙX]W[Y[ + if self._is_async_client: -B[H]Xܙ\ۜWݘ\X[\WY[ [XHX[[]\ݘ[YOT]ԙ\ۜJX[ -JK[XHX[[]\ݘ[YOT[X[P۝^ -X[ -JK -B\Έ\HHB\[YZW\ -Y\Y\Έ\[WK۝Z[\YΈ\[K[N[K\W^N -H O\[WN\˘\[ -۝Z[\YB]\ˆȜH\[H۝[Xܙ] ^۝Z[\YHK -Y\Y\B]] -\\Y[[ܞW[ZKZY]\K\\Y[[ܞK\\Y[[ܞH]\ݘ[YOS[ -K -K] -\\Y[[ܞW[ZKZY]\KY\[W\YWYXYZW\  -N[[؎[HH]\\Y[[ܞJ\WY[ ZY]\W[ۜ[[ XK -B܈ܙX]H[\ۜWݘ\X[ܙX]\[[؋]ܘ]ܙ\ۜHN\ۜHHܙX]J[[H ]\Y\Y\VȜH\\۝[]ȟWK -B\\\ۜK\J -K\] -\Y HB܈ܙX]H[\ۜWݘ\X[ܙX]\[[؋]X[Z[ܙ\ۜHN]ܙX]J[[H ]\Y\Y\VȜH\\۝[X[Z[ȟWK -H\X[N\\X[KX[ [] -X[HB\\\OHȝ[[ XH - ܈ܙX]H[[˝[Y\ -N\\Xܙ] ][[ X[ܙX]K[\˚\țY\Y\ȗJBY\\[ܘ][X[Z[ܙ\ۜWY^\ܙ[XZ[Y[[ܞW]\J -H OۙN\WY[ HܙX]W\[Y[ + async def create_async_with_memory( + **kwargs: Any, + ) -> Any: + return await self._create_with_memory_async(original_create, **kwargs) -B[H]Xܙ\ۜWݘ\X[\WY[ [XHX[\[[]\ݘ[YOT]ԙ\ۜJX[ -JK[XHX[[]\ݘ[YOP\[X[P۝^ -X[ -JK -B\Έ\HHB\[YZW\ -Y\Y\Έ\[WK۝Z[\YΈ\[K[N[K\W^N -H O\[WN\˘\[ -۝Z[\YB]\ˆȜH\[H۝[Xܙ] ^۝Z[\YHK -Y\Y\B\[Y[ܘ]ܙX]N[JH OۙN\ۜHH]Z]ܙX]J[[H ]\Y\Y\VȜH\\۝[]ȟWK -B\\\ۜK\J -K\] -\Y HB\[Yۜ[YWX[JX[W۝^[JH OۙN\[]X[W۝^\X[N\\X[KX[ [] -X[HB]] -\\Y[[ܞW[ZKZY]\K\\Y[[ܞK\\Y[[ܞH]\ݘ[YOS[ -K -K] -\\Y[[ܞW[ZKZY]\KY\[W\YWYXYZW\  -N[[؎[HH]\\Y[[ܞJ\WY[ ZY]\W[ۜ[[ XK -B܈ܙX]H[\ۜWݘ\X[ܙX]\[[؋]ܘ]ܙ\ۜHN\[[˜[[ܘ]ܙX]JJB܈ܙX]H[\ۜWݘ\X[ܙX]\[[؋]X[Z[ܙ\ۜHNX[W۝^HܙX]J[[H ]\Y\Y\VȜH\\۝[X[Z[ȟWK -B\[[˜[ۜ[YWX[JX[W۝^ -JB\\\OHȝ[[ XH - ܈ܙX]H[[˝[Y\ -N\\Xܙ] ][[ X[ܙX]K[\˚\țY\Y\ȗJB]\ X\˜\[Y]^J]PSTSTUSӗUB]\ X\˘\[[˜\[Y\ܙX[\[[ZW]\W\[ZY]\J]H OۙN\Έ\HHBܚ]\Έ\\V[ۘ[KWHHB[Y\Y\Έ\\[WWHHB\[YZW\ -Y\Y\Έ\[WK۝Z[\YΈ\[K[N[K\W^N -H O\[WN\˘\[ -۝Z[\YB]\ˆȜH\[H۝[Xܙ] ^۝Z[\YHK -Y\Y\B\[YZWYY[[ܞJY[[K۝Z[\YΈ۝[\WY[ۘ[K\[K -H OۙNܚ]\˘\[ + return create_async_with_memory -۝Z[\Y\WY ۝[ -JBY[Wܙ\]Y\ -\]Y\ \]Y\ -H O \ۜNHHۋY\]Y\ ۝[ -B[Y\Y\˘\[ -VțY\Y\ȗJB]\ \ۜJ \]Y\\\]Y\ XY\^Ș۝[ ]\H\X][ۋڜۈKۏ^ˆY]\ ]\ؚX] \][ۈܙX]Y [[ ]\X\Ȏˆˆ[^ Y\YHȜH\\[۝[ȟK[\ܙX\ۈBKK -BY[H \[Y[ -[ܝZ [[ܝ -[Wܙ\]Y\ -JB\WY[H\[[RJ\W^OH[ZK]\Y[ZY[ -BN]] -\\Y[[ܞW[ZKZY]\K\\Y[[ܞK\\Y[[ܞH]\ݘ[YOS[ -K -K] -\\Y[[ܞW[ZKZY]\KY\[W\]YZW\  -K] -\\Y[[ܞW[ZKZY]\KYY[[ܞW]YZWYY[[ܞK -K\[˘]\[XܙUYB -H\]Y\[˜[\Y[\[^\ȋ[[YU\[Bܘ\Y[HH]\\Y[[ܞJ\WY[ ZY]\W[ۜ[[ \X[YY[[ܞOH[^\ȊK -BܙX]HHX[\[\][ۗܙX]Jܘ\Y ] -B\Hˆ[[ ]\Y\Y\ȎȜH\\۝[]]HY\YHWKBY]OHܛX[\ۜHH]Z]ܙX]J -\B\\\ۜKYOH]\ ]\[Y] [] -]ȊN]ܙ\ۜHH]Z]ܙX]J -\B\\]ܙ\ۜK\J -KYOH]\ ]\[N\ۜW۝^HܙX]J -\B\\[X \]Z]XJ\ۜW۝^ -B\[]\ۜW۝^\X[Z[ܙ\ۜN\\X[Z[ܙ\ۜK]\HOH ]Z]ܘ\Y Z]ٛܗؘXܛ[\ -B]Z]\[[˜Y\ - -B˘X + def create_sync_with_memory( + **kwargs: Any, + ) -> Any: + return self._create_with_memory_sync(original_create, **kwargs) -B[[YW\[Hˆ\[ˆ܈\[[]YY\X\\[˘]YܞK[[YU\[BB\\\OHȝ[[ \X[B\\ܚ]\OHˆ -[[ \X[۝\][ێXY ][[ \X[\\]]HY\YH -BB\\[[Y\Y\HOH B\\[Y\Y\VVȘ۝[HOHXܙ] ][[ \X[\\[[YW\[OHB[[N]Z]\WY[ J -B]\ X\˘\[[\NYۛܙV[\Y YXܘ]ܗB\[Y\\Y\[Y[]\ۛWٛܗ[XY[[ - -H OۙN\WY[ ܚY[[ܙX]HHܙX]W\[Y[ - -B\Έ\HHBܚ]\Έ\\V[ۘ[KWHHB\[YZW\ -Y\Y\Έ\[WK۝Z[\YΈ\[K[N[K\W^N -H O\[WN\˘\[ -۝Z[\YB]\ˆȜH\[H۝[Xܙ] ^۝Z[\YHK -Y\Y\B\[YZWYY[[ܞJY[[K۝Z[\YΈ۝[\WY[ۘ[K\[K -H OۙNܚ]\˘\[ - -۝Z[\Y\WY ۝[ -JB]] -\\Y[[ܞW[ZKZY]\K\\Y[[ܞK\\Y[[ܞH]\ݘ[YOS[ -K -K] -\\Y[[ܞW[ZKZY]\KY\[W\YWYXYZW\  -K] -\\Y[[ܞW[ZKZY]\KYY[[ܞWYWYXYZWYY[[ܞK -N[[N[HH]\\Y[[ܞJ\WY[ ZY]\W[ۜ[[ XHYY[[ܞOH[^\ȊK -B[[؎[HH]\\Y[[ܞJ\WY[ ZY]\W[ۜ[[ XYY[[ܞOH[^\ȊK -B]Z][[؋] \][ۜ˘ܙX]J[[H ]\Y\Y\VȜH\\۝[]]H[[Y\YHWK -B]Z][[KZ]ٛܗؘXܛ[\ -B]Z][[؋Z]ٛܗؘXܛ[\ -B\\\WY[ ] \][ۜ˘ܙX]H\ܚY[[ܙX]B\\\OHȝ[[ XB\\ܚ]\OHˆ -[[ X۝\][ێXY ][[ X\\]]H[[Y\YH -BB\\ܚY[[ܙX]K[[OH B \ No newline at end of file + return create_sync_with_memory + + async def _create_with_memory_async( + self, + original_create: Any, + **kwargs: Any, + ) -> Any: + """Async version of create with memory injection.""" + messages = kwargs.get("messages", []) + + if self._options.add_memory == "always": + user_message = get_last_user_message(messages) + if user_message and user_message.strip(): + content = ( + get_conversation_content(messages) + if self._options.custom_id + else user_message + ) + custom_id = ( + f"conversation:{self._options.custom_id}" + if self._options.custom_id + else None + ) + + # Create background task for memory storage + task = asyncio.create_task( + add_memory_tool( + self._supermemory_client, + self._container_tag, + content, + custom_id, + self._logger, + ) + ) + + # Track the task and set up cleanup + self._background_tasks.add(task) + task.add_done_callback(self._background_tasks.discard) + + # Log any exceptions but don't fail the main request + def handle_task_exception(task_obj): + try: + if task_obj.exception() is not None: + exception = task_obj.exception() + if isinstance( + exception, + (SupermemoryNetworkError, SupermemoryAPIError), + ): + self._logger.warn( + "Background memory storage failed", + { + "error": str(exception), + "type": type(exception).__name__, + }, + ) + else: + self._logger.error( + "Unexpected error in background memory storage", + { + "error": str(exception), + "type": type(exception).__name__, + }, + ) + except asyncio.CancelledError: + self._logger.debug("Memory storage task was cancelled") + + task.add_done_callback(handle_task_exception) + + if self._options.mode != "profile": + user_message = get_last_user_message(messages) + if not user_message: + self._logger.debug("No user message found, skipping memory search") + return await original_create(**kwargs) + + self._logger.info( + "Starting memory search", + { + "container_tag": self._container_tag, + "conversation_id": self._options.custom_id, + "mode": self._options.mode, + }, + ) + + enhanced_messages = await add_system_prompt( + messages, + self._container_tag, + self._logger, + self._options.mode, + self._get_api_key(), + ) + + kwargs["messages"] = enhanced_messages + return await original_create(**kwargs) + + def _create_with_memory_sync( + self, + original_create: Any, + **kwargs: Any, + ) -> Any: + """Sync version of create with memory injection.""" + # For sync clients, we implement a simplified version without background tasks + messages = kwargs.get("messages", []) + + # Handle memory addition synchronously if needed + if self._options.add_memory == "always": + user_message = get_last_user_message(messages) + if user_message and user_message.strip(): + content = ( + get_conversation_content(messages) + if self._options.custom_id + else user_message + ) + custom_id = ( + f"conversation:{self._options.custom_id}" + if self._options.custom_id + else None + ) + + # Use asyncio.run() for the memory addition + try: + asyncio.run( + add_memory_tool( + self._supermemory_client, + self._container_tag, + content, + custom_id, + self._logger, + ) + ) + except RuntimeError as e: + if "cannot be called from a running event loop" in str(e): + # We're in an async context, log warning and skip memory saving + self._logger.warn( + "Cannot save memory in sync client from async context", + {"error": str(e)}, + ) + else: + raise + except SupermemoryNetworkError as e: + # Network errors are expected, log as warning + self._logger.warn("Network error saving memory", {"error": str(e)}) + except (SupermemoryAPIError, SupermemoryMemoryOperationError) as e: + # API/memory errors are concerning, log as error + self._logger.error("Failed to save memory", {"error": str(e)}) + except Exception as e: + # Unexpected errors should be investigated + self._logger.error( + "Unexpected error saving memory", + {"error": str(e), "type": type(e).__name__}, + ) + + # Handle memory search and injection + if self._options.mode != "profile": + user_message = get_last_user_message(messages) + if not user_message: + self._logger.debug("No user message found, skipping memory search") + return original_create(**kwargs) + + self._logger.info( + "Starting memory search", + { + "container_tag": self._container_tag, + "conversation_id": self._options.custom_id, + "mode": self._options.mode, + }, + ) + + # Use asyncio.run() for memory search and injection + try: + enhanced_messages = asyncio.run( + add_system_prompt( + messages, + self._container_tag, + self._logger, + self._options.mode, + self._get_api_key(), + ) + ) + except RuntimeError as e: + if "cannot be called from a running event loop" in str(e): + # We're in an async context, run in a separate thread + import concurrent.futures + + with concurrent.futures.ThreadPoolExecutor(max_workers=1) as executor: + future = executor.submit( + asyncio.run, + add_system_prompt( + messages, + self._container_tag, + self._logger, + self._options.mode, + self._get_api_key(), + ), + ) + enhanced_messages = future.result() + else: + raise + + kwargs["messages"] = enhanced_messages + return original_create(**kwargs) + + async def wait_for_background_tasks(self, timeout: Optional[float] = 10.0) -> None: + """ + Wait for all background memory storage tasks to complete. + + Args: + timeout: Maximum time to wait in seconds. None for no timeout. + + Raises: + asyncio.TimeoutError: If tasks don't complete within timeout + """ + if not self._background_tasks: + return + + self._logger.debug( + f"Waiting for {len(self._background_tasks)} background tasks to complete" + ) + + try: + if timeout is not None: + await asyncio.wait_for( + asyncio.gather(*self._background_tasks, return_exceptions=True), + timeout=timeout, + ) + else: + await asyncio.gather(*self._background_tasks, return_exceptions=True) + + self._logger.debug("All background tasks completed") + except asyncio.TimeoutError: + self._logger.warn( + f"Background tasks did not complete within {timeout}s timeout" + ) + # Cancel remaining tasks + tasks_to_cancel = [ + task for task in self._background_tasks if not task.done() + ] + for task in tasks_to_cancel: + task.cancel() + + if tasks_to_cancel: + await asyncio.gather(*tasks_to_cancel, return_exceptions=True) + raise + + def cancel_background_tasks(self) -> None: + """Cancel all pending background tasks.""" + cancelled_count = 0 + for task in self._background_tasks: + if not task.done(): + task.cancel() + cancelled_count += 1 + + if cancelled_count > 0: + self._logger.debug(f"Cancelled {cancelled_count} pending background tasks") + + async def __aenter__(self): + """Async context manager entry.""" + return self + + async def __aexit__(self, exc_type, exc_val, exc_tb): + """Async context manager exit - wait for background tasks.""" + try: + await self.wait_for_background_tasks(timeout=5.0) + except asyncio.TimeoutError: + self._logger.warn("Some background memory tasks did not complete on exit") + + def __enter__(self): + """Sync context manager entry.""" + return self + + def __exit__(self, exc_type, exc_val, exc_tb): + """Sync context manager exit - attempt to wait for background tasks.""" + if self._background_tasks: + try: + # Try to wait for background tasks in sync context + asyncio.run(self.wait_for_background_tasks(timeout=5.0)) + except RuntimeError as e: + if "cannot be called from a running event loop" in str(e): + # In async context, just cancel the tasks + self._logger.warn( + "Cannot wait for background tasks in sync context from async environment. " + "Use async context manager or call wait_for_background_tasks() manually." + ) + self.cancel_background_tasks() + else: + raise + except asyncio.TimeoutError: + self._logger.warn( + "Some background memory tasks did not complete on exit" + ) + self.cancel_background_tasks() + + def __getattr__(self, name: str) -> Any: + """Delegate all other attributes to the wrapped client.""" + if name in {"with_raw_response", "with_streaming_response"}: + value = self._create_client_variant_facade(name) + setattr(self, name, value) + return value + return getattr(self._client, name) + + +def with_supermemory( + openai_client: Union[OpenAI, AsyncOpenAI], + options: OpenAIMiddlewareOptions, +) -> Union[OpenAI, AsyncOpenAI]: + """ + Wraps an OpenAI client with SuperMemory middleware to automatically inject relevant memories + into the system prompt based on the user's message content. + + This middleware searches the supermemory API for relevant memories using the container tag + and user message, then either appends memories to an existing system prompt or creates + a new system prompt with the memories. + + Args: + openai_client: The OpenAI client to wrap with SuperMemory middleware + options: Configuration options for the middleware (container_tag and custom_id are required) + + Returns: + An OpenAI client with SuperMemory middleware injected + + Example: + ```python + from supermemory_openai import with_supermemory, OpenAIMiddlewareOptions + from openai import OpenAI + + # Create OpenAI client with supermemory middleware + openai = OpenAI(api_key=os.getenv("OPENAI_API_KEY")) + openai_with_supermemory = with_supermemory( + openai, + OpenAIMiddlewareOptions( + container_tag="user-123", + custom_id="conversation-456", + mode="full", + add_memory="always" + ) + ) + + # Use normally - memories will be automatically injected + response = await openai_with_supermemory.chat.completions.create( + model="gpt-4", + messages=[ + {"role": "user", "content": "What's my favorite programming language?"} + ] + ) + ``` + + Raises: + ValueError: When SUPERMEMORY_API_KEY environment variable is not set + Exception: When supermemory API request fails + """ + wrapper = SupermemoryOpenAIWrapper(openai_client, options) + # Return the wrapper, which delegates all attributes to the original client + return cast(Union[OpenAI, AsyncOpenAI], wrapper) diff --git a/packages/openai-sdk-python/tests/test_client_isolation.py b/packages/openai-sdk-python/tests/test_client_isolation.py index 700ac4cf..a0a07ace 100644 --- a/packages/openai-sdk-python/tests/test_client_isolation.py +++ b/packages/openai-sdk-python/tests/test_client_isolation.py @@ -3,12 +3,18 @@ from __future__ import annotations import asyncio +import gc +import inspect +import json import os +import warnings from types import SimpleNamespace from typing import Any, Generator, Literal, Optional from unittest.mock import AsyncMock, Mock, patch +import httpx import pytest +from openai import AsyncOpenAI from supermemory_openai import OpenAIMiddlewareOptions, with_supermemory @@ -113,6 +119,35 @@ def response_variant_creates(client: Any, name: str) -> list[Any]: ] +REAL_ASYNC_COMPLETION_PATHS = ( + "normal", + "completions.raw", + "chat.raw", + "client.raw", + "completions.streaming", + "chat.streaming", + "client.streaming", +) + + +def real_async_completion_create(client: Any, path: str) -> Any: + if path == "normal": + return client.chat.completions.create + if path == "completions.raw": + return client.chat.completions.with_raw_response.create + if path == "chat.raw": + return client.chat.with_raw_response.completions.create + if path == "client.raw": + return client.with_raw_response.chat.completions.create + if path == "completions.streaming": + return client.chat.completions.with_streaming_response.create + if path == "chat.streaming": + return client.chat.with_streaming_response.completions.create + if path == "client.streaming": + return client.with_streaming_response.chat.completions.create + raise AssertionError(f"Unknown completion path: {path}") + + @pytest.fixture(autouse=True) # type: ignore[untyped-decorator] def supermemory_api_key() -> Generator[None, None, None]: with patch.dict(os.environ, {"SUPERMEMORY_API_KEY": "test-key"}): @@ -330,6 +365,121 @@ def test_async_raw_and_streaming_response_prefixes_remain_memory_aware() -> None assert "secret-tenant-b" in str(create.call_args.kwargs["messages"]) +@pytest.mark.parametrize("path", REAL_ASYNC_COMPLETION_PATHS) +@pytest.mark.asyncio +async def test_real_async_openai_paths_use_async_middleware(path: str) -> None: + lookups: list[str] = [] + writes: list[tuple[str, Optional[str], str]] = [] + sent_messages: list[list[Any]] = [] + + async def fake_prompt( + messages: list[Any], + container_tag: str, + logger: Any, + mode: Any, + api_key: str, + ) -> list[Any]: + lookups.append(container_tag) + return [ + {"role": "system", "content": f"secret-{container_tag}"}, + *messages, + ] + + async def fake_add_memory( + client: Any, + container_tag: str, + content: str, + custom_id: Optional[str], + logger: Any, + ) -> None: + writes.append((container_tag, custom_id, content)) + + def handle_request(request: httpx.Request) -> httpx.Response: + body = json.loads(request.content) + sent_messages.append(body["messages"]) + return httpx.Response( + 200, + request=request, + headers={"content-type": "application/json"}, + json={ + "id": "chatcmpl-test", + "object": "chat.completion", + "created": 0, + "model": "gpt-test", + "choices": [ + { + "index": 0, + "message": {"role": "assistant", "content": "ok"}, + "finish_reason": "stop", + } + ], + }, + ) + + http_client = httpx.AsyncClient(transport=httpx.MockTransport(handle_request)) + base_client = AsyncOpenAI(api_key="openai-test", http_client=http_client) + + try: + with patch( + "supermemory_openai.middleware.supermemory.Supermemory", + return_value=Mock(), + ), patch( + "supermemory_openai.middleware.add_system_prompt", + new=fake_prompt, + ), patch( + "supermemory_openai.middleware.add_memory_tool", + new=fake_add_memory, + ), warnings.catch_warnings( + record=True + ) as caught: + warnings.simplefilter("always", RuntimeWarning) + wrapped: Any = with_supermemory( + base_client, + middleware_options("tenant-real", add_memory="always"), + ) + create = real_async_completion_create(wrapped, path) + kwargs = { + "model": "gpt-test", + "messages": [{"role": "user", "content": "private message"}], + } + + if path == "normal": + response = await create(**kwargs) + assert response.id == "chatcmpl-test" + elif path.endswith(".raw"): + raw_response = await create(**kwargs) + assert raw_response.parse().id == "chatcmpl-test" + else: + response_context = create(**kwargs) + assert not inspect.isawaitable(response_context) + async with response_context as streaming_response: + assert streaming_response.status_code == 200 + + await wrapped.wait_for_background_tasks() + await asyncio.sleep(0) + gc.collect() + + runtime_warnings = [ + warning + for warning in caught + if issubclass(warning.category, RuntimeWarning) + ] + + assert lookups == ["tenant-real"] + assert writes == [ + ( + "tenant-real", + "conversation:thread-tenant-real", + "User: private message", + ) + ] + assert len(sent_messages) == 1 + assert sent_messages[0][0]["content"] == "secret-tenant-real" + assert runtime_warnings == [] + finally: + await base_client.close() + + @pytest.mark.asyncio # type: ignore[untyped-decorator] async def test_shared_async_client_saves_only_for_selected_tenant() -> None: base_client, original_create = create_async_client()