Redis MCP Server

Instructions

Implementation Reference

• src/tools/list.py:11-21 (handler)

  The handler function for the 'lpush' MCP tool. Decorated with @mcp.tool(), it pushes a value to the left side of a Redis list using r.lpush(), optionally sets an expiration, and returns a success or error message.

  @mcp.tool()
  async def lpush(name: str, value: FieldT, expire: Optional[int] = None) -> str:
      """Push a value onto the left of a Redis list and optionally set an expiration time."""
      try:
          r = RedisConnectionManager.get_connection()
          r.lpush(name, value)
          if expire:
              r.expire(name, expire)
          return f"Value '{value}' pushed to the left of list '{name}'."
      except RedisError as e:
          return f"Error pushing value to list '{name}': {str(e)}"

• src/common/server.py:6-17 (registration)

  The registration mechanism for all tools, including 'lpush'. Imports all modules in src/tools package, triggering the @mcp.tool() decorators to register the tools 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", "dotenv", "numpy", "aiohttp"])
  
  # Load tools
  load_tools()