diff --git a/examples/tasks/.envrc b/examples/tasks/.envrc index 87a7dfef9..7c90adf43 100644 --- a/examples/tasks/.envrc +++ b/examples/tasks/.envrc @@ -1,10 +1,11 @@ # FastMCP Tasks Example Environment Configuration -# This file is loaded by direnv (https://direnv.net/) when you cd into this directory -# Run `direnv allow` to enable automatic environment loading +# Loaded by direnv (https://direnv.net/) when you cd into this directory. +# Run `direnv allow` to enable automatic loading — or just `source .envrc`. -# Configure Docket backend URL -# Use Redis backend (requires docker-compose up) -export FASTMCP_DOCKET_URL=redis://localhost:24242/0 +# In-process worker on an in-memory backend: no Redis, nothing to start. +# This is the default the example runs on. +export FASTMCP_DOCKET_URL=memory:// -# Or uncomment to use memory:// for single-process testing -# export FASTMCP_DOCKET_URL=memory:// +# For distributed workers across separate processes (the `fastmcp tasks worker` +# CLI), point at Redis instead and run `docker compose up -d` first: +# export FASTMCP_DOCKET_URL=redis://localhost:24242/0 diff --git a/examples/tasks/README.md b/examples/tasks/README.md index 8013968d5..81f9285f3 100644 --- a/examples/tasks/README.md +++ b/examples/tasks/README.md @@ -1,60 +1,75 @@ -# FastMCP Tasks Example +# FastMCP Background Tasks Example -Demonstrates background task execution with Docket, including progress tracking, distributed backends, and CLI worker management. +A runnable client/server pair for SEP-2663 background tasks. The server exposes +one `task=True` tool that reports progress as it works; the client drives it +three ways — transparently, through an explicit handle, and several at once in +parallel. -## Setup +This runs on the in-memory backend by default, so there's nothing to install or +start beyond the two processes. + +## Run it + +In one terminal, start the server: ```bash -# From the fastmcp root directory -uv sync +uv sync # from the fastmcp root, once +python examples/tasks/server.py # listens on http://127.0.0.1:8000/mcp +``` -# Start Redis +In another terminal, drive it from the client: + +```bash +# Transparent — call_tool runs the background task and returns its result +python examples/tasks/client.py --duration 8 + +# Explicit handle — returns immediately, poll it yourself, then collect +python examples/tasks/client.py handle --duration 6 + +# Parallel — fire several tasks at once and watch them overlap +python examples/tasks/client.py parallel +python examples/tasks/client.py parallel 8 6 4 2 +``` + +The `parallel` run is the one to watch: four tasks of decreasing duration all +start at once and total wall-clock tracks the *longest* task rather than the +sum, because the worker runs them concurrently. + +## How it works + +The server enables tasks with one line: + +```python +mcp = FastMCP("Tasks Example") +mcp.add_extension(TasksExtension()) +``` + +The client opts in by importing `fastmcp_tasks` (which it does to use +`call_tool_task`). That single import enables task support for every `Client` +in the process — without it, a `Client` never advertises the tasks capability, +so the server would run the calls synchronously. + +## Distributed workers (optional) + +The default `memory://` backend runs the worker in the server process. To run +workers as separate processes, point Docket at Redis and start it first: + +```bash cd examples/tasks docker compose up -d +export FASTMCP_DOCKET_URL=redis://localhost:24242/0 # or: direnv allow -# Load environment (or source .envrc manually) -direnv allow - -# Run the server -fastmcp run server.py +python server.py # in one terminal +fastmcp tasks worker server.py # extra worker(s) in others ``` -For single-process mode without Redis, set `FASTMCP_DOCKET_URL=memory://` (note: CLI workers won't work). +| Backend | Workers | +| ------------ | ------------------------------- | +| `memory://` | in-process only (default) | +| `redis://…` | distributed across processes | -## Running the Client +## Learn more -```bash -# Background execution with progress callbacks -python examples/tasks/client.py --duration 10 - -# Immediate execution (blocks) -python examples/tasks/client.py immediate --duration 5 -``` - -## Starting Additional Workers - -With Redis, you can run additional workers to process tasks in parallel: - -```bash -fastmcp tasks worker server.py - -# Configure via environment: -export FASTMCP_DOCKET_CONCURRENCY=20 -fastmcp tasks worker server.py -``` - -**Backend options:** -- `memory://` - Single-process only (default) -- `redis://` - Distributed, multi-process (Redis or Valkey) - -## Environment Variables - -| Variable | Default | Description | -|----------|---------|-------------| -| `FASTMCP_DOCKET_URL` | `memory://` | Docket backend URL | - -## Learn More - -- [FastMCP Tasks Documentation](https://gofastmcp.com/docs/tasks) -- [Docket Documentation](https://github.com/PrefectHQ/docket) -- [MCP Task Protocol (SEP-1686)](https://spec.modelcontextprotocol.io/specification/architecture/tasks/) +- [Server background tasks](https://gofastmcp.com/servers/tasks) +- [Client background tasks](https://gofastmcp.com/clients/tasks) +- [Docket](https://github.com/PrefectHQ/docket) diff --git a/examples/tasks/client.py b/examples/tasks/client.py index 1c039d6f7..fe93ea23b 100644 --- a/examples/tasks/client.py +++ b/examples/tasks/client.py @@ -1,28 +1,24 @@ -""" -FastMCP Tasks Example Client (SEP-2663) +"""FastMCP background-tasks example client (SEP-2663). -Demonstrates the two client task surfaces: +Start the server first (`python examples/tasks/server.py`), then run any of the +commands below against it over HTTP. -- Transparent: `client.call_tool(...)` drives the background task to completion - under the hood and returns the tool's real result. The caller writes ordinary - tool-call code and never sees that the server ran the call as a task. -- Explicit handle: `call_tool_task(...)` returns a `ToolTask` immediately, so the - client can do other work and poll the task itself before collecting the result. + # Transparent: call_tool drives the background task and returns its result + python examples/tasks/client.py --duration 8 -Usage: - # Make sure environment is configured (source .envrc or use direnv) - source .envrc + # Explicit handle: return immediately, poll it yourself, then collect + python examples/tasks/client.py handle --duration 6 - # Transparent background task (default) - python client.py --duration 10 + # Parallel: fire several tasks at once and watch them overlap + python examples/tasks/client.py parallel - # Return-quickly handle, driven by the client - python client.py handle --duration 5 +Importing `fastmcp_tasks` (below) enables client task support for every Client +in the process — without it, a Client never advertises the tasks capability and +the server runs its calls synchronously. """ import asyncio -import sys -from pathlib import Path +import time from typing import Annotated import cyclopts @@ -30,94 +26,111 @@ from mcp_types import TextContent from rich.console import Console from fastmcp.client import Client -from fastmcp_tasks import call_tool_task +from fastmcp_tasks import call_tool_task # importing enables client task support + +SERVER_URL = "http://127.0.0.1:8000/mcp" console = Console() -app = cyclopts.App(name="tasks-client", help="FastMCP Tasks Example Client") +app = cyclopts.App(name="tasks-client", help="FastMCP background-tasks example client") -def load_server(): - """Load the example server.""" - examples_dir = Path(__file__).parent.parent.parent - if str(examples_dir) not in sys.path: - sys.path.insert(0, str(examples_dir)) - - import examples.tasks.server as server_module - - return server_module.mcp +def _text(result) -> str: + assert isinstance(result.content[0], TextContent) + return result.content[0].text @app.default async def transparent( - duration: Annotated[ - int, - cyclopts.Parameter(help="Duration of computation in seconds (1-60)"), - ] = 10, + duration: Annotated[int, cyclopts.Parameter(help="Seconds (1-60)")] = 8, ): - """Call the tool transparently: the client drives the task to completion.""" - if duration < 1 or duration > 60: - console.print("[red]Error: Duration must be between 1 and 60 seconds[/red]") - sys.exit(1) + """Call the tool transparently: the client drives the task to completion. - server = load_server() - - console.print(f"\n[bold]Calling slow_computation(duration={duration})[/bold]") - console.print("Mode: [cyan]Transparent (server may run it as a task)[/cyan]\n") - - # mode="auto" negotiates the modern protocol, so the server may run the call - # as a background task; the client resolves it transparently. - async with Client(server, mode="auto") as client: + The server runs `slow_computation` as a background task, but `call_tool` + polls it under the hood and returns the tool's real result — the calling + code looks exactly like an ordinary synchronous tool call. + """ + async with Client(SERVER_URL, mode="auto") as client: + console.print(f"\n[bold]Transparent call[/bold] (duration={duration})\n") + started = time.perf_counter() result = await client.call_tool( "slow_computation", - arguments={"duration": duration}, + {"label": "transparent", "duration": duration}, ) - - console.print("\n[bold]Result:[/bold]") - assert isinstance(result.content[0], TextContent) - console.print(f" {result.content[0].text}") + console.print(f"[green]{_text(result)}[/green]") + console.print(f"[dim]elapsed {time.perf_counter() - started:.1f}s[/dim]") @app.command async def handle( - duration: Annotated[ - int, - cyclopts.Parameter(help="Duration of computation in seconds (1-60)"), - ] = 5, + duration: Annotated[int, cyclopts.Parameter(help="Seconds (1-60)")] = 6, ): - """Use the explicit handle: return immediately, then drive the task.""" - if duration < 1 or duration > 60: - console.print("[red]Error: Duration must be between 1 and 60 seconds[/red]") - sys.exit(1) - - server = load_server() - - console.print(f"\n[bold]Calling slow_computation(duration={duration})[/bold]") - console.print("Mode: [cyan]Explicit ToolTask handle[/cyan]\n") - - async with Client(server, mode="auto") as client: + """Use the explicit handle: return immediately, then drive the task yourself.""" + async with Client(SERVER_URL, mode="auto") as client: + console.print(f"\n[bold]Explicit handle[/bold] (duration={duration})\n") task = await call_tool_task( - client, - "slow_computation", - arguments={"duration": duration}, + client, "slow_computation", {"label": "handle", "duration": duration} ) - console.print(f"Task started: [cyan]{task.task_id}[/cyan]\n") - # Do other work while the task runs in the background. - for i in range(3): - await asyncio.sleep(0.5) + # Do other work while the task runs, checking its status as you go. + while True: status = await task.status() + if status.status in ("completed", "failed", "cancelled"): + break + console.print(f"[dim]still {status.status}: {status.status_message}[/dim]") + await asyncio.sleep(1) + + result = await task.result() + console.print(f"\n[green]{_text(result)}[/green]") + + +@app.command +async def parallel( + durations: Annotated[ + list[int] | None, + cyclopts.Parameter(help="One task per duration (default: 5 4 3 2)"), + ] = None, +): + """Fire several background tasks at once and drive them concurrently. + + Each `call_tool_task` returns immediately, so we start every task before + awaiting any of them. The worker runs them in parallel, so total wall-clock + tracks the *longest* task, not the sum — proof the work actually overlaps. + """ + durations = durations or [5, 4, 3, 2] + + async with Client(SERVER_URL, mode="auto") as client: + console.print(f"\n[bold]Parallel tasks[/bold]: durations={durations}\n") + started = time.perf_counter() + + # Start every task up front — none of these await completion. + tasks = [ + await call_tool_task( + client, + "slow_computation", + {"label": f"task-{i}({d}s)", "duration": d}, + ) + for i, d in enumerate(durations) + ] + for task in tasks: + console.print(f" started [cyan]{task.task_id}[/cyan]") + + # Await them together; results print as each task finishes. + async def collect(task): + result = await task.result() console.print( - f"[dim]Client doing other work... ({i + 1}/3) " - f"— task is {status.status}[/dim]" + f"[green]✓[/green] {_text(result)} " + f"[dim](+{time.perf_counter() - started:.1f}s)[/dim]" ) - console.print("\n[dim]Waiting for the final result...[/dim]") - result = await task.result() + console.print() + await asyncio.gather(*(collect(task) for task in tasks)) - console.print("\n[bold]Result:[/bold]") - assert isinstance(result.content[0], TextContent) - console.print(f" {result.content[0].text}") + total = time.perf_counter() - started + console.print( + f"\n[bold]All {len(tasks)} tasks done in {total:.1f}s[/bold] " + f"[dim](longest single task: {max(durations)}s)[/dim]" + ) if __name__ == "__main__": diff --git a/examples/tasks/server.py b/examples/tasks/server.py index 8405ab717..745009ef7 100644 --- a/examples/tasks/server.py +++ b/examples/tasks/server.py @@ -1,79 +1,65 @@ -""" -FastMCP Tasks Example Server +"""FastMCP background-tasks example server (SEP-2663). -Demonstrates background task execution with progress tracking using Docket. +Run this in one terminal, then drive it from `client.py` in another. It exposes +one `task=True` tool that reports progress as it works, so you can watch the +client poll a real background task over HTTP. -Setup: - 1. Start Redis: docker compose up -d - 2. Load environment: source .envrc - 3. Run server: fastmcp run server.py + # From the fastmcp root (memory:// backend, no Redis needed): + python examples/tasks/server.py -The example uses Redis by default to demonstrate distributed task execution -and the fastmcp tasks CLI commands. +The server listens on http://localhost:8000/mcp. The tasks extension runs its +Docket worker in-process on the default `memory://` backend, so several tasks +submitted at once execute concurrently (worker concurrency defaults to 10). +Point `FASTMCP_DOCKET_URL` at Redis to distribute work across separate worker +processes instead — see README.md. """ import asyncio import logging +from datetime import timedelta from typing import Annotated -from docket import Logged - from fastmcp import FastMCP from fastmcp.dependencies import Progress +from fastmcp.utilities.tasks import TaskConfig from fastmcp_tasks import TasksExtension -# Configure logging -logging.basicConfig(level=logging.INFO) -logger = logging.getLogger(__name__) +logging.basicConfig(level=logging.INFO, format="%(asctime)s %(message)s") +logger = logging.getLogger("tasks-example") -# Create server and enable background tasks (SEP-2663). The extension reads the -# FASTMCP_DOCKET_* environment for its backend (memory:// by default, Redis for -# distributed execution). +# Enable SEP-2663 background tasks. With no arguments the extension reads the +# FASTMCP_DOCKET_* environment and falls back to an in-process memory:// worker. mcp = FastMCP("Tasks Example") mcp.add_extension(TasksExtension()) -@mcp.tool(task=True) +# A short poll interval keeps the example snappy: the client observes each +# task finishing within ~1s. The default is 5s, tuned for real workloads. +@mcp.tool(task=TaskConfig(poll_interval=timedelta(seconds=1))) async def slow_computation( - duration: Annotated[int, Logged], + label: Annotated[str, "A name for this run, echoed back in progress logs"], + duration: Annotated[int, "How many seconds the computation should take (1-60)"], progress: Progress = Progress(), ) -> str: + """Spend `duration` seconds working, reporting progress once per second. + + Marked `task=True`, so a task-aware client runs it in the background and + polls for progress and the final result instead of blocking on the call. """ - Perform a slow computation that takes `duration` seconds. + if not 1 <= duration <= 60: + raise ValueError("duration must be between 1 and 60 seconds") - This tool demonstrates progress tracking with background tasks. - It logs progress every 1-2 seconds and reports progress via Docket. - - Args: - duration: Number of seconds the computation should take (1-60) - - Returns: - A completion message with the total duration - """ - if duration < 1 or duration > 60: - raise ValueError("Duration must be between 1 and 60 seconds") - - logger.info(f"Starting slow computation for {duration} seconds") - - # Set total progress units + logger.info("[%s] starting — %ds", label, duration) await progress.set_total(duration) - # Process each second - for i in range(duration): - # Sleep for 1 second + for elapsed in range(1, duration + 1): await asyncio.sleep(1) - - # Update progress - elapsed = i + 1 - remaining = duration - elapsed await progress.increment() - await progress.set_message( - f"Working... {elapsed}/{duration}s ({remaining}s remaining)" - ) + await progress.set_message(f"{label}: {elapsed}/{duration}s") - # Log every 1-2 seconds - if elapsed % 2 == 0 or elapsed == duration: - logger.info(f"Progress: {elapsed}/{duration}s") + logger.info("[%s] done", label) + return f"{label} finished in {duration}s" - logger.info(f"Completed computation in {duration} seconds") - return f"Computation completed successfully in {duration} seconds!" + +if __name__ == "__main__": + mcp.run(transport="http", host="127.0.0.1", port=8000)