rpush

Instructions

Implementation Reference

• src/tools/list.py:24-34 (handler)

  The handler function for the 'rpush' tool. It connects to Redis, pushes a value onto the right of a list, optionally sets an expiration, and returns a success/error message.

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

• src/tools/list.py:24-34 (schema)

  The input schema for rpush is defined by the function signature (name: str, value: FieldT, expire: Optional[int]) via the @mcp.tool() decorator from FastMCP, which auto-generates the JSON schema. Return type is str.

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

• src/tools/list.py:24-24 (registration)

  Registration of the 'rpush' tool via the @mcp.tool() decorator on the FastMCP server instance (mcp) in src/common/server.py. The load_tools() function in src/common/server.py dynamically imports src.tools.list (and all modules in src/tools), which triggers the decorator registration.

  @mcp.tool()

• src/common/connection.py:18-110 (helper)

  RedisConnectionManager.get_connection() provides the Redis client instance used by the rpush handler to execute the RPUSH command.

  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