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. Install the extra even if you already have SQLAlchemy: SQLAlchemy 2.1 no longer pulls in greenlet by itself, and its async engine cannot run without it.

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.

Failing those, a result whose error holds a non-empty string is a soft error too, with that string as the message: {"played": False, "error": "audio output stalled"} is recorded as error: audio output stalled. Only a string with something other than whitespace counts, so {"error": None}, {"error": ""} and a statistic such as {"error": 0.02} stay ok. A dict or list there ({"error": {"code": 1}}) is not flagged either; catch that shape with a flag key, a status or your own classifier. error_keys names the keys to check, and error_keys=() turns the check off:

from fastmcp_feedback.instrumentation import instrument, soft_error_classifier

# Tools that report {"failure_reason": "..."} instead
instrument(app, result_classifier=soft_error_classifier(error_keys=("failure_reason",)))

# As before 2026.10.01: an error string alone is not a soft error
instrument(app, result_classifier=soft_error_classifier(error_keys=()))

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).

Upgrading to 2026.10.01: the error key check is on by default, so a result that carries a non-empty error string with no failing status or flag (such as {"status": "unknown_target", "error": "no object named Cube"}, or {"error": "..."} alone) is now a soft error where earlier releases recorded it as ok. Counts of NOT ok and outcome = 'soft_error' can rise after upgrading. Results that also have a failing status or flag keep their old message, such as success=False: ..., and are counted once. Pass soft_error_classifier(error_keys=()) to keep the old behavior.

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. A key used to join events to calls or to each other must be unique for at least the retention period, including across restarts. A per-boot counter (speech-1, speech-2, ...) starts over on every restart, and the second boot's speech-1 joins to the first boot's events. Use a UUID, or a token chosen at startup plus a counter (speech-<boot token>-<n>).
• 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)

To join a call to its events in SQL, have the enricher write the key into extra and match on it. This is how mcspeak reads the phases of one speak call, with SQLite:

SELECT e.occurred_at, e.attrs FROM ffb_events e
JOIN ffb_tool_calls c ON json_extract(c.extra, '$.speech_id') = e.key
WHERE e.kind = 'speech.phase' AND c.id = ?
ORDER BY e.occurred_at;

On Postgres the join condition is c.extra->>'speech_id' = e.key. Events recorded inside the call also carry its id, so e.call_id = c.id joins those without a key. ffb_tool_calls has no call_id column; its primary key is id, and ffb_events.call_id holds that value. Order events by occurred_at: ids are random UUIDs and do not sort in recording order.

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.

The prompt and the answer are not stored unless you opt in, since chat text can hold anything a user typed. With capture_llm_text=True on the middleware, record_llm_call(..., prompt=..., completion=...) stores them as prompt and completion attributes, redacted and cut to max_llm_text_chars (by default 8 * max_error_chars, so 4000 characters):

from fastmcp_feedback.instrumentation import instrument

mw = instrument(app, capture_llm_text=True)
mw.record_llm_call(
    model="claude-opus-4",
    prompt="Why does the render time out?",
    completion="The scene has 40 million polygons.",
)

With the default capture_llm_text=False both are discarded, and so are prompt and completion keys passed in attrs.

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. id is a UUID string in both tables, so sort by occurred_at (events) or started_at (calls), never by id. call_id refers to ffb_tool_calls.id but has no foreign key, 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.

Embeddings: finding similar reports and errors

"Has anyone reported this before?" is hard to answer with LIKE. An EmbeddingSink turns the text in the records into vectors and stores them, so you can search by meaning. It embeds four kinds of text:

source_type	Text	source_id
feedback	title and description of a feedback.submitted event	the feedback ref
call_error	"{tool}: {error_type}: {error_message}" of calls with outcome error or soft_error	the call id
event	the values of error_excerpt, error, message, reason in an event's attrs (event_text_keys=)	the event id
llm	"prompt: ...\ncompletion: ..." of an llm.call event, when capture_llm_text=True	the event id

An llm.call without prompt or completion counts as a plain event, so a failed LLM call's error is still embedded. A list or tuple value is embedded one item per line, a dict as key: value lines sorted by key (at any depth), rather than as its Python repr.

It needs a DatabaseSink to store into and an embedder. OpenAIEmbedder talks to any OpenAI-compatible /v1/embeddings endpoint (OpenAI, Ollama, LiteLLM, vLLM):

import os

from fastmcp_feedback import add_feedback_tools
from fastmcp_feedback.instrumentation import (
    DatabaseSink,
    EmbeddingSink,
    OpenAIEmbedder,
    instrument,
)

embedder = OpenAIEmbedder(
    os.environ["EMBED_URL"],               # e.g. https://api.example.com/v1
    api_key=os.environ["EMBED_API_KEY"],
    model="mxbai-embed-large",
    dim=1024,
)
db = DatabaseSink("sqlite+aiosqlite:///calls.db", create_tables=True, embedding_dim=1024)
mw = instrument(app, [db, EmbeddingSink(embedder, db)])
add_feedback_tools(app, database_url="sqlite:///feedback.db", instrumentation=mw)

Then search:

for hit in await mw.similar("render timed out", k=5):
    print(f"{hit['score']:.2f}", hit["source_type"], hit["source_id"], hit["text"][:60])

# Reports and errors like feedback 42, leaving 42 itself out.
related = await mw.similar_feedback("42", source_types=["feedback", "call_error"])

Each hit has source_type, source_id, text (what was embedded), score and created_at. score is 1 minus the cosine distance, clamped to [0, 1]: 1 is the same direction, 0 unrelated or opposite. min_score= drops weaker hits, and source_types= limits the search. Only vectors from the embedder's model are compared. similar_feedback uses the stored vector of that feedback, so it embeds nothing, and returns [] until the feedback has been embedded. Neither ever raises; on any failure they log and return [].

On PostgreSQL the search runs in the database (ORDER BY embedding <=> :q) on an HNSW index. pgvector applies source_types and the model filter after the index scan, so a narrow filter over a large table can return fewer than k hits; SET hnsw.iterative_scan = relaxed_order (pgvector 0.8) fixes that. Elsewhere it compares in Python against the newest scan_limit (5000) rows, and logs once when there are more.

It stays off the call path. write only picks out the text and puts it on the sink's own bounded queue (max_queue, 1000), so a slow embedding backend never holds up the dispatcher or a tool call. A background task embeds up to batch_size (32) texts per request, each request limited to timeout (30 s), and upserts the rows. A full queue drops texts and counts them in sink.dropped; failures of the embedder or the database are logged and counted in sink.errors, never raised. A text whose row already holds the same text (by text_hash) is not embedded again; a changed text replaces its row. await sink.flush() waits for the queue, and mw.aclose() lets it drain for up to timeout before stopping.

Nothing new leaves the process unredacted. Text is redacted with the middleware's Redactor (the sink adopts it when passed to instrument) and cut to max_chars (4000) before it is sent to the embedder or stored, on top of the redaction every record already had.

Feedback flows as an event. With instrumentation=mw, submit_feedback records a feedback.submitted event keyed by the feedback id, with type, title, description and submitter attributes. If you have your own feedback tools, record the same event with your own ref and it is embedded the same way:

mw.record_event(
    "feedback.submitted",
    key="bug-7Q2X",
    attrs={"type": "bug", "title": "Render hangs", "description": "Stuck at 99%"},
)

Storage. Vectors go to ffb_embeddings: id, source_type, source_id, model, dim, text_hash (sha256 of text), text, created_at and embedding, unique on (source_type, source_id, model). embedding is a pgvector vector(dim) on PostgreSQL and a JSON list of floats elsewhere. DatabaseSink(embedding_dim=...) sets dim and must equal the embedder's dim, or EmbeddingSink raises ValueError. Retention covers it: prune() deletes embeddings older than the cutoff, except those of feedback, which are kept like the calls linked to feedback.

On PostgreSQL this needs the embeddings extra (pgvector) and asyncpg:

uv add "fastmcp-feedback[instrumentation,embeddings]" asyncpg

With create_tables=True, the table is created when an EmbeddingSink first writes, not before, so hosts that do not embed never need pgvector. On PostgreSQL it first runs CREATE EXTENSION IF NOT EXISTS vector. Creating an extension needs a superuser or the database owner on most setups; without that it logs a warning, and so does the table creation if the extension is still missing. The sink keeps working and counts the failed writes, and retries the creation at most once a minute.

