A tool that returns in a few milliseconds needs no ceremony. A tool that runs for several seconds does: without feedback, the client looks frozen, and a user (or a model) cannot tell a slow success from a hang. MCP solves this with progress notifications and log messages that stream back to the client while the tool is still running. This article builds a FastMCP tool that reports progress as it works, and a client that renders a live progress bar from those notifications.

The tool counts primes in a range, reporting progress after each chunk. It is real work (not a sleep loop), so the progress reflects actual computation. Over the HTTP transport the notifications arrive incrementally through the streamable-HTTP channel, which is what lets the bar advance during the call. The stack is Mac-native: uv and make.

What you will build

  • work.py: pure primality and chunked-counting logic, testable on its own.
  • A count_primes tool that reports progress and log messages through the request Context.
  • A pytest suite that captures the streamed notifications and asserts they arrive.
  • scripts/watch.py: a client that draws a live progress bar over HTTP, plus make targets to run and deploy it.

Prerequisites

  • macOS 13+ with Homebrew (brew.sh).
  • uv 0.5+ — brew install uv; verify uv --version.
  • Xcode Command Line Tools (xcode-select --install) for make.
  • Comfort with async/await. The tool and the client handlers are coroutines.

Everything runs through uv and make.

How progress works in MCP

Two moving parts:

  1. The server adds a Context parameter to a tool and calls await ctx.report_progress(progress, total, message) (and optionally await ctx.info(...) for log lines) as it works.
  2. The client passes a progress_handler (and optionally a log_handler) so it receives each notification and can update a UI.

Over the HTTP transport these notifications are delivered on the same streaming channel as the eventual result, so they show up as they are sent, not batched at the end.

Step 1: Scaffold and lock down hygiene

Create the files

mkdir -p progress-streaming-fastmcp-server-macos
cd progress-streaming-fastmcp-server-macos
touch .gitignore

Add the code: .gitignore

__pycache__/
*.py[cod]
.venv/
.uv/
.pytest_cache/
.ruff_cache/
.mypy_cache/
logs/
*.log
*.pid
.DS_Store

Detailed breakdown

  • Standard uv/macOS hygiene. This project has no secrets or generated data to ignore beyond the usual caches and logs.

Step 2: Initialize the project with uv

Create the files

uv init --name macmcp --no-workspace
rm -f main.py hello.py
uv add fastmcp
uv add --dev pytest pytest-asyncio

Add the code: pyproject.toml

[project]
name = "macmcp"
version = "0.1.0"
description = "A FastMCP tool that streams progress while it runs"
readme = "README.md"
requires-python = ">=3.12"
dependencies = [
    "fastmcp>=3.4.3",
]

[dependency-groups]
dev = [
    "pytest>=9.1.1",
    "pytest-asyncio>=1.4.0",
]

Detailed breakdown

  • fastmcp provides the Context object the tool uses to report progress and the client progress_handler/log_handler hooks.

Step 3: The work, kept pure

Separate the computation from the reporting. work.py counts primes in chunks and yields intermediate results, with no dependency on MCP. That keeps the math unit-testable and the tool a thin loop.

Create the file

touch work.py

Add the code: work.py

"""Pure, dependency-free work: primality and chunked prime counting.

Keeping the computation here (separate from the MCP tool) means it can be unit
tested on its own, and the tool stays a thin loop that reports progress.
"""

from collections.abc import Iterator


def is_prime(n: int) -> bool:
    """Trial-division primality test (deliberately not the fastest)."""
    if n < 2:
        return False
    if n < 4:
        return True
    if n % 2 == 0:
        return False
    i = 3
    while i * i <= n:
        if n % i == 0:
            return False
        i += 2
    return True


