diff --git a/backend/routers/mcp_sse.py b/backend/routers/mcp_sse.py
index 7734095a18..ec133d29b7 100644
--- a/backend/routers/mcp_sse.py
+++ b/backend/routers/mcp_sse.py
@@ -260,7 +260,13 @@ def invalid_mcp_auth_exception(
             "type": "object",
             "properties": {
                 "query": {"type": "string", "description": "Natural language search query"},
-                "limit": {"type": "integer", "description": "Maximum number of results to return", "default": 10},
+                "limit": {
+                    "type": "integer",
+                    "description": "Maximum number of results to return",
+                    "default": 10,
+                    "minimum": 1,
+                    "maximum": 20,
+                },
             },
             "required": ["query"],
         },
@@ -540,33 +546,35 @@ def execute_tool(user_id: str, tool_name: str, arguments: dict) -> dict:
         if not query:
             raise ToolExecutionError("query is required")
 
-        limit = arguments.get("limit", 10)
+        try:
+            limit = parse_mcp_int(arguments.get("limit"), "limit", default=10, minimum=1, maximum=20)
+        except ValueError as e:
+            raise ToolExecutionError(str(e), code=-32602)
+        fetch_limit = min(limit * 3, 60)
 
-        matches = vector_db.find_similar_memories(user_id, query, threshold=0.0, limit=limit)
+        matches = vector_db.find_similar_memories(user_id, query, threshold=0.0, limit=fetch_limit)
         if not matches:
             return {"memories": []}
 
-        memory_ids = [m['memory_id'] for m in matches]
+        memory_ids = [m.get('memory_id') for m in matches if m.get('memory_id')]
+        if not memory_ids:
+            return {"memories": []}
         memories = memories_db.get_memories_by_ids(user_id, memory_ids)
 
-        # Build score lookup and filter out memories the user rejected or that were
-        # superseded/invalidated, then truncate locked content. Mirrors the REST MCP
-        # path (routers/mcp.py) so the SSE tool never surfaces stale/rejected facts.
-        score_map = {m['memory_id']: m.get('score', 0) for m in matches}
+        # Mirror the REST MCP path so SSE search never surfaces rejected, locked,
+        # or superseded facts, while fetching extra candidates before filtering.
+        score_map = {m.get('memory_id'): m.get('score', 0) for m in matches if m.get('memory_id')}
         results = []
         for mem in memories:
-            if mem.get('user_review') is False or mem.get('invalid_at') is not None:
+            if mem.get('user_review') is False or mem.get('is_locked', False) or mem.get('invalid_at') is not None:
                 continue
-            if mem.get('is_locked', False):
-                content = mem.get('content', '')
-                mem['content'] = (content[:70] + '...') if len(content) > 70 else content
             mem['relevance_score'] = round(score_map.get(mem.get('id'), 0), 4)
             results.append(mem)
 
         # Sort by relevance
         results.sort(key=lambda x: x.get('relevance_score', 0), reverse=True)
 
-        return {"memories": results}
+        return {"memories": results[:limit]}
 
     elif tool_name == "search_conversations":
         query = arguments.get("query")
diff --git a/backend/tests/unit/test_lock_bypass_fixes.py b/backend/tests/unit/test_lock_bypass_fixes.py
index b6b3f65f50..fee553a50f 100644
--- a/backend/tests/unit/test_lock_bypass_fixes.py
+++ b/backend/tests/unit/test_lock_bypass_fixes.py
@@ -662,6 +662,61 @@ def test_mcp_sse_redacts_locked(self):
         assert convs[0]['structured']['title'] == 'Test Conversation'
         assert len(convs[1]['structured']['action_items']) == 1
 
+    def test_mcp_sse_search_memories_filters_locked_and_backfills_limit(self):
+        """MCP SSE search_memories must match REST filtering before applying the requested limit."""
+        import database.memories as memories_db
+        import database.vector_db as vector_db
+
+        vector_db.find_similar_memories = MagicMock(
+            return_value=[
+                {'score': 1.0},
+                {'memory_id': 'locked', 'score': 0.99},
+                {'memory_id': 'rejected', 'score': 0.98},
+                {'memory_id': 'invalidated', 'score': 0.97},
+                {'memory_id': 'visible-1', 'score': 0.70},
+                {'memory_id': 'visible-2', 'score': 0.60},
+                {'memory_id': 'visible-3', 'score': 0.50},
+            ]
+        )
+        locked = _make_memory(locked=True, memory_id='locked')
+        locked['content'] = 'LOCKED_SECRET_MEMORY'
+        rejected = _make_memory(memory_id='rejected')
+        rejected['content'] = 'REJECTED_MEMORY'
+        rejected['user_review'] = False
+        invalidated = _make_memory(memory_id='invalidated')
+        invalidated['content'] = 'INVALIDATED_MEMORY'
+        invalidated['invalid_at'] = '2026-06-10T00:00:00+00:00'
+        memories_db.get_memories_by_ids = MagicMock(
+            return_value=[
+                locked,
+                rejected,
+                invalidated,
+                _make_memory(memory_id='visible-1'),
+                _make_memory(memory_id='visible-2'),
+                _make_memory(memory_id='visible-3'),
+            ]
+        )
+
+        from routers.mcp_sse import execute_tool
+
+        result = execute_tool('test-uid', 'search_memories', {'query': 'memory', 'limit': 2})
+
+        assert [memory['id'] for memory in result['memories']] == ['visible-1', 'visible-2']
+        assert 'LOCKED_SECRET_MEMORY' not in str(result)
+        assert 'REJECTED_MEMORY' not in str(result)
+        assert 'INVALIDATED_MEMORY' not in str(result)
+        vector_db.find_similar_memories.assert_called_once_with('test-uid', 'memory', threshold=0.0, limit=6)
+
+    def test_mcp_sse_search_memories_schema_documents_limit_bounds(self):
+        """MCP clients should see the same limit bounds enforced by execute_tool."""
+        from routers.mcp_sse import MCP_TOOLS
+
+        search_memories = next(tool for tool in MCP_TOOLS if tool['name'] == 'search_memories')
+        limit_schema = search_memories['inputSchema']['properties']['limit']
+
+        assert limit_schema['minimum'] == 1
+        assert limit_schema['maximum'] == 20
+
 
 # =============================================================================
 # Test users.py endpoints