Upgrading: ffb_embeddings is new in 2026.09.30. build_metadata() includes it (pass embedding_dim=None to leave it out). Hosts on Alembic need a migration; for PostgreSQL, with the default prefix and 1024 dimensions:

CREATE EXTENSION IF NOT EXISTS vector;

CREATE TABLE ffb_embeddings (
    id VARCHAR(36) NOT NULL,
    source_type VARCHAR(16) NOT NULL,
    source_id VARCHAR(255) NOT NULL,
    model VARCHAR(128) NOT NULL,
    dim INTEGER NOT NULL,
    text_hash VARCHAR(64) NOT NULL,
    text TEXT NOT NULL,
    created_at TIMESTAMP WITH TIME ZONE NOT NULL,
    embedding VECTOR(1024) NOT NULL,
    PRIMARY KEY (id),
    CONSTRAINT ffb_embeddings_source_uq UNIQUE (source_type, source_id, model)
);
CREATE INDEX ffb_embeddings_hnsw ON ffb_embeddings USING hnsw (embedding vector_cosine_ops);
CREATE INDEX ix_ffb_embeddings_created_at ON ffb_embeddings (created_at);
CREATE INDEX ix_ffb_embeddings_source_id ON ffb_embeddings (source_id);
CREATE INDEX ix_ffb_embeddings_source_type ON ffb_embeddings (source_type);

Wrap it in op.execute(...) in the migration's upgrade(), and DROP TABLE ffb_embeddings in downgrade(). Autogenerate also works but renders the column as fastmcp_feedback.instrumentation.storage.EmbeddingVector(dim=1024), which ties the migration to this package.

Backfilling existing data

The sink embeds records as they arrive. Calls, events and feedback stored before it was running can be embedded afterwards with backfill_embeddings, which reads them back from the DatabaseSink's tables:

from fastmcp_feedback.instrumentation import backfill_embeddings

counts = await backfill_embeddings(mw.embedding_sink)
# {"feedback": 4, "call_error": 90, "event": 12, "llm": 0, "skipped": 0, "errors": 0}

It picks up the same four kinds of text as the live sink: calls with outcome error or soft_error, feedback.submitted events, llm.call events with a prompt or completion, and other events with any of the sink's event_text_keys. Calls recorded before the outcome column existed (2026.09.27.4) have it NULL; those with an error_message count as errors.

Each text goes through the sink's own extraction, redactor and max_chars, so it comes out with the same text, text_hash and source_id the live sink would give that record. That makes it idempotent: a second run embeds nothing and reports the rows as skipped, and records arriving later dedupe against what was backfilled. When the middleware has a custom Redactor, build the sink with it (EmbeddingSink(..., redactor=mw.redactor)), or use mw.embedding_sink, which already has it.

sources= narrows it (default: the sink's sources), since= and until= select rows by timestamp (naive means UTC), and limit= embeds at most that many texts, oldest first; running it again carries on where it stopped. Rows are read batch_size (500) at a time and embedded in chunks of the sink's batch_size, so a large table never sits in memory. A failed chunk is logged and counted in errors, never raised; only a missing table or bad arguments raise. dry_run=True counts what would be embedded without calling the embedder.

Feedback kept in your own tables rather than as events goes through backfill_feedback, which makes the same text a feedback.submitted event with that ref would:

from fastmcp_feedback.instrumentation import backfill_feedback

rows = [("bug-7Q2X", "Render hangs", "Stuck at 99%"), ("bug-8K1D", "Crash", None)]
await backfill_feedback(mw.embedding_sink, rows)  # (ref, title, description)

From a shell, the same over a database URL. The API key comes from FFB_EMBED_API_KEY (or the variable --api-key-env names), never a flag:

export FFB_EMBED_API_KEY=...
python -m fastmcp_feedback.instrumentation.backfill \
    --database-url postgresql+asyncpg://user:pass@db/app \
    --embed-base-url https://api.example.com/v1 \
    --model mxbai-embed-large --dim 1024 --prefix ffb_ \
    --since 2026-09-01 --sources call_error,feedback --dry-run

It prints the counts as JSON and exits 1 if any text failed to embed, 2 on a setup error. --create-tables creates ffb_embeddings (and on PostgreSQL the vector extension) when it is missing; without it a missing table is an error. --help lists the rest (--until, --limit, --batch-size, --max-chars, --event-text-keys).

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