supermemory/packages/agent-framework-python/tests/test_middleware.py
Dhravya 984297b62d
Add Supermemory integration for Microsoft Agent Framework (#775)
## Summary

This PR introduces comprehensive Supermemory integration for the Microsoft Agent Framework, providing three complementary approaches to add persistent memory capabilities to agents: middleware for automatic memory injection, context providers for session-based memory management, and tools for explicit memory operations.

## Key Changes

- **SupermemoryChatMiddleware**: Automatic memory injection middleware that fetches relevant memories from Supermemory before LLM calls and optionally saves conversations. Supports three modes:
  - `"profile"`: Injects all static and dynamic profile memories
  - `"query"`: Searches for memories relevant to the current user message
  - `"full"`: Combines both profile and query modes

- **SupermemoryContextProvider**: Idiomatic context provider following the Agent Framework pattern (similar to built-in Mem0 integration). Integrates with the session pipeline via `before_run()` and `after_run()` hooks for automatic memory retrieval and storage.

- **SupermemoryTools**: FunctionTool-compatible tools that agents can use for explicit memory operations:
  - `search_memories()`: Search for specific memories
  - `add_memory()`: Add new memories
  - `get_profile()`: Retrieve user profile

- **Utility Functions**: Helper functions for:
  - Memory deduplication across static, dynamic, and search result sources
  - Profile-to-markdown conversion for LLM consumption
  - Message extraction and conversation formatting
  - Logging with configurable verbosity

- **Exception Hierarchy**: Custom exceptions for better error handling:
  - `SupermemoryConfigurationError`: Missing/invalid configuration
  - `SupermemoryAPIError`: API request failures
  - `SupermemoryNetworkError`: Network connectivity issues
  - `SupermemoryMemoryOperationError`: Memory operation failures

- **Comprehensive Documentation**: README with quick start examples, configuration options, and API reference for all three integration approaches.

- **Test Suite**: Unit tests covering middleware, context provider, tools, and utility functions with proper mocking and error scenarios.

## Implementation Details

- Supports both async (aiohttp) and sync (requests) HTTP clients with automatic fallback
- Handles multiple message formats (dict, objects with attributes, content arrays)
- Configurable memory storage with optional conversation grouping via `conversation_id`
- Environment variable fallback for API key configuration (`SUPERMEMORY_API_KEY`)
- Background task management for non-blocking memory operations in middleware
- Proper async/sync compatibility for the Supermemory SDK

https://claude.ai/code/session_012idB5y6UGK3zmeFULgTc4z
2026-03-10 01:49:45 +00:00

113 lines
3.8 KiB
Python

"""Tests for Supermemory middleware."""
import pytest
from supermemory_agent_framework import (
AgentSupermemory,
SupermemoryChatMiddleware,
SupermemoryMiddlewareOptions,
)
from supermemory_agent_framework.middleware import (
_get_last_user_message,
_get_conversation_content,
)
def _make_conn(**kwargs):
kwargs.setdefault("api_key", "test-key")
kwargs.setdefault("container_tag", "user-123")
return AgentSupermemory(**kwargs)
class TestGetLastUserMessage:
def test_dict_messages(self) -> None:
messages = [
{"role": "system", "content": "You are helpful."},
{"role": "user", "content": "Hello!"},
{"role": "assistant", "content": "Hi there!"},
{"role": "user", "content": "How are you?"},
]
assert _get_last_user_message(messages) == "How are you?"
def test_no_user_message(self) -> None:
messages = [
{"role": "system", "content": "You are helpful."},
{"role": "assistant", "content": "Hi!"},
]
assert _get_last_user_message(messages) == ""
def test_empty_messages(self) -> None:
assert _get_last_user_message([]) == ""
assert _get_last_user_message(None) == ""
def test_content_parts(self) -> None:
messages = [
{
"role": "user",
"content": [
{"type": "text", "text": "Hello"},
{"type": "text", "text": "world"},
],
}
]
assert _get_last_user_message(messages) == "Hello world"
class TestGetConversationContent:
def test_basic_conversation(self) -> None:
messages = [
{"role": "user", "content": "Hello!"},
{"role": "assistant", "content": "Hi there!"},
{"role": "user", "content": "How are you?"},
]
result = _get_conversation_content(messages)
assert "User: Hello!" in result
assert "Assistant: Hi there!" in result
assert "User: How are you?" in result
class TestMiddlewareOptions:
def test_defaults(self) -> None:
options = SupermemoryMiddlewareOptions()
assert options.verbose is False
assert options.mode == "profile"
assert options.add_memory == "never"
def test_custom_options(self) -> None:
options = SupermemoryMiddlewareOptions(
verbose=True,
mode="full",
add_memory="always",
)
assert options.verbose is True
assert options.mode == "full"
assert options.add_memory == "always"
class TestMiddlewareConfiguration:
def test_accepts_connection(self) -> None:
conn = _make_conn()
middleware = SupermemoryChatMiddleware(conn)
assert middleware._container_tag == "user-123"
def test_uses_connection_client(self) -> None:
conn = _make_conn()
middleware = SupermemoryChatMiddleware(conn)
assert middleware._supermemory_client is conn.client
def test_conversation_id_from_connection(self) -> None:
conn = _make_conn(conversation_id="conv-abc")
middleware = SupermemoryChatMiddleware(conn)
assert middleware._connection.conversation_id == "conv-abc"
assert middleware._connection.custom_id == "conversation_conv-abc"
def test_auto_generated_conversation_id(self) -> None:
conn = _make_conn()
middleware = SupermemoryChatMiddleware(conn)
assert middleware._connection.conversation_id is not None
assert len(middleware._connection.conversation_id) > 0
def test_entity_context_from_connection(self) -> None:
conn = _make_conn(entity_context="User is a Python developer")
middleware = SupermemoryChatMiddleware(conn)
assert middleware._connection.entity_context == "User is a Python developer"