Structured query against the Postgres structured store. Use this for exact lookups like 'get fact with key X' or 'list all statuses'. Args: memory_type: 'events', 'facts', or 'statuses'. key: For facts — filter by key. subject:
(
self,
memory_type: str = "events",
key: str | None = None,
subject: str | None = None,
since: str | None = None,
source_agent: str | None = None,
client_id: str | None = None,
)
| 163 | # ----- Structured query ----- |
| 164 | |
| 165 | def query( |
| 166 | self, |
| 167 | memory_type: str = "events", |
| 168 | key: str | None = None, |
| 169 | subject: str | None = None, |
| 170 | since: str | None = None, |
| 171 | source_agent: str | None = None, |
| 172 | client_id: str | None = None, |
| 173 | ) -> dict: |
| 174 | """ |
| 175 | Structured query against the Postgres structured store. |
| 176 | |
| 177 | Use this for exact lookups like 'get fact with key X' or 'list all statuses'. |
| 178 | |
| 179 | Args: |
| 180 | memory_type: 'events', 'facts', or 'statuses'. |
| 181 | key: For facts — filter by key. |
| 182 | subject: For statuses — filter by subject. |
| 183 | since: For events — filter by timestamp (ISO 8601). |
| 184 | source_agent: Filter by agent. |
| 185 | client_id: Filter by client. |
| 186 | |
| 187 | Returns: |
| 188 | Dict with 'type', 'count', and 'results' list. |
| 189 | """ |
| 190 | params = {"type": memory_type} |
| 191 | if key: |
| 192 | params["key"] = key |
| 193 | if subject: |
| 194 | params["subject"] = subject |
| 195 | if since: |
| 196 | params["since"] = since |
| 197 | if source_agent: |
| 198 | params["source_agent"] = source_agent |
| 199 | if client_id: |
| 200 | params["client_id"] = client_id |
| 201 | |
| 202 | resp = self.session.get(f"{self.api_url}/memory/query", params=params) |
| 203 | resp.raise_for_status() |
| 204 | return resp.json() |
| 205 | |
| 206 | # ----- Delete a memory ----- |
| 207 |
no outgoing calls
no test coverage detected