mirror of
https://github.com/PrefectHQ/fastmcp.git
synced 2026-08-09 07:09:11 +02:00
Auto-route Docket-dependent components through Docket execution
Replace tool timeout parameter with Docket's native Timeout dependency. Components declaring Docket deps (Timeout, Retry, etc.) are now: - Registered with Docket at startup - Auto-routed through Docket even for foreground calls Known limitations to discuss: - Context not available in Docket workers - Registration bug: register_with_docket still checks supports_tasks() - Error propagation: ToolError wrapped in McpError - Performance: every call goes through Redis+worker
This commit is contained in:
Jeremiah Lowin
parent
41c56481af
commit
78cc19b029
19 changed files with 521 additions and 275 deletions
|
|
@ -174,6 +174,40 @@ from fastmcp.server import create_proxy
|
|||
proxy = create_proxy("http://example.com/mcp")
|
||||
```
|
||||
|
||||
## v3.0.0b2 (from v3.0.0b1)
|
||||
|
||||
### Breaking Changes
|
||||
|
||||
#### Tool `timeout` Parameter Removed
|
||||
|
||||
The `timeout` parameter on `@mcp.tool()` has been removed. Use Docket's `Timeout` dependency instead:
|
||||
|
||||
```python
|
||||
# Before (v3.0.0b1)
|
||||
@mcp.tool(timeout=30.0)
|
||||
async def fetch_data(url: str) -> dict:
|
||||
...
|
||||
|
||||
# After (v3.0.0b2)
|
||||
from datetime import timedelta
|
||||
from docket import Timeout
|
||||
|
||||
@mcp.tool
|
||||
async def fetch_data(
|
||||
url: str,
|
||||
timeout: Timeout = Timeout(timedelta(seconds=30)),
|
||||
) -> dict:
|
||||
...
|
||||
```
|
||||
|
||||
Benefits of the new approach:
|
||||
|
||||
- **Unified execution**: Timeouts work identically for foreground and background tasks
|
||||
- **Additional capabilities**: Access to `Retry`, `ExponentialRetry`, `ConcurrencyLimit`
|
||||
- **Auto-routing**: Components with Docket dependencies automatically route through Docket
|
||||
|
||||
Note: This is only breaking from beta 1 → beta 2. The `timeout` parameter didn't exist before beta 1.
|
||||
|
||||
## v2.14.0
|
||||
|
||||
### OpenAPI Parser Promotion
|
||||
|
|
|
|||
|
|
@ -810,20 +810,53 @@ Documentation: [Lifespan](/servers/lifespan)
|
|||
|
||||
---
|
||||
|
||||
## Tool Timeout
|
||||
The features above were released in **v3.0.0b1**.
|
||||
|
||||
Tools can limit foreground execution time with a `timeout` parameter ([#2872](https://github.com/jlowin/fastmcp/pull/2872)):
|
||||
---
|
||||
|
||||
## v3.0 Beta 2 Features
|
||||
|
||||
---
|
||||
|
||||
## Auto-Docket Execution
|
||||
|
||||
Components with Docket dependencies (like `Timeout`, `Retry`, `ExponentialRetry`) automatically route through Docket, ensuring consistent behavior for both foreground and background execution.
|
||||
|
||||
```python
|
||||
@mcp.tool(timeout=30.0)
|
||||
async def fetch_data(url: str) -> dict:
|
||||
"""Fetch with 30-second timeout."""
|
||||
from datetime import timedelta
|
||||
from docket import Timeout, ExponentialRetry
|
||||
from fastmcp import FastMCP
|
||||
|
||||
mcp = FastMCP("Server")
|
||||
|
||||
@mcp.tool
|
||||
async def fetch_data(
|
||||
url: str,
|
||||
timeout: Timeout = Timeout(timedelta(seconds=30)),
|
||||
retry: ExponentialRetry = ExponentialRetry(max_attempts=3),
|
||||
) -> dict:
|
||||
"""Fetch with timeout and automatic retry."""
|
||||
...
|
||||
```
|
||||
|
||||
When exceeded, clients receive MCP error code `-32000`. Both sync and async tools are supported—sync functions run in thread pools so the timeout applies regardless of execution model.
|
||||
When FastMCP detects these Docket dependencies in a component's function signature, it:
|
||||
|
||||
Note: This timeout applies to foreground execution only. Background tasks (`task=True`) execute in Docket workers where this timeout isn't enforced.
|
||||
1. **Registers the component with Docket at startup** - even if `task=True` wasn't specified
|
||||
2. **Routes foreground calls through Docket** - ensuring timeouts/retries work correctly
|
||||
3. **Uses the same code path for background tasks** - unified execution model
|
||||
|
||||
This replaces the previous `timeout` parameter on `@mcp.tool()`. The Docket approach has several advantages:
|
||||
|
||||
- **Unified behavior**: Timeouts and retries work identically for foreground and background execution
|
||||
- **Battle-tested implementation**: Docket's timeout/retry logic handles edge cases properly
|
||||
- **Composable**: Multiple dependencies can be combined (timeout + retry + concurrency limits)
|
||||
- **Declarative**: Dependencies in the function signature clearly document requirements
|
||||
|
||||
Supported Docket dependencies:
|
||||
- `Timeout` - Cancel execution after a duration
|
||||
- `Retry` / `ExponentialRetry` - Retry on failure with configurable backoff
|
||||
- `ConcurrencyLimit` - Limit concurrent executions
|
||||
- `CurrentDocket` / `CurrentWorker` / `CurrentExecution` - Access Docket runtime context
|
||||
|
||||
---
|
||||
|
||||
|
|
|
|||
|
|
@ -115,12 +115,6 @@ def search_products_implementation(query: str, category: str | None = None) -> l
|
|||
Optional meta information about the tool. This data is passed through to the MCP client as the `meta` field of the client-side tool object and can be used for custom metadata, versioning, or other application-specific purposes.
|
||||
</ParamField>
|
||||
|
||||
<ParamField body="timeout" type="float | None">
|
||||
<VersionBadge version="3.0.0" />
|
||||
|
||||
Execution timeout in seconds. If the tool takes longer than this to complete, an MCP error is returned to the client. See [Timeouts](#timeouts) for details.
|
||||
</ParamField>
|
||||
|
||||
<ParamField body="version" type="str | int | None">
|
||||
<VersionBadge version="3.0.0" />
|
||||
|
||||
|
|
@ -779,57 +773,51 @@ def divide(a: float, b: float) -> float:
|
|||
|
||||
When `mask_error_details=True`, only error messages from `ToolError` will include details, other exceptions will be converted to a generic message.
|
||||
|
||||
## Timeouts
|
||||
## Timeouts and Retries
|
||||
|
||||
<VersionBadge version="3.0.0" />
|
||||
|
||||
Tools can specify a `timeout` parameter to limit how long execution can take. When the timeout is exceeded, the client receives an MCP error and the tool stops processing. This protects your server from unexpectedly slow operations that could block resources or leave clients waiting indefinitely.
|
||||
FastMCP supports timeouts and retries through Docket's dependency injection system. When you use Docket dependencies like `Timeout` or `Retry` in your tool functions, FastMCP automatically routes execution through Docket to honor them.
|
||||
|
||||
```python
|
||||
from datetime import timedelta
|
||||
from docket import Timeout, ExponentialRetry
|
||||
from fastmcp import FastMCP
|
||||
|
||||
mcp = FastMCP()
|
||||
|
||||
@mcp.tool(timeout=30.0)
|
||||
async def fetch_data(url: str) -> dict:
|
||||
"""Fetch data with a 30-second timeout."""
|
||||
# If this takes longer than 30 seconds,
|
||||
# the client receives an MCP error
|
||||
@mcp.tool
|
||||
async def fetch_data(
|
||||
url: str,
|
||||
timeout: Timeout = Timeout(timedelta(seconds=30)),
|
||||
retry: ExponentialRetry = ExponentialRetry(max_attempts=3)
|
||||
) -> dict:
|
||||
"""Fetch data with a 30-second timeout and automatic retries."""
|
||||
...
|
||||
```
|
||||
|
||||
Timeouts are specified in seconds as a float. When a tool exceeds its timeout, FastMCP returns an MCP error with code `-32000` and a message indicating which tool timed out and how long it ran. Both sync and async tools support timeouts—sync functions run in thread pools, so the timeout applies to the entire operation regardless of execution model.
|
||||
When FastMCP detects Docket dependencies in your function signature, it runs the tool through Docket's execution engine even for foreground (non-background) calls. This gives you:
|
||||
|
||||
- **Unified behavior**: Timeouts and retries work the same whether the tool runs as a foreground call or a background task
|
||||
- **Battle-tested implementation**: Docket's execution model handles cancellation, retry scheduling, and failure tracking
|
||||
- **Additional features**: Access to concurrency limits, progress reporting, and more
|
||||
|
||||
<Note>
|
||||
Tools must explicitly opt-in to timeouts. There is no server-level default timeout setting.
|
||||
Using Docket dependencies requires installing FastMCP with the tasks extra: `pip install 'fastmcp[tasks]'`
|
||||
</Note>
|
||||
|
||||
### Timeouts vs Background Tasks
|
||||
### Available Dependencies
|
||||
|
||||
Timeouts apply to **foreground execution**—when a tool runs directly in response to a client request. They protect your server from tools that unexpectedly hang due to network issues, resource contention, or other transient problems.
|
||||
Docket provides several dependencies you can use in your tools:
|
||||
|
||||
<Warning>
|
||||
The `timeout` parameter does **not** apply to background tasks. When a tool runs as a background task (`task=True`), execution happens in a Docket worker where the FastMCP timeout is not enforced.
|
||||
| Dependency | Purpose |
|
||||
|------------|---------|
|
||||
| `Timeout(timedelta)` | Cancel execution if it exceeds the duration |
|
||||
| `Retry` / `ExponentialRetry` | Automatically retry on failure |
|
||||
| `ConcurrencyLimit(n)` | Limit concurrent executions |
|
||||
| `Progress()` | Report progress during execution |
|
||||
|
||||
For task timeouts, use Docket's `Timeout` dependency directly in your function signature:
|
||||
|
||||
```python
|
||||
from datetime import timedelta
|
||||
from docket import Timeout
|
||||
|
||||
@mcp.tool(task=True)
|
||||
async def long_running_task(
|
||||
data: str,
|
||||
timeout: Timeout = Timeout(timedelta(minutes=10))
|
||||
) -> str:
|
||||
"""Task with a 10-minute timeout enforced by Docket."""
|
||||
...
|
||||
```
|
||||
|
||||
See the [Docket documentation](https://chrisguidry.github.io/docket/dependencies/#task-timeouts) for more on task timeouts and retries.
|
||||
</Warning>
|
||||
|
||||
When a tool times out, FastMCP logs a warning suggesting task mode. For operations you know will be long-running, use `task=True` instead—background tasks offload work to distributed workers and let clients poll for progress.
|
||||
See the [Docket documentation](https://chrisguidry.github.io/docket/dependencies/) for full details on each dependency.
|
||||
|
||||
## Component Visibility
|
||||
|
||||
|
|
|
|||
|
|
@ -313,6 +313,15 @@ class FunctionPrompt(Prompt):
|
|||
logger.exception(f"Error rendering prompt {self.name}")
|
||||
raise PromptError(f"Error rendering prompt {self.name}.") from e
|
||||
|
||||
@property
|
||||
def docket_callable(self) -> Callable[..., Any]:
|
||||
"""Return the callable that would be registered with Docket.
|
||||
|
||||
FunctionPrompt returns self.fn (the underlying function with user's
|
||||
Depends parameters for Docket to resolve).
|
||||
"""
|
||||
return self.fn
|
||||
|
||||
def register_with_docket(self, docket: Docket) -> None:
|
||||
"""Register this prompt with docket for background execution.
|
||||
|
||||
|
|
@ -321,7 +330,7 @@ class FunctionPrompt(Prompt):
|
|||
"""
|
||||
if not self.task_config.supports_tasks():
|
||||
return
|
||||
docket.register(self.fn, names=[self.key])
|
||||
docket.register(self.docket_callable, names=[self.key])
|
||||
|
||||
async def add_to_docket(
|
||||
self,
|
||||
|
|
|
|||
|
|
@ -357,17 +357,25 @@ class Prompt(FastMCPComponent):
|
|||
task_meta=task_meta,
|
||||
)
|
||||
if task_result:
|
||||
return task_result
|
||||
return task_result # type: ignore[return-value]
|
||||
|
||||
# Synchronous execution
|
||||
result = await self.render(arguments)
|
||||
return self.convert_result(result)
|
||||
|
||||
@property
|
||||
def docket_callable(self) -> Callable[..., Any]:
|
||||
"""Return the callable that would be registered with Docket.
|
||||
|
||||
For base Prompt, this is self.render. FunctionPrompt overrides to return self.fn.
|
||||
"""
|
||||
return self.render
|
||||
|
||||
def register_with_docket(self, docket: Docket) -> None:
|
||||
"""Register this prompt with docket for background execution."""
|
||||
if not self.task_config.supports_tasks():
|
||||
return
|
||||
docket.register(self.render, names=[self.key])
|
||||
docket.register(self.docket_callable, names=[self.key])
|
||||
|
||||
async def add_to_docket( # type: ignore[override]
|
||||
self,
|
||||
|
|
|
|||
|
|
@ -217,6 +217,15 @@ class FunctionResource(Resource):
|
|||
|
||||
return result
|
||||
|
||||
@property
|
||||
def docket_callable(self) -> Callable[..., Any]:
|
||||
"""Return the callable that would be registered with Docket.
|
||||
|
||||
FunctionResource returns self.fn (the underlying function with user's
|
||||
Depends parameters for Docket to resolve).
|
||||
"""
|
||||
return self.fn
|
||||
|
||||
def register_with_docket(self, docket: Docket) -> None:
|
||||
"""Register this resource with docket for background execution.
|
||||
|
||||
|
|
@ -225,7 +234,7 @@ class FunctionResource(Resource):
|
|||
"""
|
||||
if not self.task_config.supports_tasks():
|
||||
return
|
||||
docket.register(self.fn, names=[self.key])
|
||||
docket.register(self.docket_callable, names=[self.key])
|
||||
|
||||
|
||||
def resource(
|
||||
|
|
|
|||
|
|
@ -349,7 +349,7 @@ class Resource(FastMCPComponent):
|
|||
component=self, task_type="resource", arguments=None, task_meta=task_meta
|
||||
)
|
||||
if task_result:
|
||||
return task_result
|
||||
return task_result # type: ignore[return-value]
|
||||
|
||||
# Synchronous execution - convert result to ResourceResult
|
||||
result = await self.read()
|
||||
|
|
@ -383,11 +383,19 @@ class Resource(FastMCPComponent):
|
|||
base_key = self.make_key(str(self.uri))
|
||||
return f"{base_key}@{self.version or ''}"
|
||||
|
||||
@property
|
||||
def docket_callable(self) -> Callable[..., Any]:
|
||||
"""Return the callable that would be registered with Docket.
|
||||
|
||||
For base Resource, this is self.read. FunctionResource overrides to return self.fn.
|
||||
"""
|
||||
return self.read
|
||||
|
||||
def register_with_docket(self, docket: Docket) -> None:
|
||||
"""Register this resource with docket for background execution."""
|
||||
if not self.task_config.supports_tasks():
|
||||
return
|
||||
docket.register(self.read, names=[self.key])
|
||||
docket.register(self.docket_callable, names=[self.key])
|
||||
|
||||
async def add_to_docket( # type: ignore[override]
|
||||
self,
|
||||
|
|
|
|||
|
|
@ -230,7 +230,7 @@ class ResourceTemplate(FastMCPComponent):
|
|||
component=self, task_type="template", arguments=params, task_meta=task_meta
|
||||
)
|
||||
if task_result:
|
||||
return task_result
|
||||
return task_result # type: ignore[return-value]
|
||||
|
||||
# Synchronous execution - create resource and read directly
|
||||
# Call resource.read() not resource._read() to avoid task routing on ephemeral resource
|
||||
|
|
@ -287,11 +287,19 @@ class ResourceTemplate(FastMCPComponent):
|
|||
base_key = self.make_key(self.uri_template)
|
||||
return f"{base_key}@{self.version or ''}"
|
||||
|
||||
@property
|
||||
def docket_callable(self) -> Callable[..., Any]:
|
||||
"""Return the callable that would be registered with Docket.
|
||||
|
||||
For base ResourceTemplate, this is self.read. FunctionResourceTemplate overrides to return self.fn.
|
||||
"""
|
||||
return self.read
|
||||
|
||||
def register_with_docket(self, docket: Docket) -> None:
|
||||
"""Register this template with docket for background execution."""
|
||||
if not self.task_config.supports_tasks():
|
||||
return
|
||||
docket.register(self.read, names=[self.key])
|
||||
docket.register(self.docket_callable, names=[self.key])
|
||||
|
||||
async def add_to_docket( # type: ignore[override]
|
||||
self,
|
||||
|
|
@ -363,7 +371,7 @@ class FunctionResourceTemplate(ResourceTemplate):
|
|||
component=self, task_type="template", arguments=params, task_meta=task_meta
|
||||
)
|
||||
if task_result:
|
||||
return task_result
|
||||
return task_result # type: ignore[return-value]
|
||||
|
||||
# Synchronous execution - call read() directly, skip resource creation
|
||||
result = await self.read(arguments=params)
|
||||
|
|
@ -419,6 +427,15 @@ class FunctionResourceTemplate(ResourceTemplate):
|
|||
|
||||
return result
|
||||
|
||||
@property
|
||||
def docket_callable(self) -> Callable[..., Any]:
|
||||
"""Return the callable that would be registered with Docket.
|
||||
|
||||
FunctionResourceTemplate returns self.fn (the underlying function with user's
|
||||
Depends parameters for Docket to resolve).
|
||||
"""
|
||||
return self.fn
|
||||
|
||||
def register_with_docket(self, docket: Docket) -> None:
|
||||
"""Register this template with docket for background execution.
|
||||
|
||||
|
|
@ -427,7 +444,7 @@ class FunctionResourceTemplate(ResourceTemplate):
|
|||
"""
|
||||
if not self.task_config.supports_tasks():
|
||||
return
|
||||
docket.register(self.fn, names=[self.key])
|
||||
docket.register(self.docket_callable, names=[self.key])
|
||||
|
||||
async def add_to_docket(
|
||||
self,
|
||||
|
|
|
|||
|
|
@ -54,6 +54,7 @@ __all__ = [
|
|||
"get_server",
|
||||
"is_docket_available",
|
||||
"require_docket",
|
||||
"requires_docket_execution",
|
||||
"resolve_dependencies",
|
||||
"transform_context_annotations",
|
||||
"without_injected_parameters",
|
||||
|
|
@ -102,6 +103,61 @@ def require_docket(feature: str) -> None:
|
|||
)
|
||||
|
||||
|
||||
def requires_docket_execution(component: Any) -> bool:
|
||||
"""Check if a component has Docket-specific dependencies requiring Docket execution.
|
||||
|
||||
Docket provides dependencies like Timeout, Retry, ExponentialRetry that
|
||||
integrate with its task lifecycle. If a component's callable uses these,
|
||||
it must be executed through Docket for them to function properly.
|
||||
|
||||
Args:
|
||||
component: A FastMCP component (Tool, Resource, Prompt, etc.) with a
|
||||
docket_callable property.
|
||||
|
||||
Returns:
|
||||
True if the component's docket_callable uses Docket-specific dependencies.
|
||||
"""
|
||||
callable_fn = getattr(component, "docket_callable", None)
|
||||
if callable_fn is None:
|
||||
return False
|
||||
|
||||
if not is_docket_available():
|
||||
return False # Can't have Docket deps without Docket installed
|
||||
|
||||
try:
|
||||
from docket.dependencies import (
|
||||
ConcurrencyLimit,
|
||||
CurrentDocket,
|
||||
CurrentExecution,
|
||||
CurrentWorker,
|
||||
ExponentialRetry,
|
||||
Retry,
|
||||
Timeout,
|
||||
)
|
||||
from docket.dependencies import (
|
||||
get_dependency_parameters as _get_dep_params,
|
||||
)
|
||||
except ImportError:
|
||||
return False
|
||||
|
||||
# Docket dependency types that require Docket execution
|
||||
DOCKET_TYPES = (
|
||||
Timeout,
|
||||
Retry,
|
||||
ExponentialRetry,
|
||||
ConcurrencyLimit,
|
||||
CurrentDocket,
|
||||
CurrentWorker,
|
||||
CurrentExecution,
|
||||
)
|
||||
|
||||
# Use the DI system to find dependencies
|
||||
deps = _get_dep_params(callable_fn)
|
||||
|
||||
# Check if any dependency is a Docket type
|
||||
return any(isinstance(d, DOCKET_TYPES) for d in deps.values())
|
||||
|
||||
|
||||
# --- Dependency injection imports ---
|
||||
# Try docket first for isinstance compatibility in worker context,
|
||||
# fall back to vendored DI engine when docket is not installed.
|
||||
|
|
|
|||
|
|
@ -416,14 +416,18 @@ class Provider:
|
|||
# -------------------------------------------------------------------------
|
||||
|
||||
async def get_tasks(self) -> Sequence[FastMCPComponent]:
|
||||
"""Return components that should be registered as background tasks.
|
||||
"""Return components that should be registered with Docket.
|
||||
|
||||
Override to customize which components are task-eligible.
|
||||
Default calls list_* methods, applies provider transforms, and filters
|
||||
for components with task_config.mode != 'forbidden'.
|
||||
for components that either:
|
||||
- Have task_config.mode != 'forbidden' (explicit task support), OR
|
||||
- Have Docket-specific dependencies (Timeout, Retry, etc.)
|
||||
|
||||
Used by the server during startup to register functions with Docket.
|
||||
"""
|
||||
from fastmcp.server.dependencies import requires_docket_execution
|
||||
|
||||
# Fetch all component types in parallel
|
||||
results = await gather(
|
||||
self._list_tools(),
|
||||
|
|
@ -452,7 +456,7 @@ class Provider:
|
|||
*templates,
|
||||
*prompts,
|
||||
]
|
||||
if c.task_config.supports_tasks()
|
||||
if c.task_config.supports_tasks() or requires_docket_execution(c)
|
||||
]
|
||||
|
||||
# -------------------------------------------------------------------------
|
||||
|
|
|
|||
|
|
@ -64,7 +64,6 @@ class ToolDecoratorMixin:
|
|||
task=resolved_task,
|
||||
exclude_args=meta.exclude_args,
|
||||
serializer=meta.serializer,
|
||||
timeout=meta.timeout,
|
||||
auth=meta.auth,
|
||||
)
|
||||
else:
|
||||
|
|
@ -114,7 +113,6 @@ class ToolDecoratorMixin:
|
|||
enabled: bool = True,
|
||||
task: bool | TaskConfig | None = None,
|
||||
serializer: ToolResultSerializerType | None = None, # Deprecated
|
||||
timeout: float | None = None,
|
||||
auth: AuthCheckCallable | list[AuthCheckCallable] | None = None,
|
||||
) -> Callable[[AnyFunction], FunctionTool]: ...
|
||||
|
||||
|
|
@ -139,7 +137,6 @@ class ToolDecoratorMixin:
|
|||
enabled: bool = True,
|
||||
task: bool | TaskConfig | None = None,
|
||||
serializer: ToolResultSerializerType | None = None, # Deprecated
|
||||
timeout: float | None = None,
|
||||
auth: AuthCheckCallable | list[AuthCheckCallable] | None = None,
|
||||
) -> (
|
||||
Callable[[AnyFunction], FunctionTool]
|
||||
|
|
@ -243,7 +240,6 @@ class ToolDecoratorMixin:
|
|||
meta=meta,
|
||||
serializer=serializer,
|
||||
task=resolved_task,
|
||||
timeout=timeout,
|
||||
auth=auth,
|
||||
)
|
||||
self._add_component(tool_obj)
|
||||
|
|
@ -266,7 +262,6 @@ class ToolDecoratorMixin:
|
|||
task=task,
|
||||
exclude_args=exclude_args,
|
||||
serializer=serializer,
|
||||
timeout=timeout,
|
||||
auth=auth,
|
||||
enabled=enabled,
|
||||
)
|
||||
|
|
@ -310,6 +305,5 @@ class ToolDecoratorMixin:
|
|||
enabled=enabled,
|
||||
task=task,
|
||||
serializer=serializer,
|
||||
timeout=timeout,
|
||||
auth=auth,
|
||||
)
|
||||
|
|
|
|||
|
|
@ -447,13 +447,22 @@ class LocalProvider(
|
|||
# =========================================================================
|
||||
|
||||
async def get_tasks(self) -> Sequence[FastMCPComponent]:
|
||||
"""Return components eligible for background task execution.
|
||||
"""Return components eligible for Docket registration.
|
||||
|
||||
Returns components that:
|
||||
- Have task_config.mode != 'forbidden' (explicit task support), OR
|
||||
- Have Docket-specific dependencies (Timeout, Retry, etc.)
|
||||
|
||||
Returns components that have task_config.mode != 'forbidden'.
|
||||
This includes both FunctionTool/Resource/Prompt instances created via
|
||||
decorators and custom Tool/Resource/Prompt subclasses.
|
||||
"""
|
||||
return [c for c in self._components.values() if c.task_config.supports_tasks()]
|
||||
from fastmcp.server.dependencies import requires_docket_execution
|
||||
|
||||
return [
|
||||
c
|
||||
for c in self._components.values()
|
||||
if c.task_config.supports_tasks() or requires_docket_execution(c)
|
||||
]
|
||||
|
||||
# =========================================================================
|
||||
# Decorator methods
|
||||
|
|
|
|||
|
|
@ -1371,7 +1371,6 @@ class FastMCP(
|
|||
exclude_args: list[str] | None = None,
|
||||
meta: dict[str, Any] | None = None,
|
||||
task: bool | TaskConfig | None = None,
|
||||
timeout: float | None = None,
|
||||
auth: AuthCheckCallable | list[AuthCheckCallable] | None = None,
|
||||
) -> FunctionTool: ...
|
||||
|
||||
|
|
@ -1391,7 +1390,6 @@ class FastMCP(
|
|||
exclude_args: list[str] | None = None,
|
||||
meta: dict[str, Any] | None = None,
|
||||
task: bool | TaskConfig | None = None,
|
||||
timeout: float | None = None,
|
||||
auth: AuthCheckCallable | list[AuthCheckCallable] | None = None,
|
||||
) -> Callable[[AnyFunction], FunctionTool]: ...
|
||||
|
||||
|
|
@ -1410,7 +1408,6 @@ class FastMCP(
|
|||
exclude_args: list[str] | None = None,
|
||||
meta: dict[str, Any] | None = None,
|
||||
task: bool | TaskConfig | None = None,
|
||||
timeout: float | None = None,
|
||||
auth: AuthCheckCallable | list[AuthCheckCallable] | None = None,
|
||||
) -> (
|
||||
Callable[[AnyFunction], FunctionTool]
|
||||
|
|
@ -1479,7 +1476,6 @@ class FastMCP(
|
|||
exclude_args=exclude_args,
|
||||
meta=meta,
|
||||
task=task if task is not None else self._support_tasks_by_default,
|
||||
timeout=timeout,
|
||||
serializer=self._tool_serializer,
|
||||
auth=auth,
|
||||
)
|
||||
|
|
|
|||
|
|
@ -8,21 +8,23 @@ from __future__ import annotations
|
|||
import uuid
|
||||
from contextlib import suppress
|
||||
from datetime import datetime, timezone
|
||||
from typing import TYPE_CHECKING, Any, Literal
|
||||
from typing import Any, Literal
|
||||
|
||||
import mcp.types
|
||||
from docket.execution import ExecutionState
|
||||
from mcp.shared.exceptions import McpError
|
||||
from mcp.types import INTERNAL_ERROR, ErrorData
|
||||
|
||||
from fastmcp.prompts.prompt import Prompt, PromptResult
|
||||
from fastmcp.resources.resource import Resource, ResourceResult
|
||||
from fastmcp.resources.template import ResourceTemplate
|
||||
from fastmcp.server.dependencies import _current_docket, get_context
|
||||
from fastmcp.server.tasks.config import TaskMeta
|
||||
from fastmcp.server.tasks.keys import build_task_key
|
||||
from fastmcp.tools.tool import Tool, ToolResult
|
||||
from fastmcp.utilities.logging import get_logger
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from fastmcp.prompts.prompt import Prompt
|
||||
from fastmcp.resources.resource import Resource
|
||||
from fastmcp.resources.template import ResourceTemplate
|
||||
from fastmcp.tools.tool import Tool
|
||||
logger = get_logger(__name__)
|
||||
|
||||
# Redis mapping TTL buffer: Add 15 minutes to Docket's execution_ttl
|
||||
TASK_MAPPING_TTL_BUFFER_SECONDS = 15 * 60
|
||||
|
|
@ -155,3 +157,86 @@ async def submit_to_docket(
|
|||
pollInterval=poll_interval_ms,
|
||||
)
|
||||
)
|
||||
|
||||
|
||||
async def run_in_docket_sync(
|
||||
task_type: Literal["tool", "resource", "template", "prompt"],
|
||||
component: Tool | Resource | ResourceTemplate | Prompt,
|
||||
arguments: dict[str, Any] | None = None,
|
||||
) -> ToolResult | ResourceResult | PromptResult:
|
||||
"""Submit to Docket and wait for result (foreground semantics).
|
||||
|
||||
This is used when a component has Docket-specific dependencies (Timeout,
|
||||
Retry, etc.) but the caller didn't request background execution. We run
|
||||
through Docket to honor the dependencies, but wait for the result.
|
||||
|
||||
Args:
|
||||
task_type: Component type ("tool", "resource", "template", "prompt")
|
||||
component: The component instance
|
||||
arguments: Arguments for tool/prompt/template execution
|
||||
|
||||
Returns:
|
||||
The component's result (ToolResult, ResourceResult, or PromptResult)
|
||||
|
||||
Raises:
|
||||
McpError: If Docket is not available or execution fails
|
||||
"""
|
||||
docket = _current_docket.get()
|
||||
if docket is None:
|
||||
raise McpError(
|
||||
ErrorData(
|
||||
code=INTERNAL_ERROR,
|
||||
message="Docket dependencies require a running FastMCP server with Docket enabled",
|
||||
)
|
||||
)
|
||||
|
||||
# Generate a unique task key for this sync execution
|
||||
task_key = f"fastmcp:sync:{uuid.uuid4()}"
|
||||
fn_key = component.key
|
||||
|
||||
# Queue the function to Docket
|
||||
# Component types have different add_to_docket signatures
|
||||
if isinstance(component, Resource):
|
||||
execution = await component.add_to_docket(
|
||||
docket, fn_key=fn_key, task_key=task_key
|
||||
)
|
||||
elif isinstance(component, (Tool, ResourceTemplate)):
|
||||
args = arguments if arguments is not None else {}
|
||||
execution = await component.add_to_docket(
|
||||
docket, args, fn_key=fn_key, task_key=task_key
|
||||
)
|
||||
else: # Prompt
|
||||
execution = await component.add_to_docket(
|
||||
docket, arguments, fn_key=fn_key, task_key=task_key
|
||||
)
|
||||
|
||||
# Wait for execution to complete
|
||||
terminal_states: set[ExecutionState] = {
|
||||
ExecutionState.COMPLETED,
|
||||
ExecutionState.FAILED,
|
||||
ExecutionState.CANCELLED,
|
||||
}
|
||||
|
||||
async for event in execution.subscribe():
|
||||
if hasattr(event, "state") and event.state in terminal_states:
|
||||
break
|
||||
|
||||
# Get the result - execution is complete so this returns immediately.
|
||||
# If the task failed, get_result re-raises the stored exception (which could
|
||||
# be any exception type from the user's function), so we catch broadly.
|
||||
try:
|
||||
raw_result = await execution.get_result()
|
||||
except McpError:
|
||||
raise # Don't wrap MCP errors
|
||||
except Exception as e:
|
||||
logger.error(f"Docket execution failed for {component.key}: {e}")
|
||||
raise McpError(
|
||||
ErrorData(
|
||||
code=INTERNAL_ERROR,
|
||||
message=f"Docket execution failed: {e}",
|
||||
)
|
||||
) from e
|
||||
|
||||
# Convert raw result to appropriate MCP type
|
||||
# The convert_result method handles the transformation
|
||||
return component.convert_result(raw_result)
|
||||
|
|
|
|||
|
|
@ -11,14 +11,15 @@ import mcp.types
|
|||
from mcp.shared.exceptions import McpError
|
||||
from mcp.types import METHOD_NOT_FOUND, ErrorData
|
||||
|
||||
from fastmcp.server.dependencies import requires_docket_execution
|
||||
from fastmcp.server.tasks.config import TaskMeta
|
||||
from fastmcp.server.tasks.handlers import submit_to_docket
|
||||
from fastmcp.server.tasks.handlers import run_in_docket_sync, submit_to_docket
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from fastmcp.prompts.prompt import Prompt
|
||||
from fastmcp.resources.resource import Resource
|
||||
from fastmcp.prompts.prompt import Prompt, PromptResult
|
||||
from fastmcp.resources.resource import Resource, ResourceResult
|
||||
from fastmcp.resources.template import ResourceTemplate
|
||||
from fastmcp.tools.tool import Tool
|
||||
from fastmcp.tools.tool import Tool, ToolResult
|
||||
|
||||
TaskType = Literal["tool", "resource", "template", "prompt"]
|
||||
|
||||
|
|
@ -28,9 +29,13 @@ async def check_background_task(
|
|||
task_type: TaskType,
|
||||
arguments: dict[str, Any] | None = None,
|
||||
task_meta: TaskMeta | None = None,
|
||||
) -> mcp.types.CreateTaskResult | None:
|
||||
) -> mcp.types.CreateTaskResult | ToolResult | ResourceResult | PromptResult | None:
|
||||
"""Check task mode and submit to background if requested.
|
||||
|
||||
Also handles auto-routing for components with Docket dependencies:
|
||||
if a component uses Timeout, Retry, etc. and wasn't explicitly requested
|
||||
as a background task, we run it through Docket with sync-wait semantics.
|
||||
|
||||
Args:
|
||||
component: The MCP component
|
||||
task_type: Type of task ("tool", "resource", "template", "prompt")
|
||||
|
|
@ -38,7 +43,9 @@ async def check_background_task(
|
|||
task_meta: Task execution metadata. If provided, execute as background task.
|
||||
|
||||
Returns:
|
||||
CreateTaskResult if submitted to docket, None for sync execution
|
||||
CreateTaskResult if submitted to docket as background task,
|
||||
ToolResult/ResourceResult/PromptResult if auto-routed through Docket sync,
|
||||
None for regular sync execution
|
||||
|
||||
Raises:
|
||||
McpError: If mode="required" but no task metadata, or mode="forbidden"
|
||||
|
|
@ -67,7 +74,12 @@ async def check_background_task(
|
|||
)
|
||||
)
|
||||
|
||||
# No task metadata - synchronous execution
|
||||
# Auto-route through Docket if component has Docket dependencies
|
||||
# This ensures Timeout, Retry, etc. work even for foreground calls
|
||||
if not task_meta and requires_docket_execution(component):
|
||||
return await run_in_docket_sync(task_type, component, arguments)
|
||||
|
||||
# No task metadata - regular synchronous execution
|
||||
if not task_meta:
|
||||
return None
|
||||
|
||||
|
|
|
|||
|
|
@ -16,10 +16,8 @@ from typing import (
|
|||
runtime_checkable,
|
||||
)
|
||||
|
||||
import anyio
|
||||
import mcp.types
|
||||
from mcp.shared.exceptions import McpError
|
||||
from mcp.types import ErrorData, Icon, ToolAnnotations, ToolExecution
|
||||
from mcp.types import Icon, ToolAnnotations, ToolExecution
|
||||
|
||||
import fastmcp
|
||||
from fastmcp.decorators import resolve_task_config
|
||||
|
|
@ -75,7 +73,6 @@ class ToolMeta:
|
|||
task: bool | TaskConfig | None = None
|
||||
exclude_args: list[str] | None = None
|
||||
serializer: Any | None = None
|
||||
timeout: float | None = None
|
||||
auth: AuthCheckCallable | list[AuthCheckCallable] | None = None
|
||||
enabled: bool = True
|
||||
|
||||
|
|
@ -120,7 +117,6 @@ class FunctionTool(Tool):
|
|||
serializer: ToolResultSerializerType | None = None,
|
||||
meta: dict[str, Any] | None = None,
|
||||
task: bool | TaskConfig | None = None,
|
||||
timeout: float | None = None,
|
||||
auth: AuthCheckCallable | list[AuthCheckCallable] | None = None,
|
||||
) -> FunctionTool:
|
||||
"""Create a FunctionTool from a function.
|
||||
|
|
@ -147,7 +143,6 @@ class FunctionTool(Tool):
|
|||
meta,
|
||||
task,
|
||||
serializer,
|
||||
timeout,
|
||||
auth,
|
||||
]
|
||||
)
|
||||
|
|
@ -176,7 +171,6 @@ class FunctionTool(Tool):
|
|||
task=task,
|
||||
exclude_args=exclude_args,
|
||||
serializer=serializer,
|
||||
timeout=timeout,
|
||||
auth=auth,
|
||||
)
|
||||
|
||||
|
|
@ -240,7 +234,6 @@ class FunctionTool(Tool):
|
|||
serializer=metadata.serializer,
|
||||
meta=metadata.meta,
|
||||
task_config=task_config,
|
||||
timeout=metadata.timeout,
|
||||
auth=metadata.auth,
|
||||
)
|
||||
|
||||
|
|
@ -249,46 +242,26 @@ class FunctionTool(Tool):
|
|||
wrapper_fn = without_injected_parameters(self.fn)
|
||||
type_adapter = get_cached_typeadapter(wrapper_fn)
|
||||
|
||||
# Apply timeout if configured
|
||||
if self.timeout is not None:
|
||||
try:
|
||||
with anyio.fail_after(self.timeout):
|
||||
# Thread pool execution for sync functions, direct await for async
|
||||
if inspect.iscoroutinefunction(wrapper_fn):
|
||||
result = await type_adapter.validate_python(arguments)
|
||||
else:
|
||||
# Sync function: run in threadpool to avoid blocking
|
||||
result = await call_sync_fn_in_threadpool(
|
||||
type_adapter.validate_python, arguments
|
||||
)
|
||||
# Handle sync wrappers that return awaitables
|
||||
if inspect.isawaitable(result):
|
||||
result = await result
|
||||
except TimeoutError:
|
||||
logger.warning(
|
||||
f"Tool '{self.name}' timed out after {self.timeout}s. "
|
||||
f"Consider using task=True for long-running operations. "
|
||||
f"See https://gofastmcp.com/servers/tasks"
|
||||
)
|
||||
raise McpError(
|
||||
ErrorData(
|
||||
code=-32000,
|
||||
message=f"Tool '{self.name}' execution timed out after {self.timeout}s",
|
||||
)
|
||||
) from None
|
||||
if inspect.iscoroutinefunction(wrapper_fn):
|
||||
result = await type_adapter.validate_python(arguments)
|
||||
else:
|
||||
# No timeout: use existing execution path
|
||||
if inspect.iscoroutinefunction(wrapper_fn):
|
||||
result = await type_adapter.validate_python(arguments)
|
||||
else:
|
||||
result = await call_sync_fn_in_threadpool(
|
||||
type_adapter.validate_python, arguments
|
||||
)
|
||||
if inspect.isawaitable(result):
|
||||
result = await result
|
||||
result = await call_sync_fn_in_threadpool(
|
||||
type_adapter.validate_python, arguments
|
||||
)
|
||||
if inspect.isawaitable(result):
|
||||
result = await result
|
||||
|
||||
return self.convert_result(result)
|
||||
|
||||
@property
|
||||
def docket_callable(self) -> Callable[..., Any]:
|
||||
"""Return the callable that would be registered with Docket.
|
||||
|
||||
FunctionTool returns self.fn (the underlying function with user's
|
||||
Depends parameters for Docket to resolve).
|
||||
"""
|
||||
return self.fn
|
||||
|
||||
def register_with_docket(self, docket: Docket) -> None:
|
||||
"""Register this tool with docket for background execution.
|
||||
|
||||
|
|
@ -297,7 +270,7 @@ class FunctionTool(Tool):
|
|||
"""
|
||||
if not self.task_config.supports_tasks():
|
||||
return
|
||||
docket.register(self.fn, names=[self.key])
|
||||
docket.register(self.docket_callable, names=[self.key])
|
||||
|
||||
async def add_to_docket(
|
||||
self,
|
||||
|
|
@ -342,7 +315,6 @@ def tool(
|
|||
task: bool | TaskConfig | None = None,
|
||||
exclude_args: list[str] | None = None,
|
||||
serializer: Any | None = None,
|
||||
timeout: float | None = None,
|
||||
auth: AuthCheckCallable | list[AuthCheckCallable] | None = None,
|
||||
) -> Callable[[F], F]: ...
|
||||
@overload
|
||||
|
|
@ -361,7 +333,6 @@ def tool(
|
|||
task: bool | TaskConfig | None = None,
|
||||
exclude_args: list[str] | None = None,
|
||||
serializer: Any | None = None,
|
||||
timeout: float | None = None,
|
||||
auth: AuthCheckCallable | list[AuthCheckCallable] | None = None,
|
||||
) -> Callable[[F], F]: ...
|
||||
|
||||
|
|
@ -381,7 +352,6 @@ def tool(
|
|||
task: bool | TaskConfig | None = None,
|
||||
exclude_args: list[str] | None = None,
|
||||
serializer: Any | None = None,
|
||||
timeout: float | None = None,
|
||||
auth: AuthCheckCallable | list[AuthCheckCallable] | None = None,
|
||||
) -> Any:
|
||||
"""Standalone decorator to mark a function as an MCP tool.
|
||||
|
|
@ -413,7 +383,6 @@ def tool(
|
|||
task=resolve_task_config(task),
|
||||
exclude_args=exclude_args,
|
||||
serializer=serializer,
|
||||
timeout=timeout,
|
||||
auth=auth,
|
||||
)
|
||||
return FunctionTool.from_function(fn, metadata=tool_meta)
|
||||
|
|
@ -432,7 +401,6 @@ def tool(
|
|||
task=task,
|
||||
exclude_args=exclude_args,
|
||||
serializer=serializer,
|
||||
timeout=timeout,
|
||||
auth=auth,
|
||||
)
|
||||
target = fn.__func__ if hasattr(fn, "__func__") else fn
|
||||
|
|
|
|||
|
|
@ -149,12 +149,6 @@ class Tool(FastMCPComponent):
|
|||
AuthCheckCallable | list[AuthCheckCallable] | None,
|
||||
Field(description="Authorization checks for this tool", exclude=True),
|
||||
] = None
|
||||
timeout: Annotated[
|
||||
float | None,
|
||||
Field(
|
||||
description="Execution timeout in seconds. If None, no timeout is applied."
|
||||
),
|
||||
] = None
|
||||
|
||||
@model_validator(mode="after")
|
||||
def _validate_tool_name(self) -> Tool:
|
||||
|
|
@ -205,7 +199,6 @@ class Tool(FastMCPComponent):
|
|||
serializer: ToolResultSerializerType | None = None, # Deprecated
|
||||
meta: dict[str, Any] | None = None,
|
||||
task: bool | TaskConfig | None = None,
|
||||
timeout: float | None = None,
|
||||
auth: AuthCheckCallable | list[AuthCheckCallable] | None = None,
|
||||
) -> FunctionTool:
|
||||
"""Create a Tool from a function."""
|
||||
|
|
@ -225,7 +218,6 @@ class Tool(FastMCPComponent):
|
|||
serializer=serializer,
|
||||
meta=meta,
|
||||
task=task,
|
||||
timeout=timeout,
|
||||
auth=auth,
|
||||
)
|
||||
|
||||
|
|
@ -329,15 +321,23 @@ class Tool(FastMCPComponent):
|
|||
task_meta=task_meta,
|
||||
)
|
||||
if task_result:
|
||||
return task_result
|
||||
return task_result # type: ignore[return-value]
|
||||
|
||||
return await self.run(arguments)
|
||||
|
||||
@property
|
||||
def docket_callable(self) -> Callable[..., Any]:
|
||||
"""Return the callable that would be registered with Docket.
|
||||
|
||||
For base Tool, this is self.run. FunctionTool overrides to return self.fn.
|
||||
"""
|
||||
return self.run
|
||||
|
||||
def register_with_docket(self, docket: Docket) -> None:
|
||||
"""Register this tool with docket for background execution."""
|
||||
if not self.task_config.supports_tasks():
|
||||
return
|
||||
docket.register(self.run, names=[self.key])
|
||||
docket.register(self.docket_callable, names=[self.key])
|
||||
|
||||
async def add_to_docket( # type: ignore[override]
|
||||
self,
|
||||
|
|
|
|||
|
|
@ -1,6 +1,6 @@
|
|||
from __future__ import annotations
|
||||
|
||||
from collections.abc import Sequence
|
||||
from collections.abc import Callable, Sequence
|
||||
from typing import TYPE_CHECKING, Annotated, Any, ClassVar, TypedDict, cast
|
||||
|
||||
from mcp.types import Icon
|
||||
|
|
@ -204,6 +204,18 @@ class FastMCPComponent(FastMCPBaseModel):
|
|||
"""Create a copy of the component."""
|
||||
return self.model_copy()
|
||||
|
||||
@property
|
||||
def docket_callable(self) -> Callable[..., Any] | None:
|
||||
"""Return the callable that would be registered with Docket.
|
||||
|
||||
Subclasses override to return their specific callable:
|
||||
- Tool: self.run (or self.fn for FunctionTool)
|
||||
- Resource: self.read (or self.fn for FunctionResource)
|
||||
- Prompt: self.render (or self.fn for FunctionPrompt)
|
||||
- ResourceTemplate: self.read (or self.fn for FunctionResourceTemplate)
|
||||
"""
|
||||
return None # Base implementation; subclasses override
|
||||
|
||||
def register_with_docket(self, docket: Docket) -> None:
|
||||
"""Register this component with docket for background execution.
|
||||
|
||||
|
|
|
|||
|
|
@ -1,21 +1,29 @@
|
|||
"""Tests for tool timeout functionality."""
|
||||
"""Tests for tool timeout and Docket dependency detection."""
|
||||
|
||||
import time
|
||||
|
||||
import anyio
|
||||
import pytest
|
||||
from mcp.shared.exceptions import McpError
|
||||
from mcp.types import TextContent
|
||||
|
||||
from fastmcp import FastMCP
|
||||
from fastmcp.exceptions import ToolError
|
||||
|
||||
|
||||
class TestToolTimeout:
|
||||
"""Test tool timeout behavior."""
|
||||
def _docket_available() -> bool:
|
||||
"""Check if Docket is available."""
|
||||
try:
|
||||
import docket # noqa: F401
|
||||
|
||||
async def test_no_timeout_completes_normally_async(self):
|
||||
"""Tool without timeout completes normally (async)."""
|
||||
return True
|
||||
except ImportError:
|
||||
return False
|
||||
|
||||
|
||||
class TestToolExecution:
|
||||
"""Test basic tool execution (timeout parameter removed in v3.0.0b2)."""
|
||||
|
||||
async def test_async_tool_completes_normally(self):
|
||||
"""Async tool completes normally."""
|
||||
mcp = FastMCP()
|
||||
|
||||
@mcp.tool
|
||||
|
|
@ -27,8 +35,8 @@ class TestToolTimeout:
|
|||
assert isinstance(result.content[0], TextContent)
|
||||
assert result.content[0].text == "completed"
|
||||
|
||||
async def test_no_timeout_completes_normally_sync(self):
|
||||
"""Tool without timeout completes normally (sync)."""
|
||||
async def test_sync_tool_completes_normally(self):
|
||||
"""Sync tool completes normally."""
|
||||
mcp = FastMCP()
|
||||
|
||||
@mcp.tool
|
||||
|
|
@ -40,153 +48,149 @@ class TestToolTimeout:
|
|||
assert isinstance(result.content[0], TextContent)
|
||||
assert result.content[0].text == "completed"
|
||||
|
||||
async def test_timeout_not_reached_async(self):
|
||||
"""Async tool with timeout completes before timeout."""
|
||||
async def test_multiple_tools_complete_normally(self):
|
||||
"""Multiple tools can run and complete."""
|
||||
mcp = FastMCP()
|
||||
|
||||
@mcp.tool(timeout=5.0)
|
||||
async def fast_async_tool() -> str:
|
||||
await anyio.sleep(0.1)
|
||||
return "completed"
|
||||
|
||||
result = await mcp.call_tool("fast_async_tool")
|
||||
assert isinstance(result.content[0], TextContent)
|
||||
assert result.content[0].text == "completed"
|
||||
|
||||
async def test_timeout_not_reached_sync(self):
|
||||
"""Sync tool with timeout completes before timeout."""
|
||||
mcp = FastMCP()
|
||||
|
||||
@mcp.tool(timeout=5.0)
|
||||
def fast_sync_tool() -> str:
|
||||
time.sleep(0.1)
|
||||
return "completed"
|
||||
|
||||
result = await mcp.call_tool("fast_sync_tool")
|
||||
assert isinstance(result.content[0], TextContent)
|
||||
assert result.content[0].text == "completed"
|
||||
|
||||
async def test_async_timeout_exceeded(self):
|
||||
"""Async tool exceeds timeout and raises TimeoutError."""
|
||||
mcp = FastMCP()
|
||||
|
||||
@mcp.tool(timeout=0.2)
|
||||
async def slow_async_tool() -> str:
|
||||
await anyio.sleep(2.0)
|
||||
return "should not reach"
|
||||
|
||||
# TimeoutError is caught and converted to ToolError by FastMCP
|
||||
with pytest.raises(ToolError) as exc_info:
|
||||
await mcp.call_tool("slow_async_tool")
|
||||
|
||||
# Verify the tool raised an error (error message may be masked)
|
||||
assert exc_info.value is not None
|
||||
|
||||
async def test_sync_timeout_exceeded(self):
|
||||
"""Sync tool timeout works with CPU-bound operations."""
|
||||
pytest.skip(
|
||||
"Sync timeouts require thread pool execution (coming in future commit)"
|
||||
)
|
||||
# Note: time.sleep() blocks the event loop and cannot be interrupted
|
||||
# by anyio.fail_after(). This will work once sync functions run in
|
||||
# thread pools (commit 8c471a49).
|
||||
|
||||
async def test_timeout_error_raises_tool_error(self):
|
||||
"""Timeout error is converted to ToolError and logs warning."""
|
||||
mcp = FastMCP()
|
||||
|
||||
@mcp.tool(timeout=0.1)
|
||||
async def slow_tool() -> str:
|
||||
await anyio.sleep(1.0)
|
||||
return "never"
|
||||
|
||||
# Verify that ToolError is raised (timeout warning is logged to stderr)
|
||||
with pytest.raises(ToolError):
|
||||
await mcp.call_tool("slow_tool")
|
||||
|
||||
async def test_timeout_from_tool_from_function(self):
|
||||
"""Timeout works when using Tool.from_function()."""
|
||||
from fastmcp.tools import Tool
|
||||
|
||||
async def my_slow_tool() -> str:
|
||||
await anyio.sleep(1.0)
|
||||
return "never"
|
||||
|
||||
tool = Tool.from_function(my_slow_tool, timeout=0.1)
|
||||
mcp = FastMCP()
|
||||
mcp.add_tool(tool)
|
||||
|
||||
with pytest.raises(ToolError):
|
||||
await mcp.call_tool("my_slow_tool")
|
||||
|
||||
async def test_timeout_zero_times_out_immediately(self):
|
||||
"""Timeout of 0 times out immediately."""
|
||||
mcp = FastMCP()
|
||||
|
||||
@mcp.tool(timeout=0.0)
|
||||
async def instant_timeout() -> str:
|
||||
await anyio.sleep(0) # Give the event loop a chance to check timeout
|
||||
return "never"
|
||||
|
||||
with pytest.raises(ToolError):
|
||||
await mcp.call_tool("instant_timeout")
|
||||
|
||||
async def test_timeout_with_task_mode(self):
|
||||
"""Tool with timeout and task mode can be configured together."""
|
||||
mcp = FastMCP(tasks=True)
|
||||
|
||||
@mcp.tool(task=True, timeout=1.0)
|
||||
async def task_with_timeout() -> str:
|
||||
await anyio.sleep(0.1)
|
||||
return "completed"
|
||||
|
||||
# Tool should be registered successfully
|
||||
tools = await mcp.list_tools()
|
||||
tool = next((t for t in tools if t.name == "task_with_timeout"), None)
|
||||
assert tool is not None
|
||||
assert tool.timeout == 1.0
|
||||
assert tool.task_config.supports_tasks()
|
||||
|
||||
async def test_multiple_tools_with_different_timeouts(self):
|
||||
"""Multiple tools can have different timeout values."""
|
||||
mcp = FastMCP()
|
||||
|
||||
@mcp.tool(timeout=1.0)
|
||||
async def short_timeout() -> str:
|
||||
await anyio.sleep(0.1)
|
||||
return "short"
|
||||
|
||||
@mcp.tool(timeout=5.0)
|
||||
async def long_timeout() -> str:
|
||||
await anyio.sleep(0.1)
|
||||
return "long"
|
||||
|
||||
@mcp.tool
|
||||
async def no_timeout() -> str:
|
||||
await anyio.sleep(0.1)
|
||||
return "none"
|
||||
async def tool_one() -> str:
|
||||
await anyio.sleep(0.01)
|
||||
return "one"
|
||||
|
||||
# All should complete successfully
|
||||
result1 = await mcp.call_tool("short_timeout")
|
||||
result2 = await mcp.call_tool("long_timeout")
|
||||
result3 = await mcp.call_tool("no_timeout")
|
||||
@mcp.tool
|
||||
async def tool_two() -> str:
|
||||
await anyio.sleep(0.01)
|
||||
return "two"
|
||||
|
||||
@mcp.tool
|
||||
def tool_three() -> str:
|
||||
time.sleep(0.01)
|
||||
return "three"
|
||||
|
||||
result1 = await mcp.call_tool("tool_one")
|
||||
result2 = await mcp.call_tool("tool_two")
|
||||
result3 = await mcp.call_tool("tool_three")
|
||||
|
||||
assert isinstance(result1.content[0], TextContent)
|
||||
assert isinstance(result2.content[0], TextContent)
|
||||
assert isinstance(result3.content[0], TextContent)
|
||||
assert result1.content[0].text == "short"
|
||||
assert result2.content[0].text == "long"
|
||||
assert result3.content[0].text == "none"
|
||||
assert result1.content[0].text == "one"
|
||||
assert result2.content[0].text == "two"
|
||||
assert result3.content[0].text == "three"
|
||||
|
||||
async def test_timeout_error_converted_to_tool_error(self):
|
||||
"""Timeout errors are converted to ToolError by FastMCP."""
|
||||
mcp = FastMCP()
|
||||
|
||||
@mcp.tool(timeout=0.1)
|
||||
async def times_out() -> str:
|
||||
await anyio.sleep(1.0)
|
||||
return "never"
|
||||
class TestDocketCallableProperty:
|
||||
"""Test the docket_callable property on components."""
|
||||
|
||||
# TimeoutError should be caught and converted to ToolError
|
||||
with pytest.raises((ToolError, McpError)):
|
||||
await mcp.call_tool("times_out")
|
||||
async def test_function_tool_docket_callable_returns_fn(self):
|
||||
"""FunctionTool.docket_callable returns the underlying function."""
|
||||
from fastmcp.tools import Tool
|
||||
|
||||
async def my_tool() -> str:
|
||||
return "result"
|
||||
|
||||
tool = Tool.from_function(my_tool)
|
||||
# FunctionTool should return self.fn
|
||||
assert tool.docket_callable is not None
|
||||
# It should be callable
|
||||
assert callable(tool.docket_callable)
|
||||
|
||||
async def test_function_resource_docket_callable_returns_fn(self):
|
||||
"""FunctionResource.docket_callable returns the underlying function."""
|
||||
from fastmcp.resources import Resource
|
||||
|
||||
def my_resource() -> str:
|
||||
return "data"
|
||||
|
||||
resource = Resource.from_function(my_resource, uri="data://test")
|
||||
assert resource.docket_callable is not None
|
||||
assert callable(resource.docket_callable)
|
||||
|
||||
async def test_function_prompt_docket_callable_returns_fn(self):
|
||||
"""FunctionPrompt.docket_callable returns the underlying function."""
|
||||
from fastmcp.prompts import Prompt
|
||||
|
||||
def my_prompt(topic: str) -> str:
|
||||
return f"Write about {topic}"
|
||||
|
||||
prompt = Prompt.from_function(my_prompt)
|
||||
assert prompt.docket_callable is not None
|
||||
assert callable(prompt.docket_callable)
|
||||
|
||||
|
||||
class TestRequiresDocketExecution:
|
||||
"""Test the requires_docket_execution() function."""
|
||||
|
||||
def test_tool_without_docket_deps_returns_false(self):
|
||||
"""Tool without Docket dependencies returns False."""
|
||||
from fastmcp.server.dependencies import requires_docket_execution
|
||||
from fastmcp.tools import Tool
|
||||
|
||||
async def simple_tool(x: int) -> int:
|
||||
return x * 2
|
||||
|
||||
tool = Tool.from_function(simple_tool)
|
||||
assert requires_docket_execution(tool) is False
|
||||
|
||||
def test_tool_with_regular_deps_returns_false(self):
|
||||
"""Tool with regular (non-Docket) dependencies returns False."""
|
||||
from fastmcp.server.context import Context
|
||||
from fastmcp.server.dependencies import requires_docket_execution
|
||||
from fastmcp.tools import Tool
|
||||
|
||||
async def tool_with_context(x: int, ctx: Context) -> int:
|
||||
return x * 2
|
||||
|
||||
tool = Tool.from_function(tool_with_context)
|
||||
assert requires_docket_execution(tool) is False
|
||||
|
||||
def test_component_without_docket_callable_returns_false(self):
|
||||
"""Component without docket_callable returns False."""
|
||||
from fastmcp.server.dependencies import requires_docket_execution
|
||||
|
||||
class FakeComponent:
|
||||
pass
|
||||
|
||||
assert requires_docket_execution(FakeComponent()) is False
|
||||
|
||||
@pytest.mark.skipif(
|
||||
not _docket_available(),
|
||||
reason="Docket not installed",
|
||||
)
|
||||
def test_tool_with_timeout_dep_returns_true(self):
|
||||
"""Tool with Docket Timeout dependency returns True."""
|
||||
from datetime import timedelta
|
||||
|
||||
from docket import Timeout
|
||||
|
||||
from fastmcp.server.dependencies import requires_docket_execution
|
||||
from fastmcp.tools import Tool
|
||||
|
||||
async def tool_with_timeout(
|
||||
x: int,
|
||||
timeout: Timeout = Timeout(timedelta(seconds=30)),
|
||||
) -> int:
|
||||
return x * 2
|
||||
|
||||
tool = Tool.from_function(tool_with_timeout)
|
||||
assert requires_docket_execution(tool) is True
|
||||
|
||||
@pytest.mark.skipif(
|
||||
not _docket_available(),
|
||||
reason="Docket not installed",
|
||||
)
|
||||
def test_tool_with_retry_dep_returns_true(self):
|
||||
"""Tool with Docket Retry dependency returns True."""
|
||||
from docket import ExponentialRetry
|
||||
|
||||
from fastmcp.server.dependencies import requires_docket_execution
|
||||
from fastmcp.tools import Tool
|
||||
|
||||
async def tool_with_retry(
|
||||
x: int,
|
||||
retry: ExponentialRetry = ExponentialRetry(attempts=3),
|
||||
) -> int:
|
||||
return x * 2
|
||||
|
||||
tool = Tool.from_function(tool_with_retry)
|
||||
assert requires_docket_execution(tool) is True
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue