mirror of
https://github.com/PrefectHQ/fastmcp.git
synced 2026-08-21 04:54:17 +02:00
Rebuild guard tasks as end-and-reenter; remove imperative in-task elicit
A task tool that returns InputRequiredResult now ends its leg (freeing the worker) and stores the ask as durable state; tasks/update enqueues a fresh Docket execution (the next leg) with accumulated request_state/input_responses injected via ctx. No worker ever blocks on input, so a parked task no longer holds up shutdown. Imperative ctx.elicit() inside a task is removed and raises with guard-pattern guidance. Co-Authored-By: Claude <noreply@anthropic.com>
This commit is contained in:
Jeremiah Lowin
parent
ef29b731ea
commit
d41ff5bcd8
17 changed files with 1051 additions and 567 deletions
|
|
@ -3,7 +3,7 @@ from __future__ import annotations
|
|||
import logging
|
||||
import warnings
|
||||
import weakref
|
||||
from collections.abc import Awaitable, Callable, Generator, Mapping, Sequence
|
||||
from collections.abc import Callable, Generator, Mapping, Sequence
|
||||
from contextlib import contextmanager
|
||||
from contextvars import ContextVar, Token
|
||||
from dataclasses import dataclass
|
||||
|
|
@ -125,29 +125,17 @@ def _warn_sampling_deprecated() -> None:
|
|||
_current_context: ContextVar[Context | None] = ContextVar("context", default=None)
|
||||
|
||||
|
||||
#: Hook installed by the tasks extension (``fastmcp-tasks``) so ``ctx.elicit()``
|
||||
#: works inside a background-task worker, where there is no live request to
|
||||
#: carry the elicitation. Core ships no task engine; the extension registers a
|
||||
#: handler here at construction and ``Context._elicit_for_task`` delegates to it.
|
||||
#: ``None`` (the default) means no tasks extension is active, so in-task
|
||||
#: elicitation raises a clear install hint.
|
||||
_task_elicitation_handler: (
|
||||
Callable[[Context, str, dict[str, Any]], Awaitable[mcp_types.ElicitResult]] | None
|
||||
) = None
|
||||
|
||||
|
||||
def set_task_elicitation_handler(
|
||||
handler: Callable[[Context, str, dict[str, Any]], Awaitable[mcp_types.ElicitResult]]
|
||||
| None,
|
||||
) -> None:
|
||||
"""Install (or clear) the in-task elicitation handler.
|
||||
|
||||
Called by the tasks extension so a worker's ``ctx.elicit()`` parks an input
|
||||
request the client answers via ``tasks/update`` (SEP-2663 poll-based input).
|
||||
Passing ``None`` restores the default "requires the tasks extension" error.
|
||||
"""
|
||||
global _task_elicitation_handler
|
||||
_task_elicitation_handler = handler
|
||||
#: Error raised when a tool calls ``ctx.elicit()`` inside a background task.
|
||||
#: Background tasks gather input with the guard/return pattern (return an
|
||||
#: ``InputRequiredResult``), which the end-and-reenter machinery drives across
|
||||
#: worker legs. Imperative elicitation would require blocking a worker on a
|
||||
#: client round-trip, which end-and-reenter deliberately does not do.
|
||||
_TASK_ELICIT_ERROR = (
|
||||
"Imperative ctx.elicit() is not supported inside a background task. Gather "
|
||||
"input with the guard pattern instead: return an InputRequiredResult from "
|
||||
"the tool (with input_requests), and read ctx.input_responses / "
|
||||
"ctx.request_state when the task re-runs after the client answers."
|
||||
)
|
||||
|
||||
|
||||
TransportType = Literal["stdio", "sse", "streamable-http"]
|
||||
|
|
@ -273,6 +261,14 @@ class Context:
|
|||
self._origin_request_id: str | None = origin_request_id
|
||||
# Request-scoped state for non-serializable values (serializable=False)
|
||||
self._request_state: dict[str, Any] = {}
|
||||
# Multi-round-trip input carried in-task (SEP-2322 guard channel). A
|
||||
# foreground round recovers `input_responses`/`request_state` from the
|
||||
# wire request; a worker has no wire request, so the tasks extension's
|
||||
# in-task loop sets these between rounds and the properties below fall
|
||||
# back to them. The guard tool reads `ctx.input_responses` identically
|
||||
# in both modes — only the transport differs (task store vs wire params).
|
||||
self._task_input_responses: mcp_types.InputResponses | None = None
|
||||
self._task_request_state: str | None = None
|
||||
|
||||
@property
|
||||
def is_background_task(self) -> bool:
|
||||
|
|
@ -441,9 +437,14 @@ class Context:
|
|||
keys match the `input_requests` map the tool minted; each value is the
|
||||
client's result for that request (an `ElicitResult`, `CreateMessageResult`,
|
||||
or `ListRootsResult`).
|
||||
|
||||
In a background task there is no wire request, so this falls back to the
|
||||
responses the in-task guard loop delivered (see the tasks extension).
|
||||
"""
|
||||
params = self._input_response_params()
|
||||
return params.input_responses if params else None
|
||||
if params is not None and params.input_responses is not None:
|
||||
return params.input_responses
|
||||
return self._task_input_responses
|
||||
|
||||
@property
|
||||
def request_state(self) -> str | None:
|
||||
|
|
@ -455,9 +456,14 @@ class Context:
|
|||
before the tool runs, so tampering is rejected before this is read).
|
||||
`None` on the initial round. Use it to carry a small amount of computed
|
||||
state across rounds without re-deriving it.
|
||||
|
||||
In a background task there is no wire request, so this falls back to the
|
||||
state the in-task guard loop re-injected (see the tasks extension).
|
||||
"""
|
||||
params = self._input_response_params()
|
||||
return params.request_state if params else None
|
||||
if params is not None and params.request_state is not None:
|
||||
return params.request_state
|
||||
return self._task_request_state
|
||||
|
||||
@property
|
||||
def lifespan_context(self) -> dict[str, Any]:
|
||||
|
|
@ -1349,9 +1355,10 @@ class Context:
|
|||
``value`` field. Same scope rules as ``response_title``.
|
||||
|
||||
Note:
|
||||
This method works transparently in both request and background task
|
||||
contexts. In background task mode (SEP-1686), it will set the task
|
||||
status to "input_required" and wait for the client to provide input.
|
||||
Imperative elicitation is not available inside a background task
|
||||
(calling it there raises a ``ToolError``). A task gathers input with
|
||||
the guard pattern: return an ``InputRequiredResult`` and read
|
||||
``ctx.input_responses`` / ``ctx.request_state`` when the task re-runs.
|
||||
"""
|
||||
if response_type is None and fastmcp.settings.deprecation_warnings:
|
||||
warnings.warn(
|
||||
|
|
@ -1371,24 +1378,22 @@ class Context:
|
|||
)
|
||||
|
||||
if self.is_background_task:
|
||||
# Background task mode: use task-aware elicitation
|
||||
result = await self._elicit_for_task(
|
||||
message=message,
|
||||
schema=config.schema,
|
||||
)
|
||||
else:
|
||||
# Foreground push path: server-initiated elicitation needs a
|
||||
# back-channel, which the 2026-07-28 era removed (SEP-2577). Raise a
|
||||
# clear era-aware error before hitting the wire instead of the SDK's
|
||||
# opaque "Method not found". Handshake-era behavior is unchanged.
|
||||
if self._is_modern_protocol():
|
||||
raise ToolError(_ELICIT_MODERN_ERROR)
|
||||
# Standard request mode: use session.elicit directly
|
||||
result = await self.session.elicit(
|
||||
message=message,
|
||||
requested_schema=config.schema,
|
||||
related_request_id=self.request_id,
|
||||
)
|
||||
# Background tasks gather input with the guard/return pattern, not
|
||||
# imperative elicitation — the worker never blocks on a client
|
||||
# round-trip. Fail fast with the guidance to use InputRequiredResult.
|
||||
raise ToolError(_TASK_ELICIT_ERROR)
|
||||
# Foreground push path: server-initiated elicitation needs a back-channel,
|
||||
# which the 2026-07-28 era removed (SEP-2577). Raise a clear era-aware
|
||||
# error before hitting the wire instead of the SDK's opaque "Method not
|
||||
# found". Handshake-era behavior is unchanged.
|
||||
if self._is_modern_protocol():
|
||||
raise ToolError(_ELICIT_MODERN_ERROR)
|
||||
# Standard request mode: use session.elicit directly
|
||||
result = await self.session.elicit(
|
||||
message=message,
|
||||
requested_schema=config.schema,
|
||||
related_request_id=self.request_id,
|
||||
)
|
||||
|
||||
if result.action == "accept":
|
||||
return handle_elicit_accept(config, result.content)
|
||||
|
|
@ -1399,48 +1404,6 @@ class Context:
|
|||
else:
|
||||
raise ValueError(f"Unexpected elicitation action: {result.action}")
|
||||
|
||||
async def _elicit_for_task(
|
||||
self,
|
||||
message: str,
|
||||
schema: dict[str, Any],
|
||||
) -> mcp_types.ElicitResult:
|
||||
"""Send an elicitation request from a background task (SEP-1686).
|
||||
|
||||
This method handles elicitation when running in a Docket worker context,
|
||||
where there's no active MCP request. It:
|
||||
1. Sets the task status to "input_required"
|
||||
2. Sends the elicitation request with task metadata
|
||||
3. Waits for the client to provide input via tasks/sendInput
|
||||
4. Returns the result and resumes task execution
|
||||
|
||||
Args:
|
||||
message: The message to display to the user
|
||||
schema: The JSON schema for the expected response
|
||||
|
||||
Returns:
|
||||
ElicitResult with the user's response
|
||||
|
||||
Raises:
|
||||
RuntimeError: If not running in a background task context
|
||||
"""
|
||||
if not self.is_background_task:
|
||||
raise RuntimeError(
|
||||
"_elicit_for_task called but not in a background task context"
|
||||
)
|
||||
|
||||
# In-task elicitation is provided by the tasks extension (SEP-2663)
|
||||
# from the `fastmcp-tasks` package, which installs the handler below.
|
||||
# Core ships no task engine, so without the extension this raises a
|
||||
# clear install hint rather than reaching a wire the worker lacks.
|
||||
handler = _task_elicitation_handler
|
||||
if handler is None:
|
||||
raise RuntimeError(
|
||||
"In-task elicitation requires the tasks extension. Install "
|
||||
"'fastmcp[tasks]' and register the tasks extension via "
|
||||
"mcp.add_extension(...)."
|
||||
)
|
||||
return await handler(self, message, schema)
|
||||
|
||||
def _make_state_key(self, key: str) -> str:
|
||||
"""Create session-prefixed key for state storage."""
|
||||
return f"{self.session_id}:{key}"
|
||||
|
|
|
|||
|
|
@ -37,6 +37,7 @@ from fastmcp.tools.base import Tool
|
|||
from fastmcp.tools.function_tool import FunctionTool, _resolve_param_hints
|
||||
from fastmcp.utilities.components import FastMCPComponent
|
||||
from fastmcp.utilities.types import get_cached_typeadapter
|
||||
from fastmcp_tasks.input_loop import reentrant_task_fn
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from docket import Docket
|
||||
|
|
@ -54,7 +55,11 @@ def register_component_with_docket(component: FastMCPComponent, docket: Docket)
|
|||
return
|
||||
|
||||
if isinstance(component, FunctionTool):
|
||||
docket.register(component.fn, names=[component.key])
|
||||
# Run the tool through the guard loop so a body that returns an
|
||||
# InputRequiredResult drives the reentrant in-task input cycle. The
|
||||
# wrapper is signature-preserving, so Docket's dependency injection is
|
||||
# unchanged for a body that never asks for input.
|
||||
docket.register(reentrant_task_fn(component.fn), names=[component.key])
|
||||
elif isinstance(component, Tool):
|
||||
docket.register(component.run, names=[component.key])
|
||||
elif isinstance(component, FunctionResource):
|
||||
|
|
|
|||
|
|
@ -16,7 +16,11 @@ from contextvars import ContextVar
|
|||
from dataclasses import dataclass
|
||||
from typing import TYPE_CHECKING
|
||||
|
||||
from fastmcp_tasks.keys import parse_task_key, task_redis_prefix
|
||||
from fastmcp_tasks.keys import (
|
||||
leg_number_from_key,
|
||||
parse_task_key,
|
||||
task_redis_prefix,
|
||||
)
|
||||
|
||||
try:
|
||||
from docket import TaskKey
|
||||
|
|
@ -109,6 +113,26 @@ def get_task_context() -> TaskContextInfo | None:
|
|||
return None
|
||||
|
||||
|
||||
def get_task_leg_number() -> int:
|
||||
"""Return the current leg number of the running task (1 outside a re-entry).
|
||||
|
||||
Each re-entry after client input runs as a fresh Docket execution under a
|
||||
per-leg key; the capture wrapper reads this to scope a leg's outstanding
|
||||
input requests so successive legs never collide in Redis.
|
||||
"""
|
||||
from fastmcp_tasks.dependencies import is_docket_available
|
||||
|
||||
if not is_docket_available():
|
||||
return 1
|
||||
|
||||
from docket.dependencies import current_execution
|
||||
|
||||
try:
|
||||
return leg_number_from_key(current_execution.get().key)
|
||||
except LookupError:
|
||||
return 1
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class TaskContextSnapshot:
|
||||
"""All context data snapshotted at task-submission time.
|
||||
|
|
@ -388,6 +412,11 @@ async def make_task_context() -> Context | None:
|
|||
id; the server prefers the one registered at submission time so mounted
|
||||
tasks resolve to the child server. No live session is attached — SEP-2663
|
||||
input and status are polled, so the worker needs no back-channel.
|
||||
|
||||
For a re-entered leg (after the client answered a guard ask), the accumulated
|
||||
per-leg state is loaded and injected so the tool reads ``ctx.input_responses``
|
||||
/ ``ctx.request_state`` identically to the foreground guard contract. Leg 1
|
||||
loads nothing (both ``None``).
|
||||
"""
|
||||
from fastmcp.server.context import Context
|
||||
from fastmcp.server.dependencies import get_server
|
||||
|
|
@ -407,4 +436,19 @@ async def make_task_context() -> Context | None:
|
|||
origin_request_id=origin_request_id,
|
||||
)
|
||||
await ctx.__aenter__()
|
||||
|
||||
docket = server._docket
|
||||
if docket is None:
|
||||
from fastmcp_tasks.dependencies import _current_docket
|
||||
|
||||
docket = _current_docket.get()
|
||||
if docket is not None:
|
||||
from fastmcp_tasks.input_store import load_pending_input
|
||||
|
||||
request_state, input_responses = await load_pending_input(
|
||||
docket, task_info.task_scope, task_info.task_id
|
||||
)
|
||||
ctx._task_request_state = request_state
|
||||
ctx._task_input_responses = input_responses
|
||||
|
||||
return ctx
|
||||
|
|
|
|||
|
|
@ -31,6 +31,7 @@ from fastmcp_tasks.context import (
|
|||
register_task_server,
|
||||
)
|
||||
from fastmcp_tasks.dependencies import _current_docket
|
||||
from fastmcp_tasks.input_store import save_current_leg, save_task_args
|
||||
from fastmcp_tasks.keys import build_task_key, task_redis_prefix
|
||||
from fastmcp_tasks.models import CreateTaskResult
|
||||
|
||||
|
|
@ -74,8 +75,9 @@ async def create_task(
|
|||
# argument-splatting match what the worker will invoke.
|
||||
component = await _registered_task_component(context, tool)
|
||||
|
||||
raw_arguments = dict(arguments or {})
|
||||
coerced = coerce_task_arguments(
|
||||
component, dict(arguments or {}), strict=_strict_input_validation()
|
||||
component, raw_arguments, strict=_strict_input_validation()
|
||||
)
|
||||
|
||||
task_id = secrets.token_urlsafe(32)
|
||||
|
|
@ -114,6 +116,13 @@ async def create_task(
|
|||
await redis.set(created_at_key, created_at, ex=ttl_seconds)
|
||||
await redis.set(poll_interval_key, str(poll_interval_ms), ex=ttl_seconds)
|
||||
|
||||
# End-and-reenter state: the raw (wire) arguments feed every leg, re-coerced
|
||||
# per leg, and the leg pointer starts at leg 1 (the base task key). A guard
|
||||
# return re-enters by enqueuing the next leg with these same arguments (see
|
||||
# handlers.enqueue_next_leg).
|
||||
await save_task_args(docket, task_scope, task_id, raw_arguments, ttl_seconds)
|
||||
await save_current_leg(docket, task_scope, task_id, task_key, 1, ttl_seconds)
|
||||
|
||||
await snapshot.save(docket, task_scope, task_id, ttl_seconds)
|
||||
|
||||
await add_component_to_docket(
|
||||
|
|
@ -150,6 +159,47 @@ def _owning_server(tool: Tool, fallback: FastMCP) -> FastMCP:
|
|||
return fallback
|
||||
|
||||
|
||||
async def registered_component_for_key(server: FastMCP, component_key: str) -> Tool:
|
||||
"""Return the Docket-registered task component matching ``component_key``.
|
||||
|
||||
``get_tasks()`` yields the same components registered with Docket (the
|
||||
underlying ``FunctionTool`` for a mounted tool, not a provider wrapper), so
|
||||
matching by ``key`` recovers the component whose calling convention agrees
|
||||
with the worker. Used when re-entering a task leg, where only the stored
|
||||
compound key (not the original ``Tool`` object) is available.
|
||||
"""
|
||||
for component in await server.get_tasks():
|
||||
if component.key == component_key and isinstance(component, Tool):
|
||||
return component
|
||||
raise MCPError(
|
||||
code=INTERNAL_ERROR,
|
||||
message=f"No task-enabled component found for {component_key!r}.",
|
||||
)
|
||||
|
||||
|
||||
async def enqueue_task_leg(
|
||||
server: FastMCP,
|
||||
docket: Docket,
|
||||
component: Tool,
|
||||
raw_arguments: dict[str, object],
|
||||
leg_key: str,
|
||||
) -> None:
|
||||
"""Enqueue a fresh Docket execution (the next leg) for a re-entered task.
|
||||
|
||||
Re-coerces the stored wire arguments (each leg validates independently, as a
|
||||
foreground retry would) and adds the component's registered callable — the
|
||||
capture wrapper — under ``leg_key``. Waits for the execution to become
|
||||
durable so a ``tasks/get`` immediately after ``tasks/update`` resolves.
|
||||
"""
|
||||
coerced = coerce_task_arguments(
|
||||
component, dict(raw_arguments), strict=_strict_input_validation()
|
||||
)
|
||||
await add_component_to_docket(
|
||||
component, docket, coerced, fn_key=component.key, task_key=leg_key
|
||||
)
|
||||
await _await_durable_creation(docket, leg_key)
|
||||
|
||||
|
||||
async def _registered_task_component(context: Context, tool: Tool) -> Tool:
|
||||
"""Return the component Docket registered for ``tool``'s key.
|
||||
|
||||
|
|
|
|||
|
|
@ -203,10 +203,10 @@ class TasksExtension(ServerExtension):
|
|||
async def lifespan(self) -> AsyncIterator[None]:
|
||||
"""Start the Docket backend/worker and install the worker-side hooks.
|
||||
|
||||
Installs core's background-context factory and in-task elicitation
|
||||
handler for the duration so a worker's ``ctx`` (progress, elicitation)
|
||||
functions, then runs the Docket lifespan. The hooks are process-global
|
||||
and refcounted: with several servers in one process (each its own
|
||||
Installs core's background-context factory and worker-server resolver for
|
||||
the duration so a worker's ``ctx`` (progress, server resolution) works,
|
||||
then runs the Docket lifespan. The hooks are process-global and
|
||||
refcounted: with several servers in one process (each its own
|
||||
runtime-tree root), the hooks stay installed until the last tasks
|
||||
extension shuts down, so one server's exit cannot strand another
|
||||
server's in-flight workers.
|
||||
|
|
@ -231,20 +231,17 @@ _active_worker_hook_holds: int = 0
|
|||
|
||||
|
||||
def _install_worker_hooks() -> None:
|
||||
from fastmcp.server.context import set_task_elicitation_handler
|
||||
from fastmcp.server.dependencies import (
|
||||
set_background_context_factory,
|
||||
set_worker_server_resolver,
|
||||
)
|
||||
from fastmcp_tasks import wire_production
|
||||
from fastmcp_tasks.context import make_task_context, resolve_worker_server
|
||||
from fastmcp_tasks.input_store import elicit_in_task
|
||||
|
||||
global _active_worker_hook_holds
|
||||
_active_worker_hook_holds += 1
|
||||
set_background_context_factory(make_task_context)
|
||||
set_worker_server_resolver(resolve_worker_server)
|
||||
set_task_elicitation_handler(elicit_in_task)
|
||||
# Enable server-side production of the claimed CreateTaskResult on tools/call
|
||||
# (the SDK ships only claim consumption). Refcounted independently but
|
||||
# installed/released in lockstep with the worker hooks.
|
||||
|
|
@ -252,7 +249,6 @@ def _install_worker_hooks() -> None:
|
|||
|
||||
|
||||
def _release_worker_hooks() -> None:
|
||||
from fastmcp.server.context import set_task_elicitation_handler
|
||||
from fastmcp.server.dependencies import (
|
||||
set_background_context_factory,
|
||||
set_worker_server_resolver,
|
||||
|
|
@ -263,7 +259,6 @@ def _release_worker_hooks() -> None:
|
|||
_active_worker_hook_holds -= 1
|
||||
if _active_worker_hook_holds <= 0:
|
||||
_active_worker_hook_holds = 0
|
||||
set_task_elicitation_handler(None)
|
||||
set_worker_server_resolver(None)
|
||||
set_background_context_factory(None)
|
||||
wire_production.uninstall()
|
||||
|
|
|
|||
|
|
@ -33,8 +33,21 @@ from fastmcp.tools.base import InputRequiredToolResult, Tool
|
|||
from fastmcp.utilities.tasks import DEFAULT_POLL_INTERVAL_MS
|
||||
from fastmcp.utilities.versions import VersionSpec
|
||||
from fastmcp_tasks.context import get_task_scope
|
||||
from fastmcp_tasks.input_store import deliver_input_responses, read_outstanding_inputs
|
||||
from fastmcp_tasks.keys import parse_task_key, task_redis_prefix
|
||||
from fastmcp_tasks.creation import enqueue_task_leg, registered_component_for_key
|
||||
from fastmcp_tasks.input_store import (
|
||||
clear_outstanding,
|
||||
load_current_leg,
|
||||
load_task_args,
|
||||
read_outstanding_inputs,
|
||||
save_current_leg,
|
||||
store_input_responses,
|
||||
translate_responses,
|
||||
)
|
||||
from fastmcp_tasks.keys import (
|
||||
leg_execution_key,
|
||||
parse_task_key,
|
||||
task_redis_prefix,
|
||||
)
|
||||
from fastmcp_tasks.models import (
|
||||
CancelTaskResult,
|
||||
GetTaskResult,
|
||||
|
|
@ -57,10 +70,6 @@ DOCKET_TO_MCP_STATE: dict[ExecutionState, str] = {
|
|||
ExecutionState.CANCELLED: "cancelled",
|
||||
}
|
||||
|
||||
_WORKING_STATES = frozenset(
|
||||
{ExecutionState.SCHEDULED, ExecutionState.QUEUED, ExecutionState.RUNNING}
|
||||
)
|
||||
|
||||
|
||||
def _task_not_found(task_id: str) -> MCPError:
|
||||
"""The single "not found" error for missing, expired, or cross-scope ids.
|
||||
|
|
@ -96,12 +105,14 @@ def _ttl_ms(docket: Docket) -> int:
|
|||
|
||||
async def _lookup_task(
|
||||
docket: Docket, task_scope: str | None, task_id: str
|
||||
) -> tuple[Any, str, str | None, int]:
|
||||
"""Resolve a task's execution and stored metadata within the caller's scope.
|
||||
) -> tuple[Any, str, int, str | None, int]:
|
||||
"""Resolve a task's current-leg execution and metadata within the scope.
|
||||
|
||||
Returns ``(execution, task_key, created_at, poll_interval_ms)``. Raises the
|
||||
shared "not found" error when the scope-prefixed metadata is absent or the
|
||||
execution has expired.
|
||||
Returns ``(execution, base_task_key, leg_number, created_at,
|
||||
poll_interval_ms)``. The execution is the *current leg* (the latest Docket
|
||||
execution), which for a re-entered task differs from the base task key.
|
||||
Raises the shared "not found" error when the scope-prefixed metadata is
|
||||
absent or the current leg's execution has expired.
|
||||
"""
|
||||
prefix = task_redis_prefix(task_scope)
|
||||
meta_key = docket.key(f"{prefix}:{task_id}")
|
||||
|
|
@ -115,11 +126,13 @@ async def _lookup_task(
|
|||
values = await redis.mget(meta_key, created_at_key, poll_key) # ty: ignore[too-many-positional-arguments]
|
||||
task_key_bytes, created_at_bytes, poll_bytes = values
|
||||
|
||||
task_key = task_key_bytes.decode("utf-8") if task_key_bytes else None
|
||||
if not task_key:
|
||||
base_task_key = task_key_bytes.decode("utf-8") if task_key_bytes else None
|
||||
if not base_task_key:
|
||||
raise _task_not_found(task_id)
|
||||
|
||||
execution = await docket.get_execution(task_key)
|
||||
current_leg_key, leg_number = await load_current_leg(docket, task_scope, task_id)
|
||||
execution_key = current_leg_key or base_task_key
|
||||
execution = await docket.get_execution(execution_key)
|
||||
if not execution:
|
||||
raise _task_not_found(task_id)
|
||||
|
||||
|
|
@ -132,7 +145,7 @@ async def _lookup_task(
|
|||
except (ValueError, UnicodeDecodeError):
|
||||
poll_interval_ms = DEFAULT_POLL_INTERVAL_MS
|
||||
|
||||
return execution, task_key, created_at, poll_interval_ms
|
||||
return execution, base_task_key, leg_number, created_at, poll_interval_ms
|
||||
|
||||
|
||||
async def _resolve_tool(server: FastMCP, task_key: str) -> Tool:
|
||||
|
|
@ -160,17 +173,21 @@ async def _resolve_tool(server: FastMCP, task_key: str) -> Tool:
|
|||
def _inline_result(tool: Tool, raw_value: Any) -> dict[str, Any]:
|
||||
"""Convert a completed task's raw return into an inlined CallToolResult dict.
|
||||
|
||||
A guard tool that returned an ``InputRequiredResult`` from inside a task is
|
||||
rejected: multi-round-trip guards need a live request to answer the prompt
|
||||
and cannot complete as a task.
|
||||
A completed task should never carry an ``InputRequiredResult``: a function
|
||||
tool's guard returns are captured by the end-and-reenter wrapper (see
|
||||
``input_loop.py``), which records the leg's outstanding requests and ends the
|
||||
leg (returning ``None``), so ``tasks/get`` reports ``input_required`` rather
|
||||
than inlining. Reaching here with a guard result means a component type the
|
||||
wrapper does not wrap (e.g. a base ``Tool``) returned one, which the task
|
||||
path cannot drive — a safety net, not an expected path.
|
||||
"""
|
||||
if isinstance(raw_value, mcp_types.InputRequiredResult | InputRequiredToolResult):
|
||||
raise MCPError(
|
||||
code=mcp_types.INTERNAL_ERROR,
|
||||
message=(
|
||||
f"Tool {tool.name!r} requested input while running as a background "
|
||||
"task. Input-required (multi-round-trip) tools need a live request "
|
||||
"to answer the prompt and cannot run as tasks."
|
||||
f"Tool {tool.name!r} returned an input-required result as a task, "
|
||||
"but its component type is not driven by the in-task guard loop. "
|
||||
"Guard-pattern tasks are supported for function tools."
|
||||
),
|
||||
)
|
||||
mcp_result = tool.convert_result(raw_value).to_mcp_result()
|
||||
|
|
@ -193,9 +210,13 @@ async def tasks_get(server: FastMCP, task_id: str) -> GetTaskResult:
|
|||
raise _task_not_found(task_id)
|
||||
|
||||
task_scope = get_task_scope()
|
||||
execution, task_key, created_at, poll_interval_ms = await _lookup_task(
|
||||
docket, task_scope, task_id
|
||||
)
|
||||
(
|
||||
execution,
|
||||
base_task_key,
|
||||
leg_number,
|
||||
created_at,
|
||||
poll_interval_ms,
|
||||
) = await _lookup_task(docket, task_scope, task_id)
|
||||
await execution.sync()
|
||||
|
||||
created_at_iso = _normalize_iso_timestamp(created_at)
|
||||
|
|
@ -218,16 +239,17 @@ async def tasks_get(server: FastMCP, task_id: str) -> GetTaskResult:
|
|||
**payload,
|
||||
)
|
||||
|
||||
# An outstanding input request outranks the Docket "running" state: the task
|
||||
# is parked in the worker waiting for tasks/update, so it is input_required.
|
||||
if execution.state in _WORKING_STATES:
|
||||
outstanding = await read_outstanding_inputs(docket, task_scope, task_id)
|
||||
if execution.state == ExecutionState.COMPLETED:
|
||||
# A guard leg ends its Docket execution and records outstanding input
|
||||
# requests to Redis: a completed leg with outstanding requests is the
|
||||
# task waiting for tasks/update (input_required), not a finished task.
|
||||
outstanding = await read_outstanding_inputs(
|
||||
docket, task_scope, task_id, leg_number
|
||||
)
|
||||
if outstanding:
|
||||
return build("input_required", input_requests=outstanding)
|
||||
|
||||
if execution.state == ExecutionState.COMPLETED:
|
||||
raw_value = await execution.get_result(timeout=timedelta(seconds=0))
|
||||
tool = await _resolve_tool(server, task_key)
|
||||
tool = await _resolve_tool(server, base_task_key)
|
||||
return build("completed", result=_inline_result(tool, raw_value))
|
||||
|
||||
if execution.state == ExecutionState.FAILED:
|
||||
|
|
@ -257,26 +279,66 @@ async def tasks_get(server: FastMCP, task_id: str) -> GetTaskResult:
|
|||
async def tasks_update(
|
||||
server: FastMCP, task_id: str, input_responses: dict[str, Any]
|
||||
) -> UpdateTaskResult:
|
||||
"""Handle ``tasks/update``: deliver input responses to the parked worker."""
|
||||
"""Handle ``tasks/update``: answer a guard leg and re-enter the task.
|
||||
|
||||
The responses are keyed by the surfaced keys ``tasks/get`` reported. Unknown
|
||||
or already-satisfied keys are ignored (SEP-2663). When at least one answer
|
||||
matches the current leg's outstanding requests, they are translated to the
|
||||
tool's own keys, stored for the next leg, and a fresh Docket execution (the
|
||||
next leg) is enqueued with the task's arguments. The worker is never blocked;
|
||||
re-entry is the whole mechanism. A stale or empty update is an idempotent
|
||||
no-op.
|
||||
"""
|
||||
docket = server._docket
|
||||
if docket is None:
|
||||
raise _task_not_found(task_id)
|
||||
|
||||
task_scope = get_task_scope()
|
||||
# Resolve within scope so a cross-scope update is a "not found", not a no-op.
|
||||
await _lookup_task(docket, task_scope, task_id)
|
||||
await deliver_input_responses(docket, task_scope, task_id, input_responses)
|
||||
_execution, base_task_key, leg_number, _created_at, _poll = await _lookup_task(
|
||||
docket, task_scope, task_id
|
||||
)
|
||||
|
||||
translated = await translate_responses(
|
||||
docket, task_scope, task_id, leg_number, input_responses
|
||||
)
|
||||
if translated is None:
|
||||
# Nothing matched the current leg's outstanding requests: the leg was
|
||||
# already answered, or the keys are unknown. Idempotent no-op.
|
||||
return UpdateTaskResult()
|
||||
|
||||
# Store the answers for the next leg to read, then enqueue that leg. Ordering
|
||||
# matters: the answers must be in Redis before the next leg's worker context
|
||||
# loads them, and current_leg must not advance to an execution that is not
|
||||
# yet durable — so enqueue (with its durable wait) precedes the pointer swap.
|
||||
await store_input_responses(docket, task_scope, task_id, translated)
|
||||
|
||||
component = await registered_component_for_key(
|
||||
server, parse_task_key(base_task_key)["component_identifier"]
|
||||
)
|
||||
raw_arguments = await load_task_args(docket, task_scope, task_id)
|
||||
next_leg = leg_number + 1
|
||||
next_leg_key = leg_execution_key(base_task_key, next_leg)
|
||||
|
||||
await enqueue_task_leg(server, docket, component, raw_arguments, next_leg_key)
|
||||
ttl_seconds = int(docket.execution_ttl.total_seconds())
|
||||
await save_current_leg(
|
||||
docket, task_scope, task_id, next_leg_key, next_leg, ttl_seconds
|
||||
)
|
||||
# The answered leg's surfaced keys are now superseded; drop them so they are
|
||||
# never reused (SEP-2663 L350).
|
||||
await clear_outstanding(docket, task_scope, task_id, leg_number)
|
||||
return UpdateTaskResult()
|
||||
|
||||
|
||||
async def tasks_cancel(server: FastMCP, task_id: str) -> CancelTaskResult:
|
||||
"""Handle ``tasks/cancel``: cooperatively cancel the task, empty ack."""
|
||||
"""Handle ``tasks/cancel``: cooperatively cancel the current leg, empty ack."""
|
||||
docket = server._docket
|
||||
if docket is None:
|
||||
raise _task_not_found(task_id)
|
||||
|
||||
task_scope = get_task_scope()
|
||||
execution, _task_key, _created_at, _poll = await _lookup_task(
|
||||
execution, _base_task_key, _leg, _created_at, _poll = await _lookup_task(
|
||||
docket, task_scope, task_id
|
||||
)
|
||||
await docket.cancel(execution.key)
|
||||
|
|
|
|||
137
fastmcp_tasks/fastmcp_tasks/input_loop.py
Normal file
137
fastmcp_tasks/fastmcp_tasks/input_loop.py
Normal file
|
|
@ -0,0 +1,137 @@
|
|||
"""The end-and-reenter capture wrapper for guard-pattern task tools.
|
||||
|
||||
A guard tool asks for input by *returning* an `InputRequiredResult` rather than
|
||||
awaiting `ctx.elicit()`. Foreground, each such return is one leg of a
|
||||
multi-round-trip: the tool returns, the client answers, the framework re-invokes
|
||||
the tool with the answers on `ctx.input_responses`. The tool body is written
|
||||
once and is oblivious to how many legs it takes.
|
||||
|
||||
As a background task the leg boundary is a *worker* boundary. This wrapper runs
|
||||
the tool body exactly once. If the body returns a real value, it is the leg's
|
||||
result. If the body returns an `InputRequiredResult`, the wrapper records the
|
||||
leg's outstanding requests (and any carried `request_state`) to Redis and
|
||||
returns — the Docket execution then completes and the worker is freed. The task
|
||||
sits in `input_required` as durable state until the client answers via
|
||||
`tasks/update`, which enqueues a fresh Docket execution (the next leg) that
|
||||
re-runs this wrapper with the accumulated state injected onto `ctx`. No worker
|
||||
is ever blocked awaiting input.
|
||||
|
||||
The wrapper preserves the wrapped callable's signature so Docket's dependency
|
||||
injection still resolves the tool's parameters (its own args, `ctx`, and any
|
||||
Docket-native dependencies) exactly as it would for the raw callable. The
|
||||
per-leg state (`ctx.input_responses` / `ctx.request_state`) is injected by the
|
||||
worker `Context` factory (`make_task_context`) before the body runs.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import functools
|
||||
import inspect
|
||||
import logging
|
||||
from typing import TYPE_CHECKING, Any
|
||||
|
||||
import mcp_types
|
||||
|
||||
from fastmcp.tools.base import InputRequiredToolResult
|
||||
from fastmcp_tasks.context import get_task_context, get_task_leg_number
|
||||
from fastmcp_tasks.input_store import store_outstanding
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from collections.abc import Awaitable, Callable
|
||||
|
||||
from docket import Docket
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
def _as_input_required(result: Any) -> mcp_types.InputRequiredResult | None:
|
||||
"""Return the `InputRequiredResult` a guard leg produced, or None.
|
||||
|
||||
A tool body may return the bare `InputRequiredResult` or the
|
||||
`InputRequiredToolResult` wrapper FastMCP uses foreground; both mean the same
|
||||
ask.
|
||||
"""
|
||||
if isinstance(result, InputRequiredToolResult):
|
||||
return result.input_required
|
||||
if isinstance(result, mcp_types.InputRequiredResult):
|
||||
return result
|
||||
return None
|
||||
|
||||
|
||||
def _serialize_requests(
|
||||
input_requests: mcp_types.InputRequests,
|
||||
) -> dict[str, dict[str, Any]]:
|
||||
"""Dump each request to the wire payload surfaced for the client to answer."""
|
||||
return {
|
||||
key: request.model_dump(by_alias=True, mode="json", exclude_none=True)
|
||||
for key, request in input_requests.items()
|
||||
}
|
||||
|
||||
|
||||
def _resolve_docket() -> Docket | None:
|
||||
"""Resolve the active Docket from the current context or worker default."""
|
||||
from fastmcp.server.dependencies import get_context
|
||||
from fastmcp_tasks.dependencies import _current_docket
|
||||
|
||||
try:
|
||||
docket = get_context().fastmcp._docket
|
||||
except RuntimeError:
|
||||
docket = None
|
||||
if docket is None:
|
||||
docket = _current_docket.get()
|
||||
return docket
|
||||
|
||||
|
||||
def reentrant_task_fn(
|
||||
fn: Callable[..., Awaitable[Any]],
|
||||
) -> Callable[..., Awaitable[Any]]:
|
||||
"""Wrap a task tool's callable to capture a guard leg's ask (end-and-reenter).
|
||||
|
||||
Signature-preserving, so Docket injects the wrapped callable's parameters
|
||||
unchanged. The body runs exactly once: a real return is the leg's result; an
|
||||
`InputRequiredResult` is captured to Redis (outstanding requests + carried
|
||||
state) and the wrapper returns, ending the leg without blocking. The next
|
||||
leg is enqueued by ``tasks/update`` when the client answers.
|
||||
"""
|
||||
|
||||
@functools.wraps(fn)
|
||||
async def wrapper(*args: Any, **kwargs: Any) -> Any:
|
||||
result = await fn(*args, **kwargs)
|
||||
input_required = _as_input_required(result)
|
||||
if input_required is None:
|
||||
return result
|
||||
|
||||
requests = input_required.input_requests or {}
|
||||
request_state = input_required.request_state
|
||||
if not requests and request_state is None:
|
||||
# A leg that asks nothing and carries nothing can never be answered;
|
||||
# treat it as the terminal result rather than an unanswerable park.
|
||||
return result
|
||||
|
||||
task_context = get_task_context()
|
||||
docket = _resolve_docket()
|
||||
if task_context is None or docket is None:
|
||||
logger.warning(
|
||||
"guard leg produced an ask outside a task worker; returning it"
|
||||
)
|
||||
return result
|
||||
|
||||
await store_outstanding(
|
||||
docket,
|
||||
task_context.task_scope,
|
||||
task_context.task_id,
|
||||
get_task_leg_number(),
|
||||
_serialize_requests(requests),
|
||||
request_state,
|
||||
)
|
||||
# The leg ends here: the Docket execution completes and the worker is
|
||||
# freed. The task is now input_required until tasks/update enqueues the
|
||||
# next leg. Return None so the completed leg carries no stray result.
|
||||
return None
|
||||
|
||||
# `functools.wraps` copies `__wrapped__`, so `inspect.signature` already
|
||||
# unwraps to `fn`; set it explicitly too, so a dependency injector reading
|
||||
# `__signature__` directly (rather than following `__wrapped__`) still sees
|
||||
# the tool's real parameters.
|
||||
wrapper.__signature__ = inspect.signature(fn) # ty: ignore[unresolved-attribute]
|
||||
return wrapper
|
||||
|
|
@ -1,142 +1,260 @@
|
|||
"""In-task input store for SEP-2663 poll-based elicitation.
|
||||
"""Per-task Redis state for SEP-2663 end-and-reenter input gathering.
|
||||
|
||||
When a background task calls ``ctx.elicit()`` it has no live request to carry the
|
||||
prompt. SEP-2663 handles this by polling: the worker parks an *input request*
|
||||
here, the task's ``tasks/get`` status flips to ``input_required`` with the
|
||||
outstanding requests, the client answers via ``tasks/update``, and the parked
|
||||
worker resumes.
|
||||
A background task gathers client input by *ending a leg* and re-entering, never
|
||||
by blocking a worker. When a `task=True` tool returns an `InputRequiredResult`,
|
||||
the leg's Docket execution completes and the worker is freed; the task's state
|
||||
lives here in Redis as `input_required`. When the client answers via
|
||||
`tasks/update`, a fresh Docket execution (the next leg) re-runs the tool with the
|
||||
accumulated state injected onto its `Context`. No worker ever waits for input.
|
||||
|
||||
This is the reworked SEP-1686 elicitation module. The Redis request/response
|
||||
mechanics — a per-request hash the poll surface reads and a per-key list the
|
||||
worker blocks on with ``BLPOP`` — are preserved. What's gone is the *push
|
||||
envelope*: the old code sent a ``notifications/tasks/status`` through the
|
||||
distributed notification queue to wake the client. Under SEP-2663 the client
|
||||
discovers the outstanding request by polling ``tasks/get``, so no push is needed.
|
||||
This module owns the durable state each task carries between legs:
|
||||
|
||||
- **args** — the original tool arguments, re-supplied to every leg.
|
||||
- **current_leg / leg** — the latest leg's Docket execution key and its number.
|
||||
- **request_state** — the opaque string a leg carried forward (SEP-2322).
|
||||
- **input_responses** — the typed answers the last `tasks/update` delivered,
|
||||
translated to the tool's own request keys.
|
||||
- **input:requests / input:map** — the current leg's outstanding requests, keyed
|
||||
by a server-minted surfaced key, plus the surfaced-key → tool-key mapping.
|
||||
|
||||
Each surfaced request key is minted fresh with high-entropy suffix and never
|
||||
reused after its response is delivered (SEP-2663 L350): a task that asks twice,
|
||||
or a leg that requests several inputs at once, surfaces distinct, independently
|
||||
answerable keys, and the tool reads its *own* keys on the next leg via the
|
||||
translated `input_responses`.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import logging
|
||||
from typing import TYPE_CHECKING, Any
|
||||
import secrets
|
||||
from typing import TYPE_CHECKING, Any, cast
|
||||
|
||||
import mcp_types
|
||||
from redis.exceptions import RedisError
|
||||
|
||||
from fastmcp_tasks.context import get_task_context
|
||||
from fastmcp_tasks.keys import task_redis_prefix
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from docket import Docket
|
||||
|
||||
from fastmcp.server.context import Context
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
# How long a parked input request (and any delivered response) lives before
|
||||
# expiring. A task blocked on input holds a worker slot, so this doubles as the
|
||||
# maximum time a worker waits for the client to answer.
|
||||
# How long a task's input state (outstanding requests and delivered responses)
|
||||
# lives before expiring. With end-and-reenter no worker is held while a task is
|
||||
# input_required, so this bounds only how long durable input state survives, not
|
||||
# any worker slot.
|
||||
INPUT_TTL_SECONDS = 3600
|
||||
|
||||
# Reconstruct a typed response from its stored `{"type": name, "data": dump}`
|
||||
# form so a re-entered leg reads a real `ElicitResult` (etc.) on
|
||||
# `ctx.input_responses`, matching the foreground guard contract.
|
||||
_RESULT_TYPE_BY_NAME: dict[str, type[mcp_types.Result]] = {
|
||||
"ElicitResult": mcp_types.ElicitResult,
|
||||
"CreateMessageResult": mcp_types.CreateMessageResult,
|
||||
"CreateMessageResultWithTools": mcp_types.CreateMessageResultWithTools,
|
||||
"ListRootsResult": mcp_types.ListRootsResult,
|
||||
}
|
||||
|
||||
def _requests_key(docket: Docket, task_scope: str | None, task_id: str) -> str:
|
||||
"""Redis hash of outstanding input requests, keyed by input key."""
|
||||
return docket.key(f"{task_redis_prefix(task_scope)}:{task_id}:input:requests")
|
||||
# Map an outstanding request's wire method to the result type its answer
|
||||
# validates into. Elicitation is the supported in-task input; the others are
|
||||
# kept complete so a client that answers one is parsed rather than dropped.
|
||||
_RESULT_TYPE_BY_METHOD: dict[str, type[mcp_types.Result]] = {
|
||||
"elicitation/create": mcp_types.ElicitResult,
|
||||
"sampling/createMessage": mcp_types.CreateMessageResult,
|
||||
"roots/list": mcp_types.ListRootsResult,
|
||||
}
|
||||
|
||||
|
||||
def _response_key(
|
||||
docket: Docket, task_scope: str | None, task_id: str, input_key: str
|
||||
def result_type_for_method(method: str) -> type[mcp_types.Result]:
|
||||
"""The result type an outstanding request's answer validates into."""
|
||||
return _RESULT_TYPE_BY_METHOD.get(method, mcp_types.ElicitResult)
|
||||
|
||||
|
||||
def _prefix(docket: Docket, task_scope: str | None, task_id: str) -> str:
|
||||
return f"{task_redis_prefix(task_scope)}:{task_id}"
|
||||
|
||||
|
||||
def _args_key(docket: Docket, task_scope: str | None, task_id: str) -> str:
|
||||
return docket.key(f"{_prefix(docket, task_scope, task_id)}:args")
|
||||
|
||||
|
||||
def _current_leg_key(docket: Docket, task_scope: str | None, task_id: str) -> str:
|
||||
return docket.key(f"{_prefix(docket, task_scope, task_id)}:current_leg")
|
||||
|
||||
|
||||
def _leg_number_key(docket: Docket, task_scope: str | None, task_id: str) -> str:
|
||||
return docket.key(f"{_prefix(docket, task_scope, task_id)}:leg")
|
||||
|
||||
|
||||
def _request_state_key(docket: Docket, task_scope: str | None, task_id: str) -> str:
|
||||
return docket.key(f"{_prefix(docket, task_scope, task_id)}:request_state")
|
||||
|
||||
|
||||
def _input_responses_key(docket: Docket, task_scope: str | None, task_id: str) -> str:
|
||||
return docket.key(f"{_prefix(docket, task_scope, task_id)}:input_responses")
|
||||
|
||||
|
||||
def _requests_key(
|
||||
docket: Docket, task_scope: str | None, task_id: str, leg: int
|
||||
) -> str:
|
||||
"""Redis list the worker blocks on for a single input key's response."""
|
||||
return docket.key(
|
||||
f"{task_redis_prefix(task_scope)}:{task_id}:input:resp:{input_key}"
|
||||
)
|
||||
"""Redis hash of a leg's outstanding input requests, keyed by surfaced key.
|
||||
|
||||
|
||||
def _elicitation_input_request(message: str, schema: dict[str, Any]) -> dict[str, Any]:
|
||||
"""Build the SEP-2663 ``InputRequest`` for an elicitation (an ElicitRequest)."""
|
||||
return {
|
||||
"method": "elicitation/create",
|
||||
"params": {"message": message, "requestedSchema": schema},
|
||||
}
|
||||
|
||||
|
||||
async def elicit_in_task(
|
||||
context: Context, message: str, schema: dict[str, Any]
|
||||
) -> mcp_types.ElicitResult:
|
||||
"""Park an elicitation request and block until the client answers it.
|
||||
|
||||
Installed as core's in-task elicitation handler by ``TasksExtension``. Parks
|
||||
an input request keyed by the task's own id (one outstanding elicitation per
|
||||
task at a time — the polling model is inherently sequential), flips the
|
||||
task's polled status to ``input_required``, and blocks on the response list.
|
||||
Returns the client's ``ElicitResult``; on timeout or a missing task context,
|
||||
returns a ``cancel`` action so the worker never hangs indefinitely.
|
||||
Scoped by leg number so a re-entered leg's fresh requests never collide with
|
||||
the answered leg's stale ones in the shared keyspace.
|
||||
"""
|
||||
task_context = get_task_context()
|
||||
if task_context is None:
|
||||
logger.warning("elicit_in_task called outside a task worker; cancelling")
|
||||
return mcp_types.ElicitResult(action="cancel", content=None)
|
||||
return docket.key(f"{_prefix(docket, task_scope, task_id)}:input:{leg}:requests")
|
||||
|
||||
docket = context.fastmcp._docket
|
||||
if docket is None:
|
||||
from fastmcp_tasks.dependencies import _current_docket
|
||||
|
||||
docket = _current_docket.get()
|
||||
if docket is None:
|
||||
return mcp_types.ElicitResult(action="cancel", content=None)
|
||||
def _map_key(docket: Docket, task_scope: str | None, task_id: str, leg: int) -> str:
|
||||
"""Redis hash mapping a leg's surfaced keys back to the tool's own keys."""
|
||||
return docket.key(f"{_prefix(docket, task_scope, task_id)}:input:{leg}:map")
|
||||
|
||||
task_scope = task_context.task_scope
|
||||
task_id = task_context.task_id
|
||||
# One elicitation outstanding per task: key the request by the task id so the
|
||||
# inputRequests map surfaced by tasks/get is stable and answerable.
|
||||
input_key = task_id
|
||||
|
||||
requests_key = _requests_key(docket, task_scope, task_id)
|
||||
response_key = _response_key(docket, task_scope, task_id, input_key)
|
||||
request_payload = _elicitation_input_request(message, schema)
|
||||
def _mint_surfaced_key(task_id: str) -> str:
|
||||
"""Mint a unique surfaced key for one outstanding request (SEP-2663 L350).
|
||||
|
||||
Namespaced by the task id and suffixed with fresh entropy so no two
|
||||
requests — across legs or within one leg — ever collide, and a key is never
|
||||
reused after its response is delivered.
|
||||
"""
|
||||
return f"{task_id}:{secrets.token_hex(8)}"
|
||||
|
||||
|
||||
def _decode(value: Any) -> str | None:
|
||||
if value is None:
|
||||
return None
|
||||
if isinstance(value, bytes):
|
||||
return value.decode("utf-8")
|
||||
return str(value)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Task arguments and leg pointer (written at create, advanced at tasks/update)
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
async def save_task_args(
|
||||
docket: Docket,
|
||||
task_scope: str | None,
|
||||
task_id: str,
|
||||
arguments: dict[str, Any],
|
||||
ttl_seconds: int,
|
||||
) -> None:
|
||||
"""Store the original tool arguments, re-supplied to every leg."""
|
||||
async with docket.redis() as redis:
|
||||
await redis.hset(requests_key, input_key, json.dumps(request_payload))
|
||||
await redis.expire(requests_key, INPUT_TTL_SECONDS)
|
||||
await redis.set(
|
||||
_args_key(docket, task_scope, task_id),
|
||||
json.dumps(arguments),
|
||||
ex=ttl_seconds,
|
||||
)
|
||||
|
||||
|
||||
async def load_task_args(
|
||||
docket: Docket, task_scope: str | None, task_id: str
|
||||
) -> dict[str, Any]:
|
||||
"""Load the stored tool arguments for a task's next leg."""
|
||||
async with docket.redis() as redis:
|
||||
raw = await redis.get(_args_key(docket, task_scope, task_id))
|
||||
decoded = _decode(raw)
|
||||
if not decoded:
|
||||
return {}
|
||||
parsed = json.loads(decoded)
|
||||
return parsed if isinstance(parsed, dict) else {}
|
||||
|
||||
|
||||
async def save_current_leg(
|
||||
docket: Docket,
|
||||
task_scope: str | None,
|
||||
task_id: str,
|
||||
leg_key: str,
|
||||
leg_number: int,
|
||||
ttl_seconds: int,
|
||||
) -> None:
|
||||
"""Record the latest leg's Docket execution key and its number."""
|
||||
async with docket.redis() as redis:
|
||||
await redis.set(
|
||||
_current_leg_key(docket, task_scope, task_id), leg_key, ex=ttl_seconds
|
||||
)
|
||||
await redis.set(
|
||||
_leg_number_key(docket, task_scope, task_id),
|
||||
str(leg_number),
|
||||
ex=ttl_seconds,
|
||||
)
|
||||
|
||||
|
||||
async def load_current_leg(
|
||||
docket: Docket, task_scope: str | None, task_id: str
|
||||
) -> tuple[str | None, int]:
|
||||
"""Return the current leg's execution key and number (defaults to 1)."""
|
||||
async with docket.redis() as redis:
|
||||
leg_key = _decode(
|
||||
await redis.get(_current_leg_key(docket, task_scope, task_id))
|
||||
)
|
||||
leg_raw = _decode(await redis.get(_leg_number_key(docket, task_scope, task_id)))
|
||||
try:
|
||||
async with docket.redis() as redis:
|
||||
result = await redis.blpop([response_key], timeout=INPUT_TTL_SECONDS)
|
||||
except (RedisError, OSError) as exc:
|
||||
logger.warning("BLPOP failed for task %s input; cancelling: %s", task_id, exc)
|
||||
result = None
|
||||
leg_number = int(leg_raw) if leg_raw else 1
|
||||
except ValueError:
|
||||
leg_number = 1
|
||||
return leg_key, leg_number
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Outstanding requests (written by the capture wrapper, read by tasks/get)
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
async def store_outstanding(
|
||||
docket: Docket,
|
||||
task_scope: str | None,
|
||||
task_id: str,
|
||||
leg: int,
|
||||
serialized_requests: dict[str, dict[str, Any]],
|
||||
request_state: str | None,
|
||||
ttl_seconds: int = INPUT_TTL_SECONDS,
|
||||
) -> None:
|
||||
"""Persist a leg's outstanding input requests plus its carried state.
|
||||
|
||||
``serialized_requests`` maps the tool's own request keys to serialized
|
||||
``InputRequest`` payloads. Each is stored under a freshly minted surfaced
|
||||
key, with the surfaced-key → tool-key mapping recorded alongside so
|
||||
``tasks/update`` can translate answers back. ``request_state`` is written
|
||||
when the leg carried one and cleared otherwise, so it travels to the next
|
||||
leg verbatim.
|
||||
"""
|
||||
requests_key = _requests_key(docket, task_scope, task_id, leg)
|
||||
map_key = _map_key(docket, task_scope, task_id, leg)
|
||||
state_key = _request_state_key(docket, task_scope, task_id)
|
||||
|
||||
async with docket.redis() as redis:
|
||||
await redis.hdel(requests_key, input_key)
|
||||
await redis.delete(response_key)
|
||||
|
||||
if not result:
|
||||
return mcp_types.ElicitResult(action="cancel", content=None)
|
||||
|
||||
_key, raw = result
|
||||
response = json.loads(raw)
|
||||
return mcp_types.ElicitResult(
|
||||
action=response.get("action", "accept"),
|
||||
content=response.get("content"),
|
||||
)
|
||||
for tool_key, payload in serialized_requests.items():
|
||||
surfaced = _mint_surfaced_key(task_id)
|
||||
await redis.hset(requests_key, surfaced, json.dumps(payload))
|
||||
await redis.hset(map_key, surfaced, tool_key)
|
||||
await redis.expire(requests_key, ttl_seconds)
|
||||
await redis.expire(map_key, ttl_seconds)
|
||||
if request_state is not None:
|
||||
await redis.set(state_key, request_state, ex=ttl_seconds)
|
||||
else:
|
||||
await redis.delete(state_key)
|
||||
|
||||
|
||||
async def read_outstanding_inputs(
|
||||
docket: Docket, task_scope: str | None, task_id: str
|
||||
docket: Docket, task_scope: str | None, task_id: str, leg: int
|
||||
) -> dict[str, Any]:
|
||||
"""Return the task's outstanding input requests, keyed by input key.
|
||||
"""Return a leg's outstanding input requests, keyed by surfaced key.
|
||||
|
||||
Empty when the task is not waiting on input. Consumed by ``tasks/get`` to
|
||||
Empty when the leg is not waiting on input. Consumed by ``tasks/get`` to
|
||||
build the ``input_required`` status and its ``inputRequests`` snapshot.
|
||||
"""
|
||||
requests_key = _requests_key(docket, task_scope, task_id)
|
||||
async with docket.redis() as redis:
|
||||
raw = await redis.hgetall(requests_key)
|
||||
raw = await redis.hgetall(_requests_key(docket, task_scope, task_id, leg))
|
||||
outstanding: dict[str, Any] = {}
|
||||
for key, value in raw.items():
|
||||
key_str = key.decode() if isinstance(key, bytes) else key
|
||||
value_str = value.decode() if isinstance(value, bytes) else value
|
||||
key_str = _decode(key)
|
||||
value_str = _decode(value)
|
||||
if key_str is None or value_str is None:
|
||||
continue
|
||||
try:
|
||||
outstanding[key_str] = json.loads(value_str)
|
||||
except json.JSONDecodeError:
|
||||
|
|
@ -144,26 +262,136 @@ async def read_outstanding_inputs(
|
|||
return outstanding
|
||||
|
||||
|
||||
async def deliver_input_responses(
|
||||
async def _read_outstanding_map(
|
||||
docket: Docket, task_scope: str | None, task_id: str, leg: int
|
||||
) -> dict[str, str]:
|
||||
"""Return the surfaced-key → tool-key mapping for a leg."""
|
||||
async with docket.redis() as redis:
|
||||
raw = await redis.hgetall(_map_key(docket, task_scope, task_id, leg))
|
||||
mapping: dict[str, str] = {}
|
||||
for key, value in raw.items():
|
||||
key_str = _decode(key)
|
||||
value_str = _decode(value)
|
||||
if key_str is None or value_str is None:
|
||||
continue
|
||||
mapping[key_str] = value_str
|
||||
return mapping
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Responses (written by tasks/update, read by the next leg's context factory)
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
async def translate_responses(
|
||||
docket: Docket,
|
||||
task_scope: str | None,
|
||||
task_id: str,
|
||||
leg: int,
|
||||
responses: dict[str, Any],
|
||||
) -> None:
|
||||
"""Deliver ``tasks/update`` responses to the parked worker(s).
|
||||
) -> dict[str, mcp_types.Result] | None:
|
||||
"""Translate a ``tasks/update`` payload into typed, tool-keyed responses.
|
||||
|
||||
For each response whose key names an outstanding request, pushes the
|
||||
response onto that key's list (waking the worker's ``BLPOP``) and removes the
|
||||
request. Responses for unknown or already-satisfied keys are ignored, as the
|
||||
spec requires.
|
||||
``responses`` is keyed by the surfaced keys the client received for ``leg``.
|
||||
Unknown or already-satisfied keys are ignored (SEP-2663). Each recognized
|
||||
answer is validated into the result type its request maps to and re-keyed to
|
||||
the tool's own request key. Returns ``None`` when nothing matched, so the
|
||||
caller can treat a stale or empty update as an idempotent no-op.
|
||||
"""
|
||||
requests_key = _requests_key(docket, task_scope, task_id)
|
||||
outstanding = await read_outstanding_inputs(docket, task_scope, task_id, leg)
|
||||
if not outstanding:
|
||||
return None
|
||||
mapping = await _read_outstanding_map(docket, task_scope, task_id, leg)
|
||||
|
||||
translated: dict[str, mcp_types.Result] = {}
|
||||
for surfaced_key, raw in responses.items():
|
||||
payload = outstanding.get(surfaced_key)
|
||||
if payload is None:
|
||||
continue
|
||||
tool_key = mapping.get(surfaced_key)
|
||||
if tool_key is None:
|
||||
continue
|
||||
method = payload.get("method", "elicitation/create")
|
||||
result_type = result_type_for_method(method)
|
||||
translated[tool_key] = result_type.model_validate(raw)
|
||||
|
||||
return translated or None
|
||||
|
||||
|
||||
async def store_input_responses(
|
||||
docket: Docket,
|
||||
task_scope: str | None,
|
||||
task_id: str,
|
||||
translated: dict[str, mcp_types.Result],
|
||||
ttl_seconds: int = INPUT_TTL_SECONDS,
|
||||
) -> None:
|
||||
"""Store translated responses for the next leg to read via ``ctx``.
|
||||
|
||||
The responses are stored typed-but-serialized (``{"type", "data"}``) so the
|
||||
next leg's context factory reconstructs real result objects keyed by the
|
||||
tool's own request keys.
|
||||
"""
|
||||
stored = {
|
||||
tool_key: {
|
||||
"type": type(result).__name__,
|
||||
"data": result.model_dump(by_alias=True, mode="json"),
|
||||
}
|
||||
for tool_key, result in translated.items()
|
||||
}
|
||||
async with docket.redis() as redis:
|
||||
for input_key, response in responses.items():
|
||||
outstanding = await redis.hget(requests_key, input_key)
|
||||
if outstanding is None:
|
||||
continue
|
||||
response_key = _response_key(docket, task_scope, task_id, input_key)
|
||||
await redis.rpush(response_key, json.dumps(response))
|
||||
await redis.expire(response_key, INPUT_TTL_SECONDS)
|
||||
await redis.hdel(requests_key, input_key)
|
||||
await redis.set(
|
||||
_input_responses_key(docket, task_scope, task_id),
|
||||
json.dumps(stored),
|
||||
ex=ttl_seconds,
|
||||
)
|
||||
|
||||
|
||||
async def clear_outstanding(
|
||||
docket: Docket, task_scope: str | None, task_id: str, leg: int
|
||||
) -> None:
|
||||
"""Drop a leg's outstanding requests and mapping once it has been answered.
|
||||
|
||||
The answered surfaced keys are never reused (a later leg mints its own), so
|
||||
a duplicate ``tasks/update`` naming them finds nothing and is a no-op.
|
||||
"""
|
||||
async with docket.redis() as redis:
|
||||
await redis.delete(_requests_key(docket, task_scope, task_id, leg))
|
||||
await redis.delete(_map_key(docket, task_scope, task_id, leg))
|
||||
|
||||
|
||||
async def load_pending_input(
|
||||
docket: Docket, task_scope: str | None, task_id: str
|
||||
) -> tuple[str | None, mcp_types.InputResponses | None]:
|
||||
"""Load the per-leg state a re-entered leg reads via ``ctx``.
|
||||
|
||||
Returns ``(request_state, input_responses)``: the opaque state carried
|
||||
forward and the typed answers keyed by the tool's own request keys. Both are
|
||||
``None`` on the first leg (nothing has been asked yet).
|
||||
"""
|
||||
async with docket.redis() as redis:
|
||||
state_raw = _decode(
|
||||
await redis.get(_request_state_key(docket, task_scope, task_id))
|
||||
)
|
||||
responses_raw = _decode(
|
||||
await redis.get(_input_responses_key(docket, task_scope, task_id))
|
||||
)
|
||||
|
||||
responses: dict[str, mcp_types.Result] | None = None
|
||||
if responses_raw:
|
||||
parsed = json.loads(responses_raw)
|
||||
if isinstance(parsed, dict):
|
||||
responses = {}
|
||||
for tool_key, entry in parsed.items():
|
||||
if not isinstance(entry, dict):
|
||||
continue
|
||||
result_type = _RESULT_TYPE_BY_NAME.get(entry.get("type", ""))
|
||||
if result_type is None:
|
||||
continue
|
||||
responses[tool_key] = result_type.model_validate(entry.get("data"))
|
||||
|
||||
# The reconstructed values are the concrete result types the tool asked for;
|
||||
# `InputResponses` is that union keyed by request key. The `Result` element
|
||||
# type erases that for the checker, so narrow at the return.
|
||||
if responses is None:
|
||||
return state_raw, None
|
||||
return state_raw, cast("mcp_types.InputResponses", responses)
|
||||
|
|
|
|||
|
|
@ -37,6 +37,44 @@ _AUTH_TAG = "auth"
|
|||
_ANON_TAG = "anon"
|
||||
_VALID_TAGS = (_AUTH_TAG, _ANON_TAG)
|
||||
|
||||
# Delimiter separating the stable base task key from a per-leg suffix. A single
|
||||
# background task runs as a sequence of Docket executions (legs): the first leg
|
||||
# uses the base key, and each re-entry (after the client answers input) enqueues
|
||||
# a fresh execution under `{base}{_LEG_DELIMITER}{n}`. The base key encodes every
|
||||
# segment with `quote(safe="")`, which percent-encodes `#` to `%23`, so a literal
|
||||
# `#` never appears inside the base key and is an unambiguous leg boundary. All
|
||||
# task-identity parsing strips the leg suffix, so the scope/task-id/component a
|
||||
# leg resolves to are identical across every leg of the same task.
|
||||
_LEG_DELIMITER = "#"
|
||||
|
||||
|
||||
def leg_execution_key(base_task_key: str, leg: int) -> str:
|
||||
"""Build the Docket execution key for a given leg of a task.
|
||||
|
||||
Leg 1 uses the bare base key (so existing single-leg behavior is unchanged);
|
||||
later legs append `#leg{n}` so each re-entry is a distinct Docket execution
|
||||
while still parsing back to the same task scope, id, and component.
|
||||
"""
|
||||
if leg <= 1:
|
||||
return base_task_key
|
||||
return f"{base_task_key}{_LEG_DELIMITER}leg{leg}"
|
||||
|
||||
|
||||
def base_task_key(execution_key: str) -> str:
|
||||
"""Strip any per-leg suffix, returning the stable base task key."""
|
||||
return execution_key.split(_LEG_DELIMITER, 1)[0]
|
||||
|
||||
|
||||
def leg_number_from_key(execution_key: str) -> int:
|
||||
"""Return the leg number a Docket execution key encodes (leg 1 = base key)."""
|
||||
_base, sep, suffix = execution_key.partition(_LEG_DELIMITER)
|
||||
if not sep:
|
||||
return 1
|
||||
try:
|
||||
return int(suffix.removeprefix("leg"))
|
||||
except ValueError:
|
||||
return 1
|
||||
|
||||
|
||||
def build_task_key(
|
||||
task_scope: str | None,
|
||||
|
|
@ -97,6 +135,9 @@ def parse_task_key(task_key: str) -> TaskKeyParts:
|
|||
>>> parse_task_key("anon:task456:tool:my_tool")
|
||||
`{'task_scope': None, 'client_task_id': 'task456', 'task_type': 'tool', 'component_identifier': 'my_tool'}`
|
||||
"""
|
||||
# A per-leg execution key (`{base}#leg{n}`) parses to the same identity as
|
||||
# its base: every leg of a task shares one scope, id, and component.
|
||||
task_key = base_task_key(task_key)
|
||||
tag, _, rest = task_key.partition(":")
|
||||
if tag not in _VALID_TAGS or not rest:
|
||||
raise ValueError(
|
||||
|
|
|
|||
|
|
@ -94,6 +94,9 @@ async def docket_lifespan(
|
|||
try:
|
||||
yield
|
||||
finally:
|
||||
# End-and-reenter never parks a worker on input, so a
|
||||
# task waiting for input holds no worker slot: cancelling
|
||||
# run_forever drains promptly regardless of task state.
|
||||
worker_task.cancel()
|
||||
with suppress(asyncio.CancelledError):
|
||||
await worker_task
|
||||
|
|
|
|||
|
|
@ -1170,7 +1170,9 @@ class TestTaskExecution:
|
|||
if hasattr(Docket, "_memory_server"):
|
||||
delattr(Docket, "_memory_server")
|
||||
|
||||
async def test_guard_result_from_task_is_rejected(self, reset_docket_memory_server):
|
||||
async def test_guard_result_from_task_parks_for_input(
|
||||
self, reset_docket_memory_server
|
||||
):
|
||||
mcp = FastMCP("guard-task")
|
||||
mcp.add_extension(TasksExtension())
|
||||
|
||||
|
|
@ -1182,14 +1184,20 @@ class TestTaskExecution:
|
|||
request_state=None,
|
||||
)
|
||||
|
||||
# A guard's `InputRequiredResult` only makes sense against a live
|
||||
# request. Submitting `book_flight` as a background task and then
|
||||
# reading it back must reject the guard result: `tasks/get` raises when
|
||||
# it tries to inline the completed task's InputRequiredResult.
|
||||
# A function-tool guard is driven as a task by the in-task reentrant
|
||||
# loop: submitting `book_flight` parks its input request on the poll
|
||||
# surface (`input_required`), where a client answers it via
|
||||
# `tasks/update`. The full round-trip lives in
|
||||
# tests/tasks/server/test_guard_reentrant.py.
|
||||
async with running_task_server(mcp):
|
||||
created = await submit_task(mcp, "book_flight", {})
|
||||
with pytest.raises(MCPError, match="background task"):
|
||||
await wait_for_task(mcp, created.task_id)
|
||||
parked = await wait_for_task(
|
||||
mcp,
|
||||
created.task_id,
|
||||
target_states=frozenset({"input_required"}),
|
||||
)
|
||||
assert parked.status == "input_required"
|
||||
assert parked.input_requests
|
||||
|
||||
|
||||
class TestHttpTransport:
|
||||
|
|
|
|||
|
|
@ -31,21 +31,16 @@ from mcp_types import (
|
|||
Implementation,
|
||||
InitializeRequestParams,
|
||||
)
|
||||
from pydantic import BaseModel
|
||||
|
||||
from fastmcp import FastMCP
|
||||
from fastmcp.exceptions import ToolError
|
||||
from fastmcp.server.auth import AccessToken
|
||||
from fastmcp.server.context import Context
|
||||
from fastmcp.server.dependencies import get_access_token
|
||||
from fastmcp.server.elicitation import (
|
||||
AcceptedElicitation,
|
||||
DeclinedElicitation,
|
||||
)
|
||||
from fastmcp_tasks import TasksExtension
|
||||
from tests.tasks.task_helpers import (
|
||||
running_task_server,
|
||||
submit_task,
|
||||
update_task,
|
||||
wait_for_task,
|
||||
)
|
||||
|
||||
|
|
@ -295,11 +290,17 @@ class TestContextClientExtensionBackgroundTask:
|
|||
|
||||
|
||||
class TestContextElicitBackgroundTask:
|
||||
"""Tests for Context.elicit() in background task mode."""
|
||||
"""Tests for Context.elicit() in background task mode.
|
||||
|
||||
async def test_elicit_raises_when_no_task_engine(self):
|
||||
"""elicit() fails fast when in a background task but no tasks extension
|
||||
is installed to answer the request."""
|
||||
Imperative elicitation is not supported inside a background task: the worker
|
||||
never blocks on a client round-trip. A task gathers input with the guard
|
||||
pattern (return an ``InputRequiredResult``), so ``ctx.elicit()`` in a task
|
||||
fails fast with guidance rather than parking a worker.
|
||||
"""
|
||||
|
||||
async def test_elicit_raises_with_guard_guidance(self):
|
||||
"""elicit() inside a background task raises a ToolError pointing to the
|
||||
guard/return pattern (InputRequiredResult)."""
|
||||
mcp = FastMCP("test")
|
||||
ctx = Context(mcp, task_id="test-task-123")
|
||||
|
||||
|
|
@ -308,7 +309,7 @@ class TestContextElicitBackgroundTask:
|
|||
|
||||
ctx._session = cast(ServerSession, MockSession())
|
||||
|
||||
with pytest.raises(RuntimeError, match="tasks extension"):
|
||||
with pytest.raises(ToolError, match="InputRequiredResult"):
|
||||
await ctx.elicit("Need input", str)
|
||||
|
||||
|
||||
|
|
@ -391,99 +392,24 @@ class TestBackgroundTaskIntegration:
|
|||
"session_unavailable": True,
|
||||
}
|
||||
|
||||
async def test_elicit_accept_flow(self):
|
||||
"""E2E: tool elicits input, client accepts via tasks/update (poll)."""
|
||||
mcp = FastMCP("elicit-accept-test")
|
||||
async def test_imperative_elicit_fails_with_guard_guidance(self):
|
||||
"""A task=True tool that calls ctx.elicit() fails with the guard-pattern
|
||||
error rather than parking a worker on a client round-trip."""
|
||||
mcp = FastMCP("elicit-forbidden")
|
||||
mcp.add_extension(TasksExtension())
|
||||
|
||||
@mcp.tool(task=True)
|
||||
async def ask_name(ctx: Context) -> str:
|
||||
result = await ctx.elicit("What is your name?", str)
|
||||
if isinstance(result, AcceptedElicitation):
|
||||
return f"Hello, {result.data}!"
|
||||
return "No name provided"
|
||||
return str(result)
|
||||
|
||||
async with running_task_server(mcp):
|
||||
created = await submit_task(mcp, "ask_name", {})
|
||||
parked = await wait_for_task(
|
||||
mcp, created.task_id, target_states=frozenset({"input_required"})
|
||||
)
|
||||
assert parked.input_requests is not None
|
||||
key = next(iter(parked.input_requests))
|
||||
await update_task(
|
||||
mcp,
|
||||
created.task_id,
|
||||
{key: {"action": "accept", "content": {"value": "Bob"}}},
|
||||
)
|
||||
final = await wait_for_task(mcp, created.task_id)
|
||||
|
||||
assert final.status == "completed"
|
||||
assert final.result is not None
|
||||
assert final.result["structuredContent"] == {"result": "Hello, Bob!"}
|
||||
|
||||
async def test_elicit_decline_flow(self):
|
||||
"""E2E: tool elicits input, client declines via tasks/update (poll)."""
|
||||
mcp = FastMCP("elicit-decline-test")
|
||||
mcp.add_extension(TasksExtension())
|
||||
|
||||
@mcp.tool(task=True)
|
||||
async def optional_input(ctx: Context) -> str:
|
||||
result = await ctx.elicit("Want to provide a name?", str)
|
||||
if isinstance(result, DeclinedElicitation):
|
||||
return "User declined"
|
||||
if isinstance(result, AcceptedElicitation):
|
||||
return f"Got: {result.data}"
|
||||
return "Cancelled"
|
||||
|
||||
async with running_task_server(mcp):
|
||||
created = await submit_task(mcp, "optional_input", {})
|
||||
parked = await wait_for_task(
|
||||
mcp, created.task_id, target_states=frozenset({"input_required"})
|
||||
)
|
||||
assert parked.input_requests is not None
|
||||
key = next(iter(parked.input_requests))
|
||||
await update_task(mcp, created.task_id, {key: {"action": "decline"}})
|
||||
final = await wait_for_task(mcp, created.task_id)
|
||||
|
||||
assert final.status == "completed"
|
||||
assert final.result is not None
|
||||
assert final.result["structuredContent"] == {"result": "User declined"}
|
||||
|
||||
async def test_elicit_with_pydantic_model(self):
|
||||
"""E2E: tool elicits structured Pydantic input via tasks/update (poll)."""
|
||||
|
||||
class UserInfo(BaseModel):
|
||||
name: str
|
||||
age: int
|
||||
|
||||
mcp = FastMCP("elicit-pydantic-test")
|
||||
mcp.add_extension(TasksExtension())
|
||||
|
||||
@mcp.tool(task=True)
|
||||
async def get_user_info(ctx: Context) -> str:
|
||||
result = await ctx.elicit("Provide user info", UserInfo)
|
||||
if isinstance(result, AcceptedElicitation):
|
||||
assert isinstance(result.data, UserInfo)
|
||||
return f"{result.data.name} is {result.data.age}"
|
||||
return "No info"
|
||||
|
||||
async with running_task_server(mcp):
|
||||
created = await submit_task(mcp, "get_user_info", {})
|
||||
parked = await wait_for_task(
|
||||
mcp, created.task_id, target_states=frozenset({"input_required"})
|
||||
)
|
||||
assert parked.input_requests is not None
|
||||
key = next(iter(parked.input_requests))
|
||||
await update_task(
|
||||
mcp,
|
||||
created.task_id,
|
||||
{key: {"action": "accept", "content": {"name": "Alice", "age": 30}}},
|
||||
)
|
||||
final = await wait_for_task(mcp, created.task_id)
|
||||
|
||||
assert final.status == "completed"
|
||||
assert final.result is not None
|
||||
assert final.result["structuredContent"] == {"result": "Alice is 30"}
|
||||
assert final.status == "failed"
|
||||
assert final.error is not None
|
||||
assert "InputRequiredResult" in final.error["message"]
|
||||
|
||||
|
||||
class TestAccessTokenInBackgroundTasks:
|
||||
|
|
|
|||
|
|
@ -26,7 +26,6 @@ from mcp.shared.exceptions import MCPError
|
|||
from fastmcp import FastMCP
|
||||
from fastmcp.client import Client
|
||||
from fastmcp.exceptions import ToolError
|
||||
from fastmcp.server import context as core_context
|
||||
from fastmcp.server.dependencies import bind_request_context
|
||||
from fastmcp.tools.base import ToolResult
|
||||
from fastmcp.utilities.tasks import TASKS_EXTENSION_ID, TaskConfig
|
||||
|
|
@ -356,6 +355,8 @@ async def test_worker_hooks_survive_sibling_server_shutdown():
|
|||
each running a TasksExtension refcount them, so the hooks clear only when
|
||||
the last extension lifespan exits.
|
||||
"""
|
||||
from fastmcp.server import dependencies as core_dependencies
|
||||
|
||||
server_a = _tasks_server()
|
||||
server_b = _tasks_server()
|
||||
|
||||
|
|
@ -363,8 +364,8 @@ async def test_worker_hooks_survive_sibling_server_shutdown():
|
|||
await stack_b.enter_async_context(server_b._lifespan_manager())
|
||||
async with AsyncExitStack() as stack_a:
|
||||
await stack_a.enter_async_context(server_a._lifespan_manager())
|
||||
assert core_context._task_elicitation_handler is not None
|
||||
assert core_dependencies._background_context_factory is not None
|
||||
# Server A has shut down; server B's workers still need the hooks.
|
||||
assert core_context._task_elicitation_handler is not None
|
||||
assert core_dependencies._background_context_factory is not None
|
||||
# The last extension exited; hooks are cleared.
|
||||
assert core_context._task_elicitation_handler is None
|
||||
assert core_dependencies._background_context_factory is None
|
||||
|
|
|
|||
174
tests/tasks/server/test_guard_reentrant.py
Normal file
174
tests/tasks/server/test_guard_reentrant.py
Normal file
|
|
@ -0,0 +1,174 @@
|
|||
"""The guard-pattern reentrant loop driven inside a background task.
|
||||
|
||||
A `task=True` tool that *returns* an `InputRequiredResult` (rather than awaiting
|
||||
`ctx.elicit()`) is the same guard authoring model FastMCP uses foreground. As a
|
||||
task, the worker drives the multi-round-trip itself: it parks the request on the
|
||||
poll surface, the client answers via `tasks/update`, and the tool is re-invoked
|
||||
with the answer on `ctx.input_responses` — identical to the foreground contract,
|
||||
only the transport differs. These tests exercise that loop end-to-end through
|
||||
the real interceptor and handlers via `task_helpers`.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from typing import Any
|
||||
|
||||
import mcp_types
|
||||
|
||||
from fastmcp import Context, FastMCP
|
||||
from fastmcp_tasks import TasksExtension
|
||||
from tests.tasks.task_helpers import (
|
||||
running_task_server,
|
||||
submit_task,
|
||||
update_task,
|
||||
wait_for_task,
|
||||
)
|
||||
|
||||
|
||||
def _elicit_request(message: str) -> mcp_types.ElicitRequest:
|
||||
return mcp_types.ElicitRequest(
|
||||
params=mcp_types.ElicitRequestFormParams(
|
||||
message=message,
|
||||
requested_schema={
|
||||
"type": "object",
|
||||
"properties": {"value": {"type": "string"}},
|
||||
},
|
||||
)
|
||||
)
|
||||
|
||||
|
||||
def _answer(responses: mcp_types.InputResponses, key: str) -> str:
|
||||
"""Read the string value a client accepted for `key` (test helper)."""
|
||||
result = responses[key]
|
||||
assert isinstance(result, mcp_types.ElicitResult)
|
||||
assert result.content is not None
|
||||
return str(result.content["value"])
|
||||
|
||||
|
||||
def _input_required(
|
||||
requests: dict[str, mcp_types.ElicitRequest],
|
||||
request_state: str | None = None,
|
||||
) -> mcp_types.InputRequiredResult:
|
||||
return mcp_types.InputRequiredResult(
|
||||
result_type="input_required",
|
||||
input_requests=requests,
|
||||
request_state=request_state,
|
||||
)
|
||||
|
||||
|
||||
async def _park_key(mcp: FastMCP, task_id: str) -> str:
|
||||
parked = await wait_for_task(
|
||||
mcp, task_id, target_states=frozenset({"input_required"})
|
||||
)
|
||||
assert parked.status == "input_required"
|
||||
assert parked.input_requests is not None
|
||||
return next(iter(parked.input_requests))
|
||||
|
||||
|
||||
async def test_guard_return_single_round_completes():
|
||||
"""A tool that returns InputRequiredResult once is driven to completion."""
|
||||
mcp = FastMCP("guard")
|
||||
mcp.add_extension(TasksExtension())
|
||||
|
||||
@mcp.tool(task=True)
|
||||
async def greet(ctx: Context) -> str | mcp_types.InputRequiredResult:
|
||||
responses = ctx.input_responses
|
||||
if responses is None:
|
||||
return _input_required({"name": _elicit_request("Your name?")})
|
||||
return f"Hello, {_answer(responses, 'name')}!"
|
||||
|
||||
async with running_task_server(mcp):
|
||||
created = await submit_task(mcp, "greet", {})
|
||||
key = await _park_key(mcp, created.task_id)
|
||||
await update_task(
|
||||
mcp,
|
||||
created.task_id,
|
||||
{key: {"action": "accept", "content": {"value": "Ada"}}},
|
||||
)
|
||||
final = await wait_for_task(mcp, created.task_id)
|
||||
|
||||
assert final.status == "completed"
|
||||
assert final.result is not None
|
||||
assert final.result["structuredContent"] == {"result": "Hello, Ada!"}
|
||||
|
||||
|
||||
async def test_guard_return_multiple_rounds_use_distinct_keys():
|
||||
"""A tool that asks twice surfaces distinct keys across rounds (SEP-2663 L350).
|
||||
|
||||
The second round's key must differ from the first's — a client that
|
||||
deduplicates by key must not suppress the second ask. Cross-round state
|
||||
travels through `request_state` (each leg's `input_responses` holds only
|
||||
that leg's answers, matching the foreground guard contract).
|
||||
"""
|
||||
mcp = FastMCP("guard")
|
||||
mcp.add_extension(TasksExtension())
|
||||
|
||||
@mcp.tool(task=True)
|
||||
async def full_name(ctx: Context) -> str | mcp_types.InputRequiredResult:
|
||||
responses = ctx.input_responses
|
||||
if responses is None:
|
||||
# Round 1: ask for the first name.
|
||||
return _input_required({"first": _elicit_request("First name?")})
|
||||
if ctx.request_state is None:
|
||||
# Round 2: carry the first name forward in request_state, ask last.
|
||||
return _input_required(
|
||||
{"last": _elicit_request("Last name?")},
|
||||
request_state=_answer(responses, "first"),
|
||||
)
|
||||
# Round 3: request_state holds the first name; responses holds the last.
|
||||
return f"{ctx.request_state} {_answer(responses, 'last')}"
|
||||
|
||||
async with running_task_server(mcp):
|
||||
created = await submit_task(mcp, "full_name", {})
|
||||
key1 = await _park_key(mcp, created.task_id)
|
||||
await update_task(
|
||||
mcp,
|
||||
created.task_id,
|
||||
{key1: {"action": "accept", "content": {"value": "Ada"}}},
|
||||
)
|
||||
key2 = await _park_key(mcp, created.task_id)
|
||||
assert key2 != key1
|
||||
await update_task(
|
||||
mcp,
|
||||
created.task_id,
|
||||
{key2: {"action": "accept", "content": {"value": "Lovelace"}}},
|
||||
)
|
||||
final = await wait_for_task(mcp, created.task_id)
|
||||
|
||||
assert final.status == "completed"
|
||||
assert final.result is not None
|
||||
assert final.result["structuredContent"] == {"result": "Ada Lovelace"}
|
||||
|
||||
|
||||
async def test_non_guard_tool_runs_once():
|
||||
"""A tool that never asks for input completes in a single invocation."""
|
||||
calls: list[int] = []
|
||||
mcp = FastMCP("guard")
|
||||
mcp.add_extension(TasksExtension())
|
||||
|
||||
@mcp.tool(task=True)
|
||||
async def square(n: int) -> int:
|
||||
calls.append(n)
|
||||
return n * n
|
||||
|
||||
async with running_task_server(mcp):
|
||||
created = await submit_task(mcp, "square", {"n": 6})
|
||||
final = await wait_for_task(mcp, created.task_id)
|
||||
|
||||
assert final.status == "completed"
|
||||
assert final.result is not None
|
||||
assert final.result["structuredContent"] == {"result": 36}
|
||||
assert calls == [6]
|
||||
|
||||
|
||||
def test_reentrant_wrapper_preserves_signature():
|
||||
"""The wrapper keeps the tool's parameters so Docket DI is unchanged."""
|
||||
import inspect
|
||||
|
||||
from fastmcp_tasks.input_loop import reentrant_task_fn
|
||||
|
||||
async def fn(n: int, ctx: Any) -> int:
|
||||
return n
|
||||
|
||||
wrapped = reentrant_task_fn(fn)
|
||||
assert list(inspect.signature(wrapped).parameters) == ["n", "ctx"]
|
||||
65
tests/tasks/server/test_reenter_shutdown.py
Normal file
65
tests/tasks/server/test_reenter_shutdown.py
Normal file
|
|
@ -0,0 +1,65 @@
|
|||
"""Shutdown regression for end-and-reenter task input.
|
||||
|
||||
The whole point of end-and-reenter is that a task waiting on client input holds
|
||||
no worker: the guard leg's Docket execution completed and the worker is free.
|
||||
This test proves it — a task parked in ``input_required`` that is never answered
|
||||
must not delay server shutdown. Under the old block-and-resume model the worker
|
||||
sat on a Redis wait for the input TTL and wedged teardown; here the lifespan
|
||||
exits promptly.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
|
||||
import mcp_types
|
||||
|
||||
from fastmcp import Context, FastMCP
|
||||
from fastmcp_tasks import TasksExtension
|
||||
from tests.tasks.task_helpers import running_task_server, submit_task, wait_for_task
|
||||
|
||||
|
||||
def _elicit_request(message: str) -> mcp_types.ElicitRequest:
|
||||
return mcp_types.ElicitRequest(
|
||||
params=mcp_types.ElicitRequestFormParams(
|
||||
message=message,
|
||||
requested_schema={
|
||||
"type": "object",
|
||||
"properties": {"value": {"type": "string"}},
|
||||
},
|
||||
)
|
||||
)
|
||||
|
||||
|
||||
async def test_parked_task_does_not_delay_shutdown():
|
||||
"""Exiting the lifespan with a task in input_required (never answered) must
|
||||
return promptly — no worker is parked awaiting input."""
|
||||
mcp = FastMCP("parked-shutdown")
|
||||
mcp.add_extension(TasksExtension())
|
||||
|
||||
@mcp.tool(task=True)
|
||||
async def greet(ctx: Context) -> str | mcp_types.InputRequiredResult:
|
||||
if ctx.input_responses is None:
|
||||
return mcp_types.InputRequiredResult(
|
||||
result_type="input_required",
|
||||
input_requests={"name": _elicit_request("Your name?")},
|
||||
request_state=None,
|
||||
)
|
||||
return "done"
|
||||
|
||||
loop = asyncio.get_event_loop()
|
||||
manager = running_task_server(mcp)
|
||||
await manager.__aenter__()
|
||||
try:
|
||||
created = await submit_task(mcp, "greet", {})
|
||||
parked = await wait_for_task(
|
||||
mcp, created.task_id, target_states=frozenset({"input_required"})
|
||||
)
|
||||
assert parked.status == "input_required"
|
||||
finally:
|
||||
# Never answer; time how long teardown takes.
|
||||
started = loop.time()
|
||||
await manager.__aexit__(None, None, None)
|
||||
elapsed = loop.time() - started
|
||||
|
||||
assert elapsed < 3.0, f"lifespan took {elapsed:.2f}s to exit with a parked task"
|
||||
|
|
@ -1,219 +0,0 @@
|
|||
"""In-task elicitation under SEP-2663 (poll-based input).
|
||||
|
||||
A background worker that calls ``ctx.elicit()`` has no live request, so SEP-2663
|
||||
parks the request and the task's ``tasks/get`` status flips to ``input_required``
|
||||
with the outstanding ``inputRequests``. The caller answers with ``tasks/update``
|
||||
and the parked worker resumes. This replaces the SEP-1686 push relay (which sent
|
||||
``elicitation/create`` over a back-channel); the accept/decline/cancel semantics,
|
||||
structured round-trips, and sequential elicitations are preserved, driven here
|
||||
in-process because there is no client task API until Phase 4.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
from dataclasses import dataclass
|
||||
from typing import Any
|
||||
|
||||
import fastmcp_tasks.input_store as input_store
|
||||
from pydantic import BaseModel
|
||||
|
||||
from fastmcp import FastMCP
|
||||
from fastmcp.server.context import Context
|
||||
from fastmcp.server.elicitation import (
|
||||
AcceptedElicitation,
|
||||
CancelledElicitation,
|
||||
DeclinedElicitation,
|
||||
)
|
||||
from fastmcp_tasks import TasksExtension
|
||||
from tests.tasks.task_helpers import (
|
||||
get_task,
|
||||
running_task_server,
|
||||
submit_task,
|
||||
update_task,
|
||||
wait_for_task,
|
||||
)
|
||||
|
||||
|
||||
async def _wait_for_input_required(server: FastMCP, task_id: str, timeout: float = 5.0):
|
||||
"""Poll until the task is waiting on input, returning the GetTaskResult."""
|
||||
deadline = asyncio.get_event_loop().time() + timeout
|
||||
while True:
|
||||
got = await get_task(server, task_id)
|
||||
if got.status == "input_required":
|
||||
return got
|
||||
if got.status in ("completed", "failed", "cancelled"):
|
||||
raise AssertionError(
|
||||
f"Task {task_id} reached {got.status!r} before requesting input"
|
||||
)
|
||||
if asyncio.get_event_loop().time() >= deadline:
|
||||
raise TimeoutError(f"Task {task_id} never requested input")
|
||||
await asyncio.sleep(0.02)
|
||||
|
||||
|
||||
async def _drive(server: FastMCP, name: str, answers: list[dict[str, Any]]) -> str:
|
||||
"""Submit a task, answer each elicitation in turn, return its result text."""
|
||||
created = await submit_task(server, name, {})
|
||||
for answer in answers:
|
||||
got = await _wait_for_input_required(server, created.task_id)
|
||||
key = next(iter(got.input_requests))
|
||||
request = got.input_requests[key]
|
||||
assert request["method"] == "elicitation/create"
|
||||
await update_task(server, created.task_id, {key: answer})
|
||||
final = await wait_for_task(server, created.task_id)
|
||||
assert final.status == "completed", final.error
|
||||
assert final.result is not None
|
||||
return final.result["content"][0]["text"]
|
||||
|
||||
|
||||
async def test_accept_answers_the_elicitation():
|
||||
mcp = FastMCP("relay-accept")
|
||||
mcp.add_extension(TasksExtension())
|
||||
|
||||
@mcp.tool(task=True)
|
||||
async def ask_name(ctx: Context) -> str:
|
||||
result = await ctx.elicit("What is your name?", str)
|
||||
if isinstance(result, AcceptedElicitation):
|
||||
return f"Hello, {result.data}!"
|
||||
return "No name"
|
||||
|
||||
async with running_task_server(mcp):
|
||||
text = await _drive(
|
||||
mcp, "ask_name", [{"action": "accept", "content": {"value": "Alice"}}]
|
||||
)
|
||||
assert text == "Hello, Alice!"
|
||||
|
||||
|
||||
async def test_decline_yields_declined_elicitation():
|
||||
mcp = FastMCP("relay-decline")
|
||||
mcp.add_extension(TasksExtension())
|
||||
|
||||
@mcp.tool(task=True)
|
||||
async def optional_input(ctx: Context) -> str:
|
||||
result = await ctx.elicit("Provide a name?", str)
|
||||
if isinstance(result, DeclinedElicitation):
|
||||
return "User declined"
|
||||
return "Other"
|
||||
|
||||
async with running_task_server(mcp):
|
||||
text = await _drive(mcp, "optional_input", [{"action": "decline"}])
|
||||
assert text == "User declined"
|
||||
|
||||
|
||||
async def test_cancel_yields_cancelled_elicitation():
|
||||
mcp = FastMCP("relay-cancel")
|
||||
mcp.add_extension(TasksExtension())
|
||||
|
||||
@mcp.tool(task=True)
|
||||
async def cancellable(ctx: Context) -> str:
|
||||
result = await ctx.elicit("Input?", str)
|
||||
if isinstance(result, CancelledElicitation):
|
||||
return "Cancelled"
|
||||
return "Not cancelled"
|
||||
|
||||
async with running_task_server(mcp):
|
||||
text = await _drive(mcp, "cancellable", [{"action": "cancel"}])
|
||||
assert text == "Cancelled"
|
||||
|
||||
|
||||
async def test_dataclass_round_trips():
|
||||
mcp = FastMCP("relay-dataclass")
|
||||
mcp.add_extension(TasksExtension())
|
||||
|
||||
@dataclass
|
||||
class UserInfo:
|
||||
name: str
|
||||
age: int
|
||||
|
||||
@mcp.tool(task=True)
|
||||
async def get_user(ctx: Context) -> str:
|
||||
result = await ctx.elicit("Provide user info", UserInfo)
|
||||
if isinstance(result, AcceptedElicitation):
|
||||
assert isinstance(result.data, UserInfo)
|
||||
return f"{result.data.name} is {result.data.age}"
|
||||
return "No info"
|
||||
|
||||
async with running_task_server(mcp):
|
||||
text = await _drive(
|
||||
mcp,
|
||||
"get_user",
|
||||
[{"action": "accept", "content": {"name": "Bob", "age": 30}}],
|
||||
)
|
||||
assert text == "Bob is 30"
|
||||
|
||||
|
||||
async def test_pydantic_model_round_trips():
|
||||
mcp = FastMCP("relay-pydantic")
|
||||
mcp.add_extension(TasksExtension())
|
||||
|
||||
class Config(BaseModel):
|
||||
host: str
|
||||
port: int
|
||||
|
||||
@mcp.tool(task=True)
|
||||
async def get_config(ctx: Context) -> str:
|
||||
result = await ctx.elicit("Server config?", Config)
|
||||
if isinstance(result, AcceptedElicitation):
|
||||
assert isinstance(result.data, Config)
|
||||
return f"{result.data.host}:{result.data.port}"
|
||||
return "No config"
|
||||
|
||||
async with running_task_server(mcp):
|
||||
text = await _drive(
|
||||
mcp,
|
||||
"get_config",
|
||||
[{"action": "accept", "content": {"host": "localhost", "port": 8080}}],
|
||||
)
|
||||
assert text == "localhost:8080"
|
||||
|
||||
|
||||
async def test_multiple_sequential_elicitations():
|
||||
mcp = FastMCP("relay-multi")
|
||||
mcp.add_extension(TasksExtension())
|
||||
|
||||
@mcp.tool(task=True)
|
||||
async def two_questions(ctx: Context) -> str:
|
||||
r1 = await ctx.elicit("First name?", str)
|
||||
r2 = await ctx.elicit("Last name?", str)
|
||||
if isinstance(r1, AcceptedElicitation) and isinstance(r2, AcceptedElicitation):
|
||||
return f"{r1.data} {r2.data}"
|
||||
return "Incomplete"
|
||||
|
||||
async with running_task_server(mcp):
|
||||
text = await _drive(
|
||||
mcp,
|
||||
"two_questions",
|
||||
[
|
||||
{"action": "accept", "content": {"value": "Jane"}},
|
||||
{"action": "accept", "content": {"value": "Doe"}},
|
||||
],
|
||||
)
|
||||
assert text == "Jane Doe"
|
||||
|
||||
|
||||
async def test_unanswered_input_times_out_to_cancel(monkeypatch):
|
||||
"""A worker that is never answered eventually resumes with a cancel.
|
||||
|
||||
The poll model has no "no handler" fast path; instead the parked worker's
|
||||
blocking wait is bounded by ``INPUT_TTL_SECONDS``. Patched short here so the
|
||||
timeout-to-cancel behaviour is testable.
|
||||
"""
|
||||
monkeypatch.setattr(input_store, "INPUT_TTL_SECONDS", 1)
|
||||
|
||||
mcp = FastMCP("relay-timeout")
|
||||
mcp.add_extension(TasksExtension())
|
||||
|
||||
@mcp.tool(task=True)
|
||||
async def needs_input(ctx: Context) -> str:
|
||||
result = await ctx.elicit("Input?", str)
|
||||
if isinstance(result, CancelledElicitation):
|
||||
return "Cancelled as expected"
|
||||
return "Other"
|
||||
|
||||
async with running_task_server(mcp):
|
||||
created = await submit_task(mcp, "needs_input", {})
|
||||
# Never answer; the worker's bounded wait resolves to cancel.
|
||||
final = await wait_for_task(mcp, created.task_id, timeout=10.0)
|
||||
assert final.status == "completed"
|
||||
assert final.result is not None
|
||||
assert final.result["content"][0]["text"] == "Cancelled as expected"
|
||||
|
|
@ -12,6 +12,7 @@ from __future__ import annotations
|
|||
|
||||
import mcp_types.methods as methods
|
||||
import pytest
|
||||
|
||||
from fastmcp_tasks import wire_production
|
||||
|
||||
_MODERN = "2026-07-28"
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue