Skip to content

strands.agent.state

Agent state management.

AgentState = JSONSerializableDict module-attribute

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
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
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]
        self._version: int = 0
        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)
        self._version += 1

    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)
        self._version += 1

    def _get_version(self) -> int:
        """Get the current version number of the store.

        The version is incremented each time set() or delete() is called.
        Consumers can compare versions to detect changes without requiring
        explicit dirty flag clearing.

        Returns:
            The current version number.
        """
        return self._version

    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
15
16
17
18
19
20
21
22
23
def __init__(self, initial_state: dict[str, Any] | None = None):
    """Initialize JSONSerializableDict."""
    self._data: dict[str, Any]
    self._version: int = 0
    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
54
55
56
57
58
59
60
61
62
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)
    self._version += 1

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
40
41
42
43
44
45
46
47
48
49
50
51
52
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
25
26
27
28
29
30
31
32
33
34
35
36
37
38
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)
    self._version += 1