Skip to content

Memory Client

agent_memory_hub.client.memory_client.MemoryClient

Client for accessing agent memory with region governance.

Source code in agent_memory_hub/client/memory_client.py 
 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
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
class MemoryClient:
    """
    Client for accessing agent memory with region governance.
    """

    def __init__(
        self,
        agent_id: str,
        session_id: str,
        region: str = DEFAULT_REGION,
        region_restricted: bool = True,
        backend: str = "adk",
        ttl_seconds: Optional[int] = None,
        alloydb_config: Optional["AlloyDBConfig"] = None,
        redis_config: Optional["RedisConfig"] = None,
        environment: str = "prod",
    ):
        """
        Initialize the MemoryClient.

        Args:
            agent_id: Unique identifier for the agent.
            session_id: Unique identifier for the session.
            region: The cloud region where memory should be stored/retrieved.
            region_restricted: If True, enforces strict region checks.
            backend: Storage backend ("adk" for GCS, "alloydb" for AlloyDB).
            ttl_seconds: Time-to-live in seconds (None = no expiry).
            alloydb_config: AlloyDB configuration (required if backend="alloydb").
            redis_config: Redis configuration (optional if backend="redis").
            environment: Environment context (e.g., "prod", "dev") for resource naming.
        """
        if not agent_id:
            raise ValueError("agent_id cannot be empty")
        if not session_id:
            raise ValueError("session_id cannot be empty")
        self.agent_id = agent_id
        self.session_id = session_id
        self.region = region
        self.region_restricted = region_restricted
        self.backend = backend
        self.ttl_seconds = ttl_seconds
        self.environment = environment
        self._tracer = get_tracer()

        if region_restricted:
            self._guard = RegionGuard(region)
            self._router = MemoryRouter(
                region_guard=self._guard,
                backend=backend,
                ttl_seconds=ttl_seconds,
                alloydb_config=alloydb_config,
                redis_config=redis_config,
                environment=environment,
            )
        else:
            # Fallback or less strict mode not fully implemented in spec, 
            # assuming guard is always used but maybe with relaxed checks if requested.
            # For now, we strictly follow the requested design where region is passed.
            self._guard = RegionGuard(region)
            self._router = MemoryRouter(
                region_guard=self._guard,
                backend=backend,
                ttl_seconds=ttl_seconds,
                alloydb_config=alloydb_config,
                redis_config=redis_config,
                environment=environment,
            )

    def write(self, value: Any, key: str = "default") -> None:
        """
        Write a value to the memory store.

        Args:
            value: The data to store.
            key: specific key or context for the memory (e.g., 'episodic', 'semantic').
        """
        with self._tracer.start_as_current_span("MemoryClient.write") as span:
            span.set_attribute("agent.id", self.agent_id)
            span.set_attribute("session.id", self.session_id)
            span.set_attribute("region", self.region)
            span.set_attribute("memory.key", key)

            # Composite key could include agent_id to namespace it
            composite_key = f"{self.agent_id}/{key}"
            self._router.write(self.session_id, composite_key, value)

    def write_model(self, memory_model: "BaseMemory") -> None:
        """
        Write a semantic memory model to the store.

        Args:
            memory_model: Pydantic model instance (EpisodicMemory, SemanticMemory, etc.)
        """
        # Ensure agent_id matches client if not set (though model has default)
        if hasattr(memory_model, "agent_id") and not memory_model.agent_id:
             memory_model.agent_id = self.agent_id

        # Use model ID or content hash as key? 
        # For now, we use a predictable key scheme: type/id
        key = f"{memory_model.__class__.__name__.lower()}/{memory_model.id}"

        # Serialize to dict
        value = memory_model.to_dict()

        with self._tracer.start_as_current_span("MemoryClient.write_model") as span:
            span.set_attribute("memory.type", memory_model.__class__.__name__)
            span.set_attribute("memory.id", memory_model.id)

            self.write(value, key=key)

    def recall(self, key: str = "default") -> Optional[Any]:
        """
        Recall a value from the memory store.

        Args:
            key: The key used during write.

        Returns:
            The stored value or None if not found.
        """
        with self._tracer.start_as_current_span("MemoryClient.recall") as span:
            span.set_attribute("agent.id", self.agent_id)
            span.set_attribute("session.id", self.session_id)
            span.set_attribute("region", self.region)
            span.set_attribute("memory.key", key)

            composite_key = f"{self.agent_id}/{key}"
            return self._router.read(self.session_id, composite_key)

__init__(agent_id, session_id, region=DEFAULT_REGION, region_restricted=True, backend='adk', ttl_seconds=None, alloydb_config=None, redis_config=None, environment='prod')

Initialize the MemoryClient.

Parameters:

Name Type Description Default
agent_id str

Unique identifier for the agent.

 required
session_id str

Unique identifier for the session.

 required
region str

The cloud region where memory should be stored/retrieved.

 DEFAULT_REGION
region_restricted bool

If True, enforces strict region checks.

 True
backend str

Storage backend ("adk" for GCS, "alloydb" for AlloyDB).

 'adk'
ttl_seconds Optional[int]

Time-to-live in seconds (None = no expiry).

 None
alloydb_config Optional[AlloyDBConfig]

AlloyDB configuration (required if backend="alloydb").

 None
redis_config Optional[RedisConfig]

Redis configuration (optional if backend="redis").

 None
environment str

Environment context (e.g., "prod", "dev") for resource naming.

 'prod'
Source code in agent_memory_hub/client/memory_client.py 
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
def __init__(
    self,
    agent_id: str,
    session_id: str,
    region: str = DEFAULT_REGION,
    region_restricted: bool = True,
    backend: str = "adk",
    ttl_seconds: Optional[int] = None,
    alloydb_config: Optional["AlloyDBConfig"] = None,
    redis_config: Optional["RedisConfig"] = None,
    environment: str = "prod",
):
    """
    Initialize the MemoryClient.

    Args:
        agent_id: Unique identifier for the agent.
        session_id: Unique identifier for the session.
        region: The cloud region where memory should be stored/retrieved.
        region_restricted: If True, enforces strict region checks.
        backend: Storage backend ("adk" for GCS, "alloydb" for AlloyDB).
        ttl_seconds: Time-to-live in seconds (None = no expiry).
        alloydb_config: AlloyDB configuration (required if backend="alloydb").
        redis_config: Redis configuration (optional if backend="redis").
        environment: Environment context (e.g., "prod", "dev") for resource naming.
    """
    if not agent_id:
        raise ValueError("agent_id cannot be empty")
    if not session_id:
        raise ValueError("session_id cannot be empty")
    self.agent_id = agent_id
    self.session_id = session_id
    self.region = region
    self.region_restricted = region_restricted
    self.backend = backend
    self.ttl_seconds = ttl_seconds
    self.environment = environment
    self._tracer = get_tracer()

    if region_restricted:
        self._guard = RegionGuard(region)
        self._router = MemoryRouter(
            region_guard=self._guard,
            backend=backend,
            ttl_seconds=ttl_seconds,
            alloydb_config=alloydb_config,
            redis_config=redis_config,
            environment=environment,
        )
    else:
        # Fallback or less strict mode not fully implemented in spec, 
        # assuming guard is always used but maybe with relaxed checks if requested.
        # For now, we strictly follow the requested design where region is passed.
        self._guard = RegionGuard(region)
        self._router = MemoryRouter(
            region_guard=self._guard,
            backend=backend,
            ttl_seconds=ttl_seconds,
            alloydb_config=alloydb_config,
            redis_config=redis_config,
            environment=environment,
        )

recall(key='default')

Recall a value from the memory store.

Parameters:

Name Type Description Default
key str

The key used during write.

 'default'

Returns:

Type Description
Optional[Any]

The stored value or None if not found.
Source code in agent_memory_hub/client/memory_client.py 
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
def recall(self, key: str = "default") -> Optional[Any]:
    """
    Recall a value from the memory store.

    Args:
        key: The key used during write.

    Returns:
        The stored value or None if not found.
    """
    with self._tracer.start_as_current_span("MemoryClient.recall") as span:
        span.set_attribute("agent.id", self.agent_id)
        span.set_attribute("session.id", self.session_id)
        span.set_attribute("region", self.region)
        span.set_attribute("memory.key", key)

        composite_key = f"{self.agent_id}/{key}"
        return self._router.read(self.session_id, composite_key)

write(value, key='default')

Write a value to the memory store.

Parameters:

Name Type Description Default
value Any

The data to store.

 required
key str

specific key or context for the memory (e.g., 'episodic', 'semantic').

 'default'
Source code in agent_memory_hub/client/memory_client.py 
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
def write(self, value: Any, key: str = "default") -> None:
    """
    Write a value to the memory store.

    Args:
        value: The data to store.
        key: specific key or context for the memory (e.g., 'episodic', 'semantic').
    """
    with self._tracer.start_as_current_span("MemoryClient.write") as span:
        span.set_attribute("agent.id", self.agent_id)
        span.set_attribute("session.id", self.session_id)
        span.set_attribute("region", self.region)
        span.set_attribute("memory.key", key)

        # Composite key could include agent_id to namespace it
        composite_key = f"{self.agent_id}/{key}"
        self._router.write(self.session_id, composite_key, value)

write_model(memory_model)

Write a semantic memory model to the store.

Parameters:

Name Type Description Default
memory_model BaseMemory

Pydantic model instance (EpisodicMemory, SemanticMemory, etc.)

 required
Source code in agent_memory_hub/client/memory_client.py 
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
def write_model(self, memory_model: "BaseMemory") -> None:
    """
    Write a semantic memory model to the store.

    Args:
        memory_model: Pydantic model instance (EpisodicMemory, SemanticMemory, etc.)
    """
    # Ensure agent_id matches client if not set (though model has default)
    if hasattr(memory_model, "agent_id") and not memory_model.agent_id:
         memory_model.agent_id = self.agent_id

    # Use model ID or content hash as key? 
    # For now, we use a predictable key scheme: type/id
    key = f"{memory_model.__class__.__name__.lower()}/{memory_model.id}"

    # Serialize to dict
    value = memory_model.to_dict()

    with self._tracer.start_as_current_span("MemoryClient.write_model") as span:
        span.set_attribute("memory.type", memory_model.__class__.__name__)
        span.set_attribute("memory.id", memory_model.id)

        self.write(value, key=key)