Require top-level keys in ConfigLoaders for schema validation

- AgentConfigLoader now requires 'agent:' top-level key
- GraphConfigLoader now requires 'graph:' top-level key
- SwarmConfigLoader now requires 'swarm:' top-level key
- Updated serialize methods to include top-level keys
- Enables proper YAML/schema validation and identification
- Breaking change: all configurations must be wrapped in appropriate top-level key

BREAKING CHANGE: Configuration structure changed
- Agent configs: wrap in 'agent:' key
- Graph configs: wrap in 'graph:' key
- Swarm configs: wrap in 'swarm:' key

Author: vawsgit

src/strands/experimental/config_loader/agent/agent_config_loader.py

@@ -76,7 +76,7 @@ def load_agent(self, config: Dict[str, Any], cache_key: Optional[str] = None) ->
         """Load an Agent from a dictionary configuration.
 
         Args:
-            config: Dictionary containing agent configuration.
+            config: Dictionary containing agent configuration with top-level 'agent' key.
             cache_key: Optional key for caching the loaded agent.
 
         Returns:
@@ -86,6 +86,14 @@ def load_agent(self, config: Dict[str, Any], cache_key: Optional[str] = None) ->
             ValueError: If required configuration is missing or invalid.
             ImportError: If specified models or tools cannot be imported.
         """
+        # Validate top-level structure
+        if "agent" not in config:
+            raise ValueError("Configuration must contain a top-level 'agent' key")
+        
+        agent_config = config["agent"]
+        if not isinstance(agent_config, dict):
+            raise ValueError("The 'agent' configuration must be a dictionary")
+
         # Check cache first
         if cache_key and cache_key in self._agent_cache:
             logger.debug("agent_cache_key=<%s> | found in cache", cache_key)
@@ -100,30 +108,30 @@ def load_agent(self, config: Dict[str, Any], cache_key: Optional[str] = None) ->
         if "structured_output_defaults" in config:
             self._structured_output_defaults = config["structured_output_defaults"]
 
-        # Extract configuration values
-        model_config = config.get("model")
-        system_prompt = config.get("system_prompt")
-        tools_config = config.get("tools", [])
-        messages_config = config.get("messages", [])
+        # Extract configuration values from agent_config
+        model_config = agent_config.get("model")
+        system_prompt = agent_config.get("system_prompt")
+        tools_config = agent_config.get("tools", [])
+        messages_config = agent_config.get("messages", [])
 
         # Note: 'prompt' field is handled by AgentAsToolWrapper, not by Agent itself
         # The Agent class doesn't have a prompt parameter - it uses system_prompt
         # The prompt field is used for tool invocation templates
 
         # Agent metadata
-        agent_id = config.get("agent_id")
-        name = config.get("name")
-        description = config.get("description")
+        agent_id = agent_config.get("agent_id")
+        name = agent_config.get("name")
+        description = agent_config.get("description")
 
         # Advanced configuration
-        callback_handler_config = config.get("callback_handler")
-        conversation_manager_config = config.get("conversation_manager")
-        record_direct_tool_call = config.get("record_direct_tool_call", True)
-        load_tools_from_directory = config.get("load_tools_from_directory", False)
-        trace_attributes = config.get("trace_attributes")
-        state_config = config.get("state")
-        hooks_config = config.get("hooks", [])
-        session_manager_config = config.get("session_manager")
+        callback_handler_config = agent_config.get("callback_handler")
+        conversation_manager_config = agent_config.get("conversation_manager")
+        record_direct_tool_call = agent_config.get("record_direct_tool_call", True)
+        load_tools_from_directory = agent_config.get("load_tools_from_directory", False)
+        trace_attributes = agent_config.get("trace_attributes")
+        state_config = agent_config.get("state")
+        hooks_config = agent_config.get("hooks", [])
+        session_manager_config = agent_config.get("session_manager")
 
         # Load model
         model = self._load_model(model_config)
