Progress reporting allows MCP tools to notify clients about the progress of long-running operations. This enables clients to display progress indicators and provide better user experience during time-consuming tasks.

Why Use Progress Reporting?

Progress reporting is valuable for:

  • User experience: Keep users informed about long-running operations
  • Progress indicators: Enable clients to show progress bars or percentages
  • Timeout prevention: Demonstrate that operations are actively progressing
  • Debugging: Track execution progress for performance analysis

Basic Usage

Use ctx.report_progress() to send progress updates to the client:

from fastmcp import FastMCP, Context
import asyncio

mcp = FastMCP("ProgressDemo")

@mcp.tool
async def process_items(items: list[str], ctx: Context) -> dict:
    """Process a list of items with progress updates."""
    total = len(items)
    results = []
    
    for i, item in enumerate(items):
        # Report progress as we process each item
        await ctx.report_progress(progress=i, total=total)
        
        # Simulate processing time
        await asyncio.sleep(0.1)
        results.append(item.upper())
    
    # Report 100% completion
    await ctx.report_progress(progress=total, total=total)
    
    return {"processed": len(results), "results": results}

Method Signature

Context Progress Method

ctx.report_progress
async method

Report progress to the client for long-running operations

Progress Patterns

Percentage-Based Progress

Report progress as a percentage (0-100):

@mcp.tool
async def download_file(url: str, ctx: Context) -> str:
    """Download a file with percentage progress."""
    total_size = 1000  # KB
    downloaded = 0
    
    while downloaded < total_size:
        # Download chunk
        chunk_size = min(50, total_size - downloaded)
        downloaded += chunk_size
        
        # Report percentage progress
        percentage = (downloaded / total_size) * 100
        await ctx.report_progress(progress=percentage, total=100)
        
        await asyncio.sleep(0.1)  # Simulate download time
    
    return f"Downloaded file from {url}"

Absolute Progress

Report progress with absolute values:

@mcp.tool
async def backup_database(ctx: Context) -> str:
    """Backup database tables with absolute progress."""
    tables = ["users", "orders", "products", "inventory", "logs"]
    
    for i, table in enumerate(tables):
        await ctx.info(f"Backing up table: {table}")
        
        # Report absolute progress
        await ctx.report_progress(progress=i + 1, total=len(tables))
        
        # Simulate backup time
        await asyncio.sleep(0.5)
    
    return "Database backup completed"

Indeterminate Progress

Report progress without a known total for operations where the endpoint is unknown:

@mcp.tool
async def scan_directory(directory: str, ctx: Context) -> dict:
    """Scan directory with indeterminate progress."""
    files_found = 0
    
    # Simulate directory scanning
    for i in range(10):  # Unknown number of files
        files_found += 1
        
        # Report progress without total for indeterminate operations
        await ctx.report_progress(progress=files_found)
        
        await asyncio.sleep(0.2)
    
    return {"files_found": files_found, "directory": directory}

Multi-Stage Operations

Break complex operations into stages with progress for each:

@mcp.tool
async def data_migration(source: str, destination: str, ctx: Context) -> str:
    """Migrate data with multi-stage progress reporting."""
    
    # Stage 1: Validation (0-25%)
    await ctx.info("Validating source data")
    for i in range(5):
        await ctx.report_progress(progress=i * 5, total=100)
        await asyncio.sleep(0.1)
    
    # Stage 2: Export (25-60%)
    await ctx.info("Exporting data from source")
    for i in range(7):
        progress = 25 + (i * 5)
        await ctx.report_progress(progress=progress, total=100)
        await asyncio.sleep(0.1)
    
    # Stage 3: Transform (60-80%)
    await ctx.info("Transforming data format")
    for i in range(4):
        progress = 60 + (i * 5)
        await ctx.report_progress(progress=progress, total=100)
        await asyncio.sleep(0.1)
    
    # Stage 4: Import (80-100%)
    await ctx.info("Importing to destination")
    for i in range(4):
        progress = 80 + (i * 5)
        await ctx.report_progress(progress=progress, total=100)
        await asyncio.sleep(0.1)
    
    # Final completion
    await ctx.report_progress(progress=100, total=100)
    
    return f"Migration from {source} to {destination} completed"

Client Requirements

Progress reporting requires clients to support progress handling:

  • Clients must send a progressToken in the initial request to receive progress updates
  • If no progress token is provided, progress calls will have no effect (they won’t error)
  • See Client Progress for details on implementing client-side progress handling