mirror of
https://github.com/PrefectHQ/fastmcp.git
synced 2026-08-12 16:49:10 +02:00
* 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>
272 lines
9.6 KiB
Python
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"
|