ceos/fastmcp
1
0
Fork 0
mirror of https://github.com/PrefectHQ/fastmcp.git synced 2026-08-09 07:09:11 +02:00
Code Issues Projects Releases Packages Wiki Activity

Rework tasks example into a runnable HTTP client/server pair

Browse source 
Server runs over HTTP on the default memory:// backend (no Redis needed); the
client drives it transparently, via an explicit handle, and with a parallel
command that fires several tasks at once to show them overlap. A 1s poll
interval keeps the demo snappy.
This commit is contained in:
Jeremiah Lowin 2026-07-23 16:44:53 -04:00
commit 733801ed6c
No known key found for this signature in database
4 changed files with 188 additions and 173 deletions
Download patch file Download diff file Expand all files Collapse all files

15
examples/tasks/.envrc
View file

@ -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

111
examples/tasks/README.md
View file

@ -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)

169
examples/tasks/client.py
View file

@ -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__":

86
examples/tasks/server.py
View file

@ -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)