strands.experimental.steering.core.context

Steering context protocols for contextual guidance.

Defines protocols for context callbacks and providers that populate steering context data used by handlers to make guidance decisions.

Architecture

SteeringContextCallback → Handler.steering_context → SteeringHandler.steer() ↓ ↓ ↓ Update local context Store in handler Access via self.steering_context

Context lifecycle

1. Handler registers context callbacks for hook events
2. Callbacks update handler's local steering_context on events
3. Handler accesses self.steering_context in steer() method
4. Context persists across calls within handler instance

Implementation

Each handler maintains its own JSONSerializableDict context. Callbacks are registered per handler instance for isolation. Providers can supply multiple callbacks for different events.

EventType = TypeVar('EventType', bound=HookEvent, contravariant=True) module-attribute

logger = logging.getLogger(__name__) module-attribute

HookEvent dataclass

Bases: BaseHookEvent

Base class for single agent hook events.

Attributes:

Name	Type	Description
agent	Agent	The agent instance that triggered this event.

Source code in strands/hooks/registry.py

@dataclass
class HookEvent(BaseHookEvent):
    """Base class for single agent hook events.

    Attributes:
        agent: The agent instance that triggered this event.
    """

    agent: "Agent"

JSONSerializableDict

A key-value store with JSON serialization validation.

Provides a dict-like interface with automatic validation that all values are JSON serializable on assignment.

Source code in strands/types/json_dict.py

class JSONSerializableDict:
    """A key-value store with JSON serialization validation.

    Provides a dict-like interface with automatic validation that all values
    are JSON serializable on assignment.
    """

    def __init__(self, initial_state: dict[str, Any] | None = None):
        """Initialize JSONSerializableDict."""
        self._data: dict[str, Any]
        if initial_state:
            self._validate_json_serializable(initial_state)
            self._data = copy.deepcopy(initial_state)
        else:
            self._data = {}

    def set(self, key: str, value: Any) -> None:
        """Set a value in the store.

        Args:
            key: The key to store the value under
            value: The value to store (must be JSON serializable)

        Raises:
            ValueError: If key is invalid, or if value is not JSON serializable
        """
        self._validate_key(key)
        self._validate_json_serializable(value)
        self._data[key] = copy.deepcopy(value)

    def get(self, key: str | None = None) -> Any:
        """Get a value or entire data.

        Args:
            key: The key to retrieve (if None, returns entire data dict)

        Returns:
            The stored value, entire data dict, or None if not found
        """
        if key is None:
            return copy.deepcopy(self._data)
        else:
            return copy.deepcopy(self._data.get(key))

    def delete(self, key: str) -> None:
        """Delete a specific key from the store.

        Args:
            key: The key to delete
        """
        self._validate_key(key)
        self._data.pop(key, None)

    def _validate_key(self, key: str) -> None:
        """Validate that a key is valid.

        Args:
            key: The key to validate

        Raises:
            ValueError: If key is invalid
        """
        if key is None:
            raise ValueError("Key cannot be None")
        if not isinstance(key, str):
            raise ValueError("Key must be a string")
        if not key.strip():
            raise ValueError("Key cannot be empty")

    def _validate_json_serializable(self, value: Any) -> None:
        """Validate that a value is JSON serializable.

        Args:
            value: The value to validate

        Raises:
            ValueError: If value is not JSON serializable
        """
        try:
            json.dumps(value)
        except (TypeError, ValueError) as e:
            raise ValueError(
                f"Value is not JSON serializable: {type(value).__name__}. "
                f"Only JSON-compatible types (str, int, float, bool, list, dict, None) are allowed."
            ) from e

__init__(initial_state=None)

Initialize JSONSerializableDict.

Source code in strands/types/json_dict.py

def __init__(self, initial_state: dict[str, Any] | None = None):
    """Initialize JSONSerializableDict."""
    self._data: dict[str, Any]
    if initial_state:
        self._validate_json_serializable(initial_state)
        self._data = copy.deepcopy(initial_state)
    else:
        self._data = {}

delete(key)

Delete a specific key from the store.

Parameters:

Name	Type	Description	Default
key	str	The key to delete	required

