@@ -133,6 +133,30 @@ The client is its own program, and it carries both halves of the client story:
   so the client declares them for itself. A published extension would ship them in
   a package that both sides import.
 
+### Dynamic resources and background work
+
+Normally `resources()` is called once, when the server is built. If your resources change while
+the server runs, set `resources_are_dynamic = True` and the server asks `resources()` again on
+every listing and lookup.
+
+Build the new resources completely before you start returning them, so a request never sees a
+half-updated set. Return them as a tuple, and the same tuple until they change, so the server can
+look a resource up by URI instead of scanning every resource on each request. If two resources
+share a URI, the one registered directly on the server wins,
+then dynamic ones in extension order, then templates.
+
+To run background work, such as a task that keeps those resources up to date, override
+`lifespan()` with an `@asynccontextmanager`. Code before its `yield` runs when the server starts,
+and code after it runs when the server stops.
+
+The server enters it after your application's own lifespan, in extension order, and leaves it in
+the reverse order. The default does nothing, so you only pay for it if you override it.
+
+!!! note
+    Some transports enter the lifespan once per connection, so the same extension can be inside
+    several at once. If a worker must run only once, coordinate that yourself.
+    [`SkillCatalog`](skills.md#reloading-on-a-timer) does this for its refresh timer.
+
 ### Intercepting `tools/call`
 
 The one interceptive hook. Override `intercept_tool_call` to observe, short-circuit,
 
@@ -204,14 +204,77 @@ are not served.
     (an error with `strict=True`). Use `FilesystemSkillProvider(Path("skills"))` instead.
 
 !!! note
-    The catalog reads and hashes every file **once**, when `load()` runs, and serves that snapshot.
-    That's how the `digest` and `size` in each skill's manifest are guaranteed to match what
-    `resources/read` returns.
+    The catalog reads and hashes every file when `load()` runs. By default, it reads files from
+    the provider again on each `resources/read` and checks their bytes against that fingerprint.
 
-    It also means the catalog is fixed for the lifetime of the server. If a file changes
-    afterwards, reading it fails with an internal error (`-32603`) until you restart the server or
-    create a new catalog. And because `load()` reads every file, a large asset directory makes
-    startup slower.
+    If a file changes afterwards, reading it fails with an internal error (`-32603`) until you
+    [reload the catalog](#reloading-skills) or restart the server. And because `load()`
+    reads every file, a large asset directory makes startup slower.
+
+### Reloading skills
+
+While you're writing a skill you'll keep editing its files, adding skills and deleting others.
+Call `reload()` to pick all of it up without restarting the server:
+
+```python
+async def refresh(skills_catalog: SkillCatalog) -> None:
+    changes = await skills_catalog.reload()
+    print(changes.added, changes.removed, changes.changed)  # SKILL.md URIs
+```
+
+`reload()` scans the whole provider again, and only when every skill has been read and validated
+does it replace everything the catalog serves, in one step. `skills/list`, `skills/get`, resource
+and directory listings, and file reads all switch together, so no request sees a half-updated
+catalog. Skills and files can be added, removed or edited. It returns a `SkillCatalogChanges`
+with the `added`, `removed` and `changed` skills.
+
+It is all or nothing. If the scan fails, `reload()` raises and the previous catalog stays installed:
+
+* the provider fails (`OSError`), or lists a path that isn't valid;
+* `strict=True` and any skill is invalid, as with `load()`;
+* a skill the catalog is serving has become invalid. This holds even without `strict`, so a typo in
+  the frontmatter mid-edit doesn't take a skill down. A skill that was already skipped at `load()`
+  stays skipped and doesn't block reloads.
+
+Calls to `reload()` run one after another.
+
+Without `auto_refresh`, file reads still go to the provider: a file that has changed or disappeared
+can fail to read even when the previous catalog stays installed. Enable `auto_refresh` to retain
+the previous files' bytes as well.
+
+#### Reloading on a timer
+
+Calling `reload()` yourself gets old fast. To have the catalog do it, turn on `auto_refresh`:
+
+```python title="server.py" hl_lines="14 15"
+--8<-- "docs_src/skills/tutorial003.py"
+```
+
+Every 30 seconds here, the catalog calls `reload()`. Automatic refresh is **off by default**. With
+`auto_refresh=True` alone, `refresh_interval` defaults to 60 seconds; it must be a finite number
+greater than zero.
+
+There's no task to start and nothing to stop: the timer runs with the server's lifespan, on stdio,
+SSE and streamable HTTP. SSE enters the lifespan once per connection, so there the timer runs only
+while a client is connected.
+
+* **A failed refresh changes nothing.** The error is logged and the previous skills keep serving.
+  The catalog tries again after the next interval.
+* **Scans never overlap.** The next interval starts after the previous scan finishes.
+
+!!! note
+    With `auto_refresh`, the catalog keeps the bytes of every served file **in memory**. That's
+    what lets the previous skills keep serving consistently, even if you edit or delete a file on
+    disk while a refresh is still running. Without it, a file edited on disk reads as an internal
+    error until the next `reload()`.
+
+!!! warning
+    Clients don't hear about a reload. They may keep a `skills/list` result for `ttl_ms`, and a
+    manifest fetched before a reload can describe different bytes than a file fetched after it.
+    Re-fetch the manifest and verify the digest before loading a changed skill.
+
+    Pagination cursors are offsets, not snapshots, so restart a paged listing if the catalog
+    changed between pages.
 
 ### Directory reads from a catalog
 
@@ -286,7 +349,9 @@ File content is returned unchanged. A file whose bytes are valid UTF-8 is return
 including any byte order mark and line endings; any other file is returned as a base64 blob. In
 both cases the content matches the `digest` and `size` in the skill's manifest. Each read is
 verified against the values recorded at load time, and a file that has changed is reported as an
-internal error (`-32603`) until you restart the server or create a new catalog.
+internal error (`-32603`) until you [reload the catalog](#reloading-skills).
+With [`auto_refresh`](#reloading-on-a-timer), reads instead return the retained snapshot's
+bytes until the next successful reload.
 
 `load()` reads and hashes every file of every valid skill once, so a large asset directory makes
 startup slower and temporarily uses memory for the largest file.
@@ -318,7 +383,7 @@ ones found while reading are refused, so a skill directory doesn't expose files
 
 The root is trusted. This is a best-effort check, not a defence against someone who can modify the
 directory while the server runs. As a backstop, the catalog refuses to serve any file whose bytes
-no longer match the digest taken at `load()`. Blocking file system calls run in a worker thread.
+no longer match the digest taken at `load()` or the last `reload()`. Blocking file system calls run in a worker thread.
 
 ## Protocol version and caching
 
@@ -333,14 +398,13 @@ You don't have to branch on protocol version yourself.
 ## What this SDK doesn't do
 
 `Skills` is a protocol adapter, not a skills store. It has no opinion on where a skill's bytes
-live or when a catalog is refreshed.
+live. `SkillCatalog` supplies `reload()` and an optional refresh timer.
 
 `SkillCatalog` covers the common case of a directory (or any other file source behind a
 `SkillProvider`), with a few limits:
 
 * It doesn't include a YAML parser. `parse_frontmatter` supplies one.
-* The catalog isn't reloaded. To publish changes, create a new catalog.
+* It doesn't watch the file system or tell clients about a reload; call `reload()` or set `auto_refresh`.
 * Empty directories aren't served by directory reads.
 
-For anything else, such as a database or a catalog that changes while the server runs, write your
-own handlers for `Skills`.
+For anything else, write your own handlers for `Skills`.
@@ -0,0 +1,26 @@
+from pathlib import Path
+
+import anyio
+import yaml
+
+from mcp.server.mcpserver import MCPServer
+from mcp.server.skills import FilesystemSkillProvider, SkillCatalog
+
+
+async def build_server(root: Path) -> MCPServer:
+    skills_catalog = await SkillCatalog.load(
+        FilesystemSkillProvider(root),
+        parse_frontmatter=yaml.safe_load,
+        auto_refresh=True,
+        refresh_interval=30,
+    )
+    return MCPServer("MCP-over-skills", extensions=[skills_catalog])
+
+
+async def main() -> None:
+    server = await build_server(Path(__file__).parent / "skills")
+    await server.run_stdio_async()
+
+
+if __name__ == "__main__":
+    anyio.run(main)
@@ -19,7 +19,8 @@
 
 from __future__ import annotations
 
-from collections.abc import Awaitable, Callable, Sequence
+from collections.abc import AsyncIterator, Awaitable, Callable, Sequence
+from contextlib import asynccontextmanager
 from dataclasses import dataclass, field
 from typing import TYPE_CHECKING, Any
 
@@ -104,6 +105,8 @@ class Extension:
 
     #: Reverse-DNS extension identifier, advertised under `ServerCapabilities.extensions`.
     identifier: str
+    resources_are_dynamic: bool = False
+    """When true, `resources()` is consulted on each resource listing or lookup."""
 
     def __init_subclass__(cls, **kwargs: Any) -> None:
         super().__init_subclass__(**kwargs)
@@ -130,6 +133,14 @@ def resources(self) -> Sequence[ResourceBinding]:
         """Resources this extension contributes (additive)."""
         return ()
 
+    @asynccontextmanager
+    async def lifespan(self) -> AsyncIterator[None]:
+        """Manage extension background work for the server's lifespan.
+
+        May be entered concurrently by transports with per-connection lifespans.
+        """
+        yield
+
     def methods(self) -> Sequence[MethodBinding]:
         """New request methods this extension serves (additive)."""
         return ()
 
@@ -2,7 +2,7 @@
 
 from __future__ import annotations
 
-from collections.abc import Callable
+from collections.abc import Callable, Mapping, Sequence
 from typing import TYPE_CHECKING, Any
 
 from mcp_types import Annotations, Icon, InputRequiredResult
@@ -25,12 +25,34 @@
 logger = get_logger(__name__)
 
 
+class _ResourceSource:
+    """A dynamic resource provider and the URI index of the tuple it last returned."""
+
+    def __init__(self, provider: Callable[[], Sequence[Resource]]) -> None:
+        self._provider = provider
+        self._indexed: Sequence[Resource] | None = None
+        self._by_uri: dict[str, Resource] = {}
+
+    def by_uri(self) -> Mapping[str, Resource]:
+        resources = self._provider()
+        if resources is self._indexed:
+            return self._by_uri
+        by_uri: dict[str, Resource] = {}
+        for resource in resources:
+            by_uri.setdefault(str(resource.uri), resource)
+        # A tuple can't change in place, so its index stays valid until the provider returns another one.
+        if isinstance(resources, tuple):
+            self._indexed, self._by_uri = resources, by_uri
+        return by_uri
+
+
 class ResourceManager:
     """Manages MCPServer resources."""
 
     def __init__(self, warn_on_duplicate_resources: bool = True, *, resources: list[Resource] | None = None):
         self._resources: dict[str, Resource] = {}
         self._templates: dict[str, ResourceTemplate] = {}
+        self._resource_sources: list[_ResourceSource] = []
         self.warn_on_duplicate_resources = warn_on_duplicate_resources
 
         for resource in resources or ():
@@ -86,6 +108,14 @@ def add_template(
         self._templates[template.uri_template] = template
         return template
 
+    def add_resource_provider(self, provider: Callable[[], Sequence[Resource]]) -> None:
+        """Register a dynamic source of concrete resources, consulted on every listing and lookup.
+
+        Resources added with `add_resource` take precedence, then providers in the order they were added.
+        A provider that returns the same tuple again is looked up by URI without rescanning it.
+        """
+        self._resource_sources.append(_ResourceSource(provider))
+
     async def get_resource(
         self, uri: AnyUrl | str, context: Context[LifespanContextT, RequestT]
     ) -> Resource | InputRequiredResult:
@@ -114,6 +144,9 @@ async def get_resource(
         # First check concrete resources
         if resource := self._resources.get(uri_str):
             return resource
+        for source in self._resource_sources:
+            if resource := source.by_uri().get(uri_str):
+                return resource
 
         # Then check templates
         for template in self._templates.values():
@@ -129,7 +162,11 @@ async def get_resource(
     def list_resources(self) -> list[Resource]:
         """List all registered resources."""
         logger.debug("Listing resources", extra={"count": len(self._resources)})
-        return list(self._resources.values())
+        resources = dict(self._resources)
+        for source in self._resource_sources:
+            for uri, resource in source.by_uri().items():
+                resources.setdefault(uri, resource)
+        return list(resources.values())
 
     def list_templates(self) -> list[ResourceTemplate]:
         """List all registered templates."""
 
@@ -5,7 +5,7 @@
 import base64
 import inspect
 from collections.abc import AsyncIterator, Awaitable, Callable, Iterable, Mapping, Sequence
-from contextlib import AbstractAsyncContextManager, asynccontextmanager
+from contextlib import AbstractAsyncContextManager, AsyncExitStack, asynccontextmanager
 from typing import Any, Generic, Literal, TypeVar, cast, overload
 
 import anyio
@@ -64,6 +64,7 @@
     Extension,
     MethodBinding,
     RequestHandler,
+    ResourceBinding,
     compose_tool_call_handler,
     validate_extension_identifier,
 )
@@ -270,6 +271,7 @@ def __init__(
         for extension in extensions or ():
             self._apply_extension(extension)
         self._install_extension_interceptor()
+        self._install_extension_lifespan()
 
     @property
     def name(self) -> str:
@@ -338,8 +340,11 @@ def _apply_extension(self, extension: Extension) -> None:
 
         for tool in extension.tools():
             self.add_tool(tool.fn, meta=tool.meta, **tool.kwargs)
-        for resource in extension.resources():
-            self.add_resource(resource.resource)
+        if extension.resources_are_dynamic:
+            self._resource_manager.add_resource_provider(_dynamic_resources(extension))
+        else:
+            for resource in extension.resources():
+                self.add_resource(resource.resource)
         for method in extension.methods():
             if self._lowlevel_server.get_request_handler(method.method) is not None:
                 raise ValueError(
@@ -367,6 +372,22 @@ def _install_extension_interceptor(self) -> None:
                 compose_tool_call_handler(self._extensions, self._handle_call_tool),
             )
 
+    def _install_extension_lifespan(self) -> None:
+        extensions = [e for e in self._extensions if type(e).lifespan is not Extension.lifespan]
+        if not extensions:
+            return
+        lifespan = self._lowlevel_server.lifespan
+
+        @asynccontextmanager
+        async def wrap(server: Server[LifespanResultT]) -> AsyncIterator[LifespanResultT]:
+            async with AsyncExitStack() as stack:
+                context = await stack.enter_async_context(lifespan(server))
+                for extension in extensions:
+                    await stack.enter_async_context(extension.lifespan())
+                yield context
+
+        self._lowlevel_server.lifespan = wrap
+
     @overload
     def run(self, transport: Literal["stdio"] = ...) -> None: ...
 
@@ -1376,6 +1397,26 @@ async def get_prompt(
             raise ValueError(str(e)) from e
 
 
+def _dynamic_resources(extension: Extension) -> Callable[[], Sequence[Resource]]:
+    """Adapt a dynamic extension's `resources()` for the resource manager.
+
+    The same tuple of bindings maps to the same tuple of resources, so the manager can reuse its
+    URI index instead of rescanning every resource on each lookup.
+    """
+    last: tuple[Sequence[ResourceBinding], tuple[Resource, ...]] | None = None
+
+    def resources() -> Sequence[Resource]:
+        nonlocal last
+        bindings = extension.resources()
+        if not isinstance(bindings, tuple):
+            return [binding.resource for binding in bindings]
+        if last is None or last[0] is not bindings:
+            last = (bindings, tuple(binding.resource for binding in bindings))
+        return last[1]
+
+    return resources
+
+
 def _version_gated(method: MethodBinding) -> RequestHandler:
     """Wrap a method handler so a request at a disallowed protocol version is rejected.