json_set
Set a JSON value at a specified path in a Redis key, with an optional expiration time.
Instructions
Set a JSON value in Redis at a given path with an optional expiration time.
Args: name: The Redis key where the JSON document is stored. path: The JSON path where the value should be set. value: The JSON value to store (as JSON string, or will be auto-converted). expire_seconds: Optional; time in seconds after which the key should expire.
Returns: A success message or an error message.
Input Schema
| Name | Required | Description | Default |
|---|---|---|---|
| name | Yes | ||
| path | Yes | ||
| value | Yes | ||
| expire_seconds | No |
Output Schema
| Name | Required | Description | Default |
|---|---|---|---|
| result | Yes |
Implementation Reference
- src/tools/json.py:10-44 (handler)The main handler function for the 'json_set' tool. Uses @mcp.tool() decorator to register as an MCP tool. Takes name, path, value, and optional expire_seconds, then sets a JSON value in Redis via redis-py's json().set() method.
async def json_set( name: str, path: str, value: str, expire_seconds: Optional[int] = None, ) -> str: """Set a JSON value in Redis at a given path with an optional expiration time. Args: name: The Redis key where the JSON document is stored. path: The JSON path where the value should be set. value: The JSON value to store (as JSON string, or will be auto-converted). expire_seconds: Optional; time in seconds after which the key should expire. Returns: A success message or an error message. """ # Try to parse the value as JSON, if it fails, treat it as a plain string try: parsed_value = json.loads(value) except (json.JSONDecodeError, TypeError): parsed_value = value try: r = RedisConnectionManager.get_connection() r.json().set(name, path, parsed_value) if expire_seconds is not None: r.expire(name, expire_seconds) return f"JSON value set at path '{path}' in '{name}'." + ( f" Expires in {expire_seconds} seconds." if expire_seconds else "" ) except RedisError as e: return f"Error setting JSON value at path '{path}' in '{name}': {str(e)}" - src/tools/json.py:6-6 (registration)Imports the 'mcp' FastMCP instance from src.common.server, which is used as a decorator (@mcp.tool()) to register json_set as an MCP tool.
from src.common.server import mcp - src/common/server.py:6-19 (registration)The load_tools() function dynamically imports all tool modules (including src/tools/json.py) via pkgutil, which causes the @mcp.tool() decorators to execute and register all tools (including json_set) with the FastMCP server.
def load_tools(): import src.tools as tools_pkg for _, module_name, _ in pkgutil.iter_modules(tools_pkg.__path__): importlib.import_module(f"src.tools.{module_name}") # Initialize FastMCP server mcp = FastMCP( "Redis MCP Server", dependencies=["redis", "python-dotenv", "numpy", "aiohttp"] ) # Load tools load_tools() - src/tools/json.py:10-14 (schema)The function signature defines the schema: name (str), path (str), value (str - JSON string auto-converted), expire_seconds (Optional[int]). Return type is str.
async def json_set( name: str, path: str, value: str, expire_seconds: Optional[int] = None,