feat: add user manager factory pattern and product API enhancements (#166)

* feat: add user manager factory pattern and product API enhancements

- Add user manager factory pattern with SQLite and MySQL backends
- Add user manager configuration to MOSConfig
- Add product API router and configuration
- Add DingDing notification integration
- Add notification service utilities
- Update OpenAPI documentation

* fix: change py to meos_tools

* fix: code reorganize

Author: fridayL

docs/openapi.json

@@ -884,7 +884,7 @@
             "type": "string",
             "title": "Session Id",
             "description": "Session ID for the MOS. This is used to distinguish between different dialogue",
-            "default": "a47d75a0-5ee8-473f-86c4-3f09073fd59f"
+            "default": "842877f4-c3f7-4c22-ad38-5950026870fe"
           },
           "chat_model": {
             "$ref": "#/components/schemas/LLMConfigFactory",
@@ -905,6 +905,10 @@
             ],
             "description": "Memory scheduler configuration for managing memory operations"
           },
+          "user_manager": {
+            "$ref": "#/components/schemas/UserManagerConfigFactory",
+            "description": "User manager configuration for database operations"
+          },
           "max_turns_window": {
             "type": "integer",
             "title": "Max Turns Window",
@@ -1370,6 +1374,25 @@
         "title": "UserListResponse",
         "description": "Response model for user list operations."
       },
