FastMCP Feedback

Instrumentation and QA feedback for FastMCP servers.

Record every tool call your server handles, and let people and models report problems with it, each with a line or two of code.

Features

• Per-Call Instrumentation - Middleware records every tool call (timing, outcome, identity, payload sizes) without slowing or breaking it
• Secret Redaction - Tokens and passwords removed from recorded arguments, results and errors by default
• One-Line Integration - Add feedback tools to any FastMCP server instantly
• Modular Mixins - Selective tool registration for fine-grained control
• Multi-Database Support - SQLite (dev), PostgreSQL, MySQL (production)
• Privacy-Compliant Analytics - Optional usage insights without sensitive data
• Type-Safe - Full Pydantic validation and type hints
• Well-Tested - 98% coverage, run against FastMCP 2.12, 2.14, 3.x and 4.x

Installation

From PyPI (Recommended)

# Using uv (recommended)
uv add fastmcp-feedback

# Using pip
pip install fastmcp-feedback

# With PostgreSQL support
uv add "fastmcp-feedback[postgresql]"

From Git (Latest Development)

# Latest from main branch
uv add git+https://git.supported.systems/fastmcp-feedback/fastmcp-feedback.git

# Specific release tag
uv add git+https://git.supported.systems/fastmcp-feedback/fastmcp-feedback.git@2026.01.12

# Specific branch (for testing PRs)
uv add git+https://git.supported.systems/fastmcp-feedback/fastmcp-feedback.git@feature-branch

Local Development Install

For contributors or local modifications:

# Clone the repository
git clone git@git.supported.systems:fastmcp-feedback/fastmcp-feedback.git
cd fastmcp-feedback

# Create virtual environment and install in editable mode
uv venv
uv pip install -e ".[dev]"

# Run tests to verify
uv run pytest

Quick Start

Basic Integration (One Line)

from fastmcp import FastMCP
from fastmcp_feedback import add_feedback_tools

app = FastMCP("My Server")
add_feedback_tools(app, database_url="sqlite:///feedback.db")

# Your server now has five tools:
# - submit_feedback: bug reports, feature requests, improvements, questions
# - list_feedback: page through feedback, filtered by type or status
# - get_feedback_statistics: counts by type and status, plus the last 7 days
# - update_feedback_status: move an item through open -> in_progress -> resolved -> closed
# - delete_feedback: remove an item

Pass a database_url. Without one the tools use an in-memory SQLite database, which is handy for tests but loses everything when the process exits.

PostgreSQL

add_feedback_tools(app, database_url="postgresql://user:pass@db.example.com/feedback")

Install the driver with uv add "fastmcp-feedback[postgresql]".

Prefixed Tool Names

add_feedback_tools(app, database_url="sqlite:///feedback.db", prefix="support")
# support_submit_feedback, support_list_feedback, ...

The prefix and tool name are joined with separator (default _), so leave the trailing underscore off the prefix.

Usage Analytics

from fastmcp_feedback import FeedbackInsights

add_feedback_tools(
    app,
    database_url="sqlite:///feedback.db",
    insights=FeedbackInsights(enabled=True),
)

Analytics are off unless you pass FeedbackInsights(enabled=True) or set FEEDBACK_INSIGHTS_ENABLED=true. When enabled they record tool usage and submission metadata (type, lengths, timing), never feedback text or contact details.

Instrumentation

Record every tool call on your server:

from fastmcp import FastMCP
from fastmcp_feedback.instrumentation import instrument

app = FastMCP("My Server")
instrument(app)  # one JSON line per tool call on stderr

Each record has the tool name, start time, duration, outcome and error type, session and request ids, argument and result sizes, and whatever your hooks add. The recording happens in middleware, so it covers every tool on the server, not just the feedback ones.

It never slows or breaks a tool call. Records go to the sinks from a background task through a bounded queue, and when the queue is full they are dropped and counted instead of waited on. Errors in a sink or a hook are logged and swallowed.

Modes

Mode	Records
off	Nothing
meta (default)	Tool, timing, outcome, identity, payload sizes
full	Also the arguments and result, redacted

Set it with mode= or the FEEDBACK_INSTRUMENTATION_MODE environment variable. meta_only_tools={"run_code", "create_token"} keeps chosen tools at meta even in full mode.

Redaction

Secrets are removed from arguments, results, error messages and hook output before anything reaches a sink. It matches keys at any depth (password, token, authorization, client_secret, *_api_key, ...) and secret-shaped values anywhere in text (JWTs, Bearer ..., sk-..., GitHub, Slack, AWS and PyPI tokens, password=... in error text). Add your own:

from fastmcp_feedback.instrumentation import Redactor, instrument

instrument(
    app,
    mode="full",
    redactor=Redactor(extra_keys=["license"], extra_patterns=[r"acme_[A-Za-z0-9]{32}"]),
)

Sinks

from fastmcp_feedback.instrumentation import CallbackSink, DatabaseSink, JsonLinesSink, instrument

instrument(app, [
    JsonLinesSink(),                                   # stderr
    DatabaseSink("sqlite+aiosqlite:///calls.db", create_tables=True),
    CallbackSink(lambda records: print(len(records))),  # sync or async
])

DatabaseSink needs the instrumentation extra (uv add "fastmcp-feedback[instrumentation]"), plus asyncpg for Postgres.

Using your own database and migrations

Pass the AsyncEngine you already have. Records go to a ffb_tool_calls table (events to ffb_events; the prefix is configurable), and nothing is created at startup unless you ask. To manage the tables with Alembic, add them to your env.py:

from fastmcp_feedback.instrumentation import build_metadata

target_metadata = [Base.metadata, build_metadata(prefix="ffb_")]

from fastmcp_feedback.instrumentation import DatabaseSink, instrument

sink = DatabaseSink(engine, prefix="ffb_")  # your engine is never disposed
instrument(app, [sink])

Errors returned as data

Many tools report failure in their result instead of raising, for example {"status": "failed", "error": "disk full"} or {"success": False, ...}. A result classifier catches these, so error rates count them. Each record has an outcome:

outcome	ok	Meaning
ok	true	The call succeeded
soft_error	false	The tool returned a result the classifier flagged
error	false	The tool raised

Soft errors get error_type="SoftError" and an error_message such as status=failed: disk full, redacted and truncated like exception text.

The default classifier flags a dict result when ok or success is exactly False, when status or state is error, failed or failure (any case), or when the tool marked the result as an MCP error (isError) without raising. The message comes from the first of error, message, detail and reason. Add your own status values, or pass any (tool, result) -> str | None:

from fastmcp_feedback.instrumentation import instrument, soft_error_classifier

# blender-mcp answers {"status": "unknown_target", ...} for a missing object
instrument(app, result_classifier=soft_error_classifier(extra_statuses=("unknown_target",)))

from fastmcp_feedback.instrumentation import instrument

def classify(tool, result):
    """None means ok; a string marks a soft error and is its message."""
    if tool == "search" and isinstance(result, dict) and result.get("hits") == []:
        return "no hits"
    return None

instrument(app, result_classifier=classify)

result_classifier=None turns classification off. The classifier only sees calls that did not raise, gets the same payload the record stores (structured content when present), and if it raises, the error is logged and the call is recorded as ok. It never changes what the client receives.

Upgrading: soft errors set ok to false, so existing WHERE NOT ok counts include them from this release on. Filter on outcome = 'error' for exceptions only. The outcome column is new in 2026.09.27.4; it is nullable, and rows written by earlier versions have NULL there. Hosts that manage the schema with Alembic need a migration adding it (with its index) to ffb_tool_calls, and a table made earlier by create_tables=True needs ALTER TABLE ffb_tool_calls ADD COLUMN outcome VARCHAR(16), since existing tables are not altered. Until the column exists, database writes fail and those records are dropped (logged).

Identity and enrichment hooks

from fastmcp_feedback.instrumentation import instrument

def who(context):
    """Who made this call. May be async."""
    user = context.fastmcp_context  # read your auth state from here
    return {"user_sub": "user-123", "caller_kind": "pat:7", "client_id": "cli"}

async def enrich(tool, args, result, context):
    """Extra indexed fields. May be async."""
    return {"job_id": (result or {}).get("job_id"), "server_version": "1.4.0"}

instrument(app, identity_resolver=who, enricher=enrich)

user_sub, caller_kind and client_id become columns; other identity keys go in identity. Enricher output goes in extra, except server_version, which has its own column. Async hooks that take longer than hook_timeout (0.25 s by default) are skipped for that call.

Linking feedback to the calls behind it

When a model reports "the render tool hung", the report is far more useful with the calls that led up to it attached. Pass the middleware to add_feedback_tools and every submit_feedback does this for you:

from fastmcp_feedback import add_feedback_tools
from fastmcp_feedback.instrumentation import DatabaseSink, instrument

mw = instrument(app, [DatabaseSink("sqlite+aiosqlite:///calls.db", create_tables=True)])
add_feedback_tools(app, database_url="sqlite:///feedback.db", instrumentation=mw)
# submit_feedback now answers {"feedback_id": ..., "linked_calls": 3, ...}

