fastmcp/tests/server/tasks/test_task_dependencies.py
Chris Guidry d41bcb2c9e
Unify background task context forwarding, fix concurrent dependency bugs (#3710)
* Unify background task context forwarding and fix concurrent dependency bugs

We've been getting a steady trickle of edge-case reports around background tasks
and contextual dependencies over the last few months (#3654, #3656, #3569). Each
one pointed at a different symptom, but they all traced back to the same area:
the way context is negotiated between the "frontend" server and Docket workers
was grown piecemeal, with each new piece of context (access tokens, HTTP headers,
origin request IDs) getting its own Redis key, its own restore function, and its
own ContextVar. This made it hard to reason about what state was available where,
and the shared-instance Dependency pattern made concurrent tasks stomp on each
other's cleanup state.

This takes a step back and reworks the whole thing as a single unified system:

- Dependency subclasses (_CurrentContext, Progress, _CurrentAccessToken, etc.)
  are now stateless factories — __aenter__ returns a fresh per-invocation
  object, so concurrent tasks never share mutable state. Fixes #3654, #3656.

- The three individual context-snapshot Redis keys (access_token, http_headers,
  origin_request_id) are collapsed into a single TaskContextSnapshot stored as
  one JSON key per task. The three _restore_task_* functions and two ContextVars
  they populated are gone.

- Sync functions like get_http_request() and get_access_token() now find the
  snapshot transparently in background tasks via a 3-tier sync fallback:
  ContextVar (set by _CurrentContext for functions with deps) → in-memory dict
  (same-process workers) → sync Redis GET (out-of-process workers). No function
  wrapping needed.

- The _wrap_for_task_http_headers hack is deleted. FunctionTool registers its
  raw function with Docket so Docket sees and resolves ALL dependencies,
  including Docket-native ones like Retry and Timeout.

- ProxyTool.from_mcp_tool() now propagates execution.taskSupport metadata from
  remote tools. Fixes #3569.

- Removed redundant _current_docket/_current_worker ContextVar management from
  Context.__aenter__/__aexit__ (they're only set in the lifespan now).

Closes #3654
Closes #3656
Closes #3569

🤖 Generated with Claude Code

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

* Address code review feedback

- _OptionalCurrentContext: guard __aexit__ against cleaning up contexts it
  didn't create (check is_background_task before delegating)
- Narrow except clauses in snapshot loading (OSError, JSONDecodeError, etc.
  instead of bare Exception)
- Fix docstrings on register_with_docket for resources/prompts/templates
- Simplify Progress: read ExecutionProgress directly from current_execution
  instead of creating and manually entering a DocketProgress wrapper

🤖 Generated with Claude Code

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

* Use pop-on-access transfer buffer instead of bounded LRU cache for snapshots

The in-memory snapshot dict is a transfer mechanism, not a cache. Entries go
in at submission and come out at the worker's first access. Using pop instead
of get means the dict only holds entries during the brief submission-to-execution
window, bounded by task concurrency (~10) rather than a 10,000-entry LRU limit.

🤖 Generated with Claude Code

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

* Drop in-memory transfer buffer, use sync Redis for all backends

Instead of maintaining an in-memory dict to bridge the async/sync gap, use
a sync Redis client directly. For memory:// backends (fakeredis), shares the
same FakeServer instance via docket._redis.get_memory_server() so data written
by the async Docket client is visible to sync reads. For real Redis, creates a
standard sync connection. No in-process state to manage at all.

🤖 Generated with Claude Code

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

* Move snapshot operations to TaskContextSnapshot methods

capture(), from_json(), to_json(), save() are now classmethod/instance methods
on the dataclass instead of free functions. Deduplicates JSON parsing that was
copy-pasted between the async and sync load paths.

🤖 Generated with Claude Code

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

* Trim implementation details from register_with_docket docstrings

🤖 Generated with Claude Code

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

* Clarify docket lookup comment in submit_to_docket

🤖 Generated with Claude Code

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

* Restore docket/worker ContextVar bridge in Context.__aenter__

Servers that own the Docket (the parent) re-set _current_docket/_current_worker
from their instance attributes when entering a Context. Mounted children skip
this (their _docket is None), so they inherit the parent's value. This is needed
for ASGI deployments where ContextVars set during the lifespan don't propagate
to request handlers.

🤖 Generated with Claude Code

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

* Key snapshot cache by task_id to prevent cross-task context leakage

Docket workers may reuse the same asyncio context for sequential tasks.
The ContextVar cache now stores (task_id, snapshot) tuples so stale entries
from previous tasks are automatically ignored.

🤖 Generated with Claude Code

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

---------

Co-authored-by: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
2026-04-03 10:48:27 -04:00

272 lines
9.6 KiB
Python

"""Tests for dependency injection in background tasks.
These tests verify that Docket's dependency system works correctly when
user functions are queued as background tasks. Dependencies like CurrentDocket(),
CurrentFastMCP(), and Depends() should be resolved in the worker context.
"""
from contextlib import asynccontextmanager
from typing import Any, cast
import pytest
from fastmcp import FastMCP
from fastmcp.client import Client
from fastmcp.dependencies import CurrentDocket, CurrentFastMCP, Depends
from fastmcp.exceptions import ToolError
@pytest.fixture
async def dependency_server():
"""Create a FastMCP server with dependency-using background tasks."""
mcp = FastMCP("dependency-test-server")
# Track dependency injection
injected_values = []
@mcp.tool(task=True)
async def tool_with_docket_dependency(docket=CurrentDocket()) -> str:
"""Background tool that uses CurrentDocket dependency."""
injected_values.append(("docket", docket))
return f"Docket: {docket is not None}"
@mcp.tool(task=True)
async def tool_with_server_dependency(server=CurrentFastMCP()) -> str:
"""Background tool that uses CurrentFastMCP dependency."""
injected_values.append(("server", server))
return f"Server: {server.name}"
@mcp.tool(task=True)
async def tool_with_custom_dependency(
value: int, multiplier: int = Depends(lambda: 10)
) -> int:
"""Background tool with custom Depends()."""
injected_values.append(("multiplier", multiplier))
return value * multiplier
@mcp.tool(task=True)
async def tool_with_multiple_dependencies(
name: str,
docket=CurrentDocket(),
server=CurrentFastMCP(),
) -> str:
"""Background tool with multiple dependencies."""
injected_values.append(("multi_docket", docket))
injected_values.append(("multi_server", server))
return f"{name} on {server.name}"
@mcp.prompt(task=True)
async def prompt_with_server_dependency(topic: str, server=CurrentFastMCP()) -> str:
"""Background prompt that uses CurrentFastMCP dependency."""
injected_values.append(("prompt_server", server))
return f"Prompt from {server.name} about {topic}"
@mcp.resource("file://data.txt", task=True)
async def resource_with_docket_dependency(docket=CurrentDocket()) -> str:
"""Background resource that uses CurrentDocket dependency."""
injected_values.append(("resource_docket", docket))
return f"Resource via Docket: {docket is not None}"
# Expose for test assertions
mcp._injected_values = injected_values # type: ignore[attr-defined] # ty:ignore[unresolved-attribute]
return mcp
async def test_background_tool_receives_docket_dependency(dependency_server):
"""Background tools can use CurrentDocket() and it resolves correctly."""
async with Client(dependency_server) as client:
task = await client.call_tool("tool_with_docket_dependency", {}, task=True)
# Verify it's background
assert not task.returned_immediately
# Get result - will execute in Docket worker
result = await task
# Verify dependency was injected
assert len(dependency_server._injected_values) == 1
dep_type, dep_value = dependency_server._injected_values[0]
assert dep_type == "docket"
assert dep_value is not None
assert "Docket: True" in result.data
async def test_background_tool_receives_server_dependency(dependency_server):
"""Background tools can use CurrentFastMCP() and get the actual FastMCP server."""
dependency_server._injected_values.clear()
async with Client(dependency_server) as client:
task = await client.call_tool("tool_with_server_dependency", {}, task=True)
# Verify background execution
assert not task.returned_immediately
result = await task
# Check the server instance was injected
assert len(dependency_server._injected_values) == 1
dep_type, dep_value = dependency_server._injected_values[0]
assert dep_type == "server"
assert dep_value is dependency_server # Same instance!
assert f"Server: {dependency_server.name}" in result.data
async def test_background_tool_receives_custom_depends(dependency_server):
"""Background tools can use Depends() with custom functions."""
dependency_server._injected_values.clear()
async with Client(dependency_server) as client:
task = await client.call_tool(
"tool_with_custom_dependency", {"value": 5}, task=True
)
assert not task.returned_immediately
result = await task
# Check dependency was resolved
assert len(dependency_server._injected_values) == 1
dep_type, dep_value = dependency_server._injected_values[0]
assert dep_type == "multiplier"
assert dep_value == 10
assert result.data == 50 # 5 * 10
async def test_background_tool_with_multiple_dependencies(dependency_server):
"""Background tools can have multiple dependencies injected simultaneously."""
dependency_server._injected_values.clear()
async with Client(dependency_server) as client:
task = await client.call_tool(
"tool_with_multiple_dependencies", {"name": "test"}, task=True
)
assert not task.returned_immediately
await task
# Both dependencies should be injected
assert len(dependency_server._injected_values) == 2
dep_types = {item[0] for item in dependency_server._injected_values}
assert "multi_docket" in dep_types
assert "multi_server" in dep_types
# Verify values
server_dep = next(
v for t, v in dependency_server._injected_values if t == "multi_server"
)
assert server_dep is dependency_server
async def test_background_prompt_receives_dependencies(dependency_server):
"""Background prompts can use dependency injection."""
dependency_server._injected_values.clear()
async with Client(dependency_server) as client:
task = await client.get_prompt(
"prompt_with_server_dependency", {"topic": "AI"}, task=True
)
assert not task.returned_immediately
await task
# Check dependency was injected
assert len(dependency_server._injected_values) == 1
dep_type, dep_value = dependency_server._injected_values[0]
assert dep_type == "prompt_server"
assert dep_value is dependency_server
async def test_background_resource_receives_dependencies(dependency_server):
"""Background resources can use dependency injection."""
dependency_server._injected_values.clear()
async with Client(dependency_server) as client:
task = await client.read_resource("file://data.txt", task=True)
assert not task.returned_immediately
await task
# Check dependency was injected
assert len(dependency_server._injected_values) == 1
dep_type, dep_value = dependency_server._injected_values[0]
assert dep_type == "resource_docket"
assert dep_value is not None
async def test_foreground_tool_dependencies_unaffected(dependency_server):
"""Synchronous tools (task=False) still get dependencies as before."""
dependency_server._injected_values.clear()
@dependency_server.tool() # task=False
async def sync_tool(server=CurrentFastMCP()) -> str:
dependency_server._injected_values.append(("sync_server", server))
return f"Sync: {server.name}"
async with Client(dependency_server) as client:
await client.call_tool("sync_tool", {})
# Should execute immediately
assert len(dependency_server._injected_values) == 1
assert dependency_server._injected_values[0][1] is dependency_server
async def test_dependency_context_managers_cleaned_up_in_background():
"""Context manager dependencies are properly cleaned up after background task."""
cleanup_called = []
mcp = FastMCP("cleanup-test")
@asynccontextmanager
async def tracked_connection():
try:
cleanup_called.append("enter")
yield "connection"
finally:
cleanup_called.append("exit")
@mcp.tool(task=True)
async def use_connection(name: str, conn: str = Depends(tracked_connection)) -> str:
assert conn == "connection"
assert "enter" in cleanup_called
assert "exit" not in cleanup_called # Still open during execution
return f"Used: {conn}"
async with Client(mcp) as client:
task = await client.call_tool("use_connection", {"name": "test"}, task=True)
result = await task
# After task completes, cleanup should have been called
assert cleanup_called == ["enter", "exit"]
assert "Used: connection" in result.data
async def test_dependency_errors_propagate_to_task_failure():
"""If dependency resolution fails, the background task should fail."""
mcp = FastMCP("error-test")
async def failing_dependency():
raise ValueError("Dependency failed!")
@mcp.tool(task=True)
async def tool_with_failing_dep(
value: str, dep: str = cast(Any, Depends(failing_dependency))
) -> str:
return f"Got: {dep}"
async with Client(mcp) as client:
task = await client.call_tool(
"tool_with_failing_dep", {"value": "test"}, task=True
)
# Task should fail due to dependency error
with pytest.raises(ToolError, match="Failed to resolve dependencies"):
await task.result()
# Verify it reached failed state
status = await task.status()
assert status.status == "failed"