Spaces:
Running
Running
File size: 5,859 Bytes
1a68f2c 2937184 1a68f2c | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 | ---
title: Progress Reporting
sidebarTitle: Progress
description: Update clients on the progress of long-running operations through the MCP context.
icon: chart-line
---
import { VersionBadge } from '/snippets/version-badge.mdx'
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:
```python {14, 21}
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
<Card icon="code" title="Context Progress Method">
<ResponseField name="ctx.report_progress" type="async method">
Report progress to the client for long-running operations
<Expandable title="Parameters">
<ResponseField name="progress" type="float">
Current progress value (e.g., 24, 0.75, 1500)
</ResponseField>
<ResponseField name="total" type="float | None" default="None">
Optional total value (e.g., 100, 1.0, 2000). When provided, clients may interpret this as enabling percentage calculation.
</ResponseField>
</Expandable>
</ResponseField>
</Card>
## Progress Patterns
### Percentage-Based Progress
Report progress as a percentage (0-100):
```python {13-14}
@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:
```python {10}
@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:
```python {11}
@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:
```python
@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](/clients/progress) for details on implementing client-side progress handling
|