sadd

Instructions

Implementation Reference

• src/tools/set.py:9-32 (handler)

  The 'sadd' tool handler function. Decorated with @mcp.tool(), it adds a value to a Redis set with optional expiration. Uses RedisConnectionManager.get_connection() to get the Redis client and calls r.sadd(name, value). Returns a success/error message string.

  @mcp.tool()
  async def sadd(name: str, value: str, expire_seconds: Optional[int] = None) -> str:
      """Add a value to a Redis set with an optional expiration time.
  
      Args:
          name: The Redis set key.
          value: The value to add to the set.
          expire_seconds: Optional; time in seconds after which the set should expire.
  
      Returns:
          A success message or an error message.
      """
      try:
          r = RedisConnectionManager.get_connection()
          r.sadd(name, value)
  
          if expire_seconds is not None:
              r.expire(name, expire_seconds)
  
          return f"Value '{value}' added successfully to set '{name}'." + (
              f" Expires in {expire_seconds} seconds." if expire_seconds else ""
          )
      except RedisError as e:
          return f"Error adding value '{value}' to set '{name}': {str(e)}"

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

  Tool registration mechanism. The load_tools() function dynamically imports all modules in src/tools, including src/tools/set.py. The FastMCP instance 'mcp' is created and tools are loaded via @mcp.tool() decorator.

  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:18-110 (helper)

  RedisConnectionManager.get_connection() is used by the sadd handler to obtain a Redis client instance.

  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:
                          _logger.error(
                              "Failed to create Entra ID credential provider: %s", e
                          )
                          raise
  
                  if REDIS_CFG["cluster_mode"]:
                      redis_class: Type[Union[Redis, RedisCluster]] = (
                          redis.cluster.RedisCluster
                      )
                      connection_params = {
                          "host": REDIS_CFG["host"],
                          "port": REDIS_CFG["port"],
                          "username": REDIS_CFG["username"],
                          "password": REDIS_CFG["password"],
                          "ssl": REDIS_CFG["ssl"],
                          "ssl_ca_path": REDIS_CFG["ssl_ca_path"],
                          "ssl_keyfile": REDIS_CFG["ssl_keyfile"],
                          "ssl_certfile": REDIS_CFG["ssl_certfile"],
                          "ssl_cert_reqs": REDIS_CFG["ssl_cert_reqs"],
                          "ssl_ca_certs": REDIS_CFG["ssl_ca_certs"],
                          "decode_responses": decode_responses,
                          "lib_name": f"redis-py(mcp-server_v{__version__})",
                          "max_connections_per_node": 10,
                      }
  
                      # Add credential provider if available
                      if credential_provider:
                          connection_params["credential_provider"] = credential_provider
                          # Note: Azure Redis Enterprise with EntraID uses plain text connections
                          # SSL setting is controlled by REDIS_SSL environment variable
                  else:
                      redis_class: Type[Union[Redis, RedisCluster]] = redis.Redis
                      connection_params = {
                          "host": REDIS_CFG["host"],
                          "port": REDIS_CFG["port"],
                          "db": REDIS_CFG["db"],
                          "username": REDIS_CFG["username"],
                          "password": REDIS_CFG["password"],
                          "ssl": REDIS_CFG["ssl"],
                          "ssl_ca_path": REDIS_CFG["ssl_ca_path"],
                          "ssl_keyfile": REDIS_CFG["ssl_keyfile"],
                          "ssl_certfile": REDIS_CFG["ssl_certfile"],
                          "ssl_cert_reqs": REDIS_CFG["ssl_cert_reqs"],
                          "ssl_ca_certs": REDIS_CFG["ssl_ca_certs"],
                          "decode_responses": decode_responses,
                          "lib_name": f"redis-py(mcp-server_v{__version__})",
                          "max_connections": 10,
                      }
  
                      # Add credential provider if available
                      if credential_provider:
                          connection_params["credential_provider"] = credential_provider
                          # Note: Azure Redis Enterprise with EntraID uses plain text connections
                          # SSL setting is controlled by REDIS_SSL environment variable
  
                  cls._instance = redis_class(**connection_params)
  
              except redis.exceptions.ConnectionError:
                  _logger.error("Failed to connect to Redis server")
                  raise
              except redis.exceptions.AuthenticationError:
                  _logger.error("Authentication failed")
                  raise
              except redis.exceptions.TimeoutError:
                  _logger.error("Connection timed out")
                  raise
              except redis.exceptions.ResponseError as e:
                  _logger.error("Response error: %s", e)
                  raise
              except redis.exceptions.RedisError as e:
                  _logger.error("Redis error: %s", e)
                  raise
              except redis.exceptions.ClusterError as e:
                  _logger.error("Redis Cluster error: %s", e)
                  raise
              except Exception as e:
                  _logger.error("Unexpected error: %s", e)
                  raise
  
          return cls._instance