Read the linked calls back, in order, with their outcomes:

for call in await mw.feedback_context(feedback_id):
    print(call["position"], call["tool"], call["outcome"], call["error_type"], call["rule"])

If you have your own feedback tools, use the same pieces directly. Feedback ids can be anything (bug-7Q2X, 42):

from fastmcp import Context

@app.tool
async def report_problem(summary: str, ctx: Context) -> dict:
    ref = save_my_feedback(summary)                 # your storage, your id
    links = await mw.capture_recent_calls(ctx, n=20)
    await mw.link_feedback(ref, links)
    return {"id": ref, "linked_calls": len(links)}

Calls are matched in two ways, and each link records which one (rule):

• session: the same MCP client session. Over HTTP this is the mcp-session-id; over stdio it is the server process, which serves one client.
• user_window: the same user_sub within the last 15 minutes (window=), for clients that open a new session per call (personal access tokens, CI). This needs an identity_resolver.

Session matches come first, and user matches fill any remaining slots. Calls still waiting in the queue are included, and with a DatabaseSink, so are calls from other workers or from before a restart.

Links live in a ffb_feedback_call_links table next to ffb_tool_calls. build_metadata() includes it, so hosts that manage the schema with Alembic need one new migration before calling link_feedback.

Stateless clients: clients on MCP protocol 2026-07-28, including Claude Code and FastMCP 4's own Client, do not open sessions over HTTP. Their calls link by user only, so configure an identity_resolver for HTTP servers. Stdio servers link by session with every client.

Busy tools: exclusion and sampling

A polling tool can make up most of the rows. Leave it out entirely, or keep a fraction of its successful calls:

from fastmcp_feedback.instrumentation import instrument

instrument(
    app,
    exclude_tools={"heartbeat"},                     # never recorded
    sample_rates={"pending_dispatches": 0.01},       # 1% of ok calls
)

Excluded tools are not recorded at all and are left out of feedback links. Sampled tools keep a random fraction of their ok calls, but failures are always recorded: a call whose outcome is error or soft_error is kept whatever the rate. Calls sampled out cost no hook or redaction work and are left out of feedback links too. Rates must be in (0, 1]; anything else raises ValueError. Neither setting changes what the client receives.

Sampled-in ok rows carry the rate in a sample_rate column; everything recorded unconditionally, including failures of sampled tools, has NULL there. So each row stands for 1 / coalesce(sample_rate, 1) calls, and this estimates the true number of calls per tool:

SELECT tool, SUM(1.0 / COALESCE(sample_rate, 1)) AS est_calls
FROM ffb_tool_calls GROUP BY tool;

Retention

ffb_tool_calls grows with every call. Give the sink a retention period and it deletes older rows as it goes:

from datetime import timedelta
from fastmcp_feedback.instrumentation import DatabaseSink, instrument

sink = DatabaseSink(engine, prefix="ffb_", retention=timedelta(days=30))
mw = instrument(app, [sink])

Pruning runs in the background writer after an insert, at most once per prune_interval (an hour by default; the first write after startup prunes). It deletes prune_batch rows (5000) per transaction, so it never holds a long lock, and a failing prune is logged without affecting the insert. Calls linked to feedback are never pruned, and neither are the links, so a report keeps its evidence. Without retention, nothing is deleted.

To prune on your own schedule, call it directly; it returns the rows deleted:

from datetime import UTC, datetime, timedelta

deleted = await sink.prune()                                          # now - retention
deleted = await sink.prune(older_than=datetime.now(UTC) - timedelta(days=7))

Automatic pruning runs inside a sink write, so it is bounded by the dispatcher's sink_timeout (10 s). Batches finished before a timeout stay deleted and the rest goes on the next run, but for a large backlog it is quicker to run await sink.prune() once yourself.

Upgrading: sample_rate is a new nullable FLOAT column in 2026.09.28. Hosts that manage the schema with Alembic need a migration adding it, and a table made earlier by create_tables=True needs ALTER TABLE ffb_tool_calls ADD COLUMN sample_rate FLOAT (DOUBLE PRECISION on Postgres also works), since existing tables are not altered. Until the column exists, database writes fail and those records are dropped (logged). Retention needs no schema change.

Events: things that aren't tool calls

Some things worth recording happen outside a tool call, or inside one without being one: a background job changing state minutes after the tool that started it returned, or an LLM request a tool makes on the caller's behalf. Record them as events:

from fastmcp_feedback.instrumentation import DatabaseSink, instrument

mw = instrument(app, [DatabaseSink("sqlite+aiosqlite:///calls.db", create_tables=True)])

@app.tool
async def submit(case: str) -> dict:
    job_id = f"job-{case}"
    mw.record_event("job.submitted", key=job_id, attrs={"case": case, "nproc": 8})
    return {"job_id": job_id}

# Later, from a watcher task or thread, with no tool call running:
mw.record_event("job.finished", key="job-wing", attrs={"duration_s": 3512, "converged": True})

record_event(kind, *, key=None, attrs=None, user_sub=None, call_id=None) returns True when the event was queued and False when it was not. It never raises and never blocks, and it takes the same path as call records: the same background queue, the same drop-and-count policy when the queue is full.

• kind is a dotted name (job.submitted, llm.call), a non-empty string of at most 128 characters.
• key is what you join on, such as a job id. It is stored as a string, at most 255 characters.
• attrs is any JSON-like dict. It is redacted like call arguments, then made JSON-safe: NaN and infinities become the strings "nan"/"inf" (Postgres JSONB rejects them), datetimes become ISO strings, anything else unknown becomes str(value).
• Inside a tool call, the event gets that call's id as call_id and its session, so it joins to the call's row in ffb_tool_calls. Outside a call, call_id is None. A task started by a tool that records after the tool has returned counts as outside. Pass call_id= to set it yourself.
• From any thread. Called off the server's event loop (a watcher thread, a sync tool running in a worker thread), the event is handed to the loop with call_soon_threadsafe. The loop is known once the server has handled a tool call or recorded something on it; before that, a call from a thread with no event loop of its own returns False and counts a drop.
• In mode off it does nothing and returns False.

Read a key's events back, oldest first, with occurred_at as UTC:

for ev in await mw.events_for("job-wing"):
    print(ev["occurred_at"], ev["kind"], ev["call_id"], ev["attrs"])

steps = await mw.events_for("job-wing", kinds=["step.started", "step.finished"], limit=100)

events_for reads the DatabaseSink and adds events still waiting in the queue, so an event is visible as soon as record_event returns. With more than limit (1000) matches it returns the newest limit. Without a DatabaseSink it answers from the last recent_events (2000) events in memory; a failing database read is logged and memory is used.

LLM calls

record_llm_call records an event of kind llm.call. Call it from inside the tool that made the request and it links to that call:

mw.record_llm_call(
    model="claude-opus-4",
    provider="anthropic",
    duration_ms=812.5,
    input_tokens=1200,
    output_tokens=90,
    attrs={"cache_read_input_tokens": 1000},
)

Its attributes use plain names, which are easier to query from SQL than dotted ones (attrs->>'model' on Postgres, json_extract(attrs, '$.model') on SQLite):

Attribute	OTel GenAI equivalent	Notes
model	gen_ai.request.model	Always set
provider	gen_ai.provider.name	When given
duration_ms	gen_ai.client.operation.duration (seconds)	When given
input_tokens	gen_ai.usage.input_tokens	When given
output_tokens	gen_ai.usage.output_tokens	When given
ok		Defaults to error is None
error		Redacted, truncated to max_error_chars
error_type	error.type	When error is an exception

Anything in attrs is kept alongside, but these names win. key= works as for record_event, for example a chat turn id.

Storage and sinks

Events go to a ffb_events table: id, occurred_at, kind, key, call_id, session_id, user_sub, caller_kind, client_id, server_version, attrs (JSON; JSONB on Postgres), indexed on occurred_at, kind, key, call_id and user_sub. It has no foreign key to ffb_tool_calls, since calls are pruned and may be written later.

Sinks get calls and events in the same batches. DatabaseSink writes each to its table, calls first in their own transaction, so a database without ffb_events still stores calls. JsonLinesSink adds "record": "tool_call" or "record": "event" to every line. A CallbackSink callback receives both kinds (record.record_type tells them apart), and MemorySink has .calls and .events next to .records.

