Skip to content
FastMCP Feedback

Instrument a FastMCP server in five minutes

In this tutorial you build a two-tool FastMCP server, add one line that records every tool call, call the tools, and read what was recorded. You need Python 3.11 or newer and uv.

  1. Create a project and install the package with the instrumentation extra, which adds the database sink:

    Terminal window
    uv init weather && cd weather
    uv add "fastmcp-feedback[instrumentation]"

  2. Create weather.py with a server, two tools and the instrumentation:

    from fastmcp import FastMCP
    from fastmcp_feedback.instrumentation import DatabaseSink, JsonLinesSink, instrument
    

    app = FastMCP("Weather")
    

    

    @app.tool
    def forecast(city: str) -> dict:
        """Tomorrow's high for a city."""
        if city == "Atlantis":
            return {"ok": False, "error": "no such city"}
        return {"city": city, "high_c": 21}
    

    

    @app.tool
    def alerts(region: str) -> list[str]:
        """Active weather alerts for a region."""
        raise RuntimeError(f"alert feed for {region} is down")
    

    

    mw = instrument(app, [
        JsonLinesSink(),  # one JSON line per record on stderr
        DatabaseSink("sqlite+aiosqlite:///calls.db", create_tables=True),
    ])

    instrument(app, sinks) adds an InstrumentationMiddleware to the server and returns it. Every tool on the server is now recorded, including tools added later.

  3. Add a few calls to the bottom of the same file. Client(app) connects to the server in-process, so there is nothing to start:

    import asyncio
    

    from fastmcp import Client
    

    

    async def main():
        async with Client(app) as client:
            await client.call_tool("forecast", {"city": "Lisbon"})
            await client.call_tool("forecast", {"city": "Atlantis"})
            try:
                await client.call_tool("alerts", {"region": "north"})
            except Exception as exc:
                print("client saw:", exc)
        await mw.aclose()  # flush queued records, close the sinks
    

    

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

  4. Run it:

    Terminal window
    uv run weather.py

    Among FastMCP’s own log output, stderr gets one line per call. Shortened, they look like this:

    {"record": "tool_call", "tool": "forecast", "outcome": "ok", "duration_ms": 1.4, "error_type": null, ...}
    {"record": "tool_call", "tool": "forecast", "outcome": "soft_error", "error_type": "SoftError", "error_message": "ok=False: no such city", ...}
    {"record": "tool_call", "tool": "alerts", "outcome": "error", "error_type": "RuntimeError", "error_message": "alert feed for north is down", ...}

    The second call did not raise, but its result said it failed, so it is recorded as a soft_error. The third raised, so it is an error. The client received exactly what it would have without the middleware.

  5. Read the same records from SQLite:

    import sqlite3
    

    rows = sqlite3.connect("calls.db").execute(
        "SELECT tool, outcome, error_message FROM ffb_tool_calls ORDER BY started_at"
    )
    for tool, outcome, message in rows:
        print(f"{tool:10} {outcome:11} {message or ''}")
    forecast   ok
    forecast   soft_error  ok=False: no such city
    alerts     error       alert feed for north is down

What you built

Section titled “What you built”

You have a server whose every tool call becomes a row in ffb_tool_calls, with its duration, its outcome (ok, soft_error or error) and a redacted error message. Recording happens after the tool returns, on a background task, so a slow or broken sink cannot slow down or fail a call.

To serve the tools to a real client, call app.run() instead of main(); nothing about the instrumentation changes.

Next

Section titled “Next”