Compare commits
7
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
472071730e | ||
|
|
9ddd929a13 | ||
|
|
052862f4fc | ||
|
|
975c24c766 | ||
|
|
a530634eb9 | ||
|
|
4b16764faa | ||
|
|
58e531019d |
@@ -5,6 +5,79 @@ All notable changes to this project will be documented in this file.
|
|||||||
The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/),
|
The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/),
|
||||||
and this project adheres to [Calendar Versioning](https://calver.org/) (YYYY.MM.DD).
|
and this project adheres to [Calendar Versioning](https://calver.org/) (YYYY.MM.DD).
|
||||||
|
|
||||||
|
## [2026.09.27.4] - 2026-09-27
|
||||||
|
|
||||||
|
Count errors that tools return as data instead of raising.
|
||||||
|
|
||||||
|
### Added
|
||||||
|
- Records have an `outcome`: `ok`, `soft_error` (the tool returned a result
|
||||||
|
flagged as a failure) or `error` (the tool raised). Soft errors get
|
||||||
|
`error_type="SoftError"` and an `error_message` built from the result, such
|
||||||
|
as `status=failed: disk full`, redacted and truncated to `max_error_chars`.
|
||||||
|
- `soft_error_classifier(statuses=..., status_keys=..., flag_keys=...,
|
||||||
|
extra_statuses=...)` builds a classifier that flags dict results where `ok`
|
||||||
|
or `success` is `False`, where `status` or `state` is `error`, `failed` or
|
||||||
|
`failure`, or that the tool marked `isError` without raising.
|
||||||
|
`DEFAULT_RESULT_CLASSIFIER` is `soft_error_classifier()`.
|
||||||
|
- `InstrumentationMiddleware(..., result_classifier=...)` takes any
|
||||||
|
`(tool, result) -> str | None`; `None` turns classification off. A
|
||||||
|
classifier that raises is logged and the call is recorded as `ok`.
|
||||||
|
- `feedback_context()` rows include `outcome`.
|
||||||
|
|
||||||
|
### Changed
|
||||||
|
- `ok` is now `outcome == "ok"`, so it is also false for soft errors. Existing
|
||||||
|
`WHERE NOT ok` queries and error rates will rise; filter on
|
||||||
|
`outcome = 'error'` for exceptions only. `CallLink.ok` follows.
|
||||||
|
- The package's own feedback tools answer `{"success": False, ...}` on failure,
|
||||||
|
so those calls are now recorded as soft errors.
|
||||||
|
|
||||||
|
### Migration
|
||||||
|
- New nullable, indexed `outcome` column (`String(16)`) on
|
||||||
|
`{prefix}tool_calls`. Hosts that manage the schema with Alembic need a
|
||||||
|
migration adding it. `create_tables=True` does not alter a table that
|
||||||
|
already exists, so add the column by hand there
|
||||||
|
(`ALTER TABLE ffb_tool_calls ADD COLUMN outcome VARCHAR(16)` plus an index).
|
||||||
|
Until the column exists, `DatabaseSink` writes fail and those records are
|
||||||
|
dropped (logged). Rows from earlier versions keep NULL.
|
||||||
|
|
||||||
|
## [2026.09.27.3] - 2026-09-27
|
||||||
|
|
||||||
|
Link a feedback item to the tool calls that preceded it.
|
||||||
|
|
||||||
|
### Added
|
||||||
|
- `InstrumentationMiddleware.capture_recent_calls(ctx, n=20)` returns the
|
||||||
|
calls that led up to now. It matches the same client session first
|
||||||
|
(`session` rule), then fills remaining slots with calls by the same
|
||||||
|
`user_sub` inside `window` (`user_window` rule, 15 minutes by default), and
|
||||||
|
each link records which rule matched.
|
||||||
|
- Calls are kept in an in-memory buffer as they happen, so calls still
|
||||||
|
waiting in the sink queue count. With a `DatabaseSink` the lookup falls back
|
||||||
|
to the database, which covers calls from other workers and from before a
|
||||||
|
restart.
|
||||||
|
- `link_feedback(ref, links)` stores links against a feedback id of any
|
||||||
|
format (`bug-7Q2X`, `42`) in a new `{prefix}feedback_call_links` table.
|
||||||
|
`build_metadata()` includes it; hosts that manage the schema with Alembic
|
||||||
|
need a new migration.
|
||||||
|
- `feedback_context(ref)` returns the linked calls in order with tool,
|
||||||
|
outcome, error type and matching rule, filling in calls not yet written
|
||||||
|
from memory.
|
||||||
|
- `add_feedback_tools(..., instrumentation=mw)`: `submit_feedback` links the
|
||||||
|
preceding calls automatically and returns `linked_calls` in its response.
|
||||||
|
- `session_key(ctx)` gives a per-client key that stays stable across calls:
|
||||||
|
the HTTP `mcp-session-id` header when present, then one key per process for
|
||||||
|
stdio, then `ctx.session_id`.
|
||||||
|
|
||||||
|
### Fixed
|
||||||
|
- With `create_tables=True`, the background writer and `link_feedback` could
|
||||||
|
both run `create_all` at once and fail with "table already exists". Table
|
||||||
|
creation is now serialized by a lock.
|
||||||
|
|
||||||
|
### Note
|
||||||
|
- Clients on MCP protocol 2026-07-28 (Claude Code, FastMCP 4's `Client`)
|
||||||
|
open no HTTP sessions, so their calls link by user only. HTTP servers should
|
||||||
|
configure an `identity_resolver`. Stdio servers link by session with every
|
||||||
|
client.
|
||||||
|
|
||||||
## [2026.09.27.2] - 2026-09-27
|
## [2026.09.27.2] - 2026-09-27
|
||||||
|
|
||||||
### Fixed
|
### Fixed
|
||||||
|
|||||||
@@ -128,7 +128,7 @@ app = FastMCP("My Server")
|
|||||||
instrument(app) # one JSON line per tool call on stderr
|
instrument(app) # one JSON line per tool call on stderr
|
||||||
```
|
```
|
||||||
|
|
||||||
Each record has the tool name, start time, duration, success or error type,
|
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
|
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,
|
add. The recording happens in middleware, so it covers every tool on the server,
|
||||||
not just the feedback ones.
|
not just the feedback ones.
|
||||||
@@ -201,6 +201,62 @@ sink = DatabaseSink(engine, prefix="ffb_") # your engine is never disposed
|
|||||||
instrument(app, [sink])
|
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`:
|
||||||
|
|
||||||
|
```python
|
||||||
|
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",)))
|
||||||
|
```
|
||||||
|
|
||||||
|
```python
|
||||||
|
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
|
### Identity and enrichment hooks
|
||||||
|
|
||||||
```python
|
```python
|
||||||
@@ -223,6 +279,63 @@ go in `identity`. Enricher output goes in `extra`, except `server_version`,
|
|||||||
which has its own column. Async hooks that take longer than `hook_timeout`
|
which has its own column. Async hooks that take longer than `hook_timeout`
|
||||||
(0.25 s by default) are skipped for that call.
|
(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:
|
||||||
|
|
||||||
|
```python
|
||||||
|
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:
|
||||||
|
|
||||||
|
```python
|
||||||
|
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`):
|
||||||
|
|
||||||
|
```python
|
||||||
|
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.
|
||||||
|
|
||||||
### Middleware order and shutdown
|
### Middleware order and shutdown
|
||||||
|
|
||||||
FastMCP runs the first middleware added as the outermost. Call `instrument()`
|
FastMCP runs the first middleware added as the outermost. Call `instrument()`
|
||||||
@@ -297,7 +410,7 @@ version passes the suite.
|
|||||||
|
|
||||||
## API Reference
|
## API Reference
|
||||||
|
|
||||||
### `add_feedback_tools(mcp, database_url=None, insights=None, prefix="", separator="_")`
|
### `add_feedback_tools(mcp, database_url=None, insights=None, prefix="", separator="_", instrumentation=None)`
|
||||||
|
|
||||||
Add all five feedback tools to a FastMCP server.
|
Add all five feedback tools to a FastMCP server.
|
||||||
|
|
||||||
@@ -307,6 +420,9 @@ Add all five feedback tools to a FastMCP server.
|
|||||||
- `insights`: a `FeedbackInsights` instance. Defaults to one with analytics disabled
|
- `insights`: a `FeedbackInsights` instance. Defaults to one with analytics disabled
|
||||||
- `prefix`: prepended to every tool name
|
- `prefix`: prepended to every tool name
|
||||||
- `separator`: placed between `prefix` and the 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](#linking-feedback-to-the-calls-behind-it))
|
||||||
|
|
||||||
### Feedback Types
|
### Feedback Types
|
||||||
|
|
||||||
@@ -330,10 +446,17 @@ Full documentation: https://fastmcp-feedback.l.supported.systems
|
|||||||
|
|
||||||
## Examples
|
## Examples
|
||||||
|
|
||||||
See the [fastmcp-feedback-showcase](https://git.supported.systems/fastmcp-feedback/fastmcp-example) repository for comprehensive examples demonstrating all capabilities:
|
The [`examples/`](https://git.supported.systems/fastmcp-feedback/fastmcp-feedback/src/branch/main/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
|
||||||
|
|
||||||
```bash
|
```bash
|
||||||
uvx fastmcp-feedback-showcase # Run full demo
|
git clone https://git.supported.systems/fastmcp-feedback/fastmcp-feedback.git
|
||||||
|
cd fastmcp-feedback
|
||||||
|
uv run python examples/simple_integration.py
|
||||||
```
|
```
|
||||||
|
|
||||||
## Contributing
|
## Contributing
|
||||||
@@ -381,5 +504,4 @@ MIT License - see [LICENSE](LICENSE) for details.
|
|||||||
- **Documentation**: https://fastmcp-feedback.l.supported.systems
|
- **Documentation**: https://fastmcp-feedback.l.supported.systems
|
||||||
- **Repository**: https://git.supported.systems/fastmcp-feedback/fastmcp-feedback
|
- **Repository**: https://git.supported.systems/fastmcp-feedback/fastmcp-feedback
|
||||||
- **Issues**: https://git.supported.systems/fastmcp-feedback/fastmcp-feedback/issues
|
- **Issues**: https://git.supported.systems/fastmcp-feedback/fastmcp-feedback/issues
|
||||||
- **Showcase**: https://git.supported.systems/fastmcp-feedback/fastmcp-example
|
|
||||||
- **FastMCP**: https://gofastmcp.com
|
- **FastMCP**: https://gofastmcp.com
|
||||||
|
|||||||
+1
-1
@@ -4,7 +4,7 @@ build-backend = "hatchling.build"
|
|||||||
|
|
||||||
[project]
|
[project]
|
||||||
name = "fastmcp-feedback"
|
name = "fastmcp-feedback"
|
||||||
version = "2026.09.27.2" # YYYYMMDD.patch versioning - update for each release
|
version = "2026.09.27.4" # YYYYMMDD.patch versioning - update for each release
|
||||||
description = "Production-ready feedback collection system for FastMCP servers"
|
description = "Production-ready feedback collection system for FastMCP servers"
|
||||||
readme = "README.md"
|
readme = "README.md"
|
||||||
license = "MIT"
|
license = "MIT"
|
||||||
|
|||||||
@@ -44,7 +44,7 @@ Quick Start:
|
|||||||
import logging
|
import logging
|
||||||
|
|
||||||
# Version information - CalVer YYYY.MM.DD
|
# Version information - CalVer YYYY.MM.DD
|
||||||
__version__ = "2026.09.27.2"
|
__version__ = "2026.09.27.4"
|
||||||
__author__ = "Ryan Malloy"
|
__author__ = "Ryan Malloy"
|
||||||
__email__ = "ryan@supported.systems"
|
__email__ = "ryan@supported.systems"
|
||||||
|
|
||||||
|
|||||||
@@ -4,8 +4,14 @@ from fastmcp_feedback.instrumentation import instrument, DatabaseSink
|
|||||||
|
|
||||||
mw = instrument(app) # every tool call as JSON lines on stderr
|
mw = instrument(app) # every tool call as JSON lines on stderr
|
||||||
mw = instrument(app, [DatabaseSink(engine)], identity_resolver=who)
|
mw = instrument(app, [DatabaseSink(engine)], identity_resolver=who)
|
||||||
|
mw = instrument(app, result_classifier=soft_error_classifier(extra_statuses=("unknown_target",)))
|
||||||
"""
|
"""
|
||||||
|
|
||||||
|
from .classify import (
|
||||||
|
DEFAULT_RESULT_CLASSIFIER,
|
||||||
|
ResultClassifier,
|
||||||
|
soft_error_classifier,
|
||||||
|
)
|
||||||
from .dispatch import RecordDispatcher
|
from .dispatch import RecordDispatcher
|
||||||
from .middleware import MODE_ENV_VAR, MODES, InstrumentationMiddleware, instrument
|
from .middleware import MODE_ENV_VAR, MODES, InstrumentationMiddleware, instrument
|
||||||
from .record import ToolCallRecord
|
from .record import ToolCallRecord
|
||||||
@@ -21,6 +27,7 @@ from .storage import DEFAULT_TABLE_PREFIX, DatabaseSink, build_metadata
|
|||||||
__all__ = [
|
__all__ = [
|
||||||
"CallbackSink",
|
"CallbackSink",
|
||||||
"DatabaseSink",
|
"DatabaseSink",
|
||||||
|
"DEFAULT_RESULT_CLASSIFIER",
|
||||||
"DEFAULT_SECRET_KEYS",
|
"DEFAULT_SECRET_KEYS",
|
||||||
"DEFAULT_SECRET_PATTERNS",
|
"DEFAULT_SECRET_PATTERNS",
|
||||||
"DEFAULT_TABLE_PREFIX",
|
"DEFAULT_TABLE_PREFIX",
|
||||||
@@ -32,8 +39,10 @@ __all__ = [
|
|||||||
"RecordDispatcher",
|
"RecordDispatcher",
|
||||||
"REDACTED",
|
"REDACTED",
|
||||||
"Redactor",
|
"Redactor",
|
||||||
|
"ResultClassifier",
|
||||||
"Sink",
|
"Sink",
|
||||||
"ToolCallRecord",
|
"ToolCallRecord",
|
||||||
"build_metadata",
|
"build_metadata",
|
||||||
"instrument",
|
"instrument",
|
||||||
|
"soft_error_classifier",
|
||||||
]
|
]
|
||||||
|
|||||||
@@ -0,0 +1,97 @@
|
|||||||
|
"""Spotting tools that report failure as data instead of raising.
|
||||||
|
|
||||||
|
A result classifier is ``(tool, result) -> str | None``. ``result`` is the
|
||||||
|
recorded payload: the structured content when there is one, otherwise the
|
||||||
|
content items. None means the call succeeded; a string marks it as a soft
|
||||||
|
error and becomes the record's ``error_message`` (after redaction).
|
||||||
|
|
||||||
|
A result the tool marked as an MCP error (``isError``) without raising reaches
|
||||||
|
the classifier as a dict with ``"isError": True``, alongside the payload's own
|
||||||
|
keys or with the payload under ``"content"``.
|
||||||
|
"""
|
||||||
|
|
||||||
|
from collections.abc import Callable, Iterable
|
||||||
|
from typing import Any
|
||||||
|
|
||||||
|
ResultClassifier = Callable[[str, Any], str | None]
|
||||||
|
|
||||||
|
MESSAGE_KEYS = ("error", "message", "detail", "reason")
|
||||||
|
ERROR_FLAG_KEYS = ("isError", "is_error")
|
||||||
|
|
||||||
|
|
||||||
|
def _message(result: dict, depth: int = 0) -> str | None:
|
||||||
|
for key in MESSAGE_KEYS:
|
||||||
|
value = result.get(key)
|
||||||
|
if value is None or value == "":
|
||||||
|
continue
|
||||||
|
if isinstance(value, dict):
|
||||||
|
# {"error": {"code": 404, "message": "..."}}
|
||||||
|
if depth == 0 and (inner := _message(value, 1)):
|
||||||
|
return inner
|
||||||
|
continue
|
||||||
|
if isinstance(value, list | tuple):
|
||||||
|
continue
|
||||||
|
return str(value)
|
||||||
|
return None
|
||||||
|
|
||||||
|
|
||||||
|
def _content_text(content: Any) -> str | None:
|
||||||
|
if isinstance(content, str):
|
||||||
|
return content
|
||||||
|
if not isinstance(content, list | tuple):
|
||||||
|
return None
|
||||||
|
texts = [
|
||||||
|
item if isinstance(item, str) else item.get("text")
|
||||||
|
for item in content
|
||||||
|
if isinstance(item, str | dict)
|
||||||
|
]
|
||||||
|
return " ".join(t for t in texts if t) or None
|
||||||
|
|
||||||
|
|
||||||
|
def _describe(label: str, detail: str | None) -> str:
|
||||||
|
return f"{label}: {detail}" if detail else label
|
||||||
|
|
||||||
|
|
||||||
|
def soft_error_classifier(
|
||||||
|
*,
|
||||||
|
statuses: Iterable[str] = ("error", "failed", "failure"),
|
||||||
|
status_keys: Iterable[str] = ("status", "state"),
|
||||||
|
flag_keys: Iterable[str] = ("ok", "success"),
|
||||||
|
extra_statuses: Iterable[str] = (),
|
||||||
|
) -> ResultClassifier:
|
||||||
|
"""Build a classifier for dict results that describe their own failure.
|
||||||
|
|
||||||
|
A dict result is a soft error when any of ``flag_keys`` is exactly False,
|
||||||
|
when any of ``status_keys`` holds one of ``statuses`` or
|
||||||
|
``extra_statuses`` (case-insensitive), or when the tool marked it
|
||||||
|
``isError``. The message is the first of ``error``, ``message``,
|
||||||
|
``detail`` and ``reason`` present, e.g. ``status=failed: disk full``.
|
||||||
|
|
||||||
|
>>> classify = soft_error_classifier(extra_statuses=("unknown_target",))
|
||||||
|
>>> classify("render", {"status": "unknown_target", "error": "no such object"})
|
||||||
|
'status=unknown_target: no such object'
|
||||||
|
"""
|
||||||
|
bad = frozenset(s.lower() for s in (*statuses, *extra_statuses))
|
||||||
|
status_keys = tuple(status_keys)
|
||||||
|
flag_keys = tuple(flag_keys)
|
||||||
|
|
||||||
|
def classify(tool: str, result: Any) -> str | None:
|
||||||
|
if not isinstance(result, dict):
|
||||||
|
return None
|
||||||
|
for key in status_keys:
|
||||||
|
value = result.get(key)
|
||||||
|
if isinstance(value, str) and value.lower() in bad:
|
||||||
|
return _describe(f"{key}={value}", _message(result))
|
||||||
|
for key in flag_keys:
|
||||||
|
if key in result and result[key] is False:
|
||||||
|
return _describe(f"{key}=False", _message(result))
|
||||||
|
for key in ERROR_FLAG_KEYS:
|
||||||
|
if result.get(key) is True:
|
||||||
|
detail = _message(result) or _content_text(result.get("content"))
|
||||||
|
return _describe("isError", detail)
|
||||||
|
return None
|
||||||
|
|
||||||
|
return classify
|
||||||
|
|
||||||
|
|
||||||
|
DEFAULT_RESULT_CLASSIFIER = soft_error_classifier()
|
||||||
@@ -0,0 +1,106 @@
|
|||||||
|
"""Linking feedback to the tool calls that came before it."""
|
||||||
|
|
||||||
|
from collections.abc import Iterable
|
||||||
|
from dataclasses import asdict, dataclass
|
||||||
|
from datetime import UTC, datetime
|
||||||
|
from typing import Any
|
||||||
|
|
||||||
|
RULE_SESSION = "session"
|
||||||
|
RULE_USER_WINDOW = "user_window"
|
||||||
|
|
||||||
|
|
||||||
|
@dataclass
|
||||||
|
class CallLink:
|
||||||
|
"""A tool call linked to a piece of feedback, and why it matched."""
|
||||||
|
|
||||||
|
call_id: str
|
||||||
|
rule: str
|
||||||
|
tool: str
|
||||||
|
started_at: datetime
|
||||||
|
ok: bool
|
||||||
|
position: int = 0
|
||||||
|
|
||||||
|
def to_dict(self) -> dict[str, Any]:
|
||||||
|
data = asdict(self)
|
||||||
|
data["started_at"] = self.started_at.isoformat()
|
||||||
|
return data
|
||||||
|
|
||||||
|
|
||||||
|
@dataclass
|
||||||
|
class Candidate:
|
||||||
|
"""The fields correlation needs, from a buffered record or a database row."""
|
||||||
|
|
||||||
|
id: str
|
||||||
|
tool: str
|
||||||
|
started_at: datetime
|
||||||
|
ok: bool
|
||||||
|
session_id: str | None
|
||||||
|
user_sub: str | None
|
||||||
|
|
||||||
|
|
||||||
|
class ResolverContext:
|
||||||
|
"""What an identity resolver receives when called during correlation.
|
||||||
|
|
||||||
|
Outside a tool call there is no MiddlewareContext, so this carries only the
|
||||||
|
``fastmcp_context`` attribute, which is what resolvers normally read.
|
||||||
|
"""
|
||||||
|
|
||||||
|
def __init__(self, fastmcp_context: Any):
|
||||||
|
self.fastmcp_context = fastmcp_context
|
||||||
|
self.message = None
|
||||||
|
|
||||||
|
|
||||||
|
def as_utc(value: datetime) -> datetime:
|
||||||
|
# SQLite drops the timezone from DateTime(timezone=True) columns.
|
||||||
|
return value.replace(tzinfo=UTC) if value.tzinfo is None else value
|
||||||
|
|
||||||
|
|
||||||
|
def select_calls(
|
||||||
|
candidates: Iterable[Candidate],
|
||||||
|
*,
|
||||||
|
session_id: str | None,
|
||||||
|
user_sub: str | None,
|
||||||
|
since: datetime,
|
||||||
|
limit: int,
|
||||||
|
) -> list[CallLink]:
|
||||||
|
"""Pick up to ``limit`` recent calls for a feedback item.
|
||||||
|
|
||||||
|
Calls from the same MCP session come first. If there are fewer than
|
||||||
|
``limit``, the rest are filled with the same user's calls since ``since``,
|
||||||
|
which covers clients that open a new session per call (PATs, CI). The
|
||||||
|
result is in chronological order, numbered from 0.
|
||||||
|
"""
|
||||||
|
unique: dict[str, Candidate] = {}
|
||||||
|
for c in candidates:
|
||||||
|
unique.setdefault(c.id, c)
|
||||||
|
newest_first = sorted(
|
||||||
|
unique.values(), key=lambda c: as_utc(c.started_at), reverse=True
|
||||||
|
)
|
||||||
|
|
||||||
|
chosen: list[tuple[Candidate, str]] = []
|
||||||
|
if session_id:
|
||||||
|
chosen += [
|
||||||
|
(c, RULE_SESSION) for c in newest_first if c.session_id == session_id
|
||||||
|
][:limit]
|
||||||
|
if user_sub and len(chosen) < limit:
|
||||||
|
taken = {c.id for c, _ in chosen}
|
||||||
|
chosen += [
|
||||||
|
(c, RULE_USER_WINDOW)
|
||||||
|
for c in newest_first
|
||||||
|
if c.id not in taken
|
||||||
|
and c.user_sub == user_sub
|
||||||
|
and as_utc(c.started_at) >= since
|
||||||
|
][: limit - len(chosen)]
|
||||||
|
|
||||||
|
chosen.sort(key=lambda pair: as_utc(pair[0].started_at))
|
||||||
|
return [
|
||||||
|
CallLink(
|
||||||
|
call_id=c.id,
|
||||||
|
rule=rule,
|
||||||
|
tool=c.tool,
|
||||||
|
started_at=as_utc(c.started_at),
|
||||||
|
ok=c.ok,
|
||||||
|
position=i,
|
||||||
|
)
|
||||||
|
for i, (c, rule) in enumerate(chosen)
|
||||||
|
]
|
||||||
@@ -6,17 +6,28 @@ import json
|
|||||||
import logging
|
import logging
|
||||||
import os
|
import os
|
||||||
import time
|
import time
|
||||||
|
import uuid
|
||||||
|
from collections import deque
|
||||||
from collections.abc import Awaitable, Callable, Iterable, Sequence
|
from collections.abc import Awaitable, Callable, Iterable, Sequence
|
||||||
from datetime import UTC, datetime
|
from datetime import UTC, datetime, timedelta
|
||||||
from typing import Any
|
from typing import Any
|
||||||
|
|
||||||
from fastmcp import FastMCP
|
from fastmcp import FastMCP
|
||||||
from fastmcp.server.middleware import Middleware, MiddlewareContext
|
from fastmcp.server.middleware import Middleware, MiddlewareContext
|
||||||
|
|
||||||
|
from .classify import DEFAULT_RESULT_CLASSIFIER, ResultClassifier
|
||||||
|
from .correlation import (
|
||||||
|
CallLink,
|
||||||
|
Candidate,
|
||||||
|
ResolverContext,
|
||||||
|
as_utc,
|
||||||
|
select_calls,
|
||||||
|
)
|
||||||
from .dispatch import RecordDispatcher
|
from .dispatch import RecordDispatcher
|
||||||
from .record import ToolCallRecord
|
from .record import ToolCallRecord
|
||||||
from .redaction import Redactor
|
from .redaction import Redactor
|
||||||
from .sinks import JsonLinesSink, Sink
|
from .sinks import JsonLinesSink, Sink
|
||||||
|
from .storage import DatabaseSink
|
||||||
|
|
||||||
logger = logging.getLogger(__name__)
|
logger = logging.getLogger(__name__)
|
||||||
|
|
||||||
@@ -43,6 +54,35 @@ def _attr(obj: Any, name: str) -> Any:
|
|||||||
return str(value) if value is not None else None
|
return str(value) if value is not None else None
|
||||||
|
|
||||||
|
|
||||||
|
# stdio serves exactly one client per process, so the process is the session.
|
||||||
|
_STDIO_SESSION = f"stdio:{os.getpid()}:{uuid.uuid4().hex[:8]}"
|
||||||
|
|
||||||
|
|
||||||
|
def session_key(ctx: Any) -> str | None:
|
||||||
|
"""A per-client key that stays the same across that client's calls.
|
||||||
|
|
||||||
|
Clients on MCP protocol 2026-07-28 (Claude Code, FastMCP 4's Client) do
|
||||||
|
not open sessions, and FastMCP then generates a new ``ctx.session_id`` for
|
||||||
|
every request. So stable signals come first: the HTTP ``mcp-session-id``
|
||||||
|
header (older protocol clients), then one key per process for stdio, then
|
||||||
|
``ctx.session_id``. With a stateless HTTP client there is no session, and
|
||||||
|
correlation relies on the same-user rule.
|
||||||
|
"""
|
||||||
|
if ctx is None:
|
||||||
|
return None
|
||||||
|
try:
|
||||||
|
from fastmcp.server.dependencies import get_http_headers
|
||||||
|
|
||||||
|
header = (get_http_headers(include_all=True) or {}).get("mcp-session-id")
|
||||||
|
if header:
|
||||||
|
return header
|
||||||
|
except Exception:
|
||||||
|
pass
|
||||||
|
if _attr(ctx, "transport") == "stdio":
|
||||||
|
return _STDIO_SESSION
|
||||||
|
return _attr(ctx, "session_id")
|
||||||
|
|
||||||
|
|
||||||
def _result_payload(result: Any) -> Any:
|
def _result_payload(result: Any) -> Any:
|
||||||
structured = getattr(result, "structured_content", None)
|
structured = getattr(result, "structured_content", None)
|
||||||
if structured is not None:
|
if structured is not None:
|
||||||
@@ -88,6 +128,15 @@ class InstrumentationMiddleware(Middleware):
|
|||||||
May be async.
|
May be async.
|
||||||
server_version: default for the ``server_version`` column.
|
server_version: default for the ``server_version`` column.
|
||||||
hook_timeout: seconds an async hook may take before it is skipped.
|
hook_timeout: seconds an async hook may take before it is skipped.
|
||||||
|
recent_calls: how many recent records to keep in memory for
|
||||||
|
correlation, including ones still waiting to reach the sinks.
|
||||||
|
correlation_timeout: seconds a database lookup during correlation may
|
||||||
|
take before it is skipped.
|
||||||
|
result_classifier: ``(tool, result) -> str | None`` run on results
|
||||||
|
of calls that did not raise. A string marks the call as a soft
|
||||||
|
error (``outcome="soft_error"``, ``ok=False``) and becomes the
|
||||||
|
error message. Defaults to ``DEFAULT_RESULT_CLASSIFIER``; None
|
||||||
|
turns it off. If it raises, the call is recorded as ok.
|
||||||
"""
|
"""
|
||||||
|
|
||||||
def __init__(
|
def __init__(
|
||||||
@@ -105,6 +154,9 @@ class InstrumentationMiddleware(Middleware):
|
|||||||
batch_size: int = 100,
|
batch_size: int = 100,
|
||||||
sink_timeout: float = 10.0,
|
sink_timeout: float = 10.0,
|
||||||
max_error_chars: int = 500,
|
max_error_chars: int = 500,
|
||||||
|
recent_calls: int = 2000,
|
||||||
|
correlation_timeout: float = 1.0,
|
||||||
|
result_classifier: ResultClassifier | None = DEFAULT_RESULT_CLASSIFIER,
|
||||||
):
|
):
|
||||||
mode = (mode or os.getenv(MODE_ENV_VAR) or "meta").lower()
|
mode = (mode or os.getenv(MODE_ENV_VAR) or "meta").lower()
|
||||||
if mode not in MODES:
|
if mode not in MODES:
|
||||||
@@ -117,6 +169,9 @@ class InstrumentationMiddleware(Middleware):
|
|||||||
self.server_version = server_version
|
self.server_version = server_version
|
||||||
self.hook_timeout = hook_timeout
|
self.hook_timeout = hook_timeout
|
||||||
self.max_error_chars = max_error_chars
|
self.max_error_chars = max_error_chars
|
||||||
|
self.correlation_timeout = correlation_timeout
|
||||||
|
self.result_classifier = result_classifier
|
||||||
|
self.recent: deque[ToolCallRecord] = deque(maxlen=recent_calls)
|
||||||
self.dispatcher = RecordDispatcher(
|
self.dispatcher = RecordDispatcher(
|
||||||
sinks if sinks is not None else [JsonLinesSink()],
|
sinks if sinks is not None else [JsonLinesSink()],
|
||||||
max_queue=max_queue,
|
max_queue=max_queue,
|
||||||
@@ -144,6 +199,7 @@ class InstrumentationMiddleware(Middleware):
|
|||||||
record = await self._build_record(
|
record = await self._build_record(
|
||||||
context, started_at, duration_ms, result, error
|
context, started_at, duration_ms, result, error
|
||||||
)
|
)
|
||||||
|
self.recent.append(record)
|
||||||
self.dispatcher.submit(record)
|
self.dispatcher.submit(record)
|
||||||
except Exception:
|
except Exception:
|
||||||
logger.warning("Failed to record tool call", exc_info=True)
|
logger.warning("Failed to record tool call", exc_info=True)
|
||||||
@@ -160,6 +216,22 @@ class InstrumentationMiddleware(Middleware):
|
|||||||
logger.warning("Instrumentation hook %r failed", hook, exc_info=True)
|
logger.warning("Instrumentation hook %r failed", hook, exc_info=True)
|
||||||
return {}
|
return {}
|
||||||
|
|
||||||
|
def _classify(self, tool: str, result: Any, payload: Any) -> str | None:
|
||||||
|
if getattr(result, "is_error", False) is True:
|
||||||
|
# FastMCP 3.4+ keeps isError on the ToolResult, not in the payload.
|
||||||
|
if isinstance(payload, dict):
|
||||||
|
payload = {**payload, "isError": True}
|
||||||
|
else:
|
||||||
|
payload = {"isError": True, "content": payload}
|
||||||
|
try:
|
||||||
|
message = self.result_classifier(tool, payload)
|
||||||
|
except Exception:
|
||||||
|
logger.warning(
|
||||||
|
"Result classifier %r failed", self.result_classifier, exc_info=True
|
||||||
|
)
|
||||||
|
return None
|
||||||
|
return None if message is None else str(message)
|
||||||
|
|
||||||
async def _build_record(
|
async def _build_record(
|
||||||
self,
|
self,
|
||||||
context: MiddlewareContext,
|
context: MiddlewareContext,
|
||||||
@@ -184,24 +256,35 @@ class InstrumentationMiddleware(Middleware):
|
|||||||
mode = "meta" if tool in self.meta_only_tools else self.mode
|
mode = "meta" if tool in self.meta_only_tools else self.mode
|
||||||
full = mode == "full"
|
full = mode == "full"
|
||||||
|
|
||||||
|
outcome = "ok"
|
||||||
error_type = error_message = None
|
error_type = error_message = None
|
||||||
if error is not None:
|
if error is not None:
|
||||||
# FastMCP wraps tool exceptions in ToolError; report the original.
|
# FastMCP wraps tool exceptions in ToolError; report the original.
|
||||||
cause = error.__cause__ or error
|
cause = error.__cause__ or error
|
||||||
|
outcome = "error"
|
||||||
error_type = type(cause).__name__
|
error_type = type(cause).__name__
|
||||||
error_message = self.redactor.redact_text(str(cause))[
|
error_message = self.redactor.redact_text(str(cause))[
|
||||||
: self.max_error_chars
|
: self.max_error_chars
|
||||||
]
|
]
|
||||||
|
elif self.result_classifier is not None and payload is not None:
|
||||||
|
soft = self._classify(tool, result, payload)
|
||||||
|
if soft is not None:
|
||||||
|
outcome = "soft_error"
|
||||||
|
error_type = "SoftError"
|
||||||
|
error_message = (
|
||||||
|
self.redactor.redact_text(soft)[: self.max_error_chars] or None
|
||||||
|
)
|
||||||
|
|
||||||
return ToolCallRecord(
|
return ToolCallRecord(
|
||||||
tool=tool,
|
tool=tool,
|
||||||
started_at=started_at,
|
started_at=started_at,
|
||||||
duration_ms=duration_ms,
|
duration_ms=duration_ms,
|
||||||
ok=error is None,
|
ok=outcome == "ok",
|
||||||
|
outcome=outcome,
|
||||||
mode=mode,
|
mode=mode,
|
||||||
error_type=error_type,
|
error_type=error_type,
|
||||||
error_message=error_message,
|
error_message=error_message,
|
||||||
session_id=_attr(ctx, "session_id"),
|
session_id=session_key(ctx),
|
||||||
request_id=_attr(ctx, "request_id"),
|
request_id=_attr(ctx, "request_id"),
|
||||||
client_id=columns.get("client_id") or _attr(ctx, "client_id"),
|
client_id=columns.get("client_id") or _attr(ctx, "client_id"),
|
||||||
user_sub=columns.get("user_sub"),
|
user_sub=columns.get("user_sub"),
|
||||||
@@ -217,6 +300,140 @@ class InstrumentationMiddleware(Middleware):
|
|||||||
extra=extra,
|
extra=extra,
|
||||||
)
|
)
|
||||||
|
|
||||||
|
@property
|
||||||
|
def database_sink(self) -> DatabaseSink | None:
|
||||||
|
"""The first DatabaseSink among the sinks, if any."""
|
||||||
|
return next(
|
||||||
|
(s for s in self.dispatcher.sinks if isinstance(s, DatabaseSink)), None
|
||||||
|
)
|
||||||
|
|
||||||
|
async def capture_recent_calls(
|
||||||
|
self,
|
||||||
|
ctx: Any = None,
|
||||||
|
n: int = 20,
|
||||||
|
*,
|
||||||
|
user_sub: str | None = None,
|
||||||
|
window: timedelta = timedelta(minutes=15),
|
||||||
|
) -> list[CallLink]:
|
||||||
|
"""The calls that led up to now, for attaching to a feedback item.
|
||||||
|
|
||||||
|
Call from inside a tool. Returns up to ``n`` calls in chronological
|
||||||
|
order: first from the current MCP session, then, if that yields fewer
|
||||||
|
than ``n``, the same user's calls within ``window``. Each link says
|
||||||
|
which rule matched. Never raises; returns [] on any failure.
|
||||||
|
|
||||||
|
Args:
|
||||||
|
ctx: the FastMCP Context. Looked up automatically when omitted.
|
||||||
|
n: maximum number of calls.
|
||||||
|
user_sub: the caller's user id. When omitted, the identity
|
||||||
|
resolver is asked, given an object whose ``fastmcp_context``
|
||||||
|
is ``ctx``.
|
||||||
|
window: how far back the same-user fallback reaches.
|
||||||
|
"""
|
||||||
|
try:
|
||||||
|
if ctx is None:
|
||||||
|
ctx = _current_context()
|
||||||
|
session_id = session_key(ctx)
|
||||||
|
if user_sub is None and self.identity_resolver is not None:
|
||||||
|
identity = await self._call_hook(
|
||||||
|
self.identity_resolver, ResolverContext(ctx)
|
||||||
|
)
|
||||||
|
user_sub = identity.get("user_sub")
|
||||||
|
since = datetime.now(UTC) - window
|
||||||
|
|
||||||
|
candidates = [
|
||||||
|
Candidate(r.id, r.tool, r.started_at, r.ok, r.session_id, r.user_sub)
|
||||||
|
for r in self.recent
|
||||||
|
]
|
||||||
|
links = select_calls(
|
||||||
|
candidates,
|
||||||
|
session_id=session_id,
|
||||||
|
user_sub=user_sub,
|
||||||
|
since=since,
|
||||||
|
limit=n,
|
||||||
|
)
|
||||||
|
sink = self.database_sink
|
||||||
|
if sink is not None and len(links) < n:
|
||||||
|
# Other workers, or calls from before a restart.
|
||||||
|
rows = await asyncio.wait_for(
|
||||||
|
sink.recent_calls(
|
||||||
|
session_id=session_id, user_sub=user_sub, since=since, limit=n
|
||||||
|
),
|
||||||
|
self.correlation_timeout,
|
||||||
|
)
|
||||||
|
candidates += [
|
||||||
|
Candidate(
|
||||||
|
r["id"],
|
||||||
|
r["tool"],
|
||||||
|
as_utc(r["started_at"]),
|
||||||
|
r["ok"],
|
||||||
|
r["session_id"],
|
||||||
|
r["user_sub"],
|
||||||
|
)
|
||||||
|
for r in rows
|
||||||
|
]
|
||||||
|
links = select_calls(
|
||||||
|
candidates,
|
||||||
|
session_id=session_id,
|
||||||
|
user_sub=user_sub,
|
||||||
|
since=since,
|
||||||
|
limit=n,
|
||||||
|
)
|
||||||
|
return links
|
||||||
|
except Exception:
|
||||||
|
logger.warning("Failed to capture recent calls", exc_info=True)
|
||||||
|
return []
|
||||||
|
|
||||||
|
async def link_feedback(self, feedback_ref: str, links: Sequence[CallLink]) -> bool:
|
||||||
|
"""Store ``links`` against a feedback id of any format (``bug-XXXX``, 42).
|
||||||
|
|
||||||
|
Needs a DatabaseSink. Replaces earlier links for the same id. Returns
|
||||||
|
False, without raising, if they could not be stored.
|
||||||
|
"""
|
||||||
|
sink = self.database_sink
|
||||||
|
if sink is None:
|
||||||
|
logger.warning("link_feedback needs a DatabaseSink; links not stored")
|
||||||
|
return False
|
||||||
|
linked_at = datetime.now(UTC)
|
||||||
|
try:
|
||||||
|
await sink.replace_links(
|
||||||
|
str(feedback_ref),
|
||||||
|
[
|
||||||
|
{
|
||||||
|
"feedback_ref": str(feedback_ref),
|
||||||
|
"call_id": link.call_id,
|
||||||
|
"position": link.position,
|
||||||
|
"rule": link.rule,
|
||||||
|
"linked_at": linked_at,
|
||||||
|
}
|
||||||
|
for link in links
|
||||||
|
],
|
||||||
|
)
|
||||||
|
return True
|
||||||
|
except Exception:
|
||||||
|
logger.warning("Failed to link feedback %s", feedback_ref, exc_info=True)
|
||||||
|
return False
|
||||||
|
|
||||||
|
async def feedback_context(self, feedback_ref: str) -> list[dict[str, Any]]:
|
||||||
|
"""The calls linked to a feedback id, in order, with their details.
|
||||||
|
|
||||||
|
Calls not yet written to the database are filled in from memory.
|
||||||
|
"""
|
||||||
|
sink = self.database_sink
|
||||||
|
if sink is None:
|
||||||
|
return []
|
||||||
|
rows = await sink.linked_calls(str(feedback_ref))
|
||||||
|
in_memory = {r.id: r for r in self.recent}
|
||||||
|
for row in rows:
|
||||||
|
if row["tool"] is None and row["call_id"] in in_memory:
|
||||||
|
record = in_memory[row["call_id"]]
|
||||||
|
for key in row:
|
||||||
|
if key not in ("call_id", "position", "rule"):
|
||||||
|
row[key] = getattr(record, key, None)
|
||||||
|
if row["started_at"] is not None:
|
||||||
|
row["started_at"] = as_utc(row["started_at"])
|
||||||
|
return rows
|
||||||
|
|
||||||
async def flush(self) -> None:
|
async def flush(self) -> None:
|
||||||
"""Wait until queued records have reached the sinks."""
|
"""Wait until queued records have reached the sinks."""
|
||||||
await self.dispatcher.flush()
|
await self.dispatcher.flush()
|
||||||
@@ -226,6 +443,15 @@ class InstrumentationMiddleware(Middleware):
|
|||||||
await self.dispatcher.aclose()
|
await self.dispatcher.aclose()
|
||||||
|
|
||||||
|
|
||||||
|
def _current_context() -> Any:
|
||||||
|
try:
|
||||||
|
from fastmcp.server.dependencies import get_context
|
||||||
|
|
||||||
|
return get_context()
|
||||||
|
except Exception:
|
||||||
|
return None
|
||||||
|
|
||||||
|
|
||||||
def instrument(app: FastMCP, sinks: Sequence[Sink] | None = None, **kwargs: Any):
|
def instrument(app: FastMCP, sinks: Sequence[Sink] | None = None, **kwargs: Any):
|
||||||
"""Add an InstrumentationMiddleware to ``app`` and return it.
|
"""Add an InstrumentationMiddleware to ``app`` and return it.
|
||||||
|
|
||||||
|
|||||||
@@ -12,6 +12,10 @@ class ToolCallRecord:
|
|||||||
|
|
||||||
``args`` and ``result`` are only populated in ``full`` mode, and are
|
``args`` and ``result`` are only populated in ``full`` mode, and are
|
||||||
already redacted when they are. Sizes are measured before redaction.
|
already redacted when they are. Sizes are measured before redaction.
|
||||||
|
|
||||||
|
``outcome`` is ``ok``, ``error`` (the tool raised) or ``soft_error`` (it
|
||||||
|
returned a result the classifier flagged). ``ok`` is ``outcome == "ok"``;
|
||||||
|
when ``outcome`` is not given it is derived from ``ok``.
|
||||||
"""
|
"""
|
||||||
|
|
||||||
tool: str
|
tool: str
|
||||||
@@ -19,6 +23,7 @@ class ToolCallRecord:
|
|||||||
duration_ms: float
|
duration_ms: float
|
||||||
ok: bool
|
ok: bool
|
||||||
mode: str
|
mode: str
|
||||||
|
outcome: str | None = None
|
||||||
id: str = field(default_factory=lambda: str(uuid.uuid4()))
|
id: str = field(default_factory=lambda: str(uuid.uuid4()))
|
||||||
error_type: str | None = None
|
error_type: str | None = None
|
||||||
error_message: str | None = None
|
error_message: str | None = None
|
||||||
@@ -35,6 +40,10 @@ class ToolCallRecord:
|
|||||||
result: Any = None
|
result: Any = None
|
||||||
extra: dict[str, Any] = field(default_factory=dict)
|
extra: dict[str, Any] = field(default_factory=dict)
|
||||||
|
|
||||||
|
def __post_init__(self) -> None:
|
||||||
|
if self.outcome is None:
|
||||||
|
self.outcome = "ok" if self.ok else "error"
|
||||||
|
|
||||||
def to_row(self) -> dict[str, Any]:
|
def to_row(self) -> dict[str, Any]:
|
||||||
"""Column values for the tool_calls table."""
|
"""Column values for the tool_calls table."""
|
||||||
return asdict(self)
|
return asdict(self)
|
||||||
|
|||||||
@@ -1,11 +1,14 @@
|
|||||||
"""Async SQLAlchemy storage for tool-call records.
|
"""Async SQLAlchemy storage for tool-call records and feedback links.
|
||||||
|
|
||||||
Hosts that manage their own schema include ``build_metadata(prefix)`` in their
|
Hosts that manage their own schema include ``build_metadata(prefix)`` in their
|
||||||
Alembic ``env.py`` (``target_metadata``) and pass their existing AsyncEngine to
|
Alembic ``env.py`` (``target_metadata``) and pass their existing AsyncEngine to
|
||||||
``DatabaseSink``. Nothing is created unless ``create_tables=True``.
|
``DatabaseSink``. Nothing is created unless ``create_tables=True``.
|
||||||
"""
|
"""
|
||||||
|
|
||||||
|
import asyncio
|
||||||
from collections.abc import Sequence
|
from collections.abc import Sequence
|
||||||
|
from datetime import datetime
|
||||||
|
from typing import Any
|
||||||
|
|
||||||
from sqlalchemy import (
|
from sqlalchemy import (
|
||||||
JSON,
|
JSON,
|
||||||
@@ -18,6 +21,10 @@ from sqlalchemy import (
|
|||||||
String,
|
String,
|
||||||
Table,
|
Table,
|
||||||
Text,
|
Text,
|
||||||
|
and_,
|
||||||
|
delete,
|
||||||
|
or_,
|
||||||
|
select,
|
||||||
)
|
)
|
||||||
from sqlalchemy.dialects.postgresql import JSONB
|
from sqlalchemy.dialects.postgresql import JSONB
|
||||||
from sqlalchemy.ext.asyncio import AsyncEngine, create_async_engine
|
from sqlalchemy.ext.asyncio import AsyncEngine, create_async_engine
|
||||||
@@ -33,38 +40,52 @@ _JSON = JSON(none_as_null=True).with_variant(JSONB(none_as_null=True), "postgres
|
|||||||
def build_metadata(
|
def build_metadata(
|
||||||
prefix: str = DEFAULT_TABLE_PREFIX, metadata: MetaData | None = None
|
prefix: str = DEFAULT_TABLE_PREFIX, metadata: MetaData | None = None
|
||||||
) -> MetaData:
|
) -> MetaData:
|
||||||
"""Return MetaData holding the ``{prefix}tool_calls`` table.
|
"""Return MetaData holding ``{prefix}tool_calls`` and ``{prefix}feedback_call_links``.
|
||||||
|
|
||||||
Pass an existing MetaData to add the table to it instead.
|
Pass an existing MetaData to add the tables to it instead.
|
||||||
"""
|
"""
|
||||||
metadata = metadata if metadata is not None else MetaData()
|
metadata = metadata if metadata is not None else MetaData()
|
||||||
name = f"{prefix}tool_calls"
|
calls = f"{prefix}tool_calls"
|
||||||
if name in metadata.tables:
|
links = f"{prefix}feedback_call_links"
|
||||||
return metadata
|
if calls not in metadata.tables:
|
||||||
Table(
|
Table(
|
||||||
name,
|
calls,
|
||||||
metadata,
|
metadata,
|
||||||
Column("id", String(36), primary_key=True),
|
Column("id", String(36), primary_key=True),
|
||||||
Column("started_at", DateTime(timezone=True), nullable=False, index=True),
|
Column("started_at", DateTime(timezone=True), nullable=False, index=True),
|
||||||
Column("duration_ms", Float, nullable=False),
|
Column("duration_ms", Float, nullable=False),
|
||||||
Column("tool", String(255), nullable=False, index=True),
|
Column("tool", String(255), nullable=False, index=True),
|
||||||
Column("mode", String(8), nullable=False),
|
Column("mode", String(8), nullable=False),
|
||||||
Column("ok", Boolean, nullable=False, index=True),
|
Column("ok", Boolean, nullable=False, index=True),
|
||||||
Column("error_type", String(255)),
|
# ok, soft_error or error. NULL on rows from before 2026.09.27.4.
|
||||||
Column("error_message", Text),
|
Column("outcome", String(16), index=True),
|
||||||
Column("session_id", String(255), index=True),
|
Column("error_type", String(255)),
|
||||||
Column("request_id", String(255)),
|
Column("error_message", Text),
|
||||||
Column("client_id", String(255)),
|
Column("session_id", String(255), index=True),
|
||||||
Column("user_sub", String(255), index=True),
|
Column("request_id", String(255)),
|
||||||
Column("caller_kind", String(64)),
|
Column("client_id", String(255)),
|
||||||
Column("identity", _JSON),
|
Column("user_sub", String(255), index=True),
|
||||||
Column("server_version", String(64), index=True),
|
Column("caller_kind", String(64)),
|
||||||
Column("args_size", Integer),
|
Column("identity", _JSON),
|
||||||
Column("result_size", Integer),
|
Column("server_version", String(64), index=True),
|
||||||
Column("args", _JSON),
|
Column("args_size", Integer),
|
||||||
Column("result", _JSON),
|
Column("result_size", Integer),
|
||||||
Column("extra", _JSON),
|
Column("args", _JSON),
|
||||||
)
|
Column("result", _JSON),
|
||||||
|
Column("extra", _JSON),
|
||||||
|
)
|
||||||
|
if links not in metadata.tables:
|
||||||
|
# No foreign key to tool_calls: a link can be written before its call
|
||||||
|
# record has left the queue, and hosts may keep calls elsewhere.
|
||||||
|
Table(
|
||||||
|
links,
|
||||||
|
metadata,
|
||||||
|
Column("feedback_ref", String(255), primary_key=True),
|
||||||
|
Column("call_id", String(36), primary_key=True, index=True),
|
||||||
|
Column("position", Integer, nullable=False),
|
||||||
|
Column("rule", String(32), nullable=False),
|
||||||
|
Column("linked_at", DateTime(timezone=True), nullable=False),
|
||||||
|
)
|
||||||
return metadata
|
return metadata
|
||||||
|
|
||||||
|
|
||||||
@@ -76,8 +97,8 @@ class DatabaseSink:
|
|||||||
``sqlite+aiosqlite:///calls.db`` or ``postgresql+asyncpg://...``.
|
``sqlite+aiosqlite:///calls.db`` or ``postgresql+asyncpg://...``.
|
||||||
An engine built from a URL is disposed on ``aclose()``; a passed-in
|
An engine built from a URL is disposed on ``aclose()``; a passed-in
|
||||||
engine is left alone.
|
engine is left alone.
|
||||||
prefix: table name prefix, so the table cannot collide with the host's.
|
prefix: table name prefix, so the tables cannot collide with the host's.
|
||||||
create_tables: create the table on first write. Off by default; hosts
|
create_tables: create the tables on first use. Off by default; hosts
|
||||||
with Alembic should generate a migration from ``build_metadata``.
|
with Alembic should generate a migration from ``build_metadata``.
|
||||||
"""
|
"""
|
||||||
|
|
||||||
@@ -92,23 +113,105 @@ class DatabaseSink:
|
|||||||
self.engine = create_async_engine(engine) if isinstance(engine, str) else engine
|
self.engine = create_async_engine(engine) if isinstance(engine, str) else engine
|
||||||
self.metadata = build_metadata(prefix)
|
self.metadata = build_metadata(prefix)
|
||||||
self.table = self.metadata.tables[f"{prefix}tool_calls"]
|
self.table = self.metadata.tables[f"{prefix}tool_calls"]
|
||||||
|
self.links = self.metadata.tables[f"{prefix}feedback_call_links"]
|
||||||
self.create_tables = create_tables
|
self.create_tables = create_tables
|
||||||
self._tables_ready = False
|
self._tables_ready = False
|
||||||
|
# The background writer and link_feedback can both reach _ready() at
|
||||||
|
# once; without the lock both run create_all and one fails.
|
||||||
|
self._tables_lock = asyncio.Lock()
|
||||||
|
|
||||||
async def ensure_tables(self) -> None:
|
async def ensure_tables(self) -> None:
|
||||||
"""Create the table if missing. Only called when create_tables=True."""
|
"""Create the tables if missing. Only called when create_tables=True."""
|
||||||
async with self.engine.begin() as conn:
|
async with self.engine.begin() as conn:
|
||||||
await conn.run_sync(self.metadata.create_all)
|
await conn.run_sync(self.metadata.create_all)
|
||||||
self._tables_ready = True
|
self._tables_ready = True
|
||||||
|
|
||||||
|
async def _ready(self) -> None:
|
||||||
|
if self.create_tables and not self._tables_ready:
|
||||||
|
async with self._tables_lock:
|
||||||
|
if not self._tables_ready:
|
||||||
|
await self.ensure_tables()
|
||||||
|
|
||||||
async def write(self, records: Sequence[ToolCallRecord]) -> None:
|
async def write(self, records: Sequence[ToolCallRecord]) -> None:
|
||||||
if not records:
|
if not records:
|
||||||
return
|
return
|
||||||
if self.create_tables and not self._tables_ready:
|
await self._ready()
|
||||||
await self.ensure_tables()
|
|
||||||
async with self.engine.begin() as conn:
|
async with self.engine.begin() as conn:
|
||||||
await conn.execute(self.table.insert(), [r.to_row() for r in records])
|
await conn.execute(self.table.insert(), [r.to_row() for r in records])
|
||||||
|
|
||||||
|
async def recent_calls(
|
||||||
|
self,
|
||||||
|
*,
|
||||||
|
session_id: str | None,
|
||||||
|
user_sub: str | None,
|
||||||
|
since: datetime,
|
||||||
|
limit: int,
|
||||||
|
) -> list[dict[str, Any]]:
|
||||||
|
"""Calls from this session, or from this user since ``since``, newest first."""
|
||||||
|
t = self.table
|
||||||
|
conditions = []
|
||||||
|
if session_id:
|
||||||
|
conditions.append(t.c.session_id == session_id)
|
||||||
|
if user_sub:
|
||||||
|
conditions.append(and_(t.c.user_sub == user_sub, t.c.started_at >= since))
|
||||||
|
if not conditions:
|
||||||
|
return []
|
||||||
|
await self._ready()
|
||||||
|
query = (
|
||||||
|
select(
|
||||||
|
t.c.id, t.c.tool, t.c.started_at, t.c.ok, t.c.session_id, t.c.user_sub
|
||||||
|
)
|
||||||
|
.where(or_(*conditions))
|
||||||
|
.order_by(t.c.started_at.desc())
|
||||||
|
.limit(limit)
|
||||||
|
)
|
||||||
|
async with self.engine.connect() as conn:
|
||||||
|
return [dict(row) for row in (await conn.execute(query)).mappings()]
|
||||||
|
|
||||||
|
async def replace_links(
|
||||||
|
self, feedback_ref: str, links: Sequence[dict[str, Any]]
|
||||||
|
) -> None:
|
||||||
|
"""Set the calls linked to ``feedback_ref``, replacing any earlier set."""
|
||||||
|
await self._ready()
|
||||||
|
async with self.engine.begin() as conn:
|
||||||
|
await conn.execute(
|
||||||
|
delete(self.links).where(self.links.c.feedback_ref == feedback_ref)
|
||||||
|
)
|
||||||
|
if links:
|
||||||
|
await conn.execute(self.links.insert(), list(links))
|
||||||
|
|
||||||
|
async def linked_calls(self, feedback_ref: str) -> list[dict[str, Any]]:
|
||||||
|
"""Links for ``feedback_ref`` in order, joined to their call records.
|
||||||
|
|
||||||
|
Call columns are None for calls not (yet) in this database.
|
||||||
|
"""
|
||||||
|
await self._ready()
|
||||||
|
link, call = self.links, self.table
|
||||||
|
query = (
|
||||||
|
select(
|
||||||
|
link.c.call_id,
|
||||||
|
link.c.position,
|
||||||
|
link.c.rule,
|
||||||
|
call.c.tool,
|
||||||
|
call.c.started_at,
|
||||||
|
call.c.duration_ms,
|
||||||
|
call.c.ok,
|
||||||
|
call.c.outcome,
|
||||||
|
call.c.error_type,
|
||||||
|
call.c.error_message,
|
||||||
|
call.c.session_id,
|
||||||
|
call.c.user_sub,
|
||||||
|
call.c.caller_kind,
|
||||||
|
call.c.server_version,
|
||||||
|
call.c.extra,
|
||||||
|
)
|
||||||
|
.select_from(link.outerjoin(call, link.c.call_id == call.c.id))
|
||||||
|
.where(link.c.feedback_ref == feedback_ref)
|
||||||
|
.order_by(link.c.position)
|
||||||
|
)
|
||||||
|
async with self.engine.connect() as conn:
|
||||||
|
return [dict(row) for row in (await conn.execute(query)).mappings()]
|
||||||
|
|
||||||
async def aclose(self) -> None:
|
async def aclose(self) -> None:
|
||||||
if self._owns_engine:
|
if self._owns_engine:
|
||||||
await self.engine.dispose()
|
await self.engine.dispose()
|
||||||
|
|||||||
@@ -1,15 +1,37 @@
|
|||||||
"""Submission mixin for feedback collection."""
|
"""Submission mixin for feedback collection."""
|
||||||
|
|
||||||
from datetime import UTC, datetime
|
from datetime import UTC, datetime
|
||||||
from typing import Any
|
from typing import TYPE_CHECKING, Any
|
||||||
|
|
||||||
from ..insights import record_feedback_submission
|
from ..insights import record_feedback_submission
|
||||||
from ..models import SubmitFeedbackRequest, create_feedback_from_request
|
from ..models import SubmitFeedbackRequest, create_feedback_from_request
|
||||||
from .base import BaseFeedbackMixin, logger, mcp_tool
|
from .base import (
|
||||||
|
BaseFeedbackMixin,
|
||||||
|
FeedbackDatabase,
|
||||||
|
FeedbackInsights,
|
||||||
|
logger,
|
||||||
|
mcp_tool,
|
||||||
|
)
|
||||||
|
|
||||||
|
if TYPE_CHECKING:
|
||||||
|
from ..instrumentation import InstrumentationMiddleware
|
||||||
|
|
||||||
|
|
||||||
class SubmissionMixin(BaseFeedbackMixin):
|
class SubmissionMixin(BaseFeedbackMixin):
|
||||||
"""Mixin for feedback submission functionality."""
|
"""Mixin for feedback submission functionality.
|
||||||
|
|
||||||
|
Pass an InstrumentationMiddleware to link each submission to the tool
|
||||||
|
calls that led up to it.
|
||||||
|
"""
|
||||||
|
|
||||||
|
def __init__(
|
||||||
|
self,
|
||||||
|
database: FeedbackDatabase,
|
||||||
|
insights: FeedbackInsights | None = None,
|
||||||
|
instrumentation: "InstrumentationMiddleware | None" = None,
|
||||||
|
):
|
||||||
|
super().__init__(database, insights)
|
||||||
|
self.instrumentation = instrumentation
|
||||||
|
|
||||||
@mcp_tool(
|
@mcp_tool(
|
||||||
description="Submit feedback about your MCP experience - Use this tool when you encounter issues, have suggestions for improvement, or want to report positive experiences. AI models should proactively use this to help improve future MCP sessions. PRIVACY: Only include information you've already shared via previous MCP calls. Ask human permission before including any new sensitive data."
|
description="Submit feedback about your MCP experience - Use this tool when you encounter issues, have suggestions for improvement, or want to report positive experiences. AI models should proactively use this to help improve future MCP sessions. PRIVACY: Only include information you've already shared via previous MCP calls. Ask human permission before including any new sensitive data."
|
||||||
@@ -62,14 +84,23 @@ class SubmissionMixin(BaseFeedbackMixin):
|
|||||||
source="mcp_tool",
|
source="mcp_tool",
|
||||||
)
|
)
|
||||||
|
|
||||||
|
linked_calls = 0
|
||||||
|
if self.instrumentation is not None:
|
||||||
|
links = await self.instrumentation.capture_recent_calls()
|
||||||
|
if await self.instrumentation.link_feedback(feedback_id, links):
|
||||||
|
linked_calls = len(links)
|
||||||
|
|
||||||
self._record_tool_usage("submit_feedback", start_time, True)
|
self._record_tool_usage("submit_feedback", start_time, True)
|
||||||
logger.info(f"Feedback submitted successfully: {feedback_id}")
|
logger.info(f"Feedback submitted successfully: {feedback_id}")
|
||||||
|
|
||||||
return {
|
response = {
|
||||||
"success": True,
|
"success": True,
|
||||||
"feedback_id": feedback_id,
|
"feedback_id": feedback_id,
|
||||||
"message": "Feedback submitted successfully",
|
"message": "Feedback submitted successfully",
|
||||||
}
|
}
|
||||||
|
if self.instrumentation is not None:
|
||||||
|
response["linked_calls"] = linked_calls
|
||||||
|
return response
|
||||||
|
|
||||||
except Exception as e:
|
except Exception as e:
|
||||||
logger.error(f"Failed to submit feedback: {e}")
|
logger.error(f"Failed to submit feedback: {e}")
|
||||||
|
|||||||
@@ -17,6 +17,7 @@ def add_feedback_tools(
|
|||||||
insights: FeedbackInsights | None = None,
|
insights: FeedbackInsights | None = None,
|
||||||
prefix: str = "",
|
prefix: str = "",
|
||||||
separator: str = "_",
|
separator: str = "_",
|
||||||
|
instrumentation=None,
|
||||||
) -> None:
|
) -> None:
|
||||||
"""Add all feedback tools to a FastMCP server.
|
"""Add all feedback tools to a FastMCP server.
|
||||||
|
|
||||||
@@ -29,6 +30,9 @@ def add_feedback_tools(
|
|||||||
insights: Optional insights instance. Creates default if not provided.
|
insights: Optional insights instance. Creates default if not provided.
|
||||||
prefix: Optional prefix for tool names.
|
prefix: Optional prefix for tool names.
|
||||||
separator: Separator between prefix and tool name.
|
separator: Separator between prefix and tool name.
|
||||||
|
instrumentation: an InstrumentationMiddleware (see ``instrument``).
|
||||||
|
When given, each submission is linked to the tool calls that led
|
||||||
|
up to it, and the response reports ``linked_calls``.
|
||||||
"""
|
"""
|
||||||
try:
|
try:
|
||||||
# Initialize database
|
# Initialize database
|
||||||
@@ -39,7 +43,7 @@ def add_feedback_tools(
|
|||||||
insights = FeedbackInsights()
|
insights = FeedbackInsights()
|
||||||
|
|
||||||
# Create mixin instances
|
# Create mixin instances
|
||||||
submission_mixin = SubmissionMixin(database, insights)
|
submission_mixin = SubmissionMixin(database, insights, instrumentation)
|
||||||
retrieval_mixin = RetrievalMixin(database, insights)
|
retrieval_mixin = RetrievalMixin(database, insights)
|
||||||
management_mixin = ManagementMixin(database, insights)
|
management_mixin = ManagementMixin(database, insights)
|
||||||
|
|
||||||
|
|||||||
+22
@@ -0,0 +1,22 @@
|
|||||||
|
"""A stdio MCP server used by test_correlation to check session linking end to end."""
|
||||||
|
|
||||||
|
from fastmcp import Context, FastMCP
|
||||||
|
|
||||||
|
from fastmcp_feedback.instrumentation import MemorySink, instrument
|
||||||
|
|
||||||
|
app = FastMCP("stdio-correlation")
|
||||||
|
mw = instrument(app, [MemorySink()])
|
||||||
|
|
||||||
|
|
||||||
|
@app.tool
|
||||||
|
def echo(text: str) -> str:
|
||||||
|
return text
|
||||||
|
|
||||||
|
|
||||||
|
@app.tool
|
||||||
|
async def report(ctx: Context) -> list:
|
||||||
|
return [[link.tool, link.rule] for link in await mw.capture_recent_calls(ctx)]
|
||||||
|
|
||||||
|
|
||||||
|
if __name__ == "__main__":
|
||||||
|
app.run(show_banner=False)
|
||||||
@@ -0,0 +1,416 @@
|
|||||||
|
"""Tests for linking feedback to the tool calls that preceded it."""
|
||||||
|
|
||||||
|
import asyncio
|
||||||
|
import contextlib
|
||||||
|
import json
|
||||||
|
import sys
|
||||||
|
from datetime import UTC, datetime, timedelta
|
||||||
|
from pathlib import Path
|
||||||
|
|
||||||
|
import fastmcp
|
||||||
|
import pytest
|
||||||
|
from fastmcp import Client, Context, FastMCP
|
||||||
|
from fastmcp.client.transports import PythonStdioTransport
|
||||||
|
from fastmcp.exceptions import ToolError
|
||||||
|
|
||||||
|
from fastmcp_feedback import add_feedback_tools
|
||||||
|
from fastmcp_feedback.instrumentation import (
|
||||||
|
DatabaseSink,
|
||||||
|
MemorySink,
|
||||||
|
instrument,
|
||||||
|
)
|
||||||
|
from fastmcp_feedback.instrumentation import middleware as mw_module
|
||||||
|
from fastmcp_feedback.instrumentation.correlation import (
|
||||||
|
RULE_SESSION,
|
||||||
|
RULE_USER_WINDOW,
|
||||||
|
Candidate,
|
||||||
|
select_calls,
|
||||||
|
)
|
||||||
|
|
||||||
|
NOW = datetime(2026, 9, 27, 12, 0, tzinfo=UTC)
|
||||||
|
|
||||||
|
# FastMCP 4's Client speaks MCP 2026-07-28, which has no sessions, so each
|
||||||
|
# in-memory call gets a new ctx.session_id and there is no header or stdio
|
||||||
|
# signal to fall back on. Calls from one test client link by user instead.
|
||||||
|
FASTMCP_4 = int(fastmcp.__version__.split(".")[0]) >= 4
|
||||||
|
SAME_CLIENT_RULE = RULE_USER_WINDOW if FASTMCP_4 else RULE_SESSION
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.fixture
|
||||||
|
async def db(tmp_path):
|
||||||
|
"""Make DatabaseSinks whose engines are always disposed, even on failure.
|
||||||
|
|
||||||
|
An undisposed aiosqlite engine keeps a worker thread alive and the test
|
||||||
|
process never exits.
|
||||||
|
"""
|
||||||
|
made = []
|
||||||
|
|
||||||
|
def factory(name="calls", cls=DatabaseSink):
|
||||||
|
sink = cls(f"sqlite+aiosqlite:///{tmp_path}/{name}.db", create_tables=True)
|
||||||
|
made.append(sink)
|
||||||
|
return sink
|
||||||
|
|
||||||
|
yield factory
|
||||||
|
for sink in made:
|
||||||
|
await sink.engine.dispose()
|
||||||
|
|
||||||
|
|
||||||
|
def cand(id, minutes_ago, session="s1", user="u1", tool="t", ok=True):
|
||||||
|
return Candidate(id, tool, NOW - timedelta(minutes=minutes_ago), ok, session, user)
|
||||||
|
|
||||||
|
|
||||||
|
class TestSelectCalls:
|
||||||
|
def test_session_first_then_user_window_chronological(self):
|
||||||
|
links = select_calls(
|
||||||
|
[cand("a", 1), cand("b", 3, session="s2"), cand("c", 2)],
|
||||||
|
session_id="s1",
|
||||||
|
user_sub="u1",
|
||||||
|
since=NOW - timedelta(minutes=15),
|
||||||
|
limit=10,
|
||||||
|
)
|
||||||
|
assert [(link.call_id, link.rule, link.position) for link in links] == [
|
||||||
|
("b", RULE_USER_WINDOW, 0),
|
||||||
|
("c", RULE_SESSION, 1),
|
||||||
|
("a", RULE_SESSION, 2),
|
||||||
|
]
|
||||||
|
|
||||||
|
def test_session_matches_take_priority_over_newer_user_calls(self):
|
||||||
|
links = select_calls(
|
||||||
|
[cand("old", 10), cand("new", 1, session="s2")],
|
||||||
|
session_id="s1",
|
||||||
|
user_sub="u1",
|
||||||
|
since=NOW - timedelta(minutes=15),
|
||||||
|
limit=1,
|
||||||
|
)
|
||||||
|
assert [(link.call_id, link.rule) for link in links] == [("old", RULE_SESSION)]
|
||||||
|
|
||||||
|
def test_user_window_excludes_old_calls_and_other_users(self):
|
||||||
|
links = select_calls(
|
||||||
|
[
|
||||||
|
cand("recent", 5, session="s2"),
|
||||||
|
cand("stale", 30, session="s2"),
|
||||||
|
cand("other", 1, session="s3", user="u2"),
|
||||||
|
],
|
||||||
|
session_id="s9",
|
||||||
|
user_sub="u1",
|
||||||
|
since=NOW - timedelta(minutes=15),
|
||||||
|
limit=10,
|
||||||
|
)
|
||||||
|
assert [link.call_id for link in links] == ["recent"]
|
||||||
|
|
||||||
|
def test_limit_keeps_the_most_recent(self):
|
||||||
|
links = select_calls(
|
||||||
|
[cand(str(i), i) for i in range(10)],
|
||||||
|
session_id="s1",
|
||||||
|
user_sub=None,
|
||||||
|
since=NOW,
|
||||||
|
limit=3,
|
||||||
|
)
|
||||||
|
assert [link.call_id for link in links] == ["2", "1", "0"]
|
||||||
|
|
||||||
|
def test_duplicates_counted_once_and_naive_times_treated_as_utc(self):
|
||||||
|
naive = Candidate(
|
||||||
|
"a",
|
||||||
|
"t",
|
||||||
|
(NOW - timedelta(minutes=1)).replace(tzinfo=None),
|
||||||
|
True,
|
||||||
|
"s1",
|
||||||
|
"u1",
|
||||||
|
)
|
||||||
|
links = select_calls(
|
||||||
|
[naive, cand("a", 1)], session_id="s1", user_sub=None, since=NOW, limit=5
|
||||||
|
)
|
||||||
|
assert len(links) == 1 and links[0].started_at.tzinfo is not None
|
||||||
|
|
||||||
|
def test_no_session_or_user_means_no_links(self):
|
||||||
|
assert (
|
||||||
|
select_calls(
|
||||||
|
[cand("a", 1)], session_id=None, user_sub=None, since=NOW, limit=5
|
||||||
|
)
|
||||||
|
== []
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
def build(sinks=None, user="u1", **kwargs):
|
||||||
|
app = FastMCP("correlated")
|
||||||
|
|
||||||
|
@app.tool
|
||||||
|
def echo(text: str) -> str:
|
||||||
|
return text
|
||||||
|
|
||||||
|
@app.tool
|
||||||
|
def render(scene: str) -> str:
|
||||||
|
raise RuntimeError(f"render of {scene} hung")
|
||||||
|
|
||||||
|
captured = {}
|
||||||
|
|
||||||
|
@app.tool
|
||||||
|
async def report(ctx: Context) -> list:
|
||||||
|
links = await mw.capture_recent_calls(ctx)
|
||||||
|
captured["links"] = links
|
||||||
|
return [link.call_id for link in links]
|
||||||
|
|
||||||
|
identity = {"user": user}
|
||||||
|
mw = instrument(
|
||||||
|
app,
|
||||||
|
sinks if sinks is not None else [MemorySink()],
|
||||||
|
identity_resolver=lambda c: {"user_sub": identity["user"]},
|
||||||
|
**kwargs,
|
||||||
|
)
|
||||||
|
return app, mw, captured, identity
|
||||||
|
|
||||||
|
|
||||||
|
class TestCaptureRecentCalls:
|
||||||
|
async def test_same_session_calls_including_failures(self):
|
||||||
|
app, mw, captured, _ = build()
|
||||||
|
async with Client(app) as client:
|
||||||
|
await client.call_tool("echo", {"text": "a"})
|
||||||
|
with contextlib.suppress(ToolError):
|
||||||
|
await client.call_tool("render", {"scene": "castle"})
|
||||||
|
await client.call_tool("report", {})
|
||||||
|
|
||||||
|
links = captured["links"]
|
||||||
|
assert [(link.tool, link.ok, link.rule) for link in links] == [
|
||||||
|
("echo", True, SAME_CLIENT_RULE),
|
||||||
|
("render", False, SAME_CLIENT_RULE),
|
||||||
|
]
|
||||||
|
|
||||||
|
async def test_new_session_per_call_falls_back_to_user_window(self):
|
||||||
|
"""PAT/CI clients open a new MCP session for every call."""
|
||||||
|
app, mw, captured, _ = build()
|
||||||
|
for text in ("one", "two"):
|
||||||
|
async with Client(app) as client:
|
||||||
|
await client.call_tool("echo", {"text": text})
|
||||||
|
async with Client(app) as client:
|
||||||
|
await client.call_tool("report", {})
|
||||||
|
|
||||||
|
assert [link.rule for link in captured["links"]] == [RULE_USER_WINDOW] * 2
|
||||||
|
|
||||||
|
async def test_other_users_calls_are_not_linked(self):
|
||||||
|
app, mw, captured, identity = build()
|
||||||
|
identity["user"] = "someone-else"
|
||||||
|
async with Client(app) as client:
|
||||||
|
await client.call_tool("echo", {"text": "not mine"})
|
||||||
|
identity["user"] = "u1"
|
||||||
|
async with Client(app) as client:
|
||||||
|
await client.call_tool("report", {})
|
||||||
|
assert captured["links"] == []
|
||||||
|
|
||||||
|
async def test_calls_still_queued_are_found(self):
|
||||||
|
release = asyncio.Event()
|
||||||
|
|
||||||
|
class Stalled:
|
||||||
|
async def write(self, records):
|
||||||
|
await release.wait()
|
||||||
|
|
||||||
|
app, mw, captured, _ = build(sinks=[Stalled()])
|
||||||
|
async with Client(app) as client:
|
||||||
|
await client.call_tool("echo", {"text": "queued"})
|
||||||
|
await client.call_tool("report", {})
|
||||||
|
release.set()
|
||||||
|
assert [link.tool for link in captured["links"]] == ["echo"]
|
||||||
|
|
||||||
|
async def test_database_fallback_after_restart(self, db):
|
||||||
|
app1, mw1, _, _ = build(sinks=[db("restart")])
|
||||||
|
async with Client(app1) as client:
|
||||||
|
await client.call_tool("echo", {"text": "before restart"})
|
||||||
|
await mw1.aclose()
|
||||||
|
|
||||||
|
# A new process: empty in-memory buffer, same database.
|
||||||
|
app2, mw2, captured, _ = build(sinks=[db("restart")])
|
||||||
|
async with Client(app2) as client:
|
||||||
|
await client.call_tool("report", {})
|
||||||
|
await mw2.aclose()
|
||||||
|
|
||||||
|
[link] = captured["links"]
|
||||||
|
assert link.tool == "echo" and link.rule == RULE_USER_WINDOW
|
||||||
|
|
||||||
|
@pytest.mark.skipif(
|
||||||
|
FASTMCP_4, reason="FastMCP 4 Client is sessionless (MCP 2026-07-28)"
|
||||||
|
)
|
||||||
|
async def test_failing_resolver_still_links_by_session(self):
|
||||||
|
app, mw, captured, _ = build()
|
||||||
|
mw.identity_resolver = lambda c: 1 / 0
|
||||||
|
async with Client(app) as client:
|
||||||
|
await client.call_tool("echo", {"text": "a"})
|
||||||
|
await client.call_tool("report", {})
|
||||||
|
assert [link.rule for link in captured["links"]] == [RULE_SESSION]
|
||||||
|
|
||||||
|
async def test_slow_database_is_skipped(self, db):
|
||||||
|
class SlowDb(DatabaseSink):
|
||||||
|
async def recent_calls(self, **kwargs):
|
||||||
|
await asyncio.sleep(5)
|
||||||
|
return []
|
||||||
|
|
||||||
|
sink = db("slow", SlowDb)
|
||||||
|
app, mw, captured, _ = build(sinks=[sink], correlation_timeout=0.05)
|
||||||
|
async with Client(app) as client:
|
||||||
|
await client.call_tool("echo", {"text": "a"})
|
||||||
|
await client.call_tool("report", {})
|
||||||
|
await mw.aclose()
|
||||||
|
# The lookup timed out, so capture returned [] rather than hanging.
|
||||||
|
assert captured["links"] == []
|
||||||
|
|
||||||
|
async def test_outside_a_tool_call_returns_empty(self):
|
||||||
|
_, mw, _, _ = build()
|
||||||
|
mw.identity_resolver = None
|
||||||
|
assert await mw.capture_recent_calls() == []
|
||||||
|
|
||||||
|
|
||||||
|
class TestLinksAndContext:
|
||||||
|
async def test_link_and_read_back_with_external_ids(self, db):
|
||||||
|
sink = db("links")
|
||||||
|
app, mw, captured, _ = build(sinks=[sink])
|
||||||
|
async with Client(app) as client:
|
||||||
|
await client.call_tool("echo", {"text": "a"})
|
||||||
|
with contextlib.suppress(ToolError):
|
||||||
|
await client.call_tool("render", {"scene": "castle"})
|
||||||
|
await client.call_tool("report", {})
|
||||||
|
|
||||||
|
assert await mw.link_feedback("bug-7Q2X", captured["links"])
|
||||||
|
await mw.flush()
|
||||||
|
context = await mw.feedback_context("bug-7Q2X")
|
||||||
|
await mw.aclose()
|
||||||
|
|
||||||
|
assert [(c["position"], c["tool"], c["ok"], c["rule"]) for c in context] == [
|
||||||
|
(0, "echo", True, SAME_CLIENT_RULE),
|
||||||
|
(1, "render", False, SAME_CLIENT_RULE),
|
||||||
|
]
|
||||||
|
assert context[1]["error_type"] == "RuntimeError"
|
||||||
|
|
||||||
|
async def test_relinking_replaces_links(self, db):
|
||||||
|
sink = db("relink")
|
||||||
|
app, mw, captured, _ = build(sinks=[sink])
|
||||||
|
async with Client(app) as client:
|
||||||
|
await client.call_tool("echo", {"text": "a"})
|
||||||
|
await client.call_tool("report", {})
|
||||||
|
links = captured["links"]
|
||||||
|
await mw.link_feedback("fb-1", links)
|
||||||
|
await mw.link_feedback("fb-1", links)
|
||||||
|
assert len(await mw.feedback_context("fb-1")) == 1
|
||||||
|
await mw.aclose()
|
||||||
|
|
||||||
|
async def test_context_fills_unflushed_calls_from_memory(self, db):
|
||||||
|
release = asyncio.Event()
|
||||||
|
|
||||||
|
class StalledDb(DatabaseSink):
|
||||||
|
async def write(self, records):
|
||||||
|
await release.wait()
|
||||||
|
|
||||||
|
sink = db("unflushed", StalledDb)
|
||||||
|
app, mw, captured, _ = build(sinks=[sink])
|
||||||
|
async with Client(app) as client:
|
||||||
|
await client.call_tool("echo", {"text": "a"})
|
||||||
|
await client.call_tool("report", {})
|
||||||
|
await mw.link_feedback("fb-2", captured["links"])
|
||||||
|
|
||||||
|
[row] = await mw.feedback_context("fb-2")
|
||||||
|
release.set()
|
||||||
|
await mw.aclose()
|
||||||
|
assert row["tool"] == "echo" and row["ok"] is True
|
||||||
|
|
||||||
|
async def test_link_without_database_sink_is_false(self):
|
||||||
|
_, mw, _, _ = build()
|
||||||
|
assert await mw.link_feedback("x", []) is False
|
||||||
|
assert await mw.feedback_context("x") == []
|
||||||
|
|
||||||
|
|
||||||
|
class TestSubmitFeedbackAutoLinks:
|
||||||
|
async def test_submission_links_preceding_calls(self, db):
|
||||||
|
app = FastMCP("app")
|
||||||
|
|
||||||
|
@app.tool
|
||||||
|
def render(scene: str) -> str:
|
||||||
|
raise RuntimeError("hung")
|
||||||
|
|
||||||
|
sink = db("submit")
|
||||||
|
mw = instrument(app, [sink], identity_resolver=lambda c: {"user_sub": "u1"})
|
||||||
|
add_feedback_tools(app, database_url="sqlite:///:memory:", instrumentation=mw)
|
||||||
|
|
||||||
|
async with Client(app) as client:
|
||||||
|
with contextlib.suppress(ToolError):
|
||||||
|
await client.call_tool("render", {"scene": "castle"})
|
||||||
|
result = await client.call_tool(
|
||||||
|
"submit_feedback",
|
||||||
|
{
|
||||||
|
"request": {
|
||||||
|
"type": "bug",
|
||||||
|
"title": "render hangs",
|
||||||
|
"description": "the render tool hung",
|
||||||
|
"submitter": "model",
|
||||||
|
}
|
||||||
|
},
|
||||||
|
)
|
||||||
|
|
||||||
|
assert result.data["linked_calls"] == 1
|
||||||
|
await mw.flush()
|
||||||
|
[row] = await mw.feedback_context(result.data["feedback_id"])
|
||||||
|
await mw.aclose()
|
||||||
|
assert (row["tool"], row["ok"], row["error_type"]) == (
|
||||||
|
"render",
|
||||||
|
False,
|
||||||
|
"RuntimeError",
|
||||||
|
)
|
||||||
|
|
||||||
|
async def test_without_instrumentation_response_unchanged(self):
|
||||||
|
app = FastMCP("plain")
|
||||||
|
add_feedback_tools(app, database_url="sqlite:///:memory:")
|
||||||
|
async with Client(app) as client:
|
||||||
|
result = await client.call_tool(
|
||||||
|
"submit_feedback",
|
||||||
|
{
|
||||||
|
"request": {
|
||||||
|
"type": "bug",
|
||||||
|
"title": "t",
|
||||||
|
"description": "no instrumentation",
|
||||||
|
"submitter": "m",
|
||||||
|
}
|
||||||
|
},
|
||||||
|
)
|
||||||
|
assert "linked_calls" not in result.data
|
||||||
|
|
||||||
|
|
||||||
|
class FakeContext:
|
||||||
|
def __init__(self, transport=None, session_id="ctx-session"):
|
||||||
|
self.transport = transport
|
||||||
|
self.session_id = session_id
|
||||||
|
|
||||||
|
|
||||||
|
class TestSessionKey:
|
||||||
|
def test_http_session_header_wins(self, monkeypatch):
|
||||||
|
import fastmcp.server.dependencies as deps
|
||||||
|
|
||||||
|
monkeypatch.setattr(
|
||||||
|
deps,
|
||||||
|
"get_http_headers",
|
||||||
|
lambda include_all=False: {"mcp-session-id": "hdr-1"},
|
||||||
|
)
|
||||||
|
assert (
|
||||||
|
mw_module.session_key(FakeContext(transport="streamable-http")) == "hdr-1"
|
||||||
|
)
|
||||||
|
|
||||||
|
def test_stdio_uses_one_key_per_process(self):
|
||||||
|
a = mw_module.session_key(FakeContext(transport="stdio", session_id="x"))
|
||||||
|
b = mw_module.session_key(FakeContext(transport="stdio", session_id="y"))
|
||||||
|
assert a == b and a.startswith("stdio:")
|
||||||
|
|
||||||
|
def test_falls_back_to_ctx_session_id(self):
|
||||||
|
assert mw_module.session_key(FakeContext()) == "ctx-session"
|
||||||
|
|
||||||
|
def test_no_context(self):
|
||||||
|
assert mw_module.session_key(None) is None
|
||||||
|
|
||||||
|
|
||||||
|
class TestStdioEndToEnd:
|
||||||
|
async def test_same_stdio_client_links_by_session(self):
|
||||||
|
"""On a real stdio transport, one client's calls link by session on every FastMCP version."""
|
||||||
|
server = (
|
||||||
|
Path(__file__).parent.parent / "fixtures" / "stdio_correlation_server.py"
|
||||||
|
)
|
||||||
|
transport = PythonStdioTransport(str(server), python_cmd=sys.executable)
|
||||||
|
async with Client(transport) as client:
|
||||||
|
await client.call_tool("echo", {"text": "a"})
|
||||||
|
await client.call_tool("echo", {"text": "b"})
|
||||||
|
result = await client.call_tool("report", {})
|
||||||
|
pairs = json.loads(result.content[0].text)
|
||||||
|
assert pairs == [["echo", RULE_SESSION], ["echo", RULE_SESSION]]
|
||||||
@@ -408,7 +408,7 @@ class TestDatabaseSink:
|
|||||||
async with engine.connect() as conn:
|
async with engine.connect() as conn:
|
||||||
tables = await conn.run_sync(lambda c: inspect(c).get_table_names())
|
tables = await conn.run_sync(lambda c: inspect(c).get_table_names())
|
||||||
await engine.dispose()
|
await engine.dispose()
|
||||||
assert tables == ["qa_tool_calls"]
|
assert sorted(tables) == ["qa_feedback_call_links", "qa_tool_calls"]
|
||||||
|
|
||||||
async def test_no_tables_created_by_default(self, tmp_path):
|
async def test_no_tables_created_by_default(self, tmp_path):
|
||||||
engine = create_async_engine(f"sqlite+aiosqlite:///{tmp_path}/host.db")
|
engine = create_async_engine(f"sqlite+aiosqlite:///{tmp_path}/host.db")
|
||||||
@@ -444,7 +444,7 @@ class TestDatabaseSink:
|
|||||||
host = MetaData()
|
host = MetaData()
|
||||||
assert build_metadata("ffb_", host) is host
|
assert build_metadata("ffb_", host) is host
|
||||||
build_metadata("ffb_", host)
|
build_metadata("ffb_", host)
|
||||||
assert list(host.tables) == ["ffb_tool_calls"]
|
assert sorted(host.tables) == ["ffb_feedback_call_links", "ffb_tool_calls"]
|
||||||
|
|
||||||
|
|
||||||
class TestIntegration:
|
class TestIntegration:
|
||||||
|
|||||||
@@ -0,0 +1,397 @@
|
|||||||
|
"""Tests for classifying error results that tools return instead of raising.
|
||||||
|
|
||||||
|
Real FastMCP servers through fastmcp.Client, as in test_instrumentation.py.
|
||||||
|
"""
|
||||||
|
|
||||||
|
import contextlib
|
||||||
|
import logging
|
||||||
|
|
||||||
|
import pytest
|
||||||
|
from fastmcp import Client, FastMCP
|
||||||
|
from fastmcp.exceptions import ToolError
|
||||||
|
from mcp.types import CallToolResult, TextContent
|
||||||
|
from sqlalchemy import select
|
||||||
|
|
||||||
|
from fastmcp_feedback import add_feedback_tools
|
||||||
|
from fastmcp_feedback.instrumentation import (
|
||||||
|
DEFAULT_RESULT_CLASSIFIER,
|
||||||
|
REDACTED,
|
||||||
|
DatabaseSink,
|
||||||
|
MemorySink,
|
||||||
|
ToolCallRecord,
|
||||||
|
instrument,
|
||||||
|
soft_error_classifier,
|
||||||
|
)
|
||||||
|
from fastmcp_feedback.mixins import submission
|
||||||
|
|
||||||
|
BMCP_TOKEN = "bmcp_" + "A1b2C3d4E5f6G7h8I9j0"
|
||||||
|
|
||||||
|
try:
|
||||||
|
from fastmcp.tools.tool import ToolResult
|
||||||
|
except ImportError: # FastMCP 4
|
||||||
|
from fastmcp.tools import ToolResult
|
||||||
|
|
||||||
|
try:
|
||||||
|
ToolResult(content=[TextContent(type="text", text="x")], is_error=True)
|
||||||
|
TOOL_RESULT_IS_ERROR = True
|
||||||
|
except TypeError: # added in FastMCP 3.4
|
||||||
|
TOOL_RESULT_IS_ERROR = False
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.fixture
|
||||||
|
async def db(tmp_path):
|
||||||
|
"""DatabaseSinks whose engines are always disposed, or the process hangs."""
|
||||||
|
made = []
|
||||||
|
|
||||||
|
def factory(name="calls"):
|
||||||
|
sink = DatabaseSink(
|
||||||
|
f"sqlite+aiosqlite:///{tmp_path}/{name}.db", create_tables=True
|
||||||
|
)
|
||||||
|
made.append(sink)
|
||||||
|
return sink
|
||||||
|
|
||||||
|
yield factory
|
||||||
|
for sink in made:
|
||||||
|
await sink.engine.dispose()
|
||||||
|
|
||||||
|
|
||||||
|
def make_app(sinks=None, **kwargs):
|
||||||
|
sink = MemorySink()
|
||||||
|
app = FastMCP("soft")
|
||||||
|
|
||||||
|
@app.tool
|
||||||
|
def render(status: str) -> dict:
|
||||||
|
return {"status": status, "error": f"render said {status}"}
|
||||||
|
|
||||||
|
@app.tool
|
||||||
|
def respond(payload: dict) -> dict:
|
||||||
|
return payload
|
||||||
|
|
||||||
|
@app.tool
|
||||||
|
def text(value: str) -> str:
|
||||||
|
return value
|
||||||
|
|
||||||
|
@app.tool
|
||||||
|
def items() -> list:
|
||||||
|
return [{"status": "failed"}]
|
||||||
|
|
||||||
|
@app.tool
|
||||||
|
def crash() -> dict:
|
||||||
|
raise RuntimeError("exploded")
|
||||||
|
|
||||||
|
@app.tool
|
||||||
|
def mcp_error() -> CallToolResult:
|
||||||
|
return CallToolResult(
|
||||||
|
content=[TextContent(type="text", text="target missing")], isError=True
|
||||||
|
)
|
||||||
|
|
||||||
|
mw = instrument(app, sinks if sinks is not None else [sink], **kwargs)
|
||||||
|
return app, sink, mw
|
||||||
|
|
||||||
|
|
||||||
|
async def run(app, mw, tool, args=None):
|
||||||
|
async with Client(app) as client:
|
||||||
|
with contextlib.suppress(ToolError):
|
||||||
|
await client.call_tool(tool, args or {})
|
||||||
|
await mw.flush()
|
||||||
|
|
||||||
|
|
||||||
|
async def record_for(tool, args=None, **kwargs) -> ToolCallRecord:
|
||||||
|
app, sink, mw = make_app(**kwargs)
|
||||||
|
await run(app, mw, tool, args)
|
||||||
|
[record] = sink.records
|
||||||
|
return record
|
||||||
|
|
||||||
|
|
||||||
|
class TestClassifier:
|
||||||
|
@pytest.mark.parametrize(
|
||||||
|
"result, expected",
|
||||||
|
[
|
||||||
|
({"status": "failed", "error": "disk full"}, "status=failed: disk full"),
|
||||||
|
({"state": "ERROR"}, "state=ERROR"),
|
||||||
|
({"ok": False, "message": "no scene"}, "ok=False: no scene"),
|
||||||
|
({"success": False, "error": "nope"}, "success=False: nope"),
|
||||||
|
({"success": False, "detail": "d", "reason": "r"}, "success=False: d"),
|
||||||
|
(
|
||||||
|
{"status": "failure", "error": {"code": 7, "message": "inner"}},
|
||||||
|
"status=failure: inner",
|
||||||
|
),
|
||||||
|
({"isError": True, "content": ["boom"]}, "isError: boom"),
|
||||||
|
(
|
||||||
|
{"is_error": True, "content": [{"type": "text", "text": "a"}]},
|
||||||
|
"isError: a",
|
||||||
|
),
|
||||||
|
],
|
||||||
|
)
|
||||||
|
def test_flagged(self, result, expected):
|
||||||
|
assert DEFAULT_RESULT_CLASSIFIER("t", result) == expected
|
||||||
|
|
||||||
|
@pytest.mark.parametrize(
|
||||||
|
"result",
|
||||||
|
[
|
||||||
|
{"status": "done"},
|
||||||
|
{"status": "unknown_target"},
|
||||||
|
{"ok": True},
|
||||||
|
{"ok": 0},
|
||||||
|
{"success": None},
|
||||||
|
{"result": False},
|
||||||
|
{"state": 3},
|
||||||
|
{"isError": False},
|
||||||
|
[{"status": "failed"}],
|
||||||
|
"failed",
|
||||||
|
None,
|
||||||
|
],
|
||||||
|
)
|
||||||
|
def test_not_flagged(self, result):
|
||||||
|
assert DEFAULT_RESULT_CLASSIFIER("t", result) is None
|
||||||
|
|
||||||
|
def test_extra_statuses(self):
|
||||||
|
classify = soft_error_classifier(extra_statuses=("unknown_target",))
|
||||||
|
assert (
|
||||||
|
classify("t", {"status": "unknown_target", "error": "no cube"})
|
||||||
|
== "status=unknown_target: no cube"
|
||||||
|
)
|
||||||
|
assert classify("t", {"status": "failed"}) == "status=failed"
|
||||||
|
|
||||||
|
def test_custom_keys_replace_defaults(self):
|
||||||
|
classify = soft_error_classifier(
|
||||||
|
statuses=("bad",), status_keys=("phase",), flag_keys=("passed",)
|
||||||
|
)
|
||||||
|
assert classify("t", {"phase": "bad"}) == "phase=bad"
|
||||||
|
assert classify("t", {"passed": False}) == "passed=False"
|
||||||
|
assert classify("t", {"status": "failed", "ok": False}) is None
|
||||||
|
|
||||||
|
|
||||||
|
class TestMiddleware:
|
||||||
|
async def test_status_failed_is_soft_error(self):
|
||||||
|
record = await record_for("render", {"status": "failed"})
|
||||||
|
assert (record.outcome, record.ok, record.error_type) == (
|
||||||
|
"soft_error",
|
||||||
|
False,
|
||||||
|
"SoftError",
|
||||||
|
)
|
||||||
|
assert record.error_message == "status=failed: render said failed"
|
||||||
|
|
||||||
|
@pytest.mark.parametrize("key", ["ok", "success"])
|
||||||
|
async def test_false_flags_are_soft_errors(self, key):
|
||||||
|
record = await record_for("respond", {"payload": {key: False, "error": "x"}})
|
||||||
|
assert (record.outcome, record.ok) == ("soft_error", False)
|
||||||
|
assert record.error_message == f"{key}=False: x"
|
||||||
|
|
||||||
|
async def test_message_is_redacted_and_truncated(self):
|
||||||
|
record = await record_for(
|
||||||
|
"respond",
|
||||||
|
{"payload": {"status": "error", "error": f"bad key {BMCP_TOKEN} " * 50}},
|
||||||
|
max_error_chars=60,
|
||||||
|
)
|
||||||
|
assert BMCP_TOKEN not in record.error_message
|
||||||
|
assert REDACTED in record.error_message
|
||||||
|
assert len(record.error_message) == 60
|
||||||
|
|
||||||
|
async def test_extra_statuses(self):
|
||||||
|
record = await record_for(
|
||||||
|
"render",
|
||||||
|
{"status": "unknown_target"},
|
||||||
|
result_classifier=soft_error_classifier(extra_statuses=("unknown_target",)),
|
||||||
|
)
|
||||||
|
assert record.outcome == "soft_error"
|
||||||
|
assert (
|
||||||
|
record.error_message == "status=unknown_target: render said unknown_target"
|
||||||
|
)
|
||||||
|
|
||||||
|
async def test_unknown_status_is_ok_by_default(self):
|
||||||
|
record = await record_for("render", {"status": "unknown_target"})
|
||||||
|
assert (record.outcome, record.ok, record.error_type) == ("ok", True, None)
|
||||||
|
|
||||||
|
@pytest.mark.parametrize(
|
||||||
|
"tool, args",
|
||||||
|
[
|
||||||
|
("text", {"value": "failed"}),
|
||||||
|
("items", {}),
|
||||||
|
("respond", {"payload": {"status": "done", "count": 3}}),
|
||||||
|
],
|
||||||
|
)
|
||||||
|
async def test_normal_results_stay_ok(self, tool, args):
|
||||||
|
record = await record_for(tool, args)
|
||||||
|
assert (record.outcome, record.ok, record.error_message) == ("ok", True, None)
|
||||||
|
|
||||||
|
async def test_none_disables(self):
|
||||||
|
record = await record_for(
|
||||||
|
"render", {"status": "failed"}, result_classifier=None
|
||||||
|
)
|
||||||
|
assert (record.outcome, record.ok, record.error_type) == ("ok", True, None)
|
||||||
|
|
||||||
|
async def test_custom_classifier(self):
|
||||||
|
seen = []
|
||||||
|
|
||||||
|
def classify(tool, result):
|
||||||
|
seen.append((tool, result))
|
||||||
|
return "empty" if result.get("count") == 0 else None
|
||||||
|
|
||||||
|
record = await record_for(
|
||||||
|
"respond", {"payload": {"count": 0}}, result_classifier=classify
|
||||||
|
)
|
||||||
|
assert seen == [("respond", {"count": 0})]
|
||||||
|
assert (record.outcome, record.error_message) == ("soft_error", "empty")
|
||||||
|
|
||||||
|
async def test_raising_classifier_records_ok(self, caplog):
|
||||||
|
def classify(tool, result):
|
||||||
|
raise KeyError("classifier bug")
|
||||||
|
|
||||||
|
app, sink, mw = make_app(result_classifier=classify)
|
||||||
|
with caplog.at_level(logging.WARNING):
|
||||||
|
async with Client(app) as client:
|
||||||
|
result = await client.call_tool("render", {"status": "failed"})
|
||||||
|
await mw.flush()
|
||||||
|
assert result.data["status"] == "failed"
|
||||||
|
[record] = sink.records
|
||||||
|
assert (record.outcome, record.ok, record.error_type) == ("ok", True, None)
|
||||||
|
assert "Result classifier" in caplog.text
|
||||||
|
|
||||||
|
async def test_classifier_not_run_on_exceptions(self):
|
||||||
|
calls = []
|
||||||
|
record = await record_for(
|
||||||
|
"crash", result_classifier=lambda t, r: calls.append(t) or "x"
|
||||||
|
)
|
||||||
|
assert calls == []
|
||||||
|
assert (record.outcome, record.ok, record.error_type) == (
|
||||||
|
"error",
|
||||||
|
False,
|
||||||
|
"RuntimeError",
|
||||||
|
)
|
||||||
|
assert record.error_message == "exploded"
|
||||||
|
|
||||||
|
async def test_mcp_error_result_without_raising(self):
|
||||||
|
"""A CallToolResult with isError, on every FastMCP version."""
|
||||||
|
record = await record_for("mcp_error")
|
||||||
|
assert (record.outcome, record.ok) == ("soft_error", False)
|
||||||
|
assert record.error_message == "isError: target missing"
|
||||||
|
|
||||||
|
@pytest.mark.skipif(not TOOL_RESULT_IS_ERROR, reason="ToolResult.is_error is 3.4+")
|
||||||
|
async def test_tool_result_is_error(self):
|
||||||
|
app = FastMCP("is_error")
|
||||||
|
|
||||||
|
@app.tool
|
||||||
|
def lookup():
|
||||||
|
return ToolResult(
|
||||||
|
content=[TextContent(type="text", text="not found")], is_error=True
|
||||||
|
)
|
||||||
|
|
||||||
|
sink = MemorySink()
|
||||||
|
mw = instrument(app, [sink])
|
||||||
|
async with Client(app) as client:
|
||||||
|
await client.call_tool("lookup", {}, raise_on_error=False)
|
||||||
|
await mw.flush()
|
||||||
|
[record] = sink.records
|
||||||
|
assert (record.outcome, record.error_message) == (
|
||||||
|
"soft_error",
|
||||||
|
"isError: not found",
|
||||||
|
)
|
||||||
|
|
||||||
|
def test_record_outcome_defaults_from_ok(self):
|
||||||
|
from datetime import UTC, datetime
|
||||||
|
|
||||||
|
now = datetime.now(UTC)
|
||||||
|
assert ToolCallRecord("t", now, 1.0, True, "meta").outcome == "ok"
|
||||||
|
assert ToolCallRecord("t", now, 1.0, False, "meta").outcome == "error"
|
||||||
|
|
||||||
|
|
||||||
|
class TestStorageAndCorrelation:
|
||||||
|
async def test_database_sink_writes_outcome(self, db):
|
||||||
|
sink = db()
|
||||||
|
app, _, mw = make_app(sinks=[sink])
|
||||||
|
async with Client(app) as client:
|
||||||
|
await client.call_tool("render", {"status": "done"})
|
||||||
|
await client.call_tool("render", {"status": "failed"})
|
||||||
|
with contextlib.suppress(ToolError):
|
||||||
|
await client.call_tool("crash", {})
|
||||||
|
await mw.flush()
|
||||||
|
|
||||||
|
t = sink.table
|
||||||
|
async with sink.engine.connect() as conn:
|
||||||
|
rows = (
|
||||||
|
await conn.execute(
|
||||||
|
select(t.c.outcome, t.c.ok, t.c.error_type).order_by(t.c.started_at)
|
||||||
|
)
|
||||||
|
).all()
|
||||||
|
assert [tuple(r) for r in rows] == [
|
||||||
|
("ok", True, None),
|
||||||
|
("soft_error", False, "SoftError"),
|
||||||
|
("error", False, "RuntimeError"),
|
||||||
|
]
|
||||||
|
async with sink.engine.connect() as conn:
|
||||||
|
[(failing,)] = (
|
||||||
|
await conn.execute(
|
||||||
|
select(t.c.tool).where(~t.c.ok, t.c.outcome != "error")
|
||||||
|
)
|
||||||
|
).all()
|
||||||
|
assert failing == "render"
|
||||||
|
|
||||||
|
async def test_links_and_feedback_context_show_soft_errors(self, db):
|
||||||
|
app, _, mw = make_app(
|
||||||
|
sinks=[db()], identity_resolver=lambda c: {"user_sub": "u1"}
|
||||||
|
)
|
||||||
|
captured = {}
|
||||||
|
|
||||||
|
@app.tool
|
||||||
|
async def report() -> int:
|
||||||
|
captured["links"] = await mw.capture_recent_calls()
|
||||||
|
await mw.link_feedback("bug-1", captured["links"])
|
||||||
|
return len(captured["links"])
|
||||||
|
|
||||||
|
async with Client(app) as client:
|
||||||
|
await client.call_tool("render", {"status": "failed"})
|
||||||
|
await client.call_tool("report", {})
|
||||||
|
|
||||||
|
links = captured["links"]
|
||||||
|
assert [(link.tool, link.ok) for link in links] == [("render", False)]
|
||||||
|
# Filled from memory or read from the database, depending on timing.
|
||||||
|
[row] = await mw.feedback_context("bug-1")
|
||||||
|
assert (row["outcome"], row["ok"], row["error_type"]) == (
|
||||||
|
"soft_error",
|
||||||
|
False,
|
||||||
|
"SoftError",
|
||||||
|
)
|
||||||
|
await mw.flush()
|
||||||
|
[row] = await mw.feedback_context("bug-1")
|
||||||
|
assert (row["tool"], row["outcome"], row["ok"]) == (
|
||||||
|
"render",
|
||||||
|
"soft_error",
|
||||||
|
False,
|
||||||
|
)
|
||||||
|
await mw.aclose()
|
||||||
|
|
||||||
|
async def test_submit_feedback_failure_is_soft_error(self, monkeypatch):
|
||||||
|
def broken(request):
|
||||||
|
raise RuntimeError("database down, password=hunter2")
|
||||||
|
|
||||||
|
monkeypatch.setattr(submission, "create_feedback_from_request", broken)
|
||||||
|
app = FastMCP("fb")
|
||||||
|
sink = MemorySink()
|
||||||
|
mw = instrument(app, [sink])
|
||||||
|
add_feedback_tools(app, database_url="sqlite:///:memory:", instrumentation=mw)
|
||||||
|
|
||||||
|
async with Client(app) as client:
|
||||||
|
result = await client.call_tool(
|
||||||
|
"submit_feedback",
|
||||||
|
{
|
||||||
|
"request": {
|
||||||
|
"type": "bug",
|
||||||
|
"title": "t",
|
||||||
|
"description": "d",
|
||||||
|
"submitter": "m",
|
||||||
|
}
|
||||||
|
},
|
||||||
|
)
|
||||||
|
await mw.flush()
|
||||||
|
assert result.data["success"] is False
|
||||||
|
[record] = sink.records
|
||||||
|
assert (record.tool, record.outcome, record.ok, record.error_type) == (
|
||||||
|
"submit_feedback",
|
||||||
|
"soft_error",
|
||||||
|
False,
|
||||||
|
"SoftError",
|
||||||
|
)
|
||||||
|
assert record.error_message == (
|
||||||
|
f"success=False: database down, password={REDACTED}"
|
||||||
|
)
|
||||||
@@ -553,7 +553,7 @@ wheels = [
|
|||||||
|
|
||||||
[[package]]
|
[[package]]
|
||||||
name = "fastmcp-feedback"
|
name = "fastmcp-feedback"
|
||||||
version = "2026.9.27.2"
|
version = "2026.9.27.4"
|
||||||
source = { editable = "." }
|
source = { editable = "." }
|
||||||
dependencies = [
|
dependencies = [
|
||||||
{ name = "fastmcp" },
|
{ name = "fastmcp" },
|
||||||
|
|||||||
Reference in New Issue
Block a user