Redis MCP Server

Instructions

Implementation Reference

• src/tools/hash.py:81-100 (handler)

  The main handler function for the 'hgetall' tool. It retrieves all fields and values from a specified Redis hash key using RedisConnectionManager, converts bytes to strings if present, and handles Redis errors by returning an error message.

  @mcp.tool()
  async def hgetall(name: str) -> dict:
      """Get all fields and values from a Redis hash.
  
      Args:
          name: The Redis hash key.
  
      Returns:
          A dictionary of field-value pairs or an error message.
      """
      try:
          r = RedisConnectionManager.get_connection()
          hash_data = r.hgetall(name)
          return (
              {k: v for k, v in hash_data.items()}
              if hash_data
              else f"Hash '{name}' is empty or does not exist."
          )
      except RedisError as e:
          return f"Error getting all fields from hash '{name}': {str(e)}"

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

  The load_tools function imports all modules in src/tools, which triggers the @mcp.tool() decorators to register the 'hgetall' tool (among others) with the MCP 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()