fix: validate gateway server config block at load time (#3050)

Move the typed-validation fix from the example into the wrapper where
the bug actually lives. load_gateway_config now validates the optional
'gateway:' server block field-by-field against the SDK GatewayConfig
dataclass (single source of truth), so misspelled keys, wrong types and
negative values are rejected with clear errors instead of silently
falling back to defaults.

Revert the example to use the typed SDK GatewayConfig instead of a local
shadowing class that bypassed validation.

Co-authored-by: Mervin Praison <MervinPraison@users.noreply.github.com>

Author: praisonai-triage-agent[bot]

examples/python/gateway_example.py

@@ -7,28 +7,8 @@
 
 from praisonaiagents import Agent, GatewayConfig, SessionConfig
 
-# Configure the gateway
-from pydantic import BaseModel, Field
-
-class ServerConfig(BaseModel):
-    host: str = "0.0.0.0"
-    port: int = 8000
-    drain_timeout: int = 30
-    reload_drain_timeout: int = 60
-    max_connections: int = 1000
-    heartbeat_interval: int = 30
-
-class GatewayConfig:
-    def __init__(self, host="0.0.0.0", port=8000, drain_timeout=30, reload_drain_timeout=60, max_connections=1000, heartbeat_interval=30, session_config=None):
-        self.host = host
-        self.port = port
-        self.drain_timeout = drain_timeout
-        self.reload_drain_timeout = reload_drain_timeout
-        self.max_connections = max_connections
-        self.heartbeat_interval = heartbeat_interval
-        self.session_config = session_config or SessionConfig()
-        self.ws_url = f"ws://{self.host}:{self.port}"
-
+# Configure the gateway using the typed SDK GatewayConfig.
+# Invalid or misspelled fields are caught by the dataclass signature.
 gateway_config = GatewayConfig(
     host="127.0.0.1",
     port=8765,
@@ -37,7 +17,7 @@ def __init__(self, host="0.0.0.0", port=8000, drain_timeout=30, reload_drain_tim
     session_config=SessionConfig(
         timeout=3600,
         max_messages=500,
-    )
+    ),
 )
 
 # Create specialized agents
@@ -61,7 +41,7 @@ def __init__(self, host="0.0.0.0", port=8000, drain_timeout=30, reload_drain_tim
     print(f"  WebSocket URL: {gateway_config.ws_url}")
     print(f"  Max Connections: {gateway_config.max_connections}")
     print()
-    
+
     # Test agent
     response = researcher.start("What are the key benefits of multi-agent systems?")
     print("Researcher Response:")

src/praisonai/praisonai/gateway/server.py

@@ -30,6 +30,42 @@
 logger = logging.getLogger(__name__)
 
 
+def _closest_match(key: str, candidates) -> Optional[str]:
+    """Return the closest candidate name to ``key`` for typo suggestions."""
+    import difflib
+
+    matches = difflib.get_close_matches(key, list(candidates), n=1, cutoff=0.6)
+    return matches[0] if matches else None
+
+
+def _value_matches_field(value: Any, field_type: Any) -> bool:
+    """Best-effort check that ``value`` is compatible with a dataclass field type.
+
+    Handles the plain and string ("from __future__ import annotations") forms of
+    the primitive/container types used by ``GatewayConfig``. bools are rejected
+    where an int is expected. Unknown/complex types are treated as valid so we
+    never reject something we cannot confidently validate.
+    """
+    type_name = field_type if isinstance(field_type, str) else getattr(field_type, "__name__", str(field_type))
+    type_name = type_name.replace("typing.", "")
+    if type_name.startswith("Optional[") and type_name.endswith("]"):
+        type_name = type_name[len("Optional["):-1]
+
+    if type_name.startswith("int"):
+        return isinstance(value, int) and not isinstance(value, bool)
+    if type_name.startswith("float"):
+        return isinstance(value, (int, float)) and not isinstance(value, bool)
+    if type_name.startswith("bool"):
+        return isinstance(value, bool)
+    if type_name.startswith("str"):
+        return isinstance(value, str)
+    if type_name.startswith("List") or type_name.startswith("list"):
+        return isinstance(value, list)
+    if type_name.startswith("Dict") or type_name.startswith("dict"):
+        return isinstance(value, dict)
+    return True
+
+
 @dataclass
 class GatewaySession:
     """A gateway session tracking a conversation between client and agent."""
@@ -1358,6 +1394,42 @@ def load_gateway_config(cls, config_path: str) -> Dict[str, Any]:
                         "(use ${{ENV_VAR}} syntax for env vars)"
                     )
 
+        # Validate the optional 'gateway:' server block field-by-field,
+        # reusing the typed GatewayConfig dataclass as the single source of
+        # truth so misspelled or wrongly-typed server settings are caught at
+        # load time instead of being silently ignored at runtime (see #3050).
+        gw_block = raw.get("gateway")
+        if gw_block is not None:
+            if not isinstance(gw_block, dict):
+                errors.append("'gateway' must be a dictionary")
+            else:
+                import dataclasses
+
+                allowed_fields = {f.name: f for f in dataclasses.fields(GatewayConfig)}
+                # session_config is a nested dataclass configured elsewhere.
+                allowed_fields.pop("session_config", None)
+                for key, value in gw_block.items():
+                    if key not in allowed_fields:
+                        suggestion = _closest_match(key, allowed_fields.keys())
+                        hint = f" (did you mean '{suggestion}'?)" if suggestion else ""
+                        errors.append(
+                            f"Unknown gateway setting '{key}'{hint}. "
+                            f"Allowed: {', '.join(sorted(allowed_fields))}"
+                        )
+                        continue
+                    if value is None:
+                        continue
+                    field_type = allowed_fields[key].type
+                    if not _value_matches_field(value, field_type):
+                        errors.append(
+                            f"Gateway setting '{key}' has invalid type "
+                            f"'{type(value).__name__}' (expected {field_type})"
+                        )
+                    elif isinstance(value, int) and not isinstance(value, bool) and value < 0:
+                        errors.append(
+                            f"Gateway setting '{key}' must not be negative (got {value})"
+                        )
+
         if errors:
             msg = (
                 f"Gateway config validation failed ({config_path}):\n"

src/praisonai/tests/integration/test_websocket_integration.py

@@ -134,6 +134,78 @@ async def test_gateway_config_valid(self, tmp_path):
         assert "channels" in cfg
         assert cfg["channels"]["telegram"]["token"] == "test123"
 
+    @pytest.mark.asyncio
+    async def test_gateway_block_rejects_unknown_key(self, tmp_path):
+        """Misspelled gateway server settings are rejected with a suggestion."""
+        try:
+            from praisonai.gateway.server import WebSocketGateway
+        except ImportError:
+            pytest.skip("praisonai.gateway not available")
+
+        bad_config = tmp_path / "bad.yaml"
+        bad_config.write_text(
+            "agents:\n  bot:\n    instructions: hi\n"
+            "channels:\n  telegram:\n    token: test123\n"
+            "gateway:\n  reconnect_timout: 60\n"
+        )
+
+        with pytest.raises(ValueError, match="reconnect_timeout"):
+            WebSocketGateway.load_gateway_config(str(bad_config))
+
+    @pytest.mark.asyncio
+    async def test_gateway_block_rejects_wrong_type(self, tmp_path):
+        """A string where an int is expected is rejected at load time."""
+        try:
+            from praisonai.gateway.server import WebSocketGateway
+        except ImportError:
+            pytest.skip("praisonai.gateway not available")
+
+        bad_config = tmp_path / "bad.yaml"
+        bad_config.write_text(
+            "agents:\n  bot:\n    instructions: hi\n"
+            "channels:\n  telegram:\n    token: test123\n"
+            "gateway:\n  port: notaport\n"
+        )
+
+        with pytest.raises(ValueError, match="invalid type"):
+            WebSocketGateway.load_gateway_config(str(bad_config))
+
+    @pytest.mark.asyncio
+    async def test_gateway_block_rejects_negative(self, tmp_path):
+        """Negative numeric server settings are rejected."""
+        try:
+            from praisonai.gateway.server import WebSocketGateway
+        except ImportError:
+            pytest.skip("praisonai.gateway not available")
+
+        bad_config = tmp_path / "bad.yaml"
+        bad_config.write_text(
+            "agents:\n  bot:\n    instructions: hi\n"
+            "channels:\n  telegram:\n    token: test123\n"
+            "gateway:\n  max_connections: -5\n"
+        )
+
+        with pytest.raises(ValueError, match="must not be negative"):
+            WebSocketGateway.load_gateway_config(str(bad_config))
+
+    @pytest.mark.asyncio
+    async def test_gateway_block_valid(self, tmp_path):
+        """A valid gateway server block loads successfully."""
+        try:
+            from praisonai.gateway.server import WebSocketGateway
+        except ImportError:
+            pytest.skip("praisonai.gateway not available")
+
+        good_config = tmp_path / "good.yaml"
+        good_config.write_text(
+            "agents:\n  bot:\n    instructions: hi\n"
+            "channels:\n  telegram:\n    token: test123\n"
+            "gateway:\n  host: 127.0.0.1\n  port: 8765\n  max_connections: 100\n"
+        )
+
+        cfg = WebSocketGateway.load_gateway_config(str(good_config))
+        assert cfg["gateway"]["port"] == 8765
+
     @pytest.mark.asyncio
     async def test_websocket_connect_and_message(self):
         """Test actual WebSocket connection to gateway server.