Dynamic Tool/Resource Filtering for FastMCP

[d-mo]

Problem

Production MCP applications need to vary tool/resource availability based on request context—user permissions, tenant IDs, feature flags, or rate limits. FastMCP provides the primitives (Context, Middleware, ASGI integration) but lacks a clear pattern for this common requirement.

Multi-tenant SaaS, enterprise deployments with RBAC, and feature-flagged rollouts all share this need: different users see different capabilities.

Solution

Add FilteringMiddleware base class to fastmcp/server/middleware/ that:

1. Extracts request-scoped metadata from ASGI scope into Context.state
2. Filters tool/resource lists via overridable methods
3. Validates access at execution time

API Design

Core Middleware (src/fastmcp/server/middleware/filtering.py)

from fastmcp.server.middleware import Middleware, MiddlewareContext, CallNext
from fastmcp.server.context import Context
from fastmcp.tools.tool import Tool
from fastmcp.resources.resource import Resource

class FilteringMiddleware(Middleware):
    """Base class for dynamic tool/resource filtering based on request context.

    Override filter_* and validate_* methods to implement custom filtering logic.
    Use ctx.get_state() to access request-scoped data injected by ASGI middleware.
    """

    async def on_list_tools(
        self,
        context: MiddlewareContext[mt.ListToolsRequest],
        call_next: CallNext[mt.ListToolsRequest, Sequence[Tool]],
    ) -> Sequence[Tool]:
        tools = await call_next(context)
        if context.fastmcp_context:
            return await self.filter_tools(tools, context.fastmcp_context)
        return tools

    async def on_list_resources(
        self,
        context: MiddlewareContext[mt.ListResourcesRequest],
        call_next: CallNext[mt.ListResourcesRequest, Sequence[Resource]],
    ) -> Sequence[Resource]:
        resources = await call_next(context)
        if context.fastmcp_context:
            return await self.filter_resources(resources, context.fastmcp_context)
        return resources

    async def on_call_tool(
        self,
        context: MiddlewareContext[mt.CallToolRequestParams],
        call_next: CallNext[mt.CallToolRequestParams, ToolResult],
    ) -> ToolResult:
        if context.fastmcp_context:
            tool_name = context.message.name
            if not await self.validate_tool_access(tool_name, context.fastmcp_context):
                return ToolResult(
                    content=[TextContent(
                        type="text",
                        text=f"Access denied: Tool '{tool_name}' is not available"
                    )],
                    isError=True
                )
        return await call_next(context)

    async def on_read_resource(
        self,
        context: MiddlewareContext[mt.ReadResourceRequestParams],
        call_next: CallNext[mt.ReadResourceRequestParams, Sequence[ReadResourceContents]],
    ) -> Sequence[ReadResourceContents]:
        if context.fastmcp_context:
            resource_uri = context.message.uri
            if not await self.validate_resource_access(resource_uri, context.fastmcp_context):
                raise PermissionError(f"Access denied: Resource '{resource_uri}' is not available")
        return await call_next(context)

    # Override these in subclasses
    async def filter_tools(self, tools: Sequence[Tool], ctx: Context) -> Sequence[Tool]:
        """Filter tools based on request context. Default: no filtering."""
        return tools

    async def filter_resources(self, resources: Sequence[Resource], ctx: Context) -> Sequence[Resource]:
        """Filter resources based on request context. Default: no filtering."""
        return resources

    async def validate_tool_access(self, tool_name: str, ctx: Context) -> bool:
        """Validate access to a specific tool. Default: allow all."""
        return True

    async def validate_resource_access(self, resource_uri: str, ctx: Context) -> bool:
        """Validate access to a specific resource. Default: allow all."""
        return True

Helper for ASGI Context Injection (src/fastmcp/server/context.py)

def inject_scope_data(scope: dict, **data: Any) -> None:
    """Inject data from ASGI scope into FastMCP context state.

    Call this from ASGI middleware to make request-scoped data
    available to FilteringMiddleware and tools.

    Args:
        scope: ASGI scope dict
        **data: Key-value pairs to inject into Context.state

    Example:
        @mcp.asgi_middleware
        async def inject_tenant(scope, receive, send):
            tenant_id = extract_tenant_from_jwt(scope)
            inject_scope_data(scope, tenant_id=tenant_id)
            await call_next(scope, receive, send)
    """
    if "fastmcp.context_data" not in scope:
        scope["fastmcp.context_data"] = {}
    scope["fastmcp.context_data"].update(data)

Update Context.__aenter__ to read from scope:

async def __aenter__(self) -> Context:
    parent_context = _current_context.get(None)
    if parent_context is not None:
        self._state = copy.deepcopy(parent_context._state)

    # Inject ASGI scope data if available
    try:
        req_ctx = self.request_context
        if hasattr(req_ctx, "asgi_request") and req_ctx.asgi_request:
            scope_data = req_ctx.asgi_request.scope.get("fastmcp.context_data", {})
            self._state.update(scope_data)
    except (ValueError, AttributeError):
        pass  # Not in HTTP context

    token = _current_context.set(self)
    self._tokens.append(token)
    return self

Usage Example: Multi-Tenant Filtering

from fastmcp import FastMCP, Context
from fastmcp.server.middleware.filtering import FilteringMiddleware
from fastmcp.server.context import inject_scope_data

mcp = FastMCP("Multi-Tenant Server")

# ASGI middleware extracts tenant from JWT
async def tenant_middleware(app):
    async def middleware(scope, receive, send):
        if scope["type"] == "http":
            tenant_id = extract_tenant_from_headers(scope["headers"])
            inject_scope_data(scope, tenant_id=tenant_id)
        await app(scope, receive, send)
    return middleware

mcp.app.middleware("http")(tenant_middleware)

# Filtering middleware checks tenant access
class TenantFilteringMiddleware(FilteringMiddleware):
    def __init__(self, tenant_tool_map: dict[str, list[str]]):
        self.tenant_tool_map = tenant_tool_map

    async def filter_tools(self, tools: Sequence[Tool], ctx: Context) -> Sequence[Tool]:
        tenant_id = ctx.get_state("tenant_id")
        allowed = self.tenant_tool_map.get(tenant_id, [])
        return [t for t in tools if t.name in allowed]

    async def validate_tool_access(self, tool_name: str, ctx: Context) -> bool:
        tenant_id = ctx.get_state("tenant_id")
        allowed = self.tenant_tool_map.get(tenant_id, [])
        return tool_name in allowed

mcp.add_middleware(TenantFilteringMiddleware({
    "tenant_a": ["basic_tool", "search"],
    "tenant_b": ["basic_tool", "search", "admin_tool"],
}))

@mcp.tool
def basic_tool(data: str) -> str:
    return f"Basic: {data}"

@mcp.tool
def admin_tool(config: str) -> str:
    return f"Admin: {config}"

Implementation Plan

Files to Create

1. src/fastmcp/server/middleware/filtering.py - New FilteringMiddleware class
2. tests/server/middleware/test_filtering.py - Comprehensive test suite
3. examples/filtering/multi_tenant.py - Multi-tenant example
4. examples/filtering/rbac.py - Role-based access control example
5. docs/servers/dynamic-filtering.mdx - Documentation page

Files to Modify

1. src/fastmcp/server/context.py - Add inject_scope_data() and scope reading to Context.__aenter__
2. src/fastmcp/server/middleware/__init__.py - Export FilteringMiddleware
3. docs.json - Add dynamic-filtering page to navigation

Tests Required

• Filter tools based on context state
• Filter resources based on context state
• Validate tool access (allow/deny)
• Validate resource access (allow/deny)
• ASGI scope data injection
• Multiple filtering middleware stacked
• Empty context state (no filtering)
• Async filter methods

Documentation Required

• Concept overview: why dynamic filtering matters
• API reference for FilteringMiddleware
• Multi-tenant example walkthrough
• RBAC example with JWT claims
• Feature flag example
• Integration with existing auth providers

Benefits

Clear pattern for a common production requirement that currently requires developers to reverse-engineer from middleware docs.

Type-safe with full IDE autocomplete and type checking.

Composable - multiple filtering middleware can be stacked (tenant filter + RBAC filter + rate limiter).

Backward compatible - purely additive, no breaking changes.

Prior Art

• FastAPI dependencies: Inject request-scoped data via function parameters
• Django middleware: request.user available throughout request lifecycle
• ASP.NET Core: HttpContext.Items for request-scoped key-value storage
• Express.js: res.locals for request-scoped data sharing

Open Questions

1. Should validate_* methods raise exceptions or return bool? (Proposal: return bool for consistency)
2. Should we provide built-in filters (by tag, by annotation)? (Proposal: start minimal, add if requested)
3. Should filters apply to prompts as well? (Proposal: yes, add on_list_prompts and on_get_prompt)