Source code in strands/types/json_dict.py

def delete(self, key: str) -> None:
    """Delete a specific key from the store.

    Args:
        key: The key to delete
    """
    self._validate_key(key)
    self._data.pop(key, None)

get(key=None)

Get a value or entire data.

Parameters:

Name	Type	Description	Default
key	str | None	The key to retrieve (if None, returns entire data dict)	None

Returns:

Type	Description
Any	The stored value, entire data dict, or None if not found

Source code in strands/types/json_dict.py

def get(self, key: str | None = None) -> Any:
    """Get a value or entire data.

    Args:
        key: The key to retrieve (if None, returns entire data dict)

    Returns:
        The stored value, entire data dict, or None if not found
    """
    if key is None:
        return copy.deepcopy(self._data)
    else:
        return copy.deepcopy(self._data.get(key))

set(key, value)

Set a value in the store.

Parameters:

Name	Type	Description	Default
key	str	The key to store the value under	required
value	Any	The value to store (must be JSON serializable)	required

Raises:

Type	Description
ValueError	If key is invalid, or if value is not JSON serializable

Source code in strands/types/json_dict.py

def set(self, key: str, value: Any) -> None:
    """Set a value in the store.

    Args:
        key: The key to store the value under
        value: The value to store (must be JSON serializable)

    Raises:
        ValueError: If key is invalid, or if value is not JSON serializable
    """
    self._validate_key(key)
    self._validate_json_serializable(value)
    self._data[key] = copy.deepcopy(value)

SteeringContext dataclass

Container for steering context data.

Source code in strands/experimental/steering/core/context.py

@dataclass
class SteeringContext:
    """Container for steering context data."""

    """Container for steering context data.

    This class should not be instantiated directly - it is intended for internal use only.
    """

    data: JSONSerializableDict = field(default_factory=JSONSerializableDict)

SteeringContextCallback

Bases: ABC, Generic[EventType]

Abstract base class for steering context update callbacks.

Source code in strands/experimental/steering/core/context.py

class SteeringContextCallback(ABC, Generic[EventType]):
    """Abstract base class for steering context update callbacks."""

    @property
    def event_type(self) -> type[HookEvent]:
        """Return the event type this callback handles."""
        for base in getattr(self.__class__, "__orig_bases__", ()):
            if get_origin(base) is SteeringContextCallback:
                return cast(type[HookEvent], get_args(base)[0])
        raise ValueError("Could not determine event type from generic parameter")

    def __call__(self, event: EventType, steering_context: "SteeringContext", **kwargs: Any) -> None:
        """Update steering context based on hook event.

        Args:
            event: The hook event that triggered the callback
            steering_context: The steering context to update
            **kwargs: Additional keyword arguments for context updates
        """
        ...

event_type property

Return the event type this callback handles.

__call__(event, steering_context, **kwargs)

Update steering context based on hook event.

Parameters:

Name	Type	Description	Default
event	EventType	The hook event that triggered the callback	required
steering_context	SteeringContext	The steering context to update	required
**kwargs	Any	Additional keyword arguments for context updates	{}

Source code in strands/experimental/steering/core/context.py

def __call__(self, event: EventType, steering_context: "SteeringContext", **kwargs: Any) -> None:
    """Update steering context based on hook event.

    Args:
        event: The hook event that triggered the callback
        steering_context: The steering context to update
        **kwargs: Additional keyword arguments for context updates
    """
    ...

SteeringContextProvider

Bases: ABC

Abstract base class for context providers that handle multiple event types.

Source code in strands/experimental/steering/core/context.py

class SteeringContextProvider(ABC):
    """Abstract base class for context providers that handle multiple event types."""

    @abstractmethod
    def context_providers(self, **kwargs: Any) -> list[SteeringContextCallback]:
        """Return list of context callbacks with event types extracted from generics."""
        ...

context_providers(**kwargs) abstractmethod

Return list of context callbacks with event types extracted from generics.

Source code in strands/experimental/steering/core/context.py

@abstractmethod
def context_providers(self, **kwargs: Any) -> list[SteeringContextCallback]:
    """Return list of context callbacks with event types extracted from generics."""
    ...