refactor: extract shared Pydantic validators

- Create shared_validators.py with validate_mount_path, validate_files, validate_env_vars, validate_volumes
- Refactor config_folders.py to use shared validators
- Refactor tool_configs.py to use shared validators
- Refactor config_profiles.py to use shared env_vars validator
- Reduce ~80 lines of duplicate validation code

apps/api/src/api/config_folders.py

@@ -8,6 +8,7 @@ from pydantic import BaseModel, Field, field_validator
 from sqlalchemy import select
 from sqlalchemy.ext.asyncio import AsyncSession
 
+from src.api.shared_validators import validate_files as _validate_files, validate_mount_path as _validate_mount_path
 from src.auth.dependencies import get_current_user_id, get_db_session
 from src.models.config_folder import ConfigFolder
 
@@ -15,9 +16,6 @@ logger = logging.getLogger(__name__)
 
 router = APIRouter(prefix="/config-folders", tags=["config-folders"])
 
-MAX_FOLDER_SIZE_MB = 10
-MAX_FOLDER_SIZE_BYTES = MAX_FOLDER_SIZE_MB * 1024 * 1024
-
 
 class ConfigFolderCreate(BaseModel):
     name: str = Field(description="Folder name (unique per user)")
@@ -28,24 +26,12 @@ class ConfigFolderCreate(BaseModel):
     @field_validator("mount_path")
     @classmethod
     def validate_mount_path(cls, v: str) -> str:
-        if not v.startswith("/"):
+        return _validate_mount_path(v)
-            raise ValueError("Mount path must be absolute (start with /)")
-        return v
 
     @field_validator("files")
     @classmethod
     def validate_files(cls, v: dict) -> dict:
-        total_size = 0
+        return _validate_files(v)
-        for path, content in v.items():
-            # Check for path traversal
-            if ".." in path or path.startswith("/"):
-                raise ValueError(f"Invalid file path: {path}")
-            total_size += len(content.encode("utf-8"))
-        
-        if total_size > MAX_FOLDER_SIZE_BYTES:
-            raise ValueError(f"Total folder size exceeds {MAX_FOLDER_SIZE_MB}MB limit")
-        
-        return v
 
 
 class ConfigFolderUpdate(BaseModel):
@@ -58,29 +44,12 @@ class ConfigFolderUpdate(BaseModel):
     @field_validator("mount_path")
     @classmethod
     def validate_mount_path(cls, v: str | None) -> str | None:
-        if v is None:
+        return _validate_mount_path(v)
-            return v
-        if not v.startswith("/"):
-            raise ValueError("Mount path must be absolute (start with /)")
-        return v
 
     @field_validator("files")
     @classmethod
     def validate_files(cls, v: dict | None) -> dict | None:
-        if v is None:
+        return _validate_files(v)
-            return v
-        
-        total_size = 0
-        for path, content in v.items():
-            # Check for path traversal
-            if ".." in path or path.startswith("/"):
-                raise ValueError(f"Invalid file path: {path}")
-            total_size += len(content.encode("utf-8"))
-        
-        if total_size > MAX_FOLDER_SIZE_BYTES:
-            raise ValueError(f"Total folder size exceeds {MAX_FOLDER_SIZE_MB}MB limit")
-        
-        return v
 
 
 class ProjectOverrideCreate(BaseModel):
@@ -90,11 +59,7 @@ class ProjectOverrideCreate(BaseModel):
     @field_validator("mount_path")
     @classmethod
     def validate_mount_path(cls, v: str | None) -> str | None:
-        if v is None:
+        return _validate_mount_path(v)
-            return v
-        if not v.startswith("/"):
-            raise ValueError("Mount path must be absolute (start with /)")
-        return v
 
 
 class ConfigFolderResponse(BaseModel):

apps/api/src/api/config_profiles.py

@@ -9,6 +9,7 @@ from sqlalchemy import select
 from sqlalchemy.ext.asyncio import AsyncSession
 from sqlalchemy.orm import selectinload
 
+from src.api.shared_validators import validate_env_vars as _validate_env_vars
 from src.auth.dependencies import get_current_user_id, get_db_session
 from src.models.config_profile import ConfigProfile, ConfigProfileInclude
 from src.models.project import Project
@@ -120,9 +121,10 @@ class ConfigProfileCreate(BaseModel):
     @field_validator("env_vars")
     @classmethod
     def validate_env_vars(cls, v: dict) -> dict:
-        if not isinstance(v, dict):
+        result = _validate_env_vars(v)
+        if result is None:
             raise ValueError("env_vars must be a JSON object")
-        return v
+        return result
 
     @field_validator("runtime_hints")
     @classmethod

apps/api/src/api/shared_validators.py

@@ -0,0 +1,99 @@
+"""Shared Pydantic validators for API schemas."""
+
+from typing import Any
+
+MAX_FOLDER_SIZE_MB = 10
+MAX_FOLDER_SIZE_BYTES = MAX_FOLDER_SIZE_MB * 1024 * 1024
+
+
+def validate_mount_path(v: str | None) -> str | None:
+    """Validate that a mount path is absolute (starts with /).
+
+    Args:
+        v: Mount path string or None.
+
+    Returns:
+        The validated path, or None if input was None.
+
+    Raises:
+        ValueError: If path is not absolute.
+    """
+    if v is None:
+        return v
+    if not v.startswith("/"):
+        raise ValueError("Mount path must be absolute (start with /)")
+    return v
+
+
+def validate_files(v: dict | None, max_size_bytes: int = MAX_FOLDER_SIZE_BYTES) -> dict | None:
+    """Validate file dict for path traversal and size limits.
+
+    Args:
+        v: Dict of {path: content} or None.
+        max_size_bytes: Maximum total size in bytes.
+
+    Returns:
+        The validated dict, or None if input was None.
+
+    Raises:
+        ValueError: If path traversal detected or size limit exceeded.
+    """
+    if v is None:
+        return v
+
+    total_size = 0
+    for path, content in v.items():
+        # Check for path traversal
+        if ".." in path or path.startswith("/"):
+            raise ValueError(f"Invalid file path: {path}")
+        total_size += len(content.encode("utf-8"))
+
+    if total_size > max_size_bytes:
+        raise ValueError(f"Total folder size exceeds {max_size_bytes // (1024 * 1024)}MB limit")
+
+    return v
+
+
+def validate_env_vars(v: dict | None) -> dict | None:
+    """Validate that environment variables is a JSON object.
+
+    Args:
+        v: Dict of env vars or None.
+
+    Returns:
+        The validated dict, or None if input was None.
+
+    Raises:
+        ValueError: If not a dict.
+    """
+    if v is None:
+        return v
+    if not isinstance(v, dict):
+        raise ValueError("environment_variables must be a JSON object")
+    return v
+
+
+def validate_volumes(v: list | None) -> list | None:
+    """Validate volume mounts list.
+
+    Args:
+        v: List of volume dicts or None.
+
+    Returns:
+        The validated list, or None if input was None.
+
+    Raises:
+        ValueError: If not a list or missing required fields.
+    """
+    if v is None:
+        return v
+    if not isinstance(v, list):
+        raise ValueError("volumes must be a JSON array")
+    for i, vol in enumerate(v):
+        if not isinstance(vol, dict):
+            raise ValueError(f"Volume at index {i} must be an object")
+        if "source" not in vol:
+            raise ValueError(f"Volume at index {i} must have 'source' field")
+        if "target" not in vol:
+            raise ValueError(f"Volume at index {i} must have 'target' field")
+    return v

apps/api/src/api/tool_configs.py

@@ -8,11 +8,9 @@ from pydantic import BaseModel, Field, field_validator
 from sqlalchemy import select
 from sqlalchemy.ext.asyncio import AsyncSession
 
+from src.api.shared_validators import validate_env_vars as _validate_env_vars, validate_volumes as _validate_volumes
 from src.auth.dependencies import get_current_user_id, get_db_session
 from src.models.tool_config import ToolConfig
-from src.models.tool_type import ToolType
-
-logger = logging.getLogger(__name__)
 
 router = APIRouter(prefix="/tool-configs", tags=["tool-configs"])
 
@@ -42,27 +40,12 @@ class ToolConfigCreate(BaseModel):
     @field_validator("environment_variables")
     @classmethod
     def validate_env_vars(cls, v: dict | None) -> dict | None:
-        if v is None:
+        return _validate_env_vars(v)
-            return v
-        if not isinstance(v, dict):
-            raise ValueError("environment_variables must be a JSON object")
-        return v
 
     @field_validator("volumes")
     @classmethod
     def validate_volumes(cls, v: list | None) -> list | None:
-        if v is None:
+        return _validate_volumes(v)
-            return v
-        if not isinstance(v, list):
-            raise ValueError("volumes must be a JSON array")
-        for i, vol in enumerate(v):
-            if not isinstance(vol, dict):
-                raise ValueError(f"Volume at index {i} must be an object")
-            if "source" not in vol:
-                raise ValueError(f"Volume at index {i} must have 'source' field")
-            if "target" not in vol:
-                raise ValueError(f"Volume at index {i} must have 'target' field")
-        return v
 
 
 class ToolConfigUpdate(BaseModel):
@@ -88,27 +71,12 @@ class ToolConfigUpdate(BaseModel):
     @field_validator("environment_variables")
     @classmethod
     def validate_env_vars(cls, v: dict | None) -> dict | None:
-        if v is None:
+        return _validate_env_vars(v)
-            return v
-        if not isinstance(v, dict):
-            raise ValueError("environment_variables must be a JSON object")
-        return v
 
     @field_validator("volumes")
     @classmethod
     def validate_volumes(cls, v: list | None) -> list | None:
-        if v is None:
+        return _validate_volumes(v)
-            return v
-        if not isinstance(v, list):
-            raise ValueError("volumes must be a JSON array")
-        for i, vol in enumerate(v):
-            if not isinstance(vol, dict):
-                raise ValueError(f"Volume at index {i} must be an object")
-            if "source" not in vol:
-                raise ValueError(f"Volume at index {i} must have 'source' field")
-            if "target" not in vol:
-                raise ValueError(f"Volume at index {i} must have 'target' field")
-        return v
 
 
 class ToolConfigResponse(BaseModel):