Progress reporting lets your server send progress updates to the client while a tool is running. Clients can use these updates to show spinners, progress bars, and “still working” UI during slow operations.
Note: ctx.report_progress(...) only sends a notification if the client provided a progress token for this request. If the client didn’t request progress, it’s a no-op.
Basic usage
from dedalus_mcp import get_context, tool
@tool(description="Process files")
async def process_files(files: list[str]) -> str:
ctx = get_context()
for i, file in enumerate(files, start=1):
await ctx.report_progress(i, total=len(files), message=f"Processing {file}")
process_file(file) # your code; if this blocks, offload or make it async
return f"Processed {len(files)} files"
Parameters
await ctx.report_progress(
progress=50, # current progress value
total=100, # optional total (for percentage)
message="Halfway done...", # optional status message
)
| Parameter | Type | Description |
|---|
progress | int | float | Current progress value |
total | int | float | None | Optional total value; enables percentage UI |
message | str | None | Optional human-friendly status text |
Example: File download
from dedalus_mcp import get_context, tool
@tool(description="Download files")
async def download_files(urls: list[str]) -> dict:
ctx = get_context()
downloaded: list[str] = []
for i, url in enumerate(urls, start=1):
await ctx.report_progress(i - 1, total=len(urls), message=f"Downloading {url}")
await ctx.info("Downloading", data={"url": url})
path = await download(url) # your code
downloaded.append(path)
await ctx.report_progress(i, total=len(urls))
return {"files": downloaded}
Example: Data processing pipeline
from dedalus_mcp import get_context, tool
@tool(description="Process large dataset")
async def process_dataset(dataset_id: str) -> dict:
ctx = get_context()
await ctx.report_progress(0, total=100, message="Loading dataset")
data = load_dataset(dataset_id) # your code
await ctx.report_progress(30, total=100, message="Transforming data")
total_items = len(data)
if total_items:
for i, item in enumerate(data, start=1):
# map item progress into the 30..70 range
progress = 30 + int((i / total_items) * 40)
await ctx.report_progress(progress, total=100)
transform(item) # your code
await ctx.report_progress(70, total=100, message="Saving results")
save_results(data) # your code
await ctx.report_progress(100, total=100, message="Done")
return {"processed": len(data)}
Example: Indeterminate progress
If you don’t know the total up front, omit total and send periodic updates:
from dedalus_mcp import get_context, tool
@tool(description="Search until found")
async def search(query: str) -> str:
ctx = get_context()
pages_searched = 0
while True:
pages_searched += 1
await ctx.report_progress(pages_searched, message=f"Searched {pages_searched} pages")
result = search_page(query, pages_searched) # your code
if result:
return result
if pages_searched > 100:
return "Not found"
Tips
- Prefer async work: progress updates are most useful when your tool is doing I/O (
async def). If you do CPU-heavy or blocking work, consider offloading it so progress notifications can still flow.
- Use
message sparingly: short messages like “Downloading…”, “Transforming…”, “Saving…” are easiest for clients to display.
- Don’t spam updates: sending progress on every tiny step can be noisy. For very large loops, you may want to report every N items.
Last modified on January 27, 2026