def prime_chunks(limit: int, chunks: int = 20) -> Iterator[tuple[int, int]]:
    """Yield (scanned_up_to, primes_found_so_far) after each chunk of [2, limit].

    Splits the range into roughly `chunks` pieces so a caller can report progress
    between them. The final tuple's counts cover the whole range.
    """
    if limit < 2:
        yield limit, 0
        return
    step = max(1, (limit - 1) // chunks)
    primes = 0
    n = 2
    while n <= limit:
        end = min(n + step, limit + 1)
        primes += sum(1 for k in range(n, end) if is_prime(k))
        yield end - 1, primes
        n = end

Detailed breakdown

  • is_prime uses plain trial division. It is intentionally not optimized, so a large limit takes long enough to make progress worth watching.
  • prime_chunks does the whole scan but pauses at chunk boundaries to yield how far it has scanned and how many primes it has found. That is the natural place for the tool to report progress. Because it is a generator with no MCP dependency, the tests drive it directly.

Step 4: The tool that reports progress

The tool adds a Context parameter, which FastMCP injects. After each chunk it reports progress and logs a line. FastMCP awaits async tools, so the reporting does not block the event loop.

Create the file

touch server.py

Add the code: server.py

"""A FastMCP server with a long-running tool that reports progress.

The `count_primes` tool scans a range and, after each chunk, streams a progress
notification and a log line back to the client. Over the HTTP transport these
arrive incrementally while the tool is still running, so a client can show a live
progress bar instead of staring at a frozen call.
"""

import os

from fastmcp import Context, FastMCP

from work import prime_chunks

mcp = FastMCP("macmcp")


@mcp.tool
async def count_primes(limit: int, ctx: Context, chunks: int = 20) -> dict:
    """Count primes in [2, limit], reporting progress after each chunk."""
    if limit < 2:
        return {"limit": limit, "primes": 0}

    await ctx.info(f"counting primes up to {limit} in {chunks} chunks")
    primes = 0
    for scanned_to, primes in prime_chunks(limit, chunks):
        await ctx.report_progress(
            progress=scanned_to - 1,          # work done so far (2..limit)
            total=limit - 1,                  # total units of work
            message=f"scanned to {scanned_to}: {primes} primes",
        )
    await ctx.info(f"found {primes} primes up to {limit}")
    return {"limit": limit, "primes": primes}


def main() -> None:
    if os.environ.get("MCP_TRANSPORT", "stdio").lower() == "http":
        mcp.run(
            transport="http",
            host=os.environ.get("MCP_HOST", "127.0.0.1"),
            port=int(os.environ.get("MCP_PORT", "8000")),
        )
    else:
        mcp.run()  # stdio


if __name__ == "__main__":
    main()

Detailed breakdown

  • ctx: Context is injected by FastMCP from the type hint; it does not appear in the tool’s public schema, so a client calls count_primes(limit=...) without passing it.
  • ctx.report_progress(progress, total, message) sends a progress notification. progress is the work done and total the work to do (here, the count of numbers scanned), so a client can compute a percentage. The message is a human-readable status.
  • ctx.info(...) streams a log message on the same channel, useful for narration a progress bar cannot carry. FastMCP also offers debug, warning, and error.
  • The for loop pauses at chunk boundaries to report, so the client hears from the tool roughly chunks times across the run rather than only at the end.

Step 5: Test that progress is actually streamed

Two test files. test_work.py checks the math directly. test_progress.py drives the tool through the in-memory client with a progress_handler, which receives the same notifications an HTTP client would, and asserts they arrive.

Create the files

mkdir -p tests
touch tests/__init__.py tests/test_work.py tests/test_progress.py pytest.ini

Add the code: pytest.ini

[pytest]
asyncio_mode = auto

Add the code: tests/test_work.py

"""Test the pure primality and chunking logic."""

from work import is_prime, prime_chunks


def test_is_prime_small_cases():
    primes = {2, 3, 5, 7, 11, 13}
    for n in range(0, 15):
        assert is_prime(n) == (n in primes)


def test_prime_chunks_final_count_is_25_up_to_100():
    # There are 25 primes <= 100.
    last = list(prime_chunks(100, chunks=10))[-1]
    assert last == (100, 25)


def test_prime_chunks_progress_is_monotonic():
    scanned = [s for s, _ in prime_chunks(100, chunks=10)]
    assert scanned == sorted(scanned)
    assert scanned[-1] == 100  # final chunk reaches the limit


def test_prime_chunks_handles_tiny_limit():
    assert list(prime_chunks(1)) == [(1, 0)]

Add the code: tests/test_progress.py

"""Test that the tool streams progress notifications during the call.

The in-memory client delivers the same progress and log notifications an HTTP
client would, so a progress_handler captures them deterministically.
"""

import pytest
from fastmcp import Client

from server import mcp


@pytest.fixture()
def captured():
    return {"progress": [], "logs": []}


def _client(captured):
    async def on_progress(progress, total, message):
        captured["progress"].append((progress, total, message))

    async def on_log(message):
        captured["logs"].append(message.data)

    return Client(mcp, progress_handler=on_progress, log_handler=on_log)


async def test_result_is_correct(captured):
    async with _client(captured) as client:
        result = await client.call_tool("count_primes", {"limit": 100, "chunks": 10})
    assert result.data == {"limit": 100, "primes": 25}


async def test_progress_events_are_streamed(captured):
    async with _client(captured) as client:
        await client.call_tool("count_primes", {"limit": 100, "chunks": 10})
    progress = captured["progress"]
    assert progress, "expected progress notifications"
    # Progress is non-decreasing and finishes at 100% (progress == total).
    values = [p for p, _, _ in progress]
    assert values == sorted(values)
    final_progress, total, _ = progress[-1]
    assert final_progress == total == 99  # units of work = limit - 1


async def test_log_messages_are_streamed(captured):
    async with _client(captured) as client:
        await client.call_tool("count_primes", {"limit": 50, "chunks": 5})
    joined = " ".join(str(d) for d in captured["logs"])
    assert "counting primes" in joined and "found" in joined


async def test_tiny_limit_short_circuits(captured):
    async with _client(captured) as client:
        result = await client.call_tool("count_primes", {"limit": 1})
    assert result.data == {"limit": 1, "primes": 0}

Detailed breakdown

  • _client wires two handlers. on_progress(progress, total, message) is the progress-handler signature FastMCP calls for each notification; on_log(message) receives log messages, whose .data holds the text.
  • test_progress_events_are_streamed asserts the events arrive, are non-decreasing, and finish at 100% (progress == total). That proves the tool reports incrementally, not just once at the end.
  • test_log_messages_are_streamed confirms the ctx.info lines reach the client’s log handler.
  • Using the in-memory client keeps these tests fast and deterministic while exercising the real notification path.

Run them:

uv run pytest -v

Step 6: The Makefile

Wrap development, a live watch target, and the launchd deployment. Plain make prints help.

Create the file

touch Makefile

Add the code: Makefile

.DEFAULT_GOAL := help

# --- Deploy configuration (launchd user agent) --------------------------------
LABEL   := com.macmcp.progress
PLIST   := $(HOME)/Library/LaunchAgents/$(LABEL).plist
UV      := $(shell command -v uv)
DIR     := $(shell pwd)
DOMAIN  := gui/$(shell id -u)

# `make watch` argument
LIMIT ?= 2000000

.PHONY: help install serve test watch deploy undeploy status logs clean

help:  ## Show this help screen
	@grep -E '^[a-zA-Z_-]+:.*?## .*$$' $(MAKEFILE_LIST) \
		| awk 'BEGIN {FS = ":.*?## "}; {printf "  \033[36m%-10s\033[0m %s\n", $$1, $$2}'

install:  ## Sync runtime and dev dependencies
	uv sync

serve:  ## Run the HTTP server in the foreground (Ctrl-C to stop)
	MCP_TRANSPORT=http uv run python server.py

test:  ## Run the test suite
	uv run pytest -v

watch:  ## Run count_primes and render a live progress bar: make watch LIMIT=2000000
	@LIMIT=$(LIMIT) uv run python scripts/watch.py

deploy:  ## Render the launchd plist and start the service
	@mkdir -p logs
	@sed -e 's#@UV@#$(UV)#g' \
	     -e 's#@DIR@#$(DIR)#g' \
	     -e 's#@PATH@#$(dir $(UV)):/usr/bin:/bin#g' \
	     deploy/$(LABEL).plist.template > $(PLIST)
	launchctl bootout $(DOMAIN) $(PLIST) 2>/dev/null || true
	launchctl bootstrap $(DOMAIN) $(PLIST)
	@echo "Deployed $(LABEL). Check: make status"

undeploy:  ## Stop the service and remove the launchd plist
	launchctl bootout $(DOMAIN) $(PLIST) 2>/dev/null || true
	rm -f $(PLIST)
	@echo "Removed $(LABEL)."

status:  ## Show whether the service is registered and running
	@launchctl print $(DOMAIN)/$(LABEL) 2>/dev/null \
		| grep -E 'state =|pid =' || echo "$(LABEL) is not loaded."

logs:  ## Tail the service log files
	@tail -n 40 -f logs/server.out.log logs/server.err.log

clean:  ## Remove caches and logs
	rm -rf .pytest_cache logs __pycache__ tests/__pycache__

Detailed breakdown

  • watch runs the client with a live progress bar; raise LIMIT to make the job longer. The launchd lifecycle matches the companion deploy article.

Step 7: The live progress client

A client that subscribes to progress and draws a bar that advances during the call.

Create the files

mkdir -p scripts deploy
touch scripts/watch.py deploy/com.macmcp.progress.plist.template

Add the code: scripts/watch.py

"""Call count_primes over HTTP and render a live progress bar.

The progress notifications arrive over the streamable-HTTP channel while the tool
is still running, so the bar advances during the call.

    LIMIT=2000000 uv run python scripts/watch.py
"""

import asyncio
import os
import sys
import time

from fastmcp import Client

URL = os.environ.get("MCP_URL", "http://127.0.0.1:8000/mcp/")
LIMIT = int(os.environ.get("LIMIT", "2000000"))

_start = time.monotonic()


async def on_progress(progress: float, total: float | None, message: str | None) -> None:
    pct = (progress / total * 100) if total else 0.0
    filled = int(pct // 5)
    bar = "#" * filled + "-" * (20 - filled)
    elapsed = time.monotonic() - _start
    sys.stdout.write(f"\r[{bar}] {pct:5.1f}%  t+{elapsed:4.1f}s  {message or ''}")
    sys.stdout.flush()


async def main() -> None:
    async with Client(URL) as client:
        result = await client.call_tool(
            "count_primes", {"limit": LIMIT}, progress_handler=on_progress
        )
    print()  # end the progress line
    print("result:", result.data)


if __name__ == "__main__":
    asyncio.run(main())

Add the code: deploy/com.macmcp.progress.plist.template

<?xml version="1.0" encoding="UTF-8"?>
<!DOCTYPE plist PUBLIC "-//Apple//DTD PLIST 1.0//EN"
  "http://www.apple.com/DTDs/PropertyList-1.0.dtd">
<plist version="1.0">
<dict>
    <key>Label</key>
    <string>com.macmcp.progress</string>

    <key>ProgramArguments</key>
    <array>
        <string>@UV@</string>
        <string>run</string>
        <string>python</string>
        <string>server.py</string>
    </array>

    <key>WorkingDirectory</key>
    <string>@DIR@</string>

    <key>EnvironmentVariables</key>
    <dict>
        <key>MCP_TRANSPORT</key>
        <string>http</string>
        <key>MCP_HOST</key>
        <string>127.0.0.1</string>
        <key>MCP_PORT</key>
        <string>8000</string>
        <key>PATH</key>
        <string>@PATH@</string>
    </dict>

    <key>RunAtLoad</key>
    <true/>

    <key>KeepAlive</key>
    <true/>

    <key>StandardOutPath</key>
    <string>@DIR@/logs/server.out.log</string>

    <key>StandardErrorPath</key>
    <string>@DIR@/logs/server.err.log</string>
</dict>
</plist>

Detailed breakdown

  • on_progress is passed per call as progress_handler=. FastMCP invokes it for every report_progress the tool sends, so the bar redraws in place with the percentage, elapsed time, and the tool’s message.
  • \r and flush() redraw the same terminal line, giving the moving-bar effect. Passing the handler to call_tool (rather than the Client) scopes it to this one call.

Step 8: Watch it run

Start the server and watch a real job report progress.

make deploy
make status                 # state = running, a live pid
make watch LIMIT=2000000
# [####################] 100.0%  t+ 2.2s  scanned to 2000000: 148933 primes
# result: {'limit': 2000000, 'primes': 148933}
make undeploy

Raise LIMIT for a longer run and the bar advances over more seconds; lower it for a quick one. The progress and log notifications arrive while the computation is happening, not after it finishes.

Troubleshooting

  • The bar jumps straight to 100%. The job finished too fast to see. Raise LIMIT (trial division makes the work grow quickly).
  • No progress appears, only the final result. The client is not passing a progress_handler, or the tool is not calling ctx.report_progress. Both are required; a handler with no reporting, or reporting with no handler, shows nothing.
  • count_primes() missing 1 required positional argument: 'ctx' in a test. Call the tool through a Client, not by importing and calling the function directly. FastMCP injects the Context only when the tool is invoked through the protocol.
  • Long jobs time out. The client has a default call timeout. Pass a larger timeout= to call_tool for very long runs, and consider chunking the work into separate calls.

Recap

You built a long-running tool that keeps the client informed:

  • The tool reports progress and logs through Context (report_progress and info) after each chunk of real work.
  • The client subscribes with a progress_handler (and optionally a log_handler) and renders a live bar from the streamed notifications.
  • Over HTTP the notifications arrive incrementally, so the call never looks frozen.
  • The tests capture the streamed events to prove progress is reported, and uv, make, and launchd run and deploy it.

Next improvements

  • Report progress from genuinely I/O-bound work (downloads, batch API calls) where feedback matters most.
  • Add cancellation: check ctx for a cancellation signal between chunks and stop early.
  • Combine with the SQLite article to persist partial results, so a long job can resume.
  • Drive the tool from an agent that surfaces progress to an end user in real time.