Encrypt task context snapshots at rest

A background task still needs to know who asked for the work, so FastMCP
captures the caller's access token and every inbound HTTP header at
submission time and writes that snapshot to the Docket backend. On a
distributed backend the credentials sit in Redis as plaintext for the
task's TTL, and a rediss:// URL protects only the wire. Set
FASTMCP_ENCRYPTION_KEY and the snapshot becomes a Fernet token instead.

Restore fails closed: a worker that cannot decrypt a snapshot fails the
task rather than run the tool with no caller.

Closes #4747

🤖 Generated with Claude Code

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

docs/more/settings.mdx

@@ -85,10 +85,11 @@ Task settings (the `FASTMCP_DOCKET_` variables) moved to the optional `fastmcp-t
 
 ## Security
 
-These control FastMCP's SSRF protection for the outbound fetches it makes during authentication (OAuth client metadata and JWKS).
+These control encryption of FastMCP's data at rest, and its SSRF protection for the outbound fetches it makes during authentication (OAuth client metadata and JWKS).
 
 | Environment Variable | Type | Default | Description |
 |---|---|---|---|
+| `FASTMCP_ENCRYPTION_KEY` | `str` | None | Key used to encrypt sensitive FastMCP data at rest. Today this protects the [task context snapshot](/servers/tasks#credentials-at-rest), which carries the submitting caller's access token and HTTP headers and lives in the Docket backend for the task's TTL. Every server and worker sharing a task queue must set the same key. When unset, the snapshot is stored as plaintext JSON. |
 | `FASTMCP_SSRF_TRUST_PROXY` | `bool` | `false` | Trust an outbound HTTP proxy for SSRF-protected fetches. When `false`, FastMCP resolves the target hostname itself and refuses to connect if it maps to a private, loopback, link-local, or reserved IP. When `true`, FastMCP routes auth metadata and JWKS fetches through the configured `HTTPS_PROXY`/`ALL_PROXY` and does not honor `NO_PROXY`; if no proxy is configured the fetch is refused. |
 
 By default, FastMCP protects its OAuth and JWKS fetches against [SSRF](https://owasp.org/www-community/attacks/Server_Side_Request_Forgery) by resolving the target hostname, rejecting any address that maps to a private, loopback, link-local, or reserved IP, and then pinning the connection to that validated IP.

docs/servers/tasks.mdx

@@ -193,6 +193,28 @@ mcp.add_extension(TasksExtension(url="redis://localhost:6379/0"))
 - **Fast**: Single-digit millisecond task pickup latency
 - **Scalable**: Add workers to distribute load across processes or machines
 
+### Credentials at Rest
+
+A background task runs long after the request that submitted it has ended, but it still needs to know who asked for the work. FastMCP captures that identity at submission time in a **task context snapshot**: the caller's access token and every inbound HTTP header, including `Authorization`. The worker restores the snapshot before the tool body runs, so `get_access_token()` and `get_http_headers()` return the submitting caller.
+
+That snapshot lives in the backend for the task's TTL. With `memory://` it never leaves the process. With Redis or Valkey it is a stored value, and by default it is stored as plaintext JSON. A `rediss://` URL encrypts the connection, not the data the backend holds. Anyone who can read the backend can read the tokens.
+
+Set `FASTMCP_ENCRYPTION_KEY` to encrypt the snapshot before it is written:
+
+```bash
+export FASTMCP_ENCRYPTION_KEY=$(python -c "import secrets; print(secrets.token_urlsafe(32))")
+```
+
+<Warning>
+Every server and worker on the same queue must set the same key. The process that restores a snapshot is rarely the one that captured it, and a worker with the wrong key cannot recover the caller.
+</Warning>
+
+With a key configured, restore **fails closed**: a worker that cannot decrypt a snapshot fails the task instead of running the tool with no identity. This matters for a tool whose behavior depends on the caller: running it as an anonymous user is worse than not running it. The failure is reported to the client as a task error, and the server log names the key mismatch.
+
+Two consequences of failing closed are worth planning for. Tasks submitted before the key was set fail when a worker with the key picks them up, so drain the queue before you roll a key out. Rotating a key does the same to tasks in flight under the old one.
+
+The key protects the snapshot only. Tool arguments and any answers a task gathers through [mid-task input](#gathering-input-mid-task) are still stored as plaintext, so treat the backend as sensitive regardless.
+
 ## Workers
 
 Every FastMCP server with task-enabled tools automatically starts an **embedded worker**. You do not need to start a separate worker process for tasks to execute.

fastmcp_slim/fastmcp/settings.py

@@ -6,7 +6,7 @@ from pathlib import Path
 from typing import Annotated, Any, Literal
 
 from platformdirs import user_data_dir
-from pydantic import Field, field_validator
+from pydantic import Field, SecretStr, field_validator
 from pydantic_settings import (
     BaseSettings,
     SettingsConfigDict,
@@ -267,6 +267,25 @@ class Settings(BaseSettings):
         ),
     ] = False
 
+    encryption_key: Annotated[
+        SecretStr | None,
+        Field(
+            description=inspect.cleandoc(
+                """
+                Key used to encrypt sensitive FastMCP data at rest. Today this
+                protects the task-context snapshot, which carries the submitting
+                caller's access token and HTTP headers and is written to the
+                Docket backend for the task's TTL. Every server and worker
+                sharing a task queue must set the same key; a worker that cannot
+                decrypt a snapshot fails the task rather than running it as an
+                anonymous caller. When unset, the snapshot is stored as
+                plaintext JSON. Any string works, because the Fernet key is
+                derived from it, but use at least 32 random characters.
+                """
+            ),
+        ),
+    ] = None
+
     server_dependencies: list[str] = Field(
         default_factory=list,
         description="List of dependencies to install in the server environment",

fastmcp_slim/pyproject.toml

@@ -92,6 +92,10 @@ openai = ["openai>=1.102.0"]
 server = [
     "fastmcp-slim[mcp]=={{ version }}",
     "authlib>=1.6.11",
+    # Fernet and the HKDF/PBKDF2 key derivation behind FASTMCP_ENCRYPTION_KEY and
+    # the OAuth proxy's storage encryption. Previously reached only through
+    # authlib.
+    "cryptography>=43.0.0",
     "cyclopts>=4.0.0",
     "griffelib>=2.0.0",
     "jsonref>=1.1.0",

fastmcp_tasks/fastmcp_tasks/context.py

@@ -16,6 +16,7 @@ from contextvars import ContextVar
 from dataclasses import dataclass
 from typing import TYPE_CHECKING
 
+from fastmcp_tasks.encryption import SnapshotDecryptionError, snapshot_codec
 from fastmcp_tasks.keys import (
     leg_number_from_key,
     parse_task_key,
@@ -226,10 +227,20 @@ class TaskContextSnapshot:
         task_id: str,
         ttl_seconds: int,
     ) -> None:
-        """Store this snapshot as a single Redis key."""
+        """Store this snapshot as a single Redis key.
+
+        The stored value is encrypted when a ``FASTMCP_ENCRYPTION_KEY`` is
+        configured: this payload carries the caller's bearer token and headers,
+        and a distributed backend keeps it where the backend's operators can
+        read it (#4747).
+        """
         key = docket.key(f"{task_redis_prefix(task_scope)}:{task_id}:snapshot")
+        codec = snapshot_codec()
+        payload = self.to_json()
+        if codec is not None:
+            payload = codec.encode(payload)
         async with docket.redis() as redis:
-            await redis.set(key, self.to_json(), ex=ttl_seconds)
+            await redis.set(key, payload, ex=ttl_seconds)
 
 
 # Cache keyed by task_id so stale entries from previous tasks in the same
@@ -285,6 +296,11 @@ async def restore_task_snapshot(key: str = TaskKey()) -> None:
     backend work transparently (#3897).  Failures are non-fatal: the task
     still runs, and sync helpers return ``None`` as they would have before
     the snapshot was captured.
+
+    A snapshot that will not decrypt is the one exception. When the deployment
+    configured an encryption key, an unreadable snapshot means the caller cannot
+    be recovered, so the task fails rather than running under no identity at all
+    (#4747).
     """
     try:
         parts = parse_task_key(key)
@@ -313,12 +329,26 @@ async def restore_task_snapshot(key: str = TaskKey()) -> None:
             )
         if raw is None:
             return
+        codec = snapshot_codec()
+        if codec is not None:
+            raw = codec.decode(raw)
         snapshot = TaskContextSnapshot.from_json(raw)
         _remember_snapshot(task_id, snapshot)
         # Restore the ambient request context (auth token, headers) so core's
         # get_access_token()/get_http_headers() see the submitting caller inside
         # the worker, exactly as a normal request would.
         _apply_snapshot_to_context(snapshot)
+    except SnapshotDecryptionError:
+        # Docket reports this to the client as a generic dependency-resolution
+        # failure, so name the cause here. A key mismatch across servers and
+        # workers is the likely reason and is not guessable from the wire error.
+        _logger.error(
+            "Failed to decrypt the task snapshot for %s. Every server and worker "
+            "on this queue must share the same FASTMCP_ENCRYPTION_KEY. The task "
+            "will fail rather than run without the submitting caller's identity.",
+            key,
+        )
+        raise
     except Exception:
         _logger.warning("Failed to restore task snapshot for %s", key, exc_info=True)
 

fastmcp_tasks/fastmcp_tasks/encryption.py

@@ -0,0 +1,109 @@
+"""Encryption of the task-context snapshot at rest.
+
+The snapshot a task carries holds the submitting caller's access token and every
+inbound HTTP header, and it lives in the Docket backend for the task's TTL. A
+distributed backend therefore keeps bearer credentials in Redis, where a
+``rediss://`` URL protects the wire but not the stored value.
+
+Setting ``FASTMCP_ENCRYPTION_KEY`` turns the stored snapshot into a Fernet token.
+The same key must reach every server and worker on the queue, because the process
+that restores a snapshot is rarely the one that captured it.
+"""
+
+from __future__ import annotations
+
+from functools import lru_cache
+
+import fastmcp
+from fastmcp.utilities.logging import get_logger
+
+logger = get_logger(__name__)
+
+# Domain separation: FASTMCP_ENCRYPTION_KEY is meant to serve other at-rest uses
+# over time, and each derives its own Fernet key from this shared material.
+_SNAPSHOT_KEY_SALT = "fastmcp-task-snapshot-key"
+
+# Below this, the material is too weak to feed HKDF, which assumes its input is
+# already high-entropy. Matches the OAuth proxy's threshold for the same reason.
+_MINIMUM_HIGH_ENTROPY_LENGTH = 12
+
+
+class SnapshotDecryptionError(Exception):
+    """A stored snapshot could not be decrypted with the configured key.
+
+    Raised for a wrong key, a tampered value, or a plaintext value written
+    before the key was configured. The restore path lets this escape so the task
+    fails, rather than running the tool as an anonymous caller.
+    """
+
+
+class SnapshotCodec:
+    """Encrypts and decrypts snapshot payloads with a key derived from material.
+
+    Any string works as ``material``. High-entropy material is stretched with
+    HKDF; a short passphrase goes through PBKDF2 instead, which is slower but
+    survives the weaker input.
+    """
+
+    def __init__(self, material: str) -> None:
+        from cryptography.fernet import Fernet
+
+        from fastmcp.server.auth.jwt_issuer import derive_jwt_key
+
+        if len(material) >= _MINIMUM_HIGH_ENTROPY_LENGTH:
+            key = derive_jwt_key(
+                high_entropy_material=material, salt=_SNAPSHOT_KEY_SALT
+            )
+        else:
+            logger.warning(
+                "The configured encryption key is shorter than %d characters; "
+                "use at least 32 random characters.",
+                _MINIMUM_HIGH_ENTROPY_LENGTH,
+            )
+            key = derive_jwt_key(low_entropy_material=material, salt=_SNAPSHOT_KEY_SALT)
+
+        self._fernet = Fernet(key=key)
+
+    def encode(self, payload: str) -> str:
+        """Return the encrypted form of a serialized snapshot."""
+        return self._fernet.encrypt(payload.encode()).decode()
+
+    def decode(self, stored: str | bytes) -> str:
+        """Return the serialized snapshot a stored value holds.
+
+        Raises ``SnapshotDecryptionError`` if the value was not produced by this
+        key, including when it is unencrypted.
+        """
+        from cryptography.fernet import InvalidToken
+
+        raw = stored.encode() if isinstance(stored, str) else stored
+        try:
+            return self._fernet.decrypt(raw).decode()
+        except InvalidToken as e:
+            raise SnapshotDecryptionError(
+                "The stored task snapshot could not be decrypted with the "
+                "configured FASTMCP_ENCRYPTION_KEY."
+            ) from e
+
+
+@lru_cache(maxsize=4)
+def _codec_for(material: str) -> SnapshotCodec:
+    """One codec per key, so the derivation cost is paid once per process.
+
+    PBKDF2 over a short passphrase takes about a second, and every task
+    submission and every restore needs a codec.
+    """
+    return SnapshotCodec(material)
+
+
+def snapshot_codec() -> SnapshotCodec | None:
+    """The codec for the configured key, or ``None`` when none is configured."""
+    key = fastmcp.settings.encryption_key
+    if key is None:
+        return None
+    return _codec_for(key.get_secret_value())
+
+
+def clear_codec_cache() -> None:
+    """Drop the cached codecs, so a changed key takes effect."""
+    _codec_for.cache_clear()

tests/tasks/server/test_snapshot_encryption.py

@@ -0,0 +1,305 @@
+"""Tests for encryption of the task-context snapshot at rest (#4747).
+
+The snapshot carries the submitting caller's access token and every inbound HTTP
+header, and it is written to the Docket backend for the task's TTL. With a
+distributed backend those credentials sit in Redis where the backend's operators
+can read them. Setting ``FASTMCP_ENCRYPTION_KEY`` makes the snapshot a Fernet
+token instead, and makes a worker that cannot decrypt one fail the task rather
+than run it as an anonymous caller.
+"""
+
+from __future__ import annotations
+
+import json
+import logging
+from collections.abc import Iterator
+
+import pytest
+from fastmcp_tasks.context import TaskContextSnapshot
+from fastmcp_tasks.encryption import (
+    SnapshotCodec,
+    SnapshotDecryptionError,
+    clear_codec_cache,
+    snapshot_codec,
+)
+from fastmcp_tasks.keys import task_redis_prefix
+from pydantic import SecretStr
+
+import fastmcp
+from fastmcp import FastMCP
+from fastmcp.server.dependencies import get_access_token
+from fastmcp_tasks import TasksExtension
+from tests.tasks.task_helpers import (
+    get_task,
+    make_access_token,
+    running_task_server,
+    submit_task,
+    wait_for_task,
+)
+
+KEY = "a-test-encryption-key-for-snapshots"
+OTHER_KEY = "a-different-test-encryption-key-entirely"
+
+
+@pytest.fixture
+def encryption_key() -> Iterator[str]:
+    """Configure the global encryption key for the duration of a test."""
+    clear_codec_cache()
+    previous = fastmcp.settings.encryption_key
+    fastmcp.settings.encryption_key = SecretStr(KEY)
+    try:
+        yield KEY
+    finally:
+        fastmcp.settings.encryption_key = previous
+        clear_codec_cache()
+
+
+@pytest.fixture
+def no_encryption_key() -> Iterator[None]:
+    """Guarantee no key is configured, whatever the ambient environment holds."""
+    clear_codec_cache()
+    previous = fastmcp.settings.encryption_key
+    fastmcp.settings.encryption_key = None
+    try:
+        yield
+    finally:
+        fastmcp.settings.encryption_key = previous
+        clear_codec_cache()
+
+
+@pytest.fixture
+def sensitive_snapshot() -> TaskContextSnapshot:
+    """A snapshot carrying a bearer token and an Authorization header."""
+    token = make_access_token("client-a", "user-1")
+    return TaskContextSnapshot(
+        access_token_json=token.model_dump_json(),
+        http_headers={"authorization": f"Bearer {token.token}", "x-trace-id": "abc"},
+        origin_request_id="req-1",
+        session_id="session-1",
+        owning_tool_name="peek",
+        owning_tool_version="1.0",
+    )
+
+
+class TestSnapshotCodec:
+    def test_round_trips_a_payload(self):
+        codec = SnapshotCodec(KEY)
+        assert codec.decode(codec.encode('{"a": 1}')) == '{"a": 1}'
+
+    def test_encoded_payload_hides_the_credentials(
+        self, sensitive_snapshot: TaskContextSnapshot
+    ):
+        encoded = SnapshotCodec(KEY).encode(sensitive_snapshot.to_json())
+        assert "token-client-a-user-1" not in encoded
+        assert "authorization" not in encoded
+
+    def test_decode_rejects_another_keys_payload(self):
+        encoded = SnapshotCodec(OTHER_KEY).encode('{"a": 1}')
+        with pytest.raises(SnapshotDecryptionError):
+            SnapshotCodec(KEY).decode(encoded)
+
+    def test_decode_rejects_plaintext(self):
+        """A snapshot written before the key was set must not be trusted."""
+        with pytest.raises(SnapshotDecryptionError):
+            SnapshotCodec(KEY).decode('{"access_token_json": null}')
+
+    def test_decode_accepts_bytes(self):
+        """Redis hands back bytes on some backends."""
+        codec = SnapshotCodec(KEY)
+        assert codec.decode(codec.encode('{"a": 1}').encode()) == '{"a": 1}'
+
+    def test_same_key_reuses_one_codec(self, encryption_key: str):
+        assert snapshot_codec() is snapshot_codec()
+
+    def test_no_codec_without_a_key(self, no_encryption_key: None):
+        assert snapshot_codec() is None
+
+
+class TestSnapshotSerialization:
+    def test_json_round_trip_preserves_every_field(
+        self, sensitive_snapshot: TaskContextSnapshot
+    ):
+        assert (
+            TaskContextSnapshot.from_json(sensitive_snapshot.to_json())
+            == sensitive_snapshot
+        )
+
+
+async def _read_stored_snapshot(mcp: FastMCP, task_scope: str, task_id: str) -> str:
+    """Return the raw stored value of a task's snapshot key."""
+    docket = mcp._docket
+    assert docket is not None
+    key = docket.key(f"{task_redis_prefix(task_scope)}:{task_id}:snapshot")
+    async with docket.redis() as redis:
+        raw = await redis.get(key)
+    assert raw is not None
+    return raw.decode() if isinstance(raw, bytes) else str(raw)
+
+
+async def _write_stored_snapshot(
+    mcp: FastMCP, task_scope: str, task_id: str, payload: str
+) -> None:
+    """Overwrite a task's stored snapshot value."""
+    docket = mcp._docket
+    assert docket is not None
+    key = docket.key(f"{task_redis_prefix(task_scope)}:{task_id}:snapshot")
+    async with docket.redis() as redis:
+        await redis.set(key, payload)
+
+
+@pytest.fixture
+def echo_token_server() -> FastMCP:
+    """A task server whose one tool reports the caller it restored."""
+    mcp = FastMCP("snapshot-encryption-test")
+    mcp.add_extension(TasksExtension())
+
+    @mcp.tool(task=True)
+    async def whoami() -> str:
+        token = get_access_token()
+        return token.token if token else "no-token"
+
+    return mcp
+
+
+class TestEncryptedSnapshotRoundTrip:
+    async def test_worker_still_sees_the_submitting_caller(
+        self, echo_token_server: FastMCP, encryption_key: str
+    ):
+        token = make_access_token("client-a", "user-1")
+
+        async with running_task_server(echo_token_server):
+            created = await submit_task(
+                echo_token_server, "whoami", {}, access_token=token
+            )
+            final = await wait_for_task(
+                echo_token_server, created.task_id, access_token=token
+            )
+
+        assert final.status == "completed"
+        assert final.result is not None
+        assert final.result["structuredContent"] == {"result": token.token}
+
+    async def test_stored_value_is_not_readable(
+        self, echo_token_server: FastMCP, encryption_key: str
+    ):
+        token = make_access_token("client-a", "user-1")
+
+        async with running_task_server(echo_token_server):
+            created = await submit_task(
+                echo_token_server, "whoami", {}, access_token=token
+            )
+            stored = await _read_stored_snapshot(
+                echo_token_server, "client-a|user-1", created.task_id
+            )
+            await wait_for_task(echo_token_server, created.task_id, access_token=token)
+
+        assert token.token not in stored
+        assert "authorization" not in stored
+        with pytest.raises(json.JSONDecodeError):
+            json.loads(stored)
+
+    async def test_undecryptable_snapshot_fails_the_task(
+        self,
+        echo_token_server: FastMCP,
+        encryption_key: str,
+        caplog: pytest.LogCaptureFixture,
+    ):
+        """Fail closed: a worker that cannot recover the caller must not run.
+
+        Running anyway would execute the tool as an anonymous caller, which for
+        an authorization-sensitive tool is worse than not running at all. Docket
+        surfaces this on the wire as a generic dependency failure, so the named
+        cause has to come from the log.
+        """
+        token = make_access_token("client-a", "user-1")
+        tampered = SnapshotCodec(OTHER_KEY).encode(TaskContextSnapshot().to_json())
+
+        with caplog.at_level(logging.ERROR, logger="fastmcp_tasks.context"):
+            async with running_task_server(echo_token_server):
+                created = await submit_task(
+                    echo_token_server, "whoami", {}, access_token=token
+                )
+                await _write_stored_snapshot(
+                    echo_token_server, "client-a|user-1", created.task_id, tampered
+                )
+                final = await wait_for_task(
+                    echo_token_server,
+                    created.task_id,
+                    access_token=token,
+                    target_states=frozenset({"failed"}),
+                )
+
+        assert final.status == "failed"
+        assert final.error is not None
+        assert "FASTMCP_ENCRYPTION_KEY" in caplog.text
+
+
+class TestUnencryptedByDefault:
+    async def test_snapshot_stays_plaintext_without_a_key(
+        self, echo_token_server: FastMCP, no_encryption_key: None
+    ):
+        """No key configured is the pre-existing contract, unchanged."""
+        token = make_access_token("client-a", "user-1")
+
+        async with running_task_server(echo_token_server):
+            created = await submit_task(
+                echo_token_server, "whoami", {}, access_token=token
+            )
+            stored = await _read_stored_snapshot(
+                echo_token_server, "client-a|user-1", created.task_id
+            )
+            final = await wait_for_task(
+                echo_token_server, created.task_id, access_token=token
+            )
+
+        assert json.loads(stored)["access_token_json"] is not None
+        assert final.status == "completed"
+
+    async def test_unreadable_snapshot_is_nonfatal_without_a_key(
+        self, echo_token_server: FastMCP, no_encryption_key: None
+    ):
+        """Without encryption a corrupt snapshot still only degrades the caller."""
+        token = make_access_token("client-a", "user-1")
+
+        async with running_task_server(echo_token_server):
+            created = await submit_task(
+                echo_token_server, "whoami", {}, access_token=token
+            )
+            await _write_stored_snapshot(
+                echo_token_server, "client-a|user-1", created.task_id, "not json"
+            )
+            final = await wait_for_task(
+                echo_token_server, created.task_id, access_token=token
+            )
+
+        assert final.status == "completed"
+        assert final.result is not None
+        assert final.result["structuredContent"] == {"result": "no-token"}
+
+
+class TestTaskStillResolvesAfterFailure:
+    async def test_failed_task_reports_an_error(
+        self, echo_token_server: FastMCP, encryption_key: str
+    ):
+        """A fail-closed task is still a well-formed `tasks/get` result."""
+        token = make_access_token("client-a", "user-1")
+        tampered = SnapshotCodec(OTHER_KEY).encode(TaskContextSnapshot().to_json())
+
+        async with running_task_server(echo_token_server):
+            created = await submit_task(
+                echo_token_server, "whoami", {}, access_token=token
+            )
+            await _write_stored_snapshot(
+                echo_token_server, "client-a|user-1", created.task_id, tampered
+            )
+            await wait_for_task(
+                echo_token_server,
+                created.task_id,
+                access_token=token,
+                target_states=frozenset({"failed"}),
+            )
+            fetched = await get_task(
+                echo_token_server, created.task_id, access_token=token
+            )
+
+        assert fetched.status == "failed"

tests/test_settings.py

@@ -19,3 +19,24 @@ def test_http_host_origin_protection_env_var(value, expected, monkeypatch):
     monkeypatch.setenv("FASTMCP_HTTP_HOST_ORIGIN_PROTECTION", value)
 
     assert Settings().http_host_origin_protection == expected
+
+
+def test_encryption_key_defaults_to_none(monkeypatch):
+    monkeypatch.delenv("FASTMCP_ENCRYPTION_KEY", raising=False)
+
+    assert Settings().encryption_key is None
+
+
+def test_encryption_key_env_var(monkeypatch):
+    monkeypatch.setenv("FASTMCP_ENCRYPTION_KEY", "s3kr1t-material")
+
+    key = Settings().encryption_key
+    assert key is not None
+    assert key.get_secret_value() == "s3kr1t-material"
+
+
+def test_encryption_key_is_not_printable(monkeypatch):
+    """A settings dump must never carry the key into a log."""
+    monkeypatch.setenv("FASTMCP_ENCRYPTION_KEY", "s3kr1t-material")
+
+    assert "s3kr1t-material" not in repr(Settings())

uv.lock

@@ -1009,6 +1009,7 @@ openai = [
 ]
 server = [
     { name = "authlib" },
+    { name = "cryptography" },
     { name = "cyclopts" },
     { name = "exceptiongroup" },
     { name = "griffelib" },
@@ -1037,6 +1038,7 @@ requires-dist = [
     { name = "authlib", marker = "extra == 'client'", specifier = ">=1.6.11" },
     { name = "authlib", marker = "extra == 'server'", specifier = ">=1.6.11" },
     { name = "azure-identity", marker = "extra == 'azure'", specifier = ">=1.16.0" },
+    { name = "cryptography", marker = "extra == 'server'", specifier = ">=43.0.0" },
     { name = "cyclopts", marker = "extra == 'server'", specifier = ">=4.0.0" },
     { name = "exceptiongroup", marker = "extra == 'client'", specifier = ">=1.2.2" },
     { name = "exceptiongroup", marker = "extra == 'mcp'", specifier = ">=1.2.2" },