Add jmespath_tools contrib module for filtering tool responses

Adds a @filterable decorator that enables JMESPath-based filtering and
projection of MCP tool results. This allows LLM clients to reduce response
sizes by extracting only the data they need.

Features:
- @filterable decorator adds a `jmespath` parameter to any tool
- ToolResult TypedDict for consistent {success, data, error} structure
- Lazy import of jmespath with helpful error message
- Works with both sync and async tool functions

Install with: pip install fastmcp[jmespath]

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude <noreply@anthropic.com>

pyproject.toml

@@ -47,11 +47,12 @@ classifiers = [
 
 [project.optional-dependencies]
 openai = ["openai>=1.102.0"]
+jmespath = ["jmespath>=1.0.0"]
 
 [dependency-groups]
 dev = [
     "dirty-equals>=0.9.0",
-    "fastmcp[openai]",
+    "fastmcp[openai,jmespath]",
     # add optional dependencies for fastmcp dev
     "fastapi>=0.115.12",
     "inline-snapshot[dirty-equals]>=0.27.2",

src/fastmcp/contrib/jmespath_tools/__init__.py

@@ -0,0 +1,3 @@
+from .jmespath_tools import JmespathParam, ToolResult, filterable
+
+__all__ = ["JmespathParam", "ToolResult", "filterable"]

src/fastmcp/contrib/jmespath_tools/example.py

@@ -0,0 +1,79 @@
+"""Example usage of the @filterable decorator with JMESPath filtering.
+
+This example shows how to add JMESPath filtering to MCP tools,
+allowing clients to reduce response size by filtering/projecting results.
+"""
+
+from fastmcp import FastMCP
+from fastmcp.contrib.jmespath_tools import ToolResult, filterable
+
+mcp = FastMCP("Example Server")
+
+
+@mcp.tool
+@filterable
+async def get_users(limit: int = 100) -> ToolResult:
+    """Get users with optional JMESPath filtering.
+
+    Examples:
+        # Get all users
+        get_users(limit=100)
+
+        # Get just usernames
+        get_users(limit=100, jmespath="data.users[*].username")
+
+        # Get only active users
+        get_users(limit=100, jmespath="data.users[?status == 'active']")
+
+        # Get summary with count
+        get_users(limit=100, jmespath="{count: length(data.users), names: data.users[*].name}")
+    """
+    # Simulate fetching users from a database
+    users = [
+        {"id": i, "username": f"user{i}", "name": f"User {i}", "status": "active"}
+        for i in range(limit)
+    ]
+    return {
+        "success": True,
+        "data": {"users": users, "total": len(users)},
+        "error": None,
+    }
+
+
+@mcp.tool
+@filterable
+async def get_logs(
+    level: str | None = None,
+    limit: int = 50,
+) -> ToolResult:
+    """Get application logs with optional JMESPath filtering.
+
+    Examples:
+        # Get all logs
+        get_logs(limit=50)
+
+        # Get just error messages
+        get_logs(jmespath="data.logs[?level == 'ERROR'].message")
+
+        # Get log summary
+        get_logs(jmespath="{total: data.count, errors: length(data.logs[?level == 'ERROR'])}")
+    """
+    # Simulate fetching logs
+    levels = ["INFO", "WARNING", "ERROR", "DEBUG"]
+    logs = [
+        {
+            "timestamp": f"2024-01-01T00:00:{i:02d}Z",
+            "level": levels[i % 4],
+            "message": f"Log message {i}",
+        }
+        for i in range(limit)
+    ]
+
+    if level:
+        logs = [log for log in logs if log["level"] == level]
+
+    return {"success": True, "data": {"logs": logs, "count": len(logs)}, "error": None}
+
+
+if __name__ == "__main__":
+    mcp.run()

src/fastmcp/contrib/jmespath_tools/jmespath_tools.py

@@ -0,0 +1,173 @@
+"""JMESPath filtering decorator for MCP tools.
+
+Adds a `jmespath` parameter to tools that allows filtering/transforming results
+before returning them, reducing response size for large datasets.
+
+Requires the `jmespath` package: `pip install fastmcp[jmespath]`
+
+Example:
+    from fastmcp import FastMCP
+    from fastmcp.contrib.jmespath_tools import filterable
+
+    mcp = FastMCP("My Server")
+
+    @mcp.tool
+    @filterable
+    async def get_logs(limit: int = 100):
+        logs = await fetch_logs(limit)
+        return {"success": True, "data": {"logs": logs}, "error": None}
+
+    # Client can now filter results:
+    # get_logs(limit=100, jmespath="data.logs[?level == 'ERROR'].message")
+"""
+
+import inspect
+from functools import wraps
+from typing import Annotated, Any
+
+from pydantic import Field
+from typing_extensions import TypedDict
+
+
+def _get_jmespath():
+    """Lazy import of jmespath with helpful error message."""
+    try:
+        import jmespath
+
+        return jmespath
+    except ImportError as e:
+        raise ImportError(
+            "jmespath package is required for @filterable decorator. "
+            "Install it with: pip install fastmcp[jmespath]"
+        ) from e
+
+
+class ToolResult(TypedDict):
+    """Standard result wrapper for filterable tools.
+
+    Using a consistent wrapper allows jmespath filtering to work
+    without breaking schema validation - the filter transforms
+    the `data` field while keeping the wrapper structure intact.
+
+    Attributes:
+        success: Whether the tool call succeeded.
+        data: The actual content - original or filtered by jmespath.
+        error: Error message if success is False, None otherwise.
+    """
+
+    success: bool
+    data: Any
+    error: str | None
+
+
+# Type alias for the jmespath parameter with description
+JmespathParam = Annotated[
+    str | None,
+    Field(
+        default=None,
+        description="JMESPath expression to filter/project the result. "
+        "Examples: 'data.items[*].name' (extract values), "
+        "'data.items[?status == `active`]' (filter items), "
+        "'{count: length(data.items), names: data.items[*].name}' (project fields)",
+    ),
+]
+
+
+def filterable(fn):
+    """Decorator that adds jmespath filtering to a tool.
+
+    The decorated tool should return a dict with at least a `data` field
+    containing the content to filter. The standard pattern is to return
+    a ToolResult dict with {success, data, error}.
+
+    The jmespath filter applies to the entire result dict, so use
+    expressions like `data.items[*].name` to access nested data.
+
+    Usage:
+        @mcp.tool
+        @filterable
+        async def get_items(limit: int = 10) -> ToolResult:
+            items = await fetch_items(limit)
+            return {"success": True, "data": {"items": items}, "error": None}
+
+        # Without filter - returns full result
+        await get_items(limit=100)
+        # Returns: {"success": true, "data": {"items": [...]}, "error": null}
+
+        # With filter - returns transformed data
+        await get_items(limit=100, jmespath="data.items[*].name")
+        # Returns: {"success": true, "data": ["name1", "name2", ...], "error": null}
+
+    Args:
+        fn: The tool function to wrap.
+
+    Returns:
+        Wrapped function with jmespath parameter added.
+    """
+    if inspect.iscoroutinefunction(fn):
+
+        @wraps(fn)
+        async def async_wrapper(
+            *args, jmespath: str | None = None, **kwargs
+        ) -> ToolResult:
+            result = await fn(*args, **kwargs)
+            return _apply_filter(result, jmespath)
+
+        wrapper = async_wrapper
+    else:
+
+        @wraps(fn)
+        def sync_wrapper(*args, jmespath: str | None = None, **kwargs) -> ToolResult:
+            result = fn(*args, **kwargs)
+            return _apply_filter(result, jmespath)
+
+        wrapper = sync_wrapper
+
+    # Update signature to include jmespath parameter
+    sig = inspect.signature(fn)
+    params = list(sig.parameters.values())
+    params.append(
+        inspect.Parameter(
+            "jmespath",
+            inspect.Parameter.KEYWORD_ONLY,
+            default=None,
+            annotation=JmespathParam,
+        )
+    )
+    wrapper.__signature__ = sig.replace(parameters=params)
+
+    # Update annotations with jmespath param and ToolResult return type
+    wrapper.__annotations__ = {
+        **{k: v for k, v in fn.__annotations__.items() if k != "return"},
+        "jmespath": JmespathParam,
+        "return": ToolResult,
+    }
+
+    return wrapper
+
+
+def _apply_filter(result: dict, jmespath_expr: str | None) -> ToolResult:
+    """Apply jmespath filter to result if provided.
+
+    Args:
+        result: The tool result dict.
+        jmespath_expr: JMESPath expression to apply, or None.
+
+    Returns:
+        ToolResult with filtered data if expression provided,
+        or original result if no expression or if result indicates failure.
+    """
+    if not jmespath_expr:
+        return result
+
+    # Don't filter failed results
+    if not result.get("success", True):
+        return result
+
+    jmespath_lib = _get_jmespath()
+
+    try:
+        filtered = jmespath_lib.search(jmespath_expr, result)
+        return {"success": True, "data": filtered, "error": None}
+    except jmespath_lib.exceptions.JMESPathError as e:
+        return {"success": False, "data": None, "error": f"JMESPath error: {e}"}

tests/contrib/test_jmespath_tools.py

@@ -0,0 +1,211 @@
+"""Tests for JMESPath filtering utilities."""
+
+import inspect
+
+from fastmcp.contrib.jmespath_tools import JmespathParam, ToolResult, filterable
+
+
+class TestFilterableDecorator:
+    """Tests for the @filterable decorator."""
+
+    def test_decorator_adds_jmespath_to_signature(self):
+        """Decorator adds jmespath parameter to function signature."""
+
+        @filterable
+        async def my_tool(limit: int = 10) -> ToolResult:
+            return {"success": True, "data": {"items": []}, "error": None}
+
+        sig = inspect.signature(my_tool)
+        assert "jmespath" in sig.parameters
+        assert sig.parameters["jmespath"].default is None
+        assert sig.parameters["jmespath"].annotation == JmespathParam
+
+    def test_decorator_updates_annotations(self):
+        """Decorator updates __annotations__ for Pydantic compatibility."""
+
+        @filterable
+        async def my_tool(limit: int = 10) -> ToolResult:
+            return {"success": True, "data": {"items": []}, "error": None}
+
+        assert "jmespath" in my_tool.__annotations__
+        assert my_tool.__annotations__["jmespath"] == JmespathParam
+        assert my_tool.__annotations__["return"] == ToolResult
+
+    async def test_without_filter_returns_original(self):
+        """Without jmespath filter, returns original ToolResult."""
+
+        @filterable
+        async def my_tool(limit: int = 10) -> ToolResult:
+            items = [{"id": i, "name": f"item-{i}"} for i in range(limit)]
+            return {
+                "success": True,
+                "data": {"count": limit, "items": items},
+                "error": None,
+            }
+
+        result = await my_tool(limit=3)
+        assert result["success"] is True
+        assert result["data"]["count"] == 3
+        assert len(result["data"]["items"]) == 3
+        assert result["error"] is None
+
+    async def test_with_filter_transforms_data(self):
+        """With jmespath filter, transforms the data field."""
+
+        @filterable
+        async def my_tool(limit: int = 10) -> ToolResult:
+            items = [{"id": i, "name": f"item-{i}"} for i in range(limit)]
+            return {
+                "success": True,
+                "data": {"count": limit, "items": items},
+                "error": None,
+            }
+
+        # Extract just names - note the filter operates on entire result
+        result = await my_tool(limit=3, jmespath="data.items[*].name")
+        assert result["success"] is True
+        assert result["data"] == ["item-0", "item-1", "item-2"]
+        assert result["error"] is None
+
+    async def test_filter_with_condition(self):
+        """Filter can apply conditions."""
+
+        @filterable
+        async def my_tool() -> ToolResult:
+            items = [
+                {"id": 1, "status": "active"},
+                {"id": 2, "status": "inactive"},
+                {"id": 3, "status": "active"},
+            ]
+            return {"success": True, "data": {"items": items}, "error": None}
+
+        result = await my_tool(jmespath="data.items[?status == 'active']")
+        assert result["success"] is True
+        assert len(result["data"]) == 2
+        assert all(item["status"] == "active" for item in result["data"])
+
+    async def test_filter_with_projection(self):
+        """Filter can create projections."""
+
+        @filterable
+        async def my_tool() -> ToolResult:
+            items = [{"id": 1}, {"id": 2}, {"id": 3}]
+            return {
+                "success": True,
+                "data": {"count": 3, "items": items},
+                "error": None,
+            }
+
+        result = await my_tool(jmespath="{total: data.count, ids: data.items[*].id}")
+        assert result["success"] is True
+        assert result["data"] == {"total": 3, "ids": [1, 2, 3]}
+
+    async def test_invalid_filter_returns_error(self):
+        """Invalid jmespath expression returns error."""
+
+        @filterable
+        async def my_tool() -> ToolResult:
+            return {"success": True, "data": {"items": []}, "error": None}
+
+        result = await my_tool(jmespath="[*.bad.syntax")
+        assert result["success"] is False
+        assert result["data"] is None
+        assert result["error"] is not None
+        assert "JMESPath error" in result["error"]
+
+    async def test_filter_on_failed_result_passthrough(self):
+        """If original result is not successful, filter is not applied."""
+
+        @filterable
+        async def my_tool() -> ToolResult:
+            return {"success": False, "data": None, "error": "Something went wrong"}
+
+        result = await my_tool(jmespath="data.items[*].name")
+        assert result["success"] is False
+        assert result["data"] is None
+        assert result["error"] == "Something went wrong"
+
+    def test_sync_wrapper_works(self):
+        """Decorator works with sync functions too."""
+
+        @filterable
+        def my_tool(limit: int = 10) -> ToolResult:
+            items = [{"id": i} for i in range(limit)]
+            return {"success": True, "data": {"items": items}, "error": None}
+
+        # Without filter
+        result = my_tool(limit=2)
+        assert result["success"] is True
+        assert len(result["data"]["items"]) == 2
+
+        # With filter
+        result = my_tool(limit=2, jmespath="data.items[*].id")
+        assert result["data"] == [0, 1]
+
+    async def test_filter_extracts_nested_field(self):
+        """Filter can extract deeply nested fields."""
+
+        @filterable
+        async def my_tool() -> ToolResult:
+            return {
+                "success": True,
+                "data": {
+                    "response": {
+                        "users": [
+                            {"profile": {"name": "Alice"}},
+                            {"profile": {"name": "Bob"}},
+                        ]
+                    }
+                },
+                "error": None,
+            }
+
+        result = await my_tool(jmespath="data.response.users[*].profile.name")
+        assert result["success"] is True
+        assert result["data"] == ["Alice", "Bob"]
+
+    async def test_filter_with_length_function(self):
+        """Filter can use JMESPath built-in functions."""
+
+        @filterable
+        async def my_tool() -> ToolResult:
+            items = [{"id": i} for i in range(5)]
+            return {"success": True, "data": {"items": items}, "error": None}
+
+        result = await my_tool(jmespath="length(data.items)")
+        assert result["success"] is True
+        assert result["data"] == 5
+
+
+class TestToolResult:
+    """Tests for the ToolResult TypedDict."""
+
+    def test_tool_result_structure(self):
+        """ToolResult has the expected structure."""
+        result: ToolResult = {"success": True, "data": {"key": "value"}, "error": None}
+
+        assert result["success"] is True
+        assert result["data"] == {"key": "value"}
+        assert result["error"] is None
+
+    def test_tool_result_error_case(self):
+        """ToolResult can represent errors."""
+        result: ToolResult = {"success": False, "data": None, "error": "Error message"}
+
+        assert result["success"] is False
+        assert result["data"] is None
+        assert result["error"] == "Error message"
+
+
+class TestJmespathImportError:
+    """Tests for jmespath import error handling."""
+
+    def test_jmespath_import_works(self):
+        """Verify jmespath is available in test environment."""
+        from fastmcp.contrib.jmespath_tools.jmespath_tools import _get_jmespath
+
+        jmespath = _get_jmespath()
+        assert jmespath is not None
+        # Test basic functionality
+        result = jmespath.search("a.b", {"a": {"b": "value"}})
+        assert result == "value"

uv.lock

@@ -572,6 +572,9 @@ dependencies = [
 ]
 
 [package.optional-dependencies]
+jmespath = [
+    { name = "jmespath" },
+]
 openai = [
     { name = "openai" },
 ]
@@ -580,7 +583,7 @@ openai = [
 dev = [
     { name = "dirty-equals" },
     { name = "fastapi" },
-    { name = "fastmcp", extra = ["openai"] },
+    { name = "fastmcp", extra = ["jmespath", "openai"] },
     { name = "inline-snapshot", extra = ["dirty-equals"] },
     { name = "ipython", version = "8.37.0", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version < '3.11'" },
     { name = "ipython", version = "9.4.0", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version >= '3.11'" },
@@ -609,6 +612,7 @@ requires-dist = [
     { name = "cyclopts", specifier = ">=4.0.0" },
     { name = "exceptiongroup", specifier = ">=1.2.2" },
     { name = "httpx", specifier = ">=0.28.1" },
+    { name = "jmespath", marker = "extra == 'jmespath'", specifier = ">=1.0.0" },
     { name = "jsonschema-path", specifier = ">=0.3.4" },
     { name = "mcp", specifier = ">=1.23.1" },
     { name = "openai", marker = "extra == 'openai'", specifier = ">=1.102.0" },
@@ -622,13 +626,13 @@ requires-dist = [
     { name = "uvicorn", specifier = ">=0.35" },
     { name = "websockets", specifier = ">=15.0.1" },
 ]
-provides-extras = ["openai"]
+provides-extras = ["jmespath", "openai"]
 
 [package.metadata.requires-dev]
 dev = [
     { name = "dirty-equals", specifier = ">=0.9.0" },
     { name = "fastapi", specifier = ">=0.115.12" },
-    { name = "fastmcp", extras = ["openai"] },
+    { name = "fastmcp", extras = ["openai", "jmespath"] },
     { name = "inline-snapshot", extras = ["dirty-equals"], specifier = ">=0.27.2" },
     { name = "ipython", specifier = ">=8.12.3" },
     { name = "pdbpp", specifier = ">=0.11.7" },
@@ -881,6 +885,15 @@ wheels = [
     { url = "https://files.pythonhosted.org/packages/b3/4a/4175a563579e884192ba6e81725fc0448b042024419be8d83aa8a80a3f44/jiter-0.10.0-cp314-cp314t-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:3aa96f2abba33dc77f79b4cf791840230375f9534e5fac927ccceb58c5e604a5", size = 354213, upload-time = "2025-05-18T19:04:41.894Z" },
 ]
 
+[[package]]
+name = "jmespath"
+version = "1.0.1"
+source = { registry = "https://pypi.org/simple" }
+sdist = { url = "https://files.pythonhosted.org/packages/00/2a/e867e8531cf3e36b41201936b7fa7ba7b5702dbef42922193f05c8976cd6/jmespath-1.0.1.tar.gz", hash = "sha256:90261b206d6defd58fdd5e85f478bf633a2901798906be2ad389150c5c60edbe", size = 25843, upload-time = "2022-06-17T18:00:12.224Z" }
+wheels = [
+    { url = "https://files.pythonhosted.org/packages/31/b4/b9b800c45527aadd64d5b442f9b932b00648617eb5d63d2c7a6587b7cafc/jmespath-1.0.1-py3-none-any.whl", hash = "sha256:02e2e4cc71b5bcab88332eebf907519190dd9e6e82107fa7f83b1003a6252980", size = 20256, upload-time = "2022-06-17T18:00:10.251Z" },
+]
+
 [[package]]
 name = "jsonschema"
 version = "4.25.0"