add openai middleware functionality for python sdk (#546)openai-middleware-python

add openai middleware functionality

fix critical type errors and linting issues

update readme with middleware documentation

1 files changed, 230 insertions, 0 deletions

diff --git a/packages/openai-sdk-python/src/supermemory_openai/utils.py b/packages/openai-sdk-python/src/supermemory_openai/utils.py
new file mode 100644
index 00000000..d9ea6845
--- /dev/null
+++ b/packages/openai-sdk-python/src/supermemory_openai/utils.py
@@ -0,0 +1,230 @@
+"""Utility functions for Supermemory OpenAI middleware."""
+
+import json
+from typing import Optional, Any, Protocol
+
+from openai.types.chat import ChatCompletionMessageParam
+
+
+class Logger(Protocol):
+    """Logger protocol for type safety."""
+
+    def debug(self, message: str, data: Optional[dict[str, Any]] = None) -> None:
+        """Log debug message."""
+        ...
+
+    def info(self, message: str, data: Optional[dict[str, Any]] = None) -> None:
+        """Log info message."""
+        ...
+
+    def warn(self, message: str, data: Optional[dict[str, Any]] = None) -> None:
+        """Log warning message."""
+        ...
+
+    def error(self, message: str, data: Optional[dict[str, Any]] = None) -> None:
+        """Log error message."""
+        ...
+
+
+class SimpleLogger:
+    """Simple logger implementation."""
+
+    def __init__(self, verbose: bool = False):
+        self.verbose: bool = verbose
+
+    def _log(self, level: str, message: str, data: Optional[dict[str, Any]] = None) -> None:
+        """Internal logging method."""
+        if not self.verbose:
+            return
+
+        log_message = f"[supermemory] {message}"
+        if data:
+            log_message += f" {json.dumps(data, indent=2)}"
+
+        if level == "error":
+            print(f"ERROR: {log_message}", flush=True)
+        elif level == "warn":
+            print(f"WARN: {log_message}", flush=True)
+        else:
+            print(log_message, flush=True)
+
+    def debug(self, message: str, data: Optional[dict[str, Any]] = None) -> None:
+        """Log debug message."""
+        self._log("debug", message, data)
+
+    def info(self, message: str, data: Optional[dict[str, Any]] = None) -> None:
+        """Log info message."""
+        self._log("info", message, data)
+
+    def warn(self, message: str, data: Optional[dict[str, Any]] = None) -> None:
+        """Log warning message."""
+        self._log("warn", message, data)
+
+    def error(self, message: str, data: Optional[dict[str, Any]] = None) -> None:
+        """Log error message."""
+        self._log("error", message, data)
+
+
+def create_logger(verbose: bool) -> Logger:
+    """Create a logger instance.
+
+    Args:
+        verbose: Whether to enable verbose logging
+
+    Returns:
+        Logger instance
+    """
+    return SimpleLogger(verbose)
+
+
+def get_last_user_message(
+    messages: list[ChatCompletionMessageParam],
+) -> str:
+    """
+    Extract the last user message from an array of chat completion messages.
+
+    Searches through the messages array in reverse order to find the most recent
+    message with role "user" and returns its content as a string.
+
+    Args:
+        messages: Array of chat completion message parameters
+
+    Returns:
+        The content of the last user message, or empty string if none found
+
+    Example:
+        ```python
+        messages = [
+            {"role": "system", "content": "You are a helpful assistant."},
+            {"role": "user", "content": "Hello there!"},
+            {"role": "assistant", "content": "Hi! How can I help you?"},
+            {"role": "user", "content": "What's the weather like?"}
+        ]
+
+        last_message = get_last_user_message(messages)
+        # Returns: "What's the weather like?"
+        ```
+    """
+    for message in reversed(messages):
+        if message.get("role") == "user":
+            content = message.get("content", "")
+            if isinstance(content, str):
+                return content
+            elif isinstance(content, list):
+                # Handle content that is an array of content parts
+                text_parts = []
+                for part in content:
+                    if isinstance(part, dict) and part.get("type") == "text":
+                        text_parts.append(part.get("text", ""))
+                    elif isinstance(part, str):
+                        text_parts.append(part)
+                return " ".join(text_parts)
+    return ""
+
+
+def get_conversation_content(
+    messages: list[ChatCompletionMessageParam],
+) -> str:
+    """
+    Convert an array of chat completion messages into a formatted conversation string.
+
+    Transforms the messages array into a readable conversation format where each
+    message is prefixed with its role (User/Assistant) and messages are separated
+    by double newlines.
+
+    Args:
+        messages: Array of chat completion message parameters
+
+    Returns:
+        Formatted conversation string with role prefixes
+
+    Example:
+        ```python
+        messages = [
+            {"role": "user", "content": "Hello!"},
+            {"role": "assistant", "content": "Hi there!"},
+            {"role": "user", "content": "How are you?"}
+        ]
+
+        conversation = get_conversation_content(messages)
+        # Returns: "User: Hello!\n\nAssistant: Hi there!\n\nUser: How are you?"
+        ```
+    """
+    conversation_parts = []
+
+    for message in messages:
+        role = message.get("role", "")
+        content = message.get("content", "")
+
+        # Format role
+        if role == "user":
+            role_display = "User"
+        elif role == "assistant":
+            role_display = "Assistant"
+        elif role == "system":
+            role_display = "System"
+        else:
+            role_display = role.capitalize()
+
+        # Extract content text
+        if isinstance(content, str):
+            content_text = content
+        elif isinstance(content, list):
+            # Handle content that is an array of content parts
+            text_parts = []
+            for part in content:
+                if isinstance(part, dict) and part.get("type") == "text":
+                    text_parts.append(part.get("text", ""))
+                elif isinstance(part, str):
+                    text_parts.append(part)
+            content_text = " ".join(text_parts)
+        else:
+            content_text = str(content)
+
+        if content_text:
+            conversation_parts.append(f"{role_display}: {content_text}")
+
+    return "\n\n".join(conversation_parts)
+
+
+def convert_profile_to_markdown(data: dict[str, Any]) -> str:
+    """
+    Convert profile data to markdown based on profile.static and profile.dynamic properties.
+
+    Args:
+        data: Profile structure data
+
+    Returns:
+        Markdown string
+
+    Example:
+        ```python
+        data = {
+            "profile": {
+                "static": ["User prefers Python", "Lives in San Francisco"],
+                "dynamic": ["Recently asked about AI"]
+            },
+            "searchResults": {
+                "results": [{"memory": "Likes coffee"}]
+            }
+        }
+
+        markdown = convert_profile_to_markdown(data)
+        # Returns formatted markdown with sections
+        ```
+    """
+    sections = []
+
+    profile = data.get("profile", {})
+    static_memories = profile.get("static", [])
+    dynamic_memories = profile.get("dynamic", [])
+
+    if static_memories:
+        sections.append("## Static Profile")
+        sections.append("\n".join(f"- {item}" for item in static_memories))
+
+    if dynamic_memories:
+        sections.append("## Dynamic Profile")
+        sections.append("\n".join(f"- {item}" for item in dynamic_memories))
+
+    return "\n\n".join(sections)