diff --git a/.github/workflows/main.yaml b/.github/workflows/main.yaml
index f8b25c261..c03794086 100644
--- a/.github/workflows/main.yaml
+++ b/.github/workflows/main.yaml
@@ -50,6 +50,10 @@ jobs:
           pip install -e .[test_full]
           pytest -v
 
+      - name: Run Stub Tests
+        shell: bash -l {0}
+        run: stubtest fsspec.spec
+
   win:
     name: pytest-win
     runs-on: windows-2022
diff --git a/README.md b/README.md
index 0969ac68b..f30e170f0 100644
--- a/README.md
+++ b/README.md
@@ -81,6 +81,11 @@ run and corresponding environment file run a set of tests from the dask
 test suite, and very minimal tests against pandas and zarr from the
 test_downstream.py module in this repo.
 
+### Type Hints
+
+fsspec uses type stub files for type hints instead of inline hints. Use ``stubtest <module>``
+to verify if type stubs are in sync with implementation.
+
 ### Code Formatting
 
 fsspec uses [Black](https://black.readthedocs.io/en/stable) to ensure
diff --git a/fsspec/spec.pyi b/fsspec/spec.pyi
new file mode 100644
index 000000000..833f2b3d4
--- /dev/null
+++ b/fsspec/spec.pyi
@@ -0,0 +1,514 @@
+# pyright: reportExplicitAny=none, reportAny=none
+
+import io
+import os
+from collections.abc import Callable, Generator
+from pathlib import Path
+from typing import (
+    IO,
+    Any,
+    ClassVar,
+    Literal,
+    Self,
+    TypeAlias,
+    TypedDict,
+    overload,
+    override,
+)
+
+from _typeshed import Incomplete, OpenBinaryMode, OpenTextMode, ReadableBuffer
+
+from fsspec.mapping import FSMap
+
+from .callbacks import DEFAULT_CALLBACK as DEFAULT_CALLBACK
+from .callbacks import Callback
+from .config import conf as conf
+from .dircache import DirCache as DirCache
+from .transaction import Transaction as Transaction
+from .utils import isfilelike as isfilelike
+from .utils import other_paths as other_paths
+from .utils import read_block as read_block
+from .utils import stringify_path as stringify_path
+from .utils import tokenize as tokenize
+
+logger: Incomplete
+
+_TPath: TypeAlias = str | os.PathLike[str] | Path
+
+class _FileInfo(TypedDict):
+    name: str
+    size: int | None
+    type: Literal["file", "directory"] | str  # noqa: PYI051
+
+def make_instance(cls: type, args: list[Any], kwargs: dict[str, Any]) -> type: ...
+
+class _Cached(type):
+    def __init__(cls, *args: Any, **kwargs: Any) -> None: ...
+    def __call__(cls, *args: Any, **kwargs: Any): ...
+
+class AbstractFileSystem(metaclass=_Cached):
+    cachable: bool
+    blocksize: Incomplete
+    sep: str
+    protocol: ClassVar[str | tuple[str, ...]]
+    async_impl: bool
+    mirror_sync_methods: bool
+    root_marker: str
+    transaction_type: type[Transaction]
+    storage_args: tuple[Any, ...]
+    storage_options: dict[str, Any]
+    dircache: Incomplete
+    def __init__(self, *args, **storage_options) -> None: ...
+    @property
+    def fsid(self) -> None: ...
+    def __dask_tokenize__(self): ...
+    def __hash__(self): ...
+    def __eq__(self, other): ...
+    def __reduce__(self): ...
+    def unstrip_protocol(self, name: str) -> str: ...
+    @classmethod
+    def current(cls) -> Self: ...
+    @property
+    def transaction(self) -> Transaction: ...
+    def start_transaction(self) -> Transaction: ...
+    def end_transaction(self) -> None: ...
+    def invalidate_cache(self, path: str | None = None) -> None: ...
+    def mkdir(
+        self, path: _TPath, create_parents: bool = True, **kwargs: Any
+    ) -> None: ...
+    def makedirs(self, path: _TPath, exist_ok: bool = False) -> None: ...
+    def rmdir(self, path: _TPath) -> None: ...
+    @overload
+    def ls(
+        self, path: _TPath, detail: Literal[True], **kwargs: Any
+    ) -> list[_FileInfo]: ...
+    @overload
+    def ls(self, path: _TPath, detail: Literal[False], **kwargs: Any) -> list[str]: ...
+    @overload
+    def ls(
+        self, path: _TPath, detail: bool = True, **kwargs: Any
+    ) -> list[str] | list[_FileInfo]: ...
+    def walk(
+        self,
+        path: _TPath,
+        maxdepth: int | None = None,
+        topdown: bool = True,
+        on_error: Literal["omit", "raise"] | Callable[[OSError], Any] = "omit",
+        **kwargs: Any,
+    ) -> Generator[Incomplete, Incomplete]: ...
+    @overload
+    def find(
+        self,
+        path: _TPath,
+        maxdepth: int | None,
+        withdirs: bool,
+        detail: Literal[True],
+        **kwargs: Any,
+    ) -> dict[str, _FileInfo]: ...
+    @overload
+    def find(
+        self,
+        path: _TPath,
+        maxdepth: int | None,
+        withdirs: bool,
+        detail: Literal[False],
+        **kwargs: Any,
+    ) -> list[str]: ...
+    @overload
+    def find(
+        self,
+        path: _TPath,
+        maxdepth: int | None = None,
+        withdirs: bool = False,
+        detail: bool = False,
+        **kwargs: Any,
+    ) -> list[str] | dict[str, _FileInfo]: ...
+    @overload
+    def du(
+        self,
+        path: _TPath,
+        total: Literal[True],
+        maxdepth: int | None = None,
+        withdirs: bool = False,
+        **kwargs: Any,
+    ) -> int: ...
+    @overload
+    def du(
+        self,
+        path: _TPath,
+        total: Literal[False],
+        maxdepth: int | None = None,
+        withdirs: bool = False,
+        **kwargs: Any,
+    ) -> dict[str, int]: ...
+    @overload
+    def du(
+        self,
+        path: _TPath,
+        total: bool = True,
+        maxdepth: int | None = None,
+        withdirs: bool = False,
+        **kwargs: Any,
+    ) -> int | dict[str, int]: ...
+    def glob(
+        self, path: _TPath, maxdepth: int | None = None, **kwargs: Any
+    ) -> list[str] | dict[str, _FileInfo]: ...
+    def exists(self, path: _TPath, **kwargs: Any) -> bool: ...
+    def lexists(self, path: _TPath, **kwargs: Any) -> bool: ...
+    def info(self, path: _TPath, **kwargs: Any) -> _FileInfo: ...
+    def checksum(self, path: _TPath) -> int: ...
+    def size(self, path: _TPath) -> int | None: ...
+    def sizes(self, paths: list[_TPath]) -> list[int | None]: ...
+    def isdir(self, path: _TPath) -> bool: ...
+    def isfile(self, path: _TPath) -> bool: ...
+    def read_text(
+        self,
+        path: _TPath,
+        encoding: str | None = None,
+        errors: str | None = None,
+        newline: str | None = None,
+        **kwargs: Any,
+    ) -> str | bytes: ...
+    def write_text(
+        self,
+        path: _TPath,
+        value: str,
+        encoding: str | None = None,
+        errors: str | None = None,
+        newline: str | None = None,
+        **kwargs: Any,
+    ) -> int: ...
+    def cat_file(
+        self,
+        path: _TPath,
+        start: int | None = None,
+        end: int | None = None,
+        **kwargs: Any,
+    ) -> str | bytes: ...
+    def pipe_file(
+        self,
+        path: _TPath,
+        value: bytes,
+        mode: Literal["create", "overwrite"] = "overwrite",
+        **kwargs: Any,
+    ) -> None: ...
+    def pipe(
+        self,
+        path: _TPath | dict[_TPath, bytes],
+        value: bytes | None = None,
+        **kwargs: Any,
+    ) -> None: ...
+    @overload
+    def cat_ranges(
+        self,
+        paths: list[_TPath],
+        starts: int | list[int],
+        ends: int | list[int],
+        max_gap: None,
+        on_error: Literal["raise"],
+        **kwargs: Any,
+    ) -> list[str | bytes]: ...
+    @overload
+    def cat_ranges(
+        self,
+        paths: list[_TPath],
+        starts: int | list[int],
+        ends: int | list[int],
+        max_gap: None,
+        on_error: Literal["return"],
+        **kwargs: Any,
+    ) -> list[str | bytes | Exception]: ...
+    @overload
+    def cat_ranges(
+        self,
+        paths: list[_TPath],
+        starts: int | list[int],
+        ends: int | list[int],
+        max_gap: None = None,
+        on_error: Literal["raise", "return"] = "return",
+        **kwargs: Any,
+    ) -> list[str | bytes] | list[str | bytes | Exception]: ...
+    def cat(
+        self,
+        path: _TPath,
+        recursive: bool = False,
+        on_error: Literal["raise", "omit", "return"] = "raise",
+        **kwargs: Any,
+    ) -> str | bytes | dict[_TPath, str | bytes]: ...
+    def get_file(
+        self,
+        rpath: _TPath,
+        lpath: _TPath,
+        callback: Callback = ...,
+        outfile: IO[bytes] | None = None,
+        **kwargs: Any,
+    ) -> None: ...
+    def get(
+        self,
+        rpath: _TPath | list[_TPath],
+        lpath: _TPath | list[_TPath],
+        recursive: bool = False,
+        callback: Callback = ...,
+        maxdepth: int | None = None,
+        **kwargs: Any,
+    ) -> None: ...
+    def put_file(
+        self,
+        lpath: _TPath,
+        rpath: _TPath,
+        callback: Callback = ...,
+        mode: Literal["create", "overwrite"] = "overwrite",
+        **kwargs: Any,
+    ) -> None: ...
+    def put(
+        self,
+        lpath: _TPath | list[_TPath],
+        rpath: _TPath | list[_TPath],
+        recursive: bool = False,
+        callback: Callback = ...,
+        maxdepth: int | None = None,
+        **kwargs: Any,
+    ) -> None: ...
+    def head(self, path: _TPath, size: int = 1024) -> str | bytes: ...
+    def tail(self, path: _TPath, size: int = 1024) -> str | bytes: ...
+    def cp_file(self, path1: _TPath, path2: _TPath, **kwargs: Any) -> None: ...
+    def copy(
+        self,
+        path1: _TPath | list[_TPath],
+        path2: _TPath | list[_TPath],
+        recursive: bool = False,
+        maxdepth: int | None = None,
+        on_error: Literal["raise", "ignore"] | None = None,
+        **kwargs: Any,
+    ) -> None: ...
+    def expand_path(
+        self,
+        path: _TPath,
+        recursive: bool = False,
+        maxdepth: int | None = None,
+        assume_literal: bool = False,
+        **kwargs: Any,
+    ) -> list[str]: ...
+    def mv(
+        self,
+        path1: _TPath | list[_TPath],
+        path2: _TPath | list[_TPath],
+        recursive: bool = False,
+        maxdepth: int | None = None,
+        **kwargs: Any,
+    ) -> None: ...
+    def rm_file(self, path: _TPath) -> None: ...
+    def rm(
+        self, path: _TPath, recursive: bool = False, maxdepth: int | None = None
+    ) -> None: ...
+    @overload
+    def open(
+        self,
+        path: _TPath,
+        mode: OpenTextMode,
+        block_size: int | None = None,
+        cache_options: dict[str, Any] | None = None,
+        compression: str | None = None,
+        **kwargs: Any,
+    ) -> io.TextIOWrapper: ...
+    @overload
+    def open(
+        self,
+        path: _TPath,
+        mode: OpenBinaryMode,
+        block_size: int | None = None,
+        cache_options: dict[str, Any] | None = None,
+        compression: str | None = None,
+        **kwargs: Any,
+    ) -> AbstractBufferedFile: ...
+    @overload
+    def open(
+        self,
+        path: _TPath,
+        mode: OpenTextMode | OpenBinaryMode = "rb",
+        block_size: int | None = None,
+        cache_options: dict[str, Any] | None = None,
+        compression: str | None = None,
+        **kwargs: Any,
+    ) -> io.TextIOWrapper | AbstractBufferedFile: ...
+    def touch(self, path: _TPath, truncate: bool = True, **kwargs: Any) -> None: ...
+    def ukey(self, path: _TPath) -> str: ...
+    def read_block(
+        self, fn: str, offset: int, length: int | None, delimiter: bytes | None = None
+    ) -> bytes: ...
+    def to_json(self, *, include_password: bool = True) -> str: ...
+    @staticmethod
+    def from_json(blob: str) -> AbstractFileSystem: ...
+    def to_dict(self, *, include_password: bool = True) -> dict[str, Any]: ...
+    @staticmethod
+    def from_dict(dct: dict[str, Any]) -> AbstractFileSystem: ...
+    def get_mapper(
+        self,
+        root: str = "",
+        check: bool = False,
+        create: bool = False,
+        missing_exceptions: tuple[Exception] | None = None,
+    ) -> FSMap: ...
+    @classmethod
+    def clear_instance_cache(cls) -> None: ...
+    def created(self, path: _TPath) -> None: ...
+    def modified(self, path: _TPath) -> None: ...
+    def tree(
+        self,
+        path: str = "/",
+        recursion_limit: int = 2,
+        max_display: int = 25,
+        display_size: bool = False,
+        prefix: str = "",
+        is_last: bool = True,
+        first: bool = True,
+        indent_size: int = 4,
+    ) -> str: ...
+    def read_bytes(
+        self,
+        path: _TPath,
+        start: int | None = None,
+        end: int | None = None,
+        **kwargs: Any,
+    ) -> str | bytes: ...
+    def write_bytes(self, path: _TPath, value: bytes, **kwargs: Any) -> None: ...
+    def makedir(
+        self, path: _TPath, create_parents: bool = True, **kwargs: Any
+    ) -> None: ...
+    def mkdirs(self, path: _TPath, exist_ok: bool = False) -> None: ...
+    @overload
+    def listdir(
+        self, path: _TPath, detail: Literal[True], **kwargs: Any
+    ) -> list[_FileInfo]: ...
+    @overload
+    def listdir(
+        self, path: _TPath, detail: Literal[False], **kwargs: Any
+    ) -> list[str]: ...
+    @overload
+    def listdir(
+        self, path: _TPath, detail: bool = True, **kwargs: Any
+    ) -> list[str] | list[_FileInfo]: ...
+    def cp(self, path1: _TPath, path2: _TPath, **kwargs: Any) -> None: ...
+    def move(self, path1: _TPath, path2: _TPath, **kwargs: Any) -> None: ...
+    def stat(self, path: _TPath, **kwargs: Any) -> _FileInfo: ...
+    @overload
+    def disk_usage(
+        self,
+        path: _TPath,
+        total: Literal[True],
+        maxdepth: int | None,
+        **kwargs: Any,
+    ) -> int: ...
+    @overload
+    def disk_usage(
+        self,
+        path: _TPath,
+        total: Literal[False],
+        maxdepth: int | None,
+        **kwargs: Any,
+    ) -> dict[str, int]: ...
+    @overload
+    def disk_usage(
+        self,
+        path: _TPath,
+        total: bool = True,
+        maxdepth: int | None = None,
+        **kwargs: Any,
+    ) -> int | dict[str, int]: ...
+    def rename(self, path1: _TPath, path2: _TPath, **kwargs: Any) -> None: ...
+    def delete(
+        self, path: _TPath, recursive: bool = False, maxdepth: int | None = None
+    ) -> None: ...
+    def upload(
+        self, lpath: _TPath, rpath: _TPath, recursive: bool = False, **kwargs: Any
+    ) -> None: ...
+    def download(
+        self, rpath: _TPath, lpath: _TPath, recursive: bool = False, **kwargs: Any
+    ) -> None: ...
+    def sign(self, path: _TPath, expiration: int = 100, **kwargs: Any) -> None: ...
+
+class AbstractBufferedFile(io.IOBase):
+    DEFAULT_BLOCK_SIZE: Incomplete
+    path: Incomplete
+    fs: Incomplete
+    mode: Incomplete
+    blocksize: Incomplete
+    loc: int
+    autocommit: Incomplete
+    end: Incomplete
+    start: Incomplete
+    kwargs: Incomplete
+    size: Incomplete
+    cache: Incomplete
+    buffer: Incomplete
+    offset: Incomplete
+    forced: bool
+    location: Incomplete
+    def __init__(
+        self,
+        fs,
+        path: _TPath,
+        mode: str = "rb",
+        block_size: str = "default",
+        autocommit: bool = True,
+        cache_type: str = "readahead",
+        cache_options=None,
+        size=None,
+        **kwargs,
+    ) -> None: ...
+    @property
+    def details(self) -> _FileInfo: ...
+    @details.setter
+    def details(self, value: _FileInfo) -> None: ...
+    @property
+    def full_name(self) -> str: ...
+    @property
+    @override
+    def closed(self) -> bool: ...
+    @closed.setter
+    def closed(self, c: bool) -> None: ...
+    @override
+    def __hash__(self) -> int: ...
+    @override
+    def __eq__(self, other: object) -> bool: ...
+    def commit(self) -> None: ...
+    def discard(self) -> None: ...
+    def info(self) -> _FileInfo: ...
+    @override
+    def tell(self) -> int: ...
+    @override
+    def seek(self, loc: int, whence: int = 0) -> int: ...
+    @override
+    def write(self, data: bytes) -> int: ...
+    @override
+    def flush(self, force: bool = False) -> None: ...
+    @override
+    def read(self, length: int = -1) -> bytes: ...
+    def readinto(self, b: ReadableBuffer) -> int: ...
+    def readuntil(self, char: bytes = b"\n", blocks: int | None = None) -> bytes: ...
+    def readline(self): ...
+    @override
+    def __next__(self) -> bytes: ...
+    @override
+    def __iter__(self) -> Self: ...
+    def readlines(self): ...
+    def readinto1(self, b: ReadableBuffer) -> int: ...
+    @override
+    def close(self) -> None: ...
+    @override
+    def readable(self) -> bool: ...
+    @override
+    def seekable(self) -> bool: ...
+    @override
+    def writable(self) -> bool: ...
+    def __reduce__(self): ...
+    @override
+    def __del__(self) -> None: ...
+    @override
+    def __enter__(self) -> Self: ...
+    @override
+    def __exit__(self, *args) -> None: ...
+
+def reopen(
+    fs, path: _TPath, mode, blocksize, loc, size, autocommit, cache_type, kwargs
+): ...
diff --git a/pyproject.toml b/pyproject.toml
index 8a77470fe..4a0143c98 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -77,6 +77,7 @@ test = [
     "pytest-recording",
     "pytest-rerunfailures",
     "pytest-asyncio !=0.22.0",
+    "mypy"
 ]
 test_full = [
     'Jinja2',