refactor: simplify outputSchema implementation based on PR feedback

- Use NotRequired[JSONSchema] instead of total=False for ToolSpec
- Remove filter_tool_specs method from Model base class
- Always filter outputSchema in streaming.py (not configurable)
- Add TODO comment referencing issue #780 for future improvements
- Remove example file and filtering tests as requested
- Update test mocks to remove filter_tool_specs references

This simplifies the implementation by unconditionally filtering outputSchema
in the event loop until proper model-specific behavior handling is
implemented in #780.

Author: Vamil Gandhi

examples/output_schema_example.py

@@ -2,7 +2,7 @@
 
 import json
 import logging
-from typing import Any, AsyncGenerator, AsyncIterable, Optional
+from typing import Any, AsyncGenerator, AsyncIterable, Optional, cast
 
 from ..models.model import Model
 from ..types._events import (
@@ -337,8 +337,16 @@ async def stream_messages(
     logger.debug("model=<%s> | streaming messages", model)
 
     messages = remove_blank_messages_content_text(messages)
-    # Filter outputschema spec based on model configuration until all models supports it.
-    filtered_tool_specs = model.filter_tool_specs(tool_specs) if tool_specs else None
+
+    # TODO(#780): Remove outputSchema filtering once model-specific behavior handling is implemented.
+    # For now, we filter out outputSchema from all tool specs to ensure compatibility with all model providers.
+    # Some providers (e.g., Bedrock) will throw validation errors if they receive unknown fields.
+    filtered_tool_specs = None
+    if tool_specs:
+        filtered_tool_specs = cast(
+            list[ToolSpec], [{k: v for k, v in spec.items() if k != "outputSchema"} for spec in tool_specs]
+        )
+
     chunks = model.stream(messages, filtered_tool_specs, system_prompt)
 
     async for event in process_stream(chunks):

src/strands/event_loop/streaming.py

@@ -2,7 +2,7 @@
 
 import abc
 import logging
-from typing import Any, AsyncGenerator, AsyncIterable, Optional, Type, TypeVar, Union, cast
+from typing import Any, AsyncGenerator, AsyncIterable, Optional, Type, TypeVar, Union
 
 from pydantic import BaseModel
 
@@ -22,27 +22,6 @@ class Model(abc.ABC):
     standardized way to configure and process requests for different AI model providers.
     """
 
-    def filter_tool_specs(self, tool_specs: Optional[list[ToolSpec]]) -> Optional[list[ToolSpec]]:
-        """Filter tool specifications based on model configuration.
-
-        By default, this removes the outputSchema field from tool specs unless
-        the model configuration explicitly enables it via `supports_tool_output_schema`.
-
-        Args:
-            tool_specs: List of tool specifications to filter.
-
-        Returns:
-            Filtered tool specifications safe for the model provider.
-        """
-        if not tool_specs:
-            return tool_specs
-
-        config = self.get_config()
-        if isinstance(config, dict) and config.get("supports_tool_output_schema", False):
-            return tool_specs
-
-        return cast(list[ToolSpec], [{k: v for k, v in spec.items() if k != "outputSchema"} for spec in tool_specs])
-
     @abc.abstractmethod
     # pragma: no cover
     def update_config(self, **model_config: Any) -> None:

src/strands/models/model.py

@@ -9,7 +9,7 @@
 from dataclasses import dataclass
 from typing import TYPE_CHECKING, Any, AsyncGenerator, Awaitable, Callable, Literal, Protocol, Union
 
-from typing_extensions import TypedDict
+from typing_extensions import NotRequired, TypedDict
 
 from .media import DocumentContent, ImageContent
 
@@ -20,7 +20,7 @@
 """Type alias for JSON Schema dictionaries."""
 
 
-class ToolSpec(TypedDict, total=False):
+class ToolSpec(TypedDict):
     """Specification for a tool that can be used by an agent.
 
     Attributes:
@@ -35,7 +35,7 @@ class ToolSpec(TypedDict, total=False):
     description: str
     inputSchema: JSONSchema
     name: str
-    outputSchema: JSONSchema
+    outputSchema: NotRequired[JSONSchema]
 
 
 class Tool(TypedDict):

src/strands/types/tools.py

@@ -49,8 +49,6 @@ async def stream(*args, **kwargs):
     mock = unittest.mock.Mock(spec=getattr(request, "param", None))
     mock.configure_mock(mock_stream=unittest.mock.MagicMock())
     mock.stream.side_effect = stream
-    mock.filter_tool_specs = lambda tool_specs: tool_specs
-    mock.get_config.return_value = {}
 
     return mock
 

tests/strands/agent/test_agent.py

@@ -33,9 +33,7 @@ def mock_time():
 
 @pytest.fixture
 def model():
-    mock = unittest.mock.Mock()
-    mock.filter_tool_specs.side_effect = lambda tool_specs: tool_specs
-    return mock
+    return unittest.mock.Mock()
 
 
 @pytest.fixture