Retention covers events too: prune() deletes events whose occurred_at is before the cutoff, in the same batches, and returns the rows deleted from both tables (each table's count is logged). Exclusion and sampling apply to tool calls only; events are always recorded. A call of a sampled tool that recorded an event is always kept, with no sample_rate, so the event's call_id has a row to join to. Events recorded inside an excluded tool have no call_id.

Upgrading: ffb_events is a new table in this release. Hosts that manage the schema with Alembic need a migration creating it (build_metadata() includes it). create_tables=True creates it on first use, since it creates missing tables. Until the table exists, event writes fail and those events are dropped (logged); tool calls and pruning are unaffected.

Middleware order and shutdown

FastMCP runs the first middleware added as the outermost. Call instrument() before adding other middleware to include their time and record calls they reject; call it last to time only the tool. On shutdown, await mw.aclose() flushes queued records and closes the sinks.

Advanced Usage

Mixin Architecture

Pick only the tool groups you want to expose:

from fastmcp_feedback import (
    ManagementMixin,
    RetrievalMixin,
    SubmissionMixin,
    get_database_session,
)

db = get_database_session("sqlite:///feedback.db")

# Public server: submission only
SubmissionMixin(db).register_tools(app)

# Reporting tools under a prefix: analytics_list_feedback, ...
RetrievalMixin(db).register_tools(app, prefix="analytics")

# Admin server gets the workflow tools
ManagementMixin(db).register_tools(admin_app, prefix="admin")

Available Mixins

Mixin	Tools	Use Case
SubmissionMixin	submit_feedback	Public feedback collection
RetrievalMixin	list_feedback, get_feedback_statistics	Dashboards, reporting
ManagementMixin	update_feedback_status, delete_feedback	Admin workflow

Multi-Tenant Pattern

# Each tenant gets its own database
def add_tenant_tools(tenant_id: str):
    db = get_database_session(f"sqlite:///data/{tenant_id}.db")
    SubmissionMixin(db).register_tools(app, prefix=f"tenant_{tenant_id}")

Server Composition

from fastmcp_feedback import create_feedback_server

feedback_server = create_feedback_server(
    "Feedback API", database_url="sqlite:///feedback.db"
)

# Tools appear on main_app as feedback_submit_feedback, feedback_list_feedback, ...
main_app.mount(feedback_server, "feedback")

mount() works on every supported FastMCP version. import_server() was removed in FastMCP 4.

Compatibility

Tested against FastMCP 2.12, 2.14, 3.x and 4.x on every commit. The dependency is declared as fastmcp>=2.12.2,<5, and the ceiling moves up once a new major version passes the suite.

API Reference

add_feedback_tools(mcp, database_url=None, insights=None, prefix="", separator="_", instrumentation=None)

Add all five feedback tools to a FastMCP server.

Parameters:

• mcp: FastMCP server instance
• database_url: SQLAlchemy URL. Defaults to in-memory SQLite (sqlite:///:memory:)
• insights: a FeedbackInsights instance. Defaults to one with analytics disabled
• prefix: prepended to every tool name
• separator: placed between prefix and the tool name
• instrumentation: the middleware returned by instrument(). When given, submit_feedback links the calls that preceded it to the new item and returns linked_calls (see Linking feedback to the calls behind it)

Feedback Types

• bug - something is broken
• feature - a new capability
• improvement - make an existing capability better
• question - unclear behavior or usage

Status Workflow

open → in_progress → resolved → closed

Documentation

Full documentation: https://fastmcp-feedback.l.supported.systems

• Quick Start Tutorial
• Architecture
• API Reference
• Integration Patterns

Examples

The examples/ directory has runnable servers, each with a walkthrough:

• simple_integration.py: the one-line integration
• mcp_server.py: a production-style server configured from environment variables
• demo_server.py: every feature, including server composition and sample data

git clone https://git.supported.systems/fastmcp-feedback/fastmcp-feedback.git
cd fastmcp-feedback
uv run python examples/simple_integration.py

Contributing

We welcome contributions! FastMCP Feedback is designed to be contributor-friendly.

Quick Contribution Setup

# Fork and clone
git clone git@git.supported.systems:YOUR_USERNAME/fastmcp-feedback.git
cd fastmcp-feedback

# Install with dev dependencies
uv venv
uv pip install -e ".[dev]"

# Create feature branch
git checkout -b feature/your-feature

# Make changes, run tests
uv run pytest
uv run ruff check .

# Submit PR

See CONTRIBUTING.md for detailed guidelines.

Versioning

This project uses calendar versioning (CalVer): YYYY.MM.DD

• 2026.01.12 = Release on January 12, 2026
• Multiple releases on same day: 2026.01.12.1, 2026.01.12.2

This makes it clear when each release was made and simplifies dependency management.

License

MIT License - see LICENSE for details.

Links

• Documentation: https://fastmcp-feedback.l.supported.systems
• Repository: https://git.supported.systems/fastmcp-feedback/fastmcp-feedback
• Issues: https://git.supported.systems/fastmcp-feedback/fastmcp-feedback/issues
• FastMCP: https://gofastmcp.com