instrument() and InstrumentationMiddleware

from fastmcp_feedback.instrumentation import InstrumentationMiddleware, instrument

instrument

instrument(app, sinks=None, **kwargs) -> InstrumentationMiddleware

Creates an InstrumentationMiddleware(sinks, **kwargs), adds it to app with app.add_middleware, and returns it.

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.

InstrumentationMiddleware

InstrumentationMiddleware(
    sinks=None, *, mode=None, redactor=None, meta_only_tools=(),
    identity_resolver=None, enricher=None, server_version=None,
    hook_timeout=0.25, max_queue=1000, batch_size=100, sink_timeout=10.0,
    max_error_chars=500, recent_calls=2000, correlation_timeout=1.0,
    result_classifier=DEFAULT_RESULT_CLASSIFIER, exclude_tools=(),
    sample_rates=None, random=random.random, recent_events=2000,
    capture_llm_text=False, max_llm_text_chars=None,
)

Parameters

Parameter	Default	Description
sinks	[JsonLinesSink()]	Where records go. See Sinks.
mode	FEEDBACK_INSTRUMENTATION_MODE, then "meta"	"off", "meta" or "full". Anything else raises ValueError.
redactor	Redactor()	Secret redaction for arguments, results, error text, hook output, event attributes and embedded text. See Redaction.
meta_only_tools	()	Tools never recorded above meta, even in full mode.
identity_resolver	None	(context) -> dict, may be async. Keys user_sub, caller_kind, client_id become columns; other keys go to identity.
enricher	None	(tool, args, result, context) -> dict, may be async. Output goes to extra, except server_version, which fills that column.
server_version	None	Default for the server_version column of calls and events.
hook_timeout	0.25	Seconds an async hook may take before it is skipped for that call.
max_queue	1000	Records waiting for the sinks before new ones are dropped and counted.
batch_size	100	Records handed to each sink per write.
sink_timeout	10.0	Seconds one sink write may take.
max_error_chars	500	Longest error message stored, after redaction.
recent_calls	2000	Call records kept in memory for feedback linking, including ones still queued.
correlation_timeout	1.0	Seconds the database lookup in capture_recent_calls may take.
result_classifier	DEFAULT_RESULT_CLASSIFIER	(tool, result) -> str | None. None turns soft-error detection off. See Classifier.
exclude_tools	()	Tools never recorded and never linked. The calls themselves are untouched.
sample_rates	None	Tool name to the fraction of its ok calls recorded, in (0, 1]. Other values raise ValueError. Failures are always recorded.
random	random.random	() -> float in [0, 1) deciding sampling; a call is kept when it returns less than the rate. For tests.
recent_events	2000	Events kept in memory for events_for.
capture_llm_text	False	Store prompt and completion passed to record_llm_call.
max_llm_text_chars	8 * max_error_chars	Longest prompt or completion stored.

Recording methods

Method	Returns	Description
record_event(kind, *, key=None, attrs=None, user_sub=None, call_id=None)	bool	Record an event. Synchronous, safe from any thread, never raises or blocks. False when not recorded: mode off, kind empty or over 128 characters, key over 255 characters, attrs not a mapping, a full queue, or no event loop.
record_llm_call(*, model, provider=None, duration_ms=None, input_tokens=None, output_tokens=None, ok=None, error=None, key=None, attrs=None, prompt=None, completion=None)	bool	Record an llm.call event. See Record LLM calls.

Reading methods (async)

Method	Returns	Description
events_for(key, *, kinds=None, limit=1000)	list[dict]	A key’s events, oldest first, the newest limit of them. Reads the DatabaseSink plus events still queued; without one, memory only. occurred_at is a UTC datetime.
capture_recent_calls(ctx=None, n=20, *, user_sub=None, window=timedelta(minutes=15))	list[CallLink]	Up to n calls that led up to now, chronological: same session first, then same user_sub within window. ctx is looked up when omitted; user_sub is asked of the identity resolver when omitted. Never raises; [] on failure.
link_feedback(feedback_ref, links)	bool	Store links for a feedback id of any format, replacing earlier ones. Needs a DatabaseSink. False instead of raising.
feedback_context(feedback_ref)	list[dict]	Linked calls in order, with call columns filled from the database or from memory for calls not yet written. Keys: call_id, position, rule, tool, started_at, duration_ms, ok, outcome, error_type, error_message, session_id, user_sub, caller_kind, server_version, extra. [] without a DatabaseSink.
similar(text, *, source_types=None, k=10, min_score=None)	list[dict]	Embedded texts most like text, best first. Needs an EmbeddingSink. Never raises.
similar_feedback(feedback_ref, *, k=10, source_types=None, min_score=None)	list[dict]	Texts most like an embedded feedback item, excluding the item. [] until it is embedded. Never raises.

Each similar result has source_type, source_id, text, score (1 minus cosine distance, in [0, 1]) and created_at.

Lifecycle methods (async)

Method	Description
flush()	Wait until queued records have been handed to the sinks. Does not wait for an EmbeddingSink’s own queue; use mw.embedding_sink.flush() for that.
aclose()	Flush, stop the background task, and close every sink that has aclose. Call on server shutdown.

Properties and attributes

Name	Description
database_sink	The first DatabaseSink among the sinks, or None.
embedding_sink	The first EmbeddingSink among the sinks, or None.
redactor	The Redactor in use.
mode	The resolved mode.
dispatcher	The RecordDispatcher. dispatcher.dropped counts records dropped (full queue, no event loop); dispatcher.sink_errors counts failed sink writes.

Constants

Name	Value
MODES	("off", "meta", "full")
MODE_ENV_VAR	"FEEDBACK_INSTRUMENTATION_MODE"
DEFAULT_TABLE_PREFIX	"ffb_"
DEFAULT_EMBEDDING_DIM	1024
EMBEDDING_SOURCES	("feedback", "call_error", "event", "llm")
DEFAULT_EVENT_TEXT_KEYS	("error_excerpt", "error", "message", "reason")