+      "UserManagerConfigFactory": {
+        "properties": {
+          "backend": {
+            "type": "string",
+            "title": "Backend",
+            "description": "Backend for user manager",
+            "default": "sqlite"
+          },
+          "config": {
+            "additionalProperties": true,
+            "type": "object",
+            "title": "Config",
+            "description": "Configuration for the user manager backend"
+          }
+        },
+        "type": "object",
+        "title": "UserManagerConfigFactory",
+        "description": "Factory for user manager configurations."
+      },
       "UserResponse": {
         "properties": {
           "code": {

examples/mem_user/user_manager_factory_example.py

@@ -0,0 +1,111 @@
+"""Example demonstrating the use of UserManagerFactory with different backends."""
+
+from memos.configs.mem_user import UserManagerConfigFactory
+from memos.mem_user.factory import UserManagerFactory
+from memos.mem_user.persistent_factory import PersistentUserManagerFactory
+
+
+def example_sqlite_default():
+    """Example: Create SQLite user manager with default settings."""
+    print("=== SQLite Default Example ===")
+
+    # Method 1: Using factory with minimal config
+    user_manager = UserManagerFactory.create_sqlite()
+
+    # Method 2: Using config factory (equivalent)
+    UserManagerConfigFactory(
+        backend="sqlite",
+        config={},  # Uses all defaults
+    )
+
+    print(f"Created user manager: {type(user_manager).__name__}")
+    print(f"Database path: {user_manager.db_path}")
+
+    # Test basic operations
+    users = user_manager.list_users()
+    print(f"Initial users: {[user.user_name for user in users]}")
+
+    user_manager.close()
+
+
+def example_sqlite_custom():
+    """Example: Create SQLite user manager with custom settings."""
+    print("\n=== SQLite Custom Example ===")
+
+    config_factory = UserManagerConfigFactory(
+        backend="sqlite", config={"db_path": "/tmp/custom_memos.db", "user_id": "admin"}
+    )
+
+    user_manager = UserManagerFactory.from_config(config_factory)
+    print(f"Created user manager: {type(user_manager).__name__}")
+    print(f"Database path: {user_manager.db_path}")
+
+    # Test operations
+    user_id = user_manager.create_user("test_user")
+    print(f"Created user: {user_id}")
+
+    user_manager.close()
+
+
+def example_mysql():
+    """Example: Create MySQL user manager."""
+    print("\n=== MySQL Example ===")
+
+    # Method 1: Using factory with parameters
+    try:
+        user_manager = UserManagerFactory.create_mysql(
+            host="localhost",
+            port=3306,
+            username="root",
+            password="your_password",  # Replace with actual password
+            database="test_memos_users",
+        )
+
+        print(f"Created user manager: {type(user_manager).__name__}")
+        print(f"Connection URL: {user_manager.connection_url}")
+
+        # Test operations
+        users = user_manager.list_users()
+        print(f"Users: {[user.user_name for user in users]}")
+
+        user_manager.close()
+
+    except Exception as e:
+        print(f"MySQL connection failed (expected if not set up): {e}")
+
+
+def example_persistent_managers():
+    """Example: Create persistent user managers with configuration storage."""
+    print("\n=== Persistent User Manager Examples ===")
+
+    # SQLite persistent manager
+    config_factory = UserManagerConfigFactory(backend="sqlite", config={})
+
+    persistent_manager = PersistentUserManagerFactory.from_config(config_factory)
+    print(f"Created persistent manager: {type(persistent_manager).__name__}")
+
+    # Test config operations
+    from memos.configs.mem_os import MOSConfig
+
+    # Create a sample config (you might need to adjust this based on MOSConfig structure)
+    try:
+        # This is a simplified example - adjust based on actual MOSConfig requirements
+        sample_config = MOSConfig()  # Use default config
+
+        # Save user config
+        success = persistent_manager.save_user_config("test_user", sample_config)
+        print(f"Config saved: {success}")
+
+        # Retrieve user config
+        retrieved_config = persistent_manager.get_user_config("test_user")
+        print(f"Config retrieved: {retrieved_config is not None}")
+
+    except Exception as e:
+        print(f"Config operations failed: {e}")
+
+    persistent_manager.close()
+
+
+if __name__ == "__main__":
+    # Run all examples
+    example_sqlite_default()

src/memos/api/config.py

@@ -254,6 +254,34 @@ def is_default_cube_config_enabled() -> bool:
         """Check if default cube config is enabled via environment variable."""
         return os.getenv("MOS_ENABLE_DEFAULT_CUBE_CONFIG", "false").lower() == "true"
 
+    @staticmethod
+    def is_dingding_bot_enabled() -> bool:
+        """Check if DingDing bot is enabled via environment variable."""
+        return os.getenv("ENABLE_DINGDING_BOT", "false").lower() == "true"
+
+    @staticmethod
+    def get_dingding_bot_config() -> dict[str, Any] | None:
+        """Get DingDing bot configuration if enabled."""
+        if not APIConfig.is_dingding_bot_enabled():
+            return None
+
+        return {
+            "enabled": True,
+            "access_token_user": os.getenv("DINGDING_ACCESS_TOKEN_USER", ""),
+            "secret_user": os.getenv("DINGDING_SECRET_USER", ""),
+            "access_token_error": os.getenv("DINGDING_ACCESS_TOKEN_ERROR", ""),
+            "secret_error": os.getenv("DINGDING_SECRET_ERROR", ""),
+            "robot_code": os.getenv("DINGDING_ROBOT_CODE", ""),
+            "app_key": os.getenv("DINGDING_APP_KEY", ""),
+            "app_secret": os.getenv("DINGDING_APP_SECRET", ""),
+            "oss_endpoint": os.getenv("OSS_ENDPOINT", ""),
+            "oss_region": os.getenv("OSS_REGION", ""),
+            "oss_bucket_name": os.getenv("OSS_BUCKET_NAME", ""),
+            "oss_access_key_id": os.getenv("OSS_ACCESS_KEY_ID", ""),
+            "oss_access_key_secret": os.getenv("OSS_ACCESS_KEY_SECRET", ""),
+            "oss_public_base_url": os.getenv("OSS_PUBLIC_BASE_URL", ""),
+        }
+
     @staticmethod
     def get_product_default_config() -> dict[str, Any]:
         """Get default configuration for Product API."""
@@ -431,10 +459,6 @@ def create_user_config(user_name: str, user_id: str) -> tuple[MOSConfig, General
             )
         else:
             raise ValueError(f"Invalid Neo4j backend: {graph_db_backend}")
-        if os.getenv("ENABLE_INTERNET", "false").lower() == "true":
-            default_cube_config.text_mem.config["internet_retriever"] = (
-                APIConfig.get_internet_config()
-            )
         default_mem_cube = GeneralMemCube(default_cube_config)
         return default_config, default_mem_cube
 

src/memos/api/routers/product_router.py

@@ -26,6 +26,7 @@
 )
 from memos.configs.mem_os import MOSConfig
 from memos.mem_os.product import MOSProduct
+from memos.memos_tools.notification_service import get_error_bot_function, get_online_bot_function
 
 
 logger = logging.getLogger(__name__)
@@ -49,8 +50,17 @@ def get_mos_product_instance():
         # Get default cube config from APIConfig (may be None if disabled)
         default_cube_config = APIConfig.get_default_cube_config()
         logger.info(f"*********initdefault_cube_config******** {default_cube_config}")
+
+        # Get DingDing bot functions
+        dingding_enabled = APIConfig.is_dingding_bot_enabled()
+        online_bot = get_online_bot_function() if dingding_enabled else None
+        error_bot = get_error_bot_function() if dingding_enabled else None
+
         MOS_PRODUCT_INSTANCE = MOSProduct(
-            default_config=mos_config, default_cube_config=default_cube_config
+            default_config=mos_config,
+            default_cube_config=default_cube_config,
+            online_bot=online_bot,
+            error_bot=error_bot,
         )
         logger.info("MOSProduct instance created successfully with inheritance architecture")
     return MOS_PRODUCT_INSTANCE
@@ -60,15 +70,15 @@ def get_mos_product_instance():
 
 
 @router.post("/configure", summary="Configure MOSProduct", response_model=SimpleResponse)
-async def set_config(config):
+def set_config(config):
     """Set MOSProduct configuration."""
     global MOS_PRODUCT_INSTANCE
     MOS_PRODUCT_INSTANCE = MOSProduct(default_config=config)
     return SimpleResponse(message="Configuration set successfully")
 
 
 @router.post("/users/register", summary="Register a new user", response_model=UserRegisterResponse)
-async def register_user(user_req: UserRegisterRequest, g: Annotated[G, Depends(get_g_object)]):
+def register_user(user_req: UserRegisterRequest, g: Annotated[G, Depends(get_g_object)]):
     """Register a new user with configuration and default cube."""
     try:
         # Set request-related information in g object
@@ -113,7 +123,7 @@ async def register_user(user_req: UserRegisterRequest, g: Annotated[G, Depends(g
 @router.get(
     "/suggestions/{user_id}", summary="Get suggestion queries", response_model=SuggestionResponse
 )
-async def get_suggestion_queries(user_id: str):
+def get_suggestion_queries(user_id: str):
     """Get suggestion queries for a specific user."""
     try:
         mos_product = get_mos_product_instance()
@@ -133,7 +143,7 @@ async def get_suggestion_queries(user_id: str):
     summary="Get suggestion queries with language",
     response_model=SuggestionResponse,
 )
-async def get_suggestion_queries_post(suggestion_req: SuggestionRequest):
+def get_suggestion_queries_post(suggestion_req: SuggestionRequest):
     """Get suggestion queries for a specific user with language preference."""
     try:
         mos_product = get_mos_product_instance()
@@ -151,7 +161,7 @@ async def get_suggestion_queries_post(suggestion_req: SuggestionRequest):
 
 
 @router.post("/get_all", summary="Get all memories for user", response_model=MemoryResponse)
-async def get_all_memories(memory_req: GetMemoryRequest):
+def get_all_memories(memory_req: GetMemoryRequest):
     """Get all memories for a specific user."""
     try:
         mos_product = get_mos_product_instance()
@@ -178,7 +188,7 @@ async def get_all_memories(memory_req: GetMemoryRequest):
 
 
 @router.post("/add", summary="add a new memory", response_model=SimpleResponse)
-async def create_memory(memory_req: MemoryCreateRequest):
+def create_memory(memory_req: MemoryCreateRequest):
     """Create a new memory for a specific user."""
     try:
         mos_product = get_mos_product_instance()
@@ -199,7 +209,7 @@ async def create_memory(memory_req: MemoryCreateRequest):
 
 
 @router.post("/search", summary="Search memories", response_model=SearchResponse)
-async def search_memories(search_req: SearchRequest):
+def search_memories(search_req: SearchRequest):
     """Search memories for a specific user."""
     try:
         mos_product = get_mos_product_instance()
@@ -219,37 +229,38 @@ async def search_memories(search_req: SearchRequest):
 
 
 @router.post("/chat", summary="Chat with MemOS")
-async def chat(chat_req: ChatRequest):
+def chat(chat_req: ChatRequest):
     """Chat with MemOS for a specific user. Returns SSE stream."""
     try:
         mos_product = get_mos_product_instance()
 
-        async def generate_chat_response():
+        def generate_chat_response():
             """Generate chat response as SSE stream."""
             try:
-                import asyncio
-
-                for chunk in mos_product.chat_with_references(
+                # Directly yield from the generator without async wrapper
+                yield from mos_product.chat_with_references(
                     query=chat_req.query,
                     user_id=chat_req.user_id,
                     cube_id=chat_req.mem_cube_id,
                     history=chat_req.history,
                     internet_search=chat_req.internet_search,
-                ):
-                    yield chunk
-                    await asyncio.sleep(0.00001)  # 50ms delay between chunks
+                )
+
             except Exception as e:
                 logger.error(f"Error in chat stream: {e}")
                 error_data = f"data: {json.dumps({'type': 'error', 'content': str(traceback.format_exc())})}\n\n"
                 yield error_data
 
         return StreamingResponse(
             generate_chat_response(),
-            media_type="text/plain",
+            media_type="text/event-stream",
             headers={
                 "Cache-Control": "no-cache",
                 "Connection": "keep-alive",
                 "Content-Type": "text/event-stream",
+                "Access-Control-Allow-Origin": "*",
+                "Access-Control-Allow-Headers": "*",
+                "Access-Control-Allow-Methods": "*",
             },
         )
 
@@ -261,7 +272,7 @@ async def generate_chat_response():
 
 
 @router.get("/users", summary="List all users", response_model=BaseResponse[list])
-async def list_users():
+def list_users():
     """List all registered users."""
     try:
         mos_product = get_mos_product_instance()
@@ -289,7 +300,7 @@ async def get_user_info(user_id: str):
 @router.get(
     "/configure/{user_id}", summary="Get MOSProduct configuration", response_model=SimpleResponse
 )
-async def get_config(user_id: str):
+def get_config(user_id: str):
     """Get MOSProduct configuration."""
     global MOS_PRODUCT_INSTANCE
     config = MOS_PRODUCT_INSTANCE.default_config
@@ -299,7 +310,7 @@ async def get_config(user_id: str):
 @router.get(
     "/users/{user_id}/config", summary="Get user configuration", response_model=BaseResponse[dict]
 )
-async def get_user_config(user_id: str):
+def get_user_config(user_id: str):
     """Get user-specific configuration."""
     try:
         mos_product = get_mos_product_instance()
@@ -323,7 +334,7 @@ async def get_user_config(user_id: str):
 @router.put(
     "/users/{user_id}/config", summary="Update user configuration", response_model=SimpleResponse
 )
-async def update_user_config(user_id: str, config_data: dict):
+def update_user_config(user_id: str, config_data: dict):
     """Update user-specific configuration."""
     try:
         mos_product = get_mos_product_instance()
@@ -348,7 +359,7 @@ async def update_user_config(user_id: str, config_data: dict):
 @router.get(
     "/instances/status", summary="Get user configuration status", response_model=BaseResponse[dict]
 )
-async def get_instance_status():
+def get_instance_status():
     """Get information about active user configurations in memory."""
     try:
         mos_product = get_mos_product_instance()
@@ -362,7 +373,7 @@ async def get_instance_status():
 
 
 @router.get("/instances/count", summary="Get active user count", response_model=BaseResponse[int])
-async def get_active_user_count():
+def get_active_user_count():
     """Get the number of active user configurations in memory."""
     try:
         mos_product = get_mos_product_instance()