MCP Server with Dynamic Per-User Tool Generation
A Model Context Protocol (MCP) server implementation using FastMCP that dynamically generates tools based on user permissions from an external API.
Features
- Dynamic Tool Generation: Tools are generated at runtime based on user-specific schemas from an external API
- Per-User Permissions: Each authenticated user receives tools tailored to their permissions
- Intelligent Caching: Both API schemas and generated tools are cached with TTL for performance
- Graceful Degradation: Falls back to static tools if API is unavailable
- Comprehensive Logging: Detailed logging at INFO, DEBUG, WARNING, and ERROR levels
- Thread-Safe: Concurrent requests from multiple users are handled safely
- Global Score Filtering: Elasticsearch queries use global minScore threshold for result quality control
Architecture
Components
-
mcp_server.py- Main server entry point- Initializes FastMCP server
- Registers static tools (add, echo, multiply)
- Sets up dynamic tool middleware
-
api_client.py- External API integrationFormSchemaClient: Fetches and parses schemas from external APIFormSchemaCache: Caches API responses with TTL- Converts API field definitions to JSON Schema format
-
dynamic_tool_manager.py- Tool lifecycle managementDynamicToolManager: Manages per-user tool storage- Thread-safe caching with TTL
- Cache statistics and invalidation
-
tool_function_factory.py- Dynamic function generation- Creates typed async functions from JSON schemas
- Generates proper
inspect.Signaturefor FastMCP validation - Maps JSON Schema types to Python types
-
tool_execution_handler.py- Tool execution backendToolExecutionRouter: Routes tool calls to appropriate handlers- Specialized handlers for different tool types
- Structured error responses
-
dynamic_tool_middleware.py- FastMCP middleware- Intercepts
list/toolsrequests - Extracts authentication tokens
- Generates and caches user-specific tools
- Returns combined static + dynamic tools
- Intercepts
Installation
-
Install dependencies:
pip install -r requirements.txt -
Configure the API URL: Edit
mcp_server.pyand set theFORM_SCHEMA_API_URLto your external API endpoint:FORM_SCHEMA_API_URL = "http://172.16.11.131/api/module/request/form"
Usage
Starting the Server
python mcp_server.py
The server will start on http://127.0.0.1:9092/mcp
Authentication
The server expects authentication via HTTP Authorization header:
Authorization: Bearer <your-token>
The token is used to:
- Fetch user-specific schema from the external API
- Generate tools with fields the user has permission to access
- Cache tools for subsequent requests
Available Tools
Static Tools (Always Available)
add(a: int, b: int) -> int- Adds two numbersecho(message: str) -> str- Echoes a messagemultiply(a: int, b: int) -> int- Multiplies two numbers
Dynamic Tools (User-Specific)
create_request(...)- Creates a request with fields based on user permissions- Parameters are dynamically generated from the external API schema
- Each user sees different parameters based on their permissions
- Required fields, enums, and types are enforced
How It Works
Request Flow
-
Client sends
list/toolsrequest withAuthorization: Bearer <token> -
Middleware intercepts the request:
- Extracts auth token from header
- Checks if tools are cached for this user
- If cached: Returns cached tools
- If not cached: Proceeds to generation
-
Tool Generation:
- Fetches schema from external API with user's token
- Parses API response into JSON Schema
- Creates typed async function with proper signature
- Converts function to FastMCP
Toolobject - Caches the tool for future requests
-
Response:
- Returns list of static tools + dynamic tools
- Client sees tools specific to their permissions
Tool Execution Flow
-
Client calls
create_requesttool with arguments -
FastMCP validates arguments against the generated function signature
-
Tool execution handler:
- Receives validated arguments
- Processes the request
- Returns structured response with ID, timestamp, status
Configuration
Cache TTL
Both schema cache and tool cache use 5-minute TTL by default. To change:
# In mcp_server.py
schema_client = FormSchemaClient(
api_url=FORM_SCHEMA_API_URL,
cache_ttl=300, # Change this value (seconds)
verbose=True
)
tool_manager = DynamicToolManager(cache_ttl_seconds=300) # Change this value
Logging Level
To change logging verbosity:
# In mcp_server.py
logging.basicConfig(
level=logging.DEBUG, # Change to DEBUG, INFO, WARNING, or ERROR
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
Testing
See TEST_SCENARIOS.md for comprehensive test scenarios including:
- User-specific tool generation
- Caching behavior
- Tool execution
- Error handling
- Validation enforcement
API Schema Format
The external API should return a response with this structure:
{
"fieldList": [
{
"name": "Subject",
"paramName": "subject",
"type": "TextFieldRest",
"required": true,
"description": "Request subject",
"groupIds": [1, 2],
"hidden": false,
"removed": false,
"inActive": false
}
]
}
Supported field types:
TextFieldRest→stringNumberFieldRest→numberDropDownFieldRest→stringwith enumMultiSelectDropDownFieldRest→arrayof strings- And more (see
api_client.pyfor full mapping)
Security Considerations
- Authentication tokens are truncated in logs and responses
- Tokens are validated on each request
- No token = static tools only (graceful degradation)
- FastMCP validates all tool arguments before execution
Troubleshooting
Server won't start
- Check that all dependencies are installed:
pip install -r requirements.txt - Verify Python version is 3.12+
- Check logs for import errors
Tools not appearing
- Verify
Authorizationheader is present and correctly formatted - Check server logs for API fetch errors
- Ensure external API is accessible
- Verify user has permissions in the external system
Cache not updating
- Default TTL is 5 minutes
- Manually clear cache: Call
tool_manager.clear_all_tools()or restart server - Check logs for cache hit/miss messages
License
[Your License Here]
Contributing
[Your Contributing Guidelines Here]