src/fastmcp/server/dependencies.py (1)

508-521: Expiration check looks correct but import could be hoisted.

The from datetime import datetime, timezone import inside the function body on line 517 is executed on every call to get_access_token() that reaches this branch. While Python caches module imports, moving it to the top-level would be cleaner and marginally faster for hot paths.

Also, verify that expires_at is always an int (epoch seconds). If it could be a float from the JWT, the int() cast on datetime.now(...) is fine, but the < comparison between int and int is safe only if both sides agree on the unit.

src/fastmcp/server/tasks/handlers.py (1)

106-113: Token write failure will abort task submission — consider whether this is intentional.

The access_token Redis write (lines 110-113) is inside the same async with docket.redis() block as the mandatory metadata writes. If model_dump_json() raises (unlikely but possible with custom serializers), the entire submit_to_docket call fails. Since the access token is supplementary (tasks can run without it), you may want to guard this write separately so that serialization issues don't prevent task submission.

♻️ Optional: isolate token write from mandatory metadata

     async with docket.redis() as redis:
         await redis.set(task_meta_key, task_key, ex=ttl_seconds)
         await redis.set(created_at_key, created_at.isoformat(), ex=ttl_seconds)
         await redis.set(poll_interval_key, str(poll_interval_ms), ex=ttl_seconds)
-        if access_token is not None:
-            await redis.set(
-                access_token_key, access_token.model_dump_json(), ex=ttl_seconds
-            )
+        if access_token is not None:
+            try:
+                await redis.set(
+                    access_token_key, access_token.model_dump_json(), ex=ttl_seconds
+                )
+            except Exception:
+                import logging
+                logging.getLogger(__name__).warning(
+                    "Failed to snapshot access token for task %s", server_task_id,
+                    exc_info=True,
+                )

src/fastmcp/server/dependencies.py (2)

508-521: Expiration check and fallback look correct, but consider hoisting the import.

The fallback logic is sound: check _task_access_token ContextVar, validate expiry, and return. The < comparison correctly treats a token whose expires_at equals now as still valid.

However, from datetime import datetime, timezone is a deferred import inside what can be a frequently called function. Since this module already imports from contextvars, contextlib, etc. at the top level, consider adding datetime to the top-level imports instead.

♻️ Hoist import to module level

 from collections.abc import AsyncGenerator, Callable
 from contextlib import AsyncExitStack, asynccontextmanager
 from contextvars import ContextVar, Token
 from dataclasses import dataclass
+from datetime import datetime, timezone
 from functools import lru_cache

Then in get_access_token():

             if task_token.expires_at is not None:
-                from datetime import datetime, timezone
-
                 if task_token.expires_at < int(datetime.now(timezone.utc).timestamp()):
                     return None

---

787-787: Consider using Token[AccessToken | None] | None instead of Any.

Both _CurrentContext._access_token_cv_token and _CurrentAccessToken._access_token_cv_token are typed as Any. Since you already import Token from contextvars (line 15), you could use the precise type for better static analysis support.

♻️ Suggested type annotation

 class _CurrentContext(Dependency):  # type: ignore[misc]
     _context: Context | None = None
-    _access_token_cv_token: Any = None
+    _access_token_cv_token: Token[AccessToken | None] | None = None

 class _CurrentAccessToken(Dependency):  # type: ignore[misc]
-    _access_token_cv_token: Any = None
+    _access_token_cv_token: Token[AccessToken | None] | None = None

Also applies to: 1066-1067