Test an instrumented server

MemorySink keeps every record in a list, which makes the instrumentation easy to assert on. Call tools through FastMCP’s in-process Client, flush, then look at sink.calls and sink.events.

import asyncio

from fastmcp import Client, FastMCP
from fastmcp_feedback.instrumentation import MemorySink, instrument


def make_app():
    app = FastMCP("Under Test")

    @app.tool
    def divide(a: float, b: float) -> float:
        return a / b

    sink = MemorySink()
    mw = instrument(app, [sink])
    return app, mw, sink


async def test_division_by_zero_is_recorded_as_error():
    app, mw, sink = make_app()
    async with Client(app) as client:
        await client.call_tool("divide", {"a": 1, "b": 2})
        try:
            await client.call_tool("divide", {"a": 1, "b": 0})
        except Exception:
            pass
    await mw.flush()

    assert [c.outcome for c in sink.calls] == ["ok", "error"]
    assert sink.calls[1].error_type == "ZeroDivisionError"
    await mw.aclose()


asyncio.run(test_division_by_zero_is_recorded_as_error())

With pytest and pytest-asyncio, drop the asyncio.run line and mark the test async.

Things worth knowing

• await mw.flush() waits until queued records have reached the sinks. Without it, an assertion can run before the background task has delivered anything.
• sink.records holds calls and events in arrival order; sink.calls and sink.events are the two kinds on their own.
• Records reflect what the middleware stores: in meta mode, args and result are None. Pass mode="full" to assert on redacted payloads.
• To test sampling, pass random= a function returning a fixed float. A call is kept when it returns less than the tool’s rate:

app = FastMCP("Sampled")
sink = MemorySink()
mw = instrument(app, [sink], sample_rates={"poll": 0.5}, random=lambda: 0.9)  # every ok call of poll is sampled out

• mw.dispatcher.dropped and mw.dispatcher.sink_errors count records dropped because the queue was full and sink writes that failed.