@@ -169,8 +177,8 @@ def load_agent(self, config: Dict[str, Any], cache_key: Optional[str] = None) ->
         )
 
         # Configure structured output if specified
-        if "structured_output" in config:
-            self._configure_agent_structured_output(agent, config["structured_output"])
+        if "structured_output" in agent_config:
+            self._configure_agent_structured_output(agent, agent_config["structured_output"])
 
         # Cache the agent if cache key provided
         if cache_key:
@@ -186,56 +194,56 @@ def serialize_agent(self, agent: Agent) -> Dict[str, Any]:
             agent: Agent instance to serialize.
 
         Returns:
-            Dictionary containing the agent's configuration.
+            Dictionary containing the agent's configuration with top-level 'agent' key.
 
         Note:
             The 'prompt' field is not serialized here as it's specific to AgentAsToolWrapper
             and not part of the core Agent configuration.
         """
-        config = {}
+        agent_config = {}
 
         # Basic configuration
         if hasattr(agent.model, "model_id"):
-            config["model"] = agent.model.model_id
+            agent_config["model"] = agent.model.model_id
         elif hasattr(agent.model, "config") and agent.model.config.get("model_id"):
-            config["model"] = agent.model.config["model_id"]
+            agent_config["model"] = agent.model.config["model_id"]
 
         if agent.system_prompt:
-            config["system_prompt"] = agent.system_prompt
+            agent_config["system_prompt"] = agent.system_prompt
 
         # Tools configuration
         if hasattr(agent, "tool_registry") and agent.tool_registry:
             tools_config = []
             for tool_name in agent.tool_names:
                 tools_config.append({"name": tool_name})
             if tools_config:
-                config["tools"] = tools_config
+                agent_config["tools"] = tools_config
 
         # Messages
         if agent.messages:
-            config["messages"] = agent.messages
+            agent_config["messages"] = agent.messages
 
         # Agent metadata
         if agent.agent_id != "default":
-            config["agent_id"] = agent.agent_id
+            agent_config["agent_id"] = agent.agent_id
         if agent.name != "Strands Agents":
-            config["name"] = agent.name
+            agent_config["name"] = agent.name
         if agent.description:
-            config["description"] = agent.description
+            agent_config["description"] = agent.description
 
         # Advanced configuration
         if agent.record_direct_tool_call is not True:
-            config["record_direct_tool_call"] = agent.record_direct_tool_call
+            agent_config["record_direct_tool_call"] = agent.record_direct_tool_call
         if agent.load_tools_from_directory is not False:
-            config["load_tools_from_directory"] = agent.load_tools_from_directory
+            agent_config["load_tools_from_directory"] = agent.load_tools_from_directory
         if agent.trace_attributes:
-            config["trace_attributes"] = agent.trace_attributes
+            agent_config["trace_attributes"] = agent.trace_attributes
 
         # State
         if agent.state and agent.state.get():
-            config["state"] = agent.state.get()
+            agent_config["state"] = agent.state.get()
 
-        return config
+        return {"agent": agent_config}
 
     def _load_model(self, model_config: Optional[Union[str, Dict[str, Any]]]) -> Optional[Model]:
         """Load a model from configuration.

src/strands/experimental/config_loader/graph/graph_config_loader.py

@@ -88,7 +88,7 @@ def load_graph(self, config: Dict[str, Any], cache_key: Optional[str] = None) ->
         """Load a Graph from YAML configuration (loaded as dictionary).
 
         Args:
-            config: Dictionary containing graph configuration.
+            config: Dictionary containing graph configuration with top-level 'graph' key.
             cache_key: Optional key for caching the loaded graph.
 
         Returns:
@@ -98,28 +98,36 @@ def load_graph(self, config: Dict[str, Any], cache_key: Optional[str] = None) ->
             ValueError: If required configuration is missing or invalid.
             ImportError: If specified models or tools cannot be imported.
         """
+        # Validate top-level structure
+        if "graph" not in config:
+            raise ValueError("Configuration must contain a top-level 'graph' key")
+        
+        graph_config = config["graph"]
+        if not isinstance(graph_config, dict):
+            raise ValueError("The 'graph' configuration must be a dictionary")
+
         # Check cache first
         if cache_key and cache_key in self._graph_cache:
             logger.debug("graph_cache_key=<%s> | found in cache", cache_key)
             return self._graph_cache[cache_key]
 
         # Validate configuration structure
-        self._validate_config(config)
+        self._validate_config(graph_config)
 
         # Load nodes
-        nodes = self._load_nodes(config.get("nodes", []))
+        nodes = self._load_nodes(graph_config.get("nodes", []))
 
         # Load edges with conditions
-        edges = self._load_edges(config.get("edges", []), nodes)
+        edges = self._load_edges(graph_config.get("edges", []), nodes)
 
         # Load entry points
-        entry_points = self._load_entry_points(config.get("entry_points", []), nodes)
+        entry_points = self._load_entry_points(graph_config.get("entry_points", []), nodes)
 
         # Extract graph configuration
-        graph_config = self._extract_graph_parameters(config)
+        graph_params = self._extract_graph_parameters(graph_config)
 
         # Create graph
-        graph = Graph(nodes=nodes, edges=edges, entry_points=entry_points, **graph_config)
+        graph = Graph(nodes=nodes, edges=edges, entry_points=entry_points, **graph_params)
 
         # Cache the graph if cache key provided
         if cache_key:
@@ -135,23 +143,23 @@ def serialize_graph(self, graph: Graph) -> Dict[str, Any]:
             graph: Graph instance to serialize.
 
         Returns:
-            Dictionary containing the graph's configuration that can be saved as YAML.
+            Dictionary containing the graph's configuration with top-level 'graph' key.
         """
-        config: Dict[str, Any] = {}
+        graph_config: Dict[str, Any] = {}
 
         # Serialize nodes
         nodes_config = []
         for node_id, node in graph.nodes.items():
             node_config = self._serialize_node(node_id, node)
             nodes_config.append(node_config)
-        config["nodes"] = nodes_config
+        graph_config["nodes"] = nodes_config
 
         # Serialize edges
         edges_config = []
         for edge in graph.edges:
             edge_config = self._serialize_edge(edge)
             edges_config.append(edge_config)
-        config["edges"] = edges_config
+        graph_config["edges"] = edges_config
 
         # Serialize entry points
         entry_points_config: List[str] = []
@@ -161,19 +169,19 @@ def serialize_graph(self, graph: Graph) -> Dict[str, Any]:
                 if node == entry_point:
                     entry_points_config.append(node_id)
                     break
-        config["entry_points"] = entry_points_config
+        graph_config["entry_points"] = entry_points_config
 
         # Serialize graph parameters (only include non-default values)
         if graph.max_node_executions is not None:
-            config["max_node_executions"] = graph.max_node_executions
+            graph_config["max_node_executions"] = graph.max_node_executions
         if graph.execution_timeout is not None:
-            config["execution_timeout"] = graph.execution_timeout
+            graph_config["execution_timeout"] = graph.execution_timeout
         if graph.node_timeout is not None:
-            config["node_timeout"] = graph.node_timeout
+            graph_config["node_timeout"] = graph.node_timeout
         if graph.reset_on_revisit is not False:
-            config["reset_on_revisit"] = graph.reset_on_revisit
+            graph_config["reset_on_revisit"] = graph.reset_on_revisit
 
-        return config
+        return {"graph": graph_config}
 
     def clear_cache(self) -> None:
         """Clear the internal graph cache."""

src/strands/experimental/config_loader/schema/SCHEMA-PLAN.md

@@ -61,7 +61,7 @@ def load_swarm(self, config: Dict[str, Any], cache_key: Optional[str] = None) ->
         """Load a Swarm from YAML configuration (loaded as dictionary).
 
         Args:
-            config: Dictionary containing swarm configuration.
+            config: Dictionary containing swarm configuration with top-level 'swarm' key.
             cache_key: Optional key for caching the loaded swarm.
 
         Returns:
@@ -71,24 +71,32 @@ def load_swarm(self, config: Dict[str, Any], cache_key: Optional[str] = None) ->
             ValueError: If required configuration is missing or invalid.
             ImportError: If specified models or tools cannot be imported.
         """
+        # Validate top-level structure
+        if "swarm" not in config:
+            raise ValueError("Configuration must contain a top-level 'swarm' key")
+        
+        swarm_config = config["swarm"]
+        if not isinstance(swarm_config, dict):
+            raise ValueError("The 'swarm' configuration must be a dictionary")
+
         # Check cache first
         if cache_key and cache_key in self._swarm_cache:
             logger.debug("swarm_cache_key=<%s> | found in cache", cache_key)
             return self._swarm_cache[cache_key]
 
         # Validate configuration structure
-        self._validate_config(config)
+        self._validate_config(swarm_config)
 
         # Extract agents configuration
-        agents_config = config.get("agents", [])
+        agents_config = swarm_config.get("agents", [])
         if not agents_config:
             raise ValueError("Swarm configuration must include 'agents' field with at least one agent")
 
         # Load agents using AgentConfigLoader
         agents = self.load_agents(agents_config)
 
         # Extract swarm parameters
-        swarm_params = self._extract_swarm_parameters(config)
+        swarm_params = self._extract_swarm_parameters(swarm_config)
 
         # Create swarm
         swarm = Swarm(nodes=agents, **swarm_params)
@@ -107,23 +115,23 @@ def serialize_swarm(self, swarm: Swarm) -> Dict[str, Any]:
             swarm: Swarm instance to serialize.
 
         Returns:
-            Dictionary containing the swarm's configuration that can be saved as YAML.
+            Dictionary containing the swarm's configuration with top-level 'swarm' key.
         """
-        config = {}
+        swarm_config = {}
 
         # Serialize swarm parameters (only include non-default values)
         if swarm.max_handoffs != 20:
-            config["max_handoffs"] = swarm.max_handoffs
+            swarm_config["max_handoffs"] = swarm.max_handoffs
         if swarm.max_iterations != 20:
-            config["max_iterations"] = swarm.max_iterations
+            swarm_config["max_iterations"] = swarm.max_iterations
         if swarm.execution_timeout != 900.0:
-            config["execution_timeout"] = swarm.execution_timeout
+            swarm_config["execution_timeout"] = swarm.execution_timeout
         if swarm.node_timeout != 300.0:
-            config["node_timeout"] = swarm.node_timeout
+            swarm_config["node_timeout"] = swarm.node_timeout
         if swarm.repetitive_handoff_detection_window != 0:
-            config["repetitive_handoff_detection_window"] = swarm.repetitive_handoff_detection_window
+            swarm_config["repetitive_handoff_detection_window"] = swarm.repetitive_handoff_detection_window
         if swarm.repetitive_handoff_min_unique_agents != 0:
-            config["repetitive_handoff_min_unique_agents"] = swarm.repetitive_handoff_min_unique_agents
+            swarm_config["repetitive_handoff_min_unique_agents"] = swarm.repetitive_handoff_min_unique_agents
 
         # Serialize agents
         agents_config = []
@@ -139,9 +147,9 @@ def serialize_swarm(self, swarm: Swarm) -> Dict[str, Any]:
             agent_config = agent_loader.serialize_agent(temp_agent)
             agents_config.append(agent_config)
 
-        config["agents"] = agents_config
+        swarm_config["agents"] = agents_config
 
-        return config
+        return {"swarm": swarm_config}
 
     def _create_clean_agent_copy(self, agent: Agent) -> Agent:
         """Create a copy of an agent without swarm coordination tools.