smembers

Instructions

Implementation Reference

• src/tools/set.py:58-73 (handler)

  The actual implementation of the smembers tool handler. It gets a Redis connection, calls r.smembers(name), and returns the members as a list or an error message.

  @mcp.tool()
  async def smembers(name: str) -> Union[str, List[str]]:
      """Get all members of a Redis set.
  
      Args:
          name: The Redis set key.
  
      Returns:
          A list of values in the set or an error message.
      """
      try:
          r = RedisConnectionManager.get_connection()
          members = r.smembers(name)
          return list(members) if members else f"Set '{name}' is empty or does not exist."
      except RedisError as e:
          return f"Error retrieving members of set '{name}': {str(e)}"

• src/tools/set.py:58-58 (registration)

  The @mcp.tool() decorator registers smembers as an MCP tool.

  @mcp.tool()

• src/common/server.py:14-16 (registration)

  The FastMCP server instance 'mcp' used by the @mcp.tool() decorator is created here.

  mcp = FastMCP(
      "Redis MCP Server", dependencies=["redis", "python-dotenv", "numpy", "aiohttp"]
  )

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

  The load_tools() function dynamically imports all modules in src/tools, which triggers the @mcp.tool() decorators and registers them.

  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/common/connection.py:1-30 (helper)

  The RedisConnectionManager used by smembers to obtain a Redis connection via get_connection().

  import logging
  from typing import Optional, Type, Union
  
  import redis
  from redis import Redis
  from redis.cluster import RedisCluster
  
  from src.common.config import REDIS_CFG, is_entraid_auth_enabled
  from src.common.entraid_auth import (
      create_credential_provider,
      EntraIDAuthenticationError,
  )
  from src.version import __version__
  
  _logger = logging.getLogger(__name__)
  
  
  class RedisConnectionManager:
      _instance: Optional[Redis] = None
  
      @classmethod
      def get_connection(cls, decode_responses=True) -> Redis:
          if cls._instance is None:
              try:
                  # Create Entra ID credential provider if configured
                  credential_provider = None
                  if is_entraid_auth_enabled():
                      try:
                          credential_provider = create_credential_provider()
                      except EntraIDAuthenticationError as e: