Simplify custom endpoint management system

- Remove unnecessary endpoint_id field (use model_id + provider_type as natural unique key)
- Remove display_name, created_at, updated_at metadata fields (not needed for server operations)
- Remove endpoint_url from Bedrock/OpenAI/Gemini (they use standard APIs)
- Keep endpoint_url only for CAII and OpenAI Compatible (custom deployments)
- Update API endpoints to use /model_id/provider_type pattern
- Simplify JSON storage structure to use model_id:provider_type keys
- Test confirmed: server starts and endpoints work correctly

Author: Khauneesh-AI

app/core/custom_endpoint_manager.py

@@ -13,7 +13,7 @@
 
 
 class CustomEndpointManager:
-    """Manager for custom model endpoints"""
+    """Manager for custom model endpoints - simplified"""
 
     def __init__(self, config_file: str = "custom_model_endpoints.json"):
         """
@@ -30,9 +30,7 @@ def _ensure_config_file_exists(self):
         if not self.config_file.exists():
             initial_config = {
                 "version": "1.0",
-                "endpoints": {},
-                "created_at": datetime.now(timezone.utc).isoformat(),
-                "last_updated": datetime.now(timezone.utc).isoformat()
+                "endpoints": {}
             }
             with open(self.config_file, 'w') as f:
                 json.dump(initial_config, f, indent=2)
@@ -48,7 +46,6 @@ def _load_config(self) -> Dict[str, Any]:
     def _save_config(self, config: Dict[str, Any]):
         """Save configuration to file"""
         try:
-            config["last_updated"] = datetime.now(timezone.utc).isoformat()
             with open(self.config_file, 'w') as f:
                 json.dump(config, f, indent=2)
         except Exception as e:
@@ -62,40 +59,40 @@ def add_endpoint(self, endpoint: CustomEndpoint) -> str:
             endpoint: Custom endpoint configuration
             
         Returns:
-            endpoint_id: The ID of the added endpoint
+            unique_key: The unique key for the added endpoint (model_id:provider_type)
             
         Raises:
             APIError: If endpoint already exists or validation fails
         """
         config = self._load_config()
 
-        # Check if endpoint_id already exists
-        if endpoint.endpoint_id in config["endpoints"]:
-            raise APIError(f"Endpoint with ID '{endpoint.endpoint_id}' already exists", 400)
+        # Use model_id + provider_type as unique key
+        unique_key = f"{endpoint.model_id}:{endpoint.provider_type}"
 
-        # Add timestamps
-        now = datetime.now(timezone.utc).isoformat()
-        endpoint.created_at = now
-        endpoint.updated_at = now
+        # Check if endpoint already exists
+        if unique_key in config["endpoints"]:
+            raise APIError(f"Endpoint for model '{endpoint.model_id}' with provider '{endpoint.provider_type}' already exists", 400)
 
         # Store endpoint configuration
-        config["endpoints"][endpoint.endpoint_id] = endpoint.model_dump()
+        config["endpoints"][unique_key] = endpoint.model_dump()
 
         self._save_config(config)
-        return endpoint.endpoint_id
+        return unique_key
 
-    def get_endpoint(self, endpoint_id: str) -> Optional[CustomEndpoint]:
+    def get_endpoint(self, model_id: str, provider_type: str) -> Optional[CustomEndpoint]:
         """
-        Get a specific custom endpoint by ID
+        Get a specific custom endpoint by model_id and provider_type
         
         Args:
-            endpoint_id: The endpoint ID to retrieve
+            model_id: The model identifier
+            provider_type: The provider type
             
         Returns:
             CustomEndpoint or None if not found
         """
         config = self._load_config()
-        endpoint_data = config["endpoints"].get(endpoint_id)
+        unique_key = f"{model_id}:{provider_type}"
+        endpoint_data = config["endpoints"].get(unique_key)
 
         if not endpoint_data:
             return None
@@ -112,12 +109,12 @@ def get_all_endpoints(self) -> List[CustomEndpoint]:
         config = self._load_config()
         endpoints = []
 
-        for endpoint_data in config["endpoints"].values():
+        for unique_key, endpoint_data in config["endpoints"].items():
             try:
                 endpoint = self._parse_endpoint(endpoint_data)
                 endpoints.append(endpoint)
             except Exception as e:
-                print(f"Warning: Failed to parse endpoint {endpoint_data.get('endpoint_id', 'unknown')}: {e}")
+                print(f"Warning: Failed to parse endpoint {unique_key}: {e}")
                 continue
 
         return endpoints
@@ -135,12 +132,13 @@ def get_endpoints_by_provider(self, provider_type: str) -> List[CustomEndpoint]:
         all_endpoints = self.get_all_endpoints()
         return [ep for ep in all_endpoints if ep.provider_type == provider_type]
 
-    def update_endpoint(self, endpoint_id: str, updated_endpoint: CustomEndpoint) -> bool:
+    def update_endpoint(self, model_id: str, provider_type: str, updated_endpoint: CustomEndpoint) -> bool:
         """
         Update an existing custom endpoint
         
         Args:
-            endpoint_id: The endpoint ID to update
+            model_id: The model identifier
+            provider_type: The provider type
             updated_endpoint: Updated endpoint configuration
             
         Returns:
@@ -151,39 +149,36 @@ def update_endpoint(self, endpoint_id: str, updated_endpoint: CustomEndpoint) ->
         """
         config = self._load_config()
 
-        if endpoint_id not in config["endpoints"]:
-            return False
-        
-        # Preserve original created_at timestamp
-        original_created_at = config["endpoints"][endpoint_id].get("created_at")
+        unique_key = f"{model_id}:{provider_type}"
 
-        # Update timestamps
-        updated_endpoint.created_at = original_created_at
-        updated_endpoint.updated_at = datetime.now(timezone.utc).isoformat()
-        updated_endpoint.endpoint_id = endpoint_id  # Ensure ID consistency
+        if unique_key not in config["endpoints"]:
+            return False
 
         # Update endpoint configuration
-        config["endpoints"][endpoint_id] = updated_endpoint.model_dump()
+        config["endpoints"][unique_key] = updated_endpoint.model_dump()
 
         self._save_config(config)
         return True
 
-    def delete_endpoint(self, endpoint_id: str) -> bool:
+    def delete_endpoint(self, model_id: str, provider_type: str) -> bool:
         """
         Delete a custom endpoint
         
         Args:
-            endpoint_id: The endpoint ID to delete
+            model_id: The model identifier
+            provider_type: The provider type
             
         Returns:
             True if deleted successfully, False if endpoint not found
         """
         config = self._load_config()
 
-        if endpoint_id not in config["endpoints"]:
+        unique_key = f"{model_id}:{provider_type}"
+        
+        if unique_key not in config["endpoints"]:
             return False
 
-        del config["endpoints"][endpoint_id]
+        del config["endpoints"][unique_key]
         self._save_config(config)
         return True
 
@@ -218,23 +213,6 @@ def _parse_endpoint(self, endpoint_data: Dict[str, Any]) -> CustomEndpoint:
         except Exception as e:
             raise APIError(f"Failed to parse endpoint configuration: {str(e)}", 500)
 
-    def validate_endpoint_id(self, endpoint_id: str) -> bool:
-        """
-        Validate endpoint ID format
-        
-        Args:
-            endpoint_id: The endpoint ID to validate
-            
-        Returns:
-            True if valid, False otherwise
-        """
-        if not endpoint_id or not isinstance(endpoint_id, str):
-            return False
-        
-        # Allow alphanumeric, hyphens, and underscores
-        import re
-        return bool(re.match(r'^[a-zA-Z0-9_-]+$', endpoint_id))
-    
     def get_endpoint_stats(self) -> Dict[str, Any]:
         """
         Get statistics about custom endpoints
@@ -245,12 +223,14 @@ def get_endpoint_stats(self) -> Dict[str, Any]:
         endpoints = self.get_all_endpoints()
 
         provider_counts = {}
+        model_list = []
         for endpoint in endpoints:
             provider_type = endpoint.provider_type
             provider_counts[provider_type] = provider_counts.get(provider_type, 0) + 1
+            model_list.append(f"{endpoint.model_id} ({endpoint.provider_type})")
 
         return {
             "total_endpoints": len(endpoints),
             "provider_counts": provider_counts,
-            "endpoint_ids": [ep.endpoint_id for ep in endpoints]
+            "models": model_list
         }

app/core/model_endpoints.py

@@ -388,17 +388,13 @@ async def collect_model_catalog() -> Dict[str, Dict[str, List[str]]]:
                 catalog[provider_key]["enabled"].append({
                     "model": endpoint.model_id,
                     "endpoint": getattr(endpoint, 'endpoint_url', ''),
-                    "custom": True,
-                    "endpoint_id": endpoint.endpoint_id,
-                    "display_name": endpoint.display_name
+                    "custom": True
                 })
             else:
                 # Other providers: just the model name with custom metadata
                 catalog[provider_key]["enabled"].append({
                     "model": endpoint.model_id,
                     "custom": True,
-                    "endpoint_id": endpoint.endpoint_id,
-                    "display_name": endpoint.display_name,
                     "provider_type": endpoint.provider_type
                 })
 

app/core/model_handlers.py

@@ -33,14 +33,8 @@ def get_custom_endpoint_config(model_id: str, provider_type: str):
         from app.core.custom_endpoint_manager import CustomEndpointManager
 
         custom_manager = CustomEndpointManager()
-        custom_endpoints = custom_manager.get_endpoints_by_provider(provider_type)
-        
-        # Find endpoint matching the model_id
-        for endpoint in custom_endpoints:
-            if endpoint.model_id == model_id:
-                return endpoint
+        return custom_manager.get_endpoint(model_id, provider_type)
 
-        return None
     except Exception as e:
         print(f"Warning: Failed to get custom endpoint config: {e}")
         return None

app/main.py

@@ -1488,19 +1488,14 @@ async def perform_upgrade():
 async def add_custom_model_endpoint(request: AddCustomEndpointRequest):
     """Add a new custom model endpoint"""
     try:
-        # Validate endpoint ID format
-        if not custom_endpoint_manager.validate_endpoint_id(request.endpoint_config.endpoint_id):
-            raise HTTPException(
-                status_code=400,
-                detail="Invalid endpoint ID. Use only alphanumeric characters, hyphens, and underscores."
-            )
-        
-        endpoint_id = custom_endpoint_manager.add_endpoint(request.endpoint_config)
+        unique_key = custom_endpoint_manager.add_endpoint(request.endpoint_config)
 
         return {
             "status": "success",
-            "message": f"Custom endpoint '{endpoint_id}' added successfully",
-            "endpoint_id": endpoint_id
+            "message": f"Custom endpoint for '{request.endpoint_config.model_id}' ({request.endpoint_config.provider_type}) added successfully",
+            "model_id": request.endpoint_config.model_id,
+            "provider_type": request.endpoint_config.provider_type,
+            "unique_key": unique_key
         }
 
     except APIError as e:
@@ -1530,17 +1525,17 @@ async def list_custom_model_endpoints(provider_type: Optional[str] = None):
         raise HTTPException(status_code=500, detail=f"Failed to list custom endpoints: {str(e)}")
 
 
-@app.get("/custom_model_endpoints/{endpoint_id}", include_in_schema=True, responses=responses,
+@app.get("/custom_model_endpoints/{model_id}/{provider_type}", include_in_schema=True, responses=responses,
          description="Get a specific custom model endpoint")
-async def get_custom_model_endpoint(endpoint_id: str):
+async def get_custom_model_endpoint(model_id: str, provider_type: str):
     """Get details of a specific custom model endpoint"""
     try:
-        endpoint = custom_endpoint_manager.get_endpoint(endpoint_id)
+        endpoint = custom_endpoint_manager.get_endpoint(model_id, provider_type)
 
         if not endpoint:
             raise HTTPException(
                 status_code=404,
-                detail=f"Custom endpoint '{endpoint_id}' not found"
+                detail=f"Custom endpoint for model '{model_id}' with provider '{provider_type}' not found"
             )
 
         return {
@@ -1554,29 +1549,29 @@ async def get_custom_model_endpoint(endpoint_id: str):
         raise HTTPException(status_code=500, detail=f"Failed to get custom endpoint: {str(e)}")
 
 
-@app.put("/custom_model_endpoints/{endpoint_id}", include_in_schema=True, responses=responses,
+@app.put("/custom_model_endpoints/{model_id}/{provider_type}", include_in_schema=True, responses=responses,
          description="Update a custom model endpoint")
-async def update_custom_model_endpoint(endpoint_id: str, request: AddCustomEndpointRequest):
+async def update_custom_model_endpoint(model_id: str, provider_type: str, request: AddCustomEndpointRequest):
     """Update an existing custom model endpoint"""
     try:
-        # Ensure the endpoint ID in the request matches the URL parameter
-        if request.endpoint_config.endpoint_id != endpoint_id:
+        # Ensure the model_id and provider_type in the request match the URL parameters
+        if request.endpoint_config.model_id != model_id or request.endpoint_config.provider_type != provider_type:
             raise HTTPException(
                 status_code=400,
-                detail="Endpoint ID in request body must match the URL parameter"
+                detail="Model ID and provider type in request body must match the URL parameters"
             )
 
-        success = custom_endpoint_manager.update_endpoint(endpoint_id, request.endpoint_config)
+        success = custom_endpoint_manager.update_endpoint(model_id, provider_type, request.endpoint_config)
 
         if not success:
             raise HTTPException(
                 status_code=404,
-                detail=f"Custom endpoint '{endpoint_id}' not found"
+                detail=f"Custom endpoint for model '{model_id}' with provider '{provider_type}' not found"
             )
 
         return {
             "status": "success",
-            "message": f"Custom endpoint '{endpoint_id}' updated successfully"
+            "message": f"Custom endpoint for '{model_id}' ({provider_type}) updated successfully"
         }
 
     except APIError as e:
@@ -1585,22 +1580,22 @@ async def update_custom_model_endpoint(endpoint_id: str, request: AddCustomEndpo
         raise HTTPException(status_code=500, detail=f"Failed to update custom endpoint: {str(e)}")
 
 
-@app.delete("/custom_model_endpoints/{endpoint_id}", include_in_schema=True, responses=responses,
+@app.delete("/custom_model_endpoints/{model_id}/{provider_type}", include_in_schema=True, responses=responses,
            description="Delete a custom model endpoint")
-async def delete_custom_model_endpoint(endpoint_id: str):
+async def delete_custom_model_endpoint(model_id: str, provider_type: str):
     """Delete a custom model endpoint"""
     try:
-        success = custom_endpoint_manager.delete_endpoint(endpoint_id)
+        success = custom_endpoint_manager.delete_endpoint(model_id, provider_type)
 
         if not success:
             raise HTTPException(
                 status_code=404,
-                detail=f"Custom endpoint '{endpoint_id}' not found"
+                detail=f"Custom endpoint for model '{model_id}' with provider '{provider_type}' not found"
             )
 
         return {
             "status": "success",
-            "message": f"Custom endpoint '{endpoint_id}' deleted successfully"
+            "message": f"Custom endpoint for '{model_id}' ({provider_type}) deleted successfully"
         }
 
     except APIError as e: