Skip to content

agent_k.toolsets.memory

Persistent key-value memory toolset for cross-agent communication.

agent_k.toolsets.memory

Memory tool helpers for AGENT-K agents.

@notice: | Memory tool helpers for AGENT-K agents.

@dev: | See module for implementation details and extension points.

@graph: id: agent_k.toolsets.memory provides: - agent_k.toolsets.memory:AgentKMemoryTool - agent_k.toolsets.memory:create_memory_backend - agent_k.toolsets.memory:prepare_memory_tool - agent_k.toolsets.memory:register_memory_tool pattern: toolset

@similar: - id: agent_k.embeddings.store when: "Vector store persistence; this module is file-backed memory tool."

@agent-guidance: do: - "Use agent_k.toolsets.memory as the canonical home for this capability." do_not: - "Create parallel modules without updating @similar or @graph."

@human-review: last-verified: 2026-01-26 owners: - agent-k-core

(c) Mike Casale 2025. Licensed under the MIT License.

AgentKMemoryTool

Bases: _MemoryBase

File-backed memory implementation for Anthropic MemoryTool.

@pattern: name: memory-tool rationale: "Provides a file-backed memory interface for agents." violations: "Bypassing this tool breaks memory consistency."

@concurrency: model: asyncio safe: false reason: "Performs filesystem mutations without locks."

Source code in agent_k/toolsets/memory.py 
 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
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
class AgentKMemoryTool(_MemoryBase):  # pragma: no cover - optional dependency
    """File-backed memory implementation for Anthropic MemoryTool.

    @pattern:
        name: memory-tool
        rationale: "Provides a file-backed memory interface for agents."
        violations: "Bypassing this tool breaks memory consistency."

    @concurrency:
        model: asyncio
        safe: false
        reason: "Performs filesystem mutations without locks."
    """

    def __init__(self, base_path: Annotated[Path | None, Doc("Base path for memory storage.")] = None) -> None:
        if _anthropic_memory is None:
            raise RuntimeError("anthropic is required to use AgentKMemoryTool")
        super().__init__()
        self._base_path = (base_path or _DEFAULT_MEMORY_DIR).expanduser().resolve()
        self._base_path.mkdir(parents=True, exist_ok=True)

    def view(self, command: Any) -> str:
        """View file contents or list directory entries."""
        with logfire.span("memory.view", path=command.path):
            try:
                path = self._resolve_path(command.path)
            except ValueError as exc:
                return f"Error: {exc}"

            if not path.exists():
                return f"Error: {command.path} not found."

            if path.is_dir():
                entries = [
                    f"{child.name}{'/' if child.is_dir() else ''}"
                    for child in sorted(path.iterdir(), key=lambda p: p.name)
                ]
                return "\n".join(entries) if entries else "(empty directory)"

            text = self._read_text(path)
            if command.view_range:
                lines = text.splitlines()
                start, end = _normalize_view_range(command.view_range, len(lines))
                return "\n".join(lines[start - 1 : end])
            return text

    def create(self, command: Any) -> str:
        """Create a file with the provided contents."""
        with logfire.span("memory.create", path=command.path):
            try:
                path = self._resolve_path(command.path)
            except ValueError as exc:
                return f"Error: {exc}"

            if path.exists():
                return f"Error: {command.path} already exists."

            if command.file_text is None:
                return "Error: file_text must be a string."

            self._write_text(path, command.file_text)
            return f"Created {command.path}."

    def str_replace(self, command: Any) -> str:
        """Replace matching text in a file."""
        with logfire.span("memory.str_replace", path=command.path):
            try:
                path = self._resolve_path(command.path)
            except ValueError as exc:
                return f"Error: {exc}"

            if not path.exists():
                return f"Error: {command.path} not found."

            if command.old_str is None or command.new_str is None:
                return "Error: old_str and new_str must be strings."

            text = self._read_text(path)
            occurrences = text.count(command.old_str)
            if occurrences == 0:
                return f'Error: "{command.old_str}" not found in {command.path}.'

            updated = text.replace(command.old_str, command.new_str)
            self._write_text(path, updated)
            return f"Replaced {occurrences} occurrence(s) in {command.path}."

    def insert(self, command: Any) -> str:
        """Insert text at a specified line in a file."""
        with logfire.span("memory.insert", path=command.path):
            try:
                path = self._resolve_path(command.path)
            except ValueError as exc:
                return f"Error: {exc}"

            if not path.exists():
                return f"Error: {command.path} not found."

            if command.insert_text is None:
                return "Error: insert_text must be a string."

            text = self._read_text(path)
            lines = text.splitlines()
            index = max(command.insert_line - 1, 0)
            index = min(index, len(lines))
            lines.insert(index, command.insert_text)
            updated = "\n".join(lines)
            if text.endswith("\n"):
                updated += "\n"
            self._write_text(path, updated)
            return f"Inserted text at line {command.insert_line} in {command.path}."

    def delete(self, command: Any) -> str:
        """Delete a file or directory."""
        with logfire.span("memory.delete", path=command.path):
            try:
                path = self._resolve_path(command.path)
            except ValueError as exc:
                return f"Error: {exc}"

            if not path.exists():
                return f"Error: {command.path} not found."

            if path.is_dir():
                shutil.rmtree(path)
            else:
                path.unlink()
            return f"Deleted {command.path}."

    def rename(self, command: Any) -> str:
        """Rename a file or directory."""
        with logfire.span("memory.rename", path=command.old_path, new_path=command.new_path):
            try:
                old_path = self._resolve_path(command.old_path)
                new_path = self._resolve_path(command.new_path)
            except ValueError as exc:
                return f"Error: {exc}"

            if not old_path.exists():
                return f"Error: {command.old_path} not found."

            new_path.parent.mkdir(parents=True, exist_ok=True)
            old_path.rename(new_path)
            return f"Renamed {command.old_path} to {command.new_path}."

    def _resolve_path(self, path: str) -> Path:
        candidate = (self._base_path / path).resolve()
        if not candidate.is_relative_to(self._base_path):
            raise ValueError(f"Path escapes memory root: {path}")
        return candidate

    def _read_text(self, path: Path) -> str:
        return path.read_text(encoding="utf-8")

    def _write_text(self, path: Path, text: str) -> None:
        path.parent.mkdir(parents=True, exist_ok=True)
        path.write_text(text, encoding="utf-8")
view 
view(command: Any) -> str

View file contents or list directory entries.

Source code in agent_k/toolsets/memory.py 
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
def view(self, command: Any) -> str:
    """View file contents or list directory entries."""
    with logfire.span("memory.view", path=command.path):
        try:
            path = self._resolve_path(command.path)
        except ValueError as exc:
            return f"Error: {exc}"

        if not path.exists():
            return f"Error: {command.path} not found."

        if path.is_dir():
            entries = [
                f"{child.name}{'/' if child.is_dir() else ''}"
                for child in sorted(path.iterdir(), key=lambda p: p.name)
            ]
            return "\n".join(entries) if entries else "(empty directory)"

        text = self._read_text(path)
        if command.view_range:
            lines = text.splitlines()
            start, end = _normalize_view_range(command.view_range, len(lines))
            return "\n".join(lines[start - 1 : end])
        return text
create 
create(command: Any) -> str

Create a file with the provided contents.

Source code in agent_k/toolsets/memory.py 
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
def create(self, command: Any) -> str:
    """Create a file with the provided contents."""
    with logfire.span("memory.create", path=command.path):
        try:
            path = self._resolve_path(command.path)
        except ValueError as exc:
            return f"Error: {exc}"

        if path.exists():
            return f"Error: {command.path} already exists."

        if command.file_text is None:
            return "Error: file_text must be a string."

        self._write_text(path, command.file_text)
        return f"Created {command.path}."
str_replace 
str_replace(command: Any) -> str

Replace matching text in a file.

Source code in agent_k/toolsets/memory.py 
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
def str_replace(self, command: Any) -> str:
    """Replace matching text in a file."""
    with logfire.span("memory.str_replace", path=command.path):
        try:
            path = self._resolve_path(command.path)
        except ValueError as exc:
            return f"Error: {exc}"

        if not path.exists():
            return f"Error: {command.path} not found."

        if command.old_str is None or command.new_str is None:
            return "Error: old_str and new_str must be strings."

        text = self._read_text(path)
        occurrences = text.count(command.old_str)
        if occurrences == 0:
            return f'Error: "{command.old_str}" not found in {command.path}.'

        updated = text.replace(command.old_str, command.new_str)
        self._write_text(path, updated)
        return f"Replaced {occurrences} occurrence(s) in {command.path}."
insert 
insert(command: Any) -> str

Insert text at a specified line in a file.

Source code in agent_k/toolsets/memory.py 
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
def insert(self, command: Any) -> str:
    """Insert text at a specified line in a file."""
    with logfire.span("memory.insert", path=command.path):
        try:
            path = self._resolve_path(command.path)
        except ValueError as exc:
            return f"Error: {exc}"

        if not path.exists():
            return f"Error: {command.path} not found."

        if command.insert_text is None:
            return "Error: insert_text must be a string."

        text = self._read_text(path)
        lines = text.splitlines()
        index = max(command.insert_line - 1, 0)
        index = min(index, len(lines))
        lines.insert(index, command.insert_text)
        updated = "\n".join(lines)
        if text.endswith("\n"):
            updated += "\n"
        self._write_text(path, updated)
        return f"Inserted text at line {command.insert_line} in {command.path}."
delete 
delete(command: Any) -> str

Delete a file or directory.

Source code in agent_k/toolsets/memory.py 
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
def delete(self, command: Any) -> str:
    """Delete a file or directory."""
    with logfire.span("memory.delete", path=command.path):
        try:
            path = self._resolve_path(command.path)
        except ValueError as exc:
            return f"Error: {exc}"

        if not path.exists():
            return f"Error: {command.path} not found."

        if path.is_dir():
            shutil.rmtree(path)
        else:
            path.unlink()
        return f"Deleted {command.path}."
rename 
rename(command: Any) -> str

Rename a file or directory.

Source code in agent_k/toolsets/memory.py 
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
def rename(self, command: Any) -> str:
    """Rename a file or directory."""
    with logfire.span("memory.rename", path=command.old_path, new_path=command.new_path):
        try:
            old_path = self._resolve_path(command.old_path)
            new_path = self._resolve_path(command.new_path)
        except ValueError as exc:
            return f"Error: {exc}"

        if not old_path.exists():
            return f"Error: {command.old_path} not found."

        new_path.parent.mkdir(parents=True, exist_ok=True)
        old_path.rename(new_path)
        return f"Renamed {command.old_path} to {command.new_path}."

create_memory_backend 

create_memory_backend(
    storage_path: Annotated[
        Path | None, Doc("Base directory for memory files.")
    ] = None,
) -> AgentKMemoryTool

Create an Anthropic-compatible memory backend.

@notice: | Creates a file-backed memory tool for Anthropic providers.

@factory-for: id: agent_k.toolsets.memory:AgentKMemoryTool rationale: "Centralizes default storage path behavior." singleton: false cache-key: storage_path

@canonical-home: for: - "memory backend construction" notes: "Use create_memory_backend to ensure defaults."

Source code in agent_k/toolsets/memory.py 
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
def create_memory_backend(
    storage_path: Annotated[Path | None, Doc("Base directory for memory files.")] = None,
) -> AgentKMemoryTool:
    """Create an Anthropic-compatible memory backend.

    @notice: |
        Creates a file-backed memory tool for Anthropic providers.

    @factory-for:
        id: agent_k.toolsets.memory:AgentKMemoryTool
        rationale: "Centralizes default storage path behavior."
        singleton: false
        cache-key: storage_path

    @canonical-home:
        for:
            - "memory backend construction"
        notes: "Use create_memory_backend to ensure defaults."
    """
    return AgentKMemoryTool(base_path=storage_path)

prepare_memory_tool async 

prepare_memory_tool(
    ctx: Annotated[
        RunContext[Any],
        Doc("Run context for tool preparation."),
    ],
) -> MemoryTool | None

Dynamically enable MemoryTool only for supported providers.

@notice: | Returns MemoryTool only for Anthropic models.

Source code in agent_k/toolsets/memory.py 
256
257
258
259
260
261
262
263
264
async def prepare_memory_tool(
    ctx: Annotated[RunContext[Any], Doc("Run context for tool preparation.")],
) -> MemoryTool | None:
    """Dynamically enable MemoryTool only for supported providers.

    @notice: |
        Returns MemoryTool only for Anthropic models.
    """
    return None if ctx.model.system != "anthropic" else MemoryTool()

register_memory_tool 

register_memory_tool(
    agent: Annotated[
        Agent[Any, Any],
        Doc("Agent instance to register the tool on."),
    ],
    memory_backend: Annotated[
        AgentKMemoryTool,
        Doc("Memory backend implementation."),
    ],
) -> None

Register the Anthropic MemoryTool handler on an agent.

@notice: | Attaches the memory tool to the agent with a plain tool handler.

@effects: state: - agent tool registry

Source code in agent_k/toolsets/memory.py 
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
def register_memory_tool(
    agent: Annotated[Agent[Any, Any], Doc("Agent instance to register the tool on.")],
    memory_backend: Annotated[AgentKMemoryTool, Doc("Memory backend implementation.")],
) -> None:
    """Register the Anthropic MemoryTool handler on an agent.

    @notice: |
        Attaches the memory tool to the agent with a plain tool handler.

    @effects:
        state:
            - agent tool registry
    """

    @agent.tool_plain(name="memory", prepare=_prepare_memory_definition)
    def memory(**command: Any) -> Any:
        return memory_backend.call(command)