feat(agents): Agent Designer Phase 3 — memory_* tools (write side)

backend/src/agents/builtin_tools/memory_spaces/__init__.py

@@ -0,0 +1,13 @@
+"""Context-bound Memory-Space tools for an Agent's bound space (Agent Designer Phase 3)."""
+
+from .tools import (
+    make_memory_list_tool,
+    make_memory_read_tool,
+    make_memory_write_tool,
+)
+
+__all__ = [
+    "make_memory_list_tool",
+    "make_memory_read_tool",
+    "make_memory_write_tool",
+]

backend/src/agents/builtin_tools/memory_spaces/tools.py

@@ -0,0 +1,130 @@
+"""Context-bound tools for an Agent's bound Memory Space (Agent Designer Phase 3).
+
+Each factory closes over the *binding's* space id and the *invoking* user's identity, so
+the returned tool can physically only address that one space as that one user — it cannot
+be re-pointed at another space. ``MemorySpaceService`` re-checks the caller's grant
+(``viewer+`` for reads, ``editor+`` for writes) on every call, so a permission revoked
+mid-session surfaces as an error tool-result on the next call rather than leaking access.
+Same closure-identity + ``asyncio.to_thread`` pattern as the artifact/spreadsheet tools
+(the codebase has no tool-execution contextvar; ``MemorySpaceService`` is sync boto3).
+
+These are injected via the invocation path's ``extra_tools`` seam only when an Agent has a
+resolved ``memory_space`` binding — agents with ``extra_tools`` are never cached, so a tool
+closed over user A's identity can never be served to user B.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import logging
+from typing import Any, Optional
+
+from strands import tool
+
+from apis.shared.memory.service import (
+    MemorySpaceError,
+    MemorySpaceNotFoundError,
+    MemorySpacePermissionError,
+    MemorySpaceService,
+)
+
+logger = logging.getLogger(__name__)
+
+
+def _error(text: str) -> dict[str, Any]:
+    return {"content": [{"text": f"❌ {text}"}], "status": "error"}
+
+
+def make_memory_list_tool(space_id: str, space_name: str, user_id: str, user_email: Optional[str]):
+    @tool
+    async def memory_list(entry_type: Optional[str] = None) -> dict[str, Any]:
+        """List the entries in your bound memory space (manifest only — no content).
+
+        Use this to see what you remember before deciding whether to `memory_read` a
+        specific entry. Returns each entry's slug, type, description, and last-updated
+        time. Optionally filter by `entry_type` ("entity", "episodic", or "fact").
+
+        Args:
+            entry_type: Optional filter — one of "entity", "episodic", "fact".
+        """
+        try:
+            entries = await asyncio.to_thread(
+                MemorySpaceService().list_entries,
+                space_id, user_id, user_email, entry_type=entry_type,
+            )
+        except MemorySpacePermissionError as exc:
+            return _error(f"You no longer have access to memory space '{space_name}': {exc}")
+        except MemorySpaceError as exc:
+            return _error(f"Could not list memory: {exc}")
+
+        summary = [
+            {"slug": e.slug, "type": e.entry_type, "description": e.description, "updated": e.updated}
+            for e in entries
+        ]
+        return {"content": [{"json": {"entries": summary}}], "status": "success"}
+
+    return memory_list
+
+
+def make_memory_read_tool(space_id: str, space_name: str, user_id: str, user_email: Optional[str]):
+    @tool
+    async def memory_read(slug: str) -> dict[str, Any]:
+        """Read the full content of one entry in your bound memory space.
+
+        Pass a `slug` from `memory_list` (or referenced in your injected memory index).
+
+        Args:
+            slug: The entry's stable id within the space (e.g. "jane-doe").
+        """
+        try:
+            body = await asyncio.to_thread(
+                MemorySpaceService().read_entry, space_id, user_id, user_email, slug
+            )
+        except MemorySpaceNotFoundError:
+            return _error(f"No memory entry '{slug}' exists in '{space_name}'.")
+        except MemorySpacePermissionError as exc:
+            return _error(f"You no longer have access to memory space '{space_name}': {exc}")
+        except MemorySpaceError as exc:
+            return _error(f"Could not read memory entry '{slug}': {exc}")
+        return {"content": [{"text": body}], "status": "success"}
+
+    return memory_read
+
+
+def make_memory_write_tool(space_id: str, space_name: str, user_id: str, user_email: Optional[str]):
+    @tool
+    async def memory_write(
+        slug: str,
+        body: str,
+        entry_type: str = "fact",
+        description: str = "",
+    ) -> dict[str, Any]:
+        """Create or replace an entry in your bound memory space (persists across sessions).
+
+        Use this to remember durable facts, people/entities, or episodic notes the user
+        will want recalled later. Writing an existing `slug` replaces that entry. Only
+        available when the agent's memory binding grants write access.
+
+        Args:
+            slug: Stable id for the entry (e.g. "jane-doe", "daily-2026-07-07").
+            body: The entry's markdown content.
+            entry_type: "entity", "episodic", or "fact" (default "fact").
+            description: Short one-line summary shown in listings.
+        """
+        try:
+            ref = await asyncio.to_thread(
+                lambda: MemorySpaceService().write_entry(
+                    space_id, user_id, user_email, slug, body,
+                    entry_type=entry_type, description=description,
+                )
+            )
+        except MemorySpacePermissionError as exc:
+            return _error(f"You don't have write access to memory space '{space_name}': {exc}")
+        except MemorySpaceError as exc:
+            return _error(f"Could not write memory entry '{slug}': {exc}")
+        return {
+            "content": [{"text": f'Saved memory entry "{ref.slug}" ({ref.entry_type}) to "{space_name}".'}],
+            "status": "success",
+        }
+
+    return memory_write

backend/src/apis/inference_api/chat/routes.py

@@ -404,6 +404,36 @@ def _build_artifact_tools(
     return tools
 
 
+def _build_memory_tools(agent_memory, user_id: str, user_email: str) -> list:
+    """Context-bound Memory-Space tools for an Agent's resolved memory binding.
+
+    ``agent_memory`` is the resolver's ``ResolvedMemoryBinding`` (or ``None``). No binding
+    → no tools. Read tools (list + read) are always exposed; the write tool only when the
+    binding grants ``readwrite`` — and the service re-checks ``editor+`` on every call, so
+    this is a UX gate, not the security boundary. Not gated on ``enabled_tools``: the
+    governing capability is the Agent's binding, not the user's tool picker.
+    """
+    if agent_memory is None:
+        return []
+
+    from agents.builtin_tools.memory_spaces import (
+        make_memory_list_tool,
+        make_memory_read_tool,
+        make_memory_write_tool,
+    )
+
+    space_id, space_name = agent_memory.space_id, agent_memory.space_name
+    tools = [
+        make_memory_list_tool(space_id, space_name, user_id, user_email),
+        make_memory_read_tool(space_id, space_name, user_id, user_email),
+    ]
+    if agent_memory.access == "readwrite":
+        tools.append(make_memory_write_tool(space_id, space_name, user_id, user_email))
+
+    logger.info(f"Created {len(tools)} memory-space tools for bound space")
+    return tools
+
+
 # ============================================================
 # Attachment Partitioning (#206)
 # ============================================================
@@ -1606,6 +1636,10 @@ async def invocations(request: InvocationRequest, current_user: User = Depends(g
                 enabled_tools=input_data.enabled_tools,
                 session_id=input_data.session_id,
                 user_id=user_id,
+            ) + _build_memory_tools(
+                agent_memory=agent_memory,
+                user_id=user_id,
+                user_email=current_user.email,
             )
 
             agent = await get_agent(

backend/tests/agents/builtin_tools/memory_spaces/test_memory_tools.py

@@ -0,0 +1,100 @@
+"""Agent Designer Phase 3 — memory_* tool factories.
+
+Each tool is closed over the bound space id + invoker identity; MemorySpaceService is
+patched. Verifies success payloads and that a revoked grant (permission error) surfaces
+as an error tool-result rather than raising.
+"""
+
+from types import SimpleNamespace
+from unittest.mock import MagicMock
+
+import pytest
+
+from agents.builtin_tools.memory_spaces import (
+    make_memory_list_tool,
+    make_memory_read_tool,
+    make_memory_write_tool,
+)
+from apis.shared.memory.service import (
+    MemorySpaceNotFoundError,
+    MemorySpacePermissionError,
+)
+
+MODULE = "agents.builtin_tools.memory_spaces.tools"
+
+
+async def _call(tool, *args, **kwargs):
+    fn = getattr(tool, "__wrapped__", None) or tool
+    return await fn(*args, **kwargs)
+
+
+def _patch_service(monkeypatch) -> MagicMock:
+    svc = MagicMock()
+    monkeypatch.setattr(f"{MODULE}.MemorySpaceService", lambda: svc)
+    return svc
+
+
+class TestMemoryList:
+    @pytest.mark.asyncio
+    async def test_lists_manifest_summary(self, monkeypatch):
+        svc = _patch_service(monkeypatch)
+        svc.list_entries.return_value = [
+            SimpleNamespace(slug="jane", entry_type="entity", description="a person", updated="2026-07-07"),
+        ]
+        tool = make_memory_list_tool("spc_1", "Brain", "u1", "u1@x.edu")
+        result = await _call(tool)
+        assert result["status"] == "success"
+        assert result["content"][0]["json"]["entries"][0]["slug"] == "jane"
+        # scoped to the bound space + invoker
+        assert svc.list_entries.call_args.args[:3] == ("spc_1", "u1", "u1@x.edu")
+
+    @pytest.mark.asyncio
+    async def test_revoked_grant_is_error_result(self, monkeypatch):
+        svc = _patch_service(monkeypatch)
+        svc.list_entries.side_effect = MemorySpacePermissionError("nope")
+        tool = make_memory_list_tool("spc_1", "Brain", "u1", "u1@x.edu")
+        result = await _call(tool)
+        assert result["status"] == "error"
+        assert "no longer have access" in result["content"][0]["text"]
+
+
+class TestMemoryRead:
+    @pytest.mark.asyncio
+    async def test_reads_body(self, monkeypatch):
+        svc = _patch_service(monkeypatch)
+        svc.read_entry.return_value = "Jane is the CFO."
+        tool = make_memory_read_tool("spc_1", "Brain", "u1", "u1@x.edu")
+        result = await _call(tool, slug="jane")
+        assert result["status"] == "success"
+        assert result["content"][0]["text"] == "Jane is the CFO."
+
+    @pytest.mark.asyncio
+    async def test_missing_entry_is_error_result(self, monkeypatch):
+        svc = _patch_service(monkeypatch)
+        svc.read_entry.side_effect = MemorySpaceNotFoundError("gone")
+        tool = make_memory_read_tool("spc_1", "Brain", "u1", "u1@x.edu")
+        result = await _call(tool, slug="ghost")
+        assert result["status"] == "error" and "No memory entry 'ghost'" in result["content"][0]["text"]
+
+
+class TestMemoryWrite:
+    @pytest.mark.asyncio
+    async def test_writes_and_confirms(self, monkeypatch):
+        svc = _patch_service(monkeypatch)
+        svc.write_entry.return_value = SimpleNamespace(slug="jane", entry_type="entity")
+        tool = make_memory_write_tool("spc_1", "Brain", "u1", "u1@x.edu")
+        result = await _call(tool, slug="jane", body="Jane is the CFO.", entry_type="entity", description="person")
+        assert result["status"] == "success"
+        assert 'Saved memory entry "jane"' in result["content"][0]["text"]
+        # write goes to the bound space as the invoker, with the given fields
+        kwargs = svc.write_entry.call_args
+        assert kwargs.args[0] == "spc_1" and kwargs.args[1] == "u1"
+        assert kwargs.kwargs["entry_type"] == "entity"
+
+    @pytest.mark.asyncio
+    async def test_write_permission_error_is_error_result(self, monkeypatch):
+        svc = _patch_service(monkeypatch)
+        svc.write_entry.side_effect = MemorySpacePermissionError("read-only")
+        tool = make_memory_write_tool("spc_1", "Brain", "u1", "u1@x.edu")
+        result = await _call(tool, slug="jane", body="x")
+        assert result["status"] == "error" and "don't have write access" in result["content"][0]["text"]

backend/tests/apis/inference_api/test_build_memory_tools.py

@@ -0,0 +1,28 @@
+"""Agent Designer Phase 3 — the _build_memory_tools extra_tools seam.
+
+None binding → no tools; read access → list+read; readwrite → list+read+write.
+"""
+
+from types import SimpleNamespace
+
+from apis.inference_api.chat.routes import _build_memory_tools
+
+
+def _binding(access):
+    return SimpleNamespace(space_id="spc_1", space_name="Brain", access=access, role="editor")
+
+
+def test_no_binding_yields_no_tools():
+    assert _build_memory_tools(None, "u1", "u1@x.edu") == []
+
+
+def test_read_access_exposes_list_and_read_only():
+    tools = _build_memory_tools(_binding("read"), "u1", "u1@x.edu")
+    names = [t.tool_name for t in tools]
+    assert names == ["memory_list", "memory_read"]
+
+
+def test_readwrite_access_adds_write():
+    tools = _build_memory_tools(_binding("readwrite"), "u1", "u1@x.edu")
+    names = [t.tool_name for t in tools]
+    assert names == ["memory_list", "memory_read", "memory_write"]