diff --git a/lib/cloudinary.rb b/lib/cloudinary.rb
index 23763c66..8900bc29 100644
--- a/lib/cloudinary.rb
+++ b/lib/cloudinary.rb
@@ -22,6 +22,7 @@ module Cloudinary
   autoload :PreloadedFile, "cloudinary/preloaded_file"
   autoload :Static, "cloudinary/static"
   autoload :CarrierWave, "cloudinary/carrier_wave"
+  autoload :Search, "cloudinary/search"
 
   CF_SHARED_CDN         = "d3jpl91pxevbkh.cloudfront.net"
   AKAMAI_SHARED_CDN     = "res.cloudinary.com"
diff --git a/lib/cloudinary/api.rb b/lib/cloudinary/api.rb
index 256e41c2..3a5efe41 100644
--- a/lib/cloudinary/api.rb
+++ b/lib/cloudinary/api.rb
@@ -1,4 +1,5 @@
 require 'rest_client'
+require 'json'
 
 class Cloudinary::Api
   class Error < CloudinaryException; end
@@ -316,7 +317,14 @@ def self.call_api(method, uri, params, options)
     # Add authentication
     api_url.sub!(%r(^(https?://)), "\\1#{api_key}:#{api_secret}@")
 
-    RestClient::Request.execute(:method => method, :url => api_url, :payload => params.reject { |k, v| v.nil? || v=="" }, :timeout => timeout, :headers => { "User-Agent" => Cloudinary::USER_AGENT }) do
+    headers = { "User-Agent" => Cloudinary::USER_AGENT }
+    if options[:content_type]== :json
+      payload = params.to_json
+      headers.merge!("Content-Type"=> 'application/json', "Accept"=> 'application/json')
+    else
+      payload = params.reject { |k, v| v.nil? || v=="" }
+    end
+    RestClient::Request.execute(:method => method, :url => api_url, :payload => payload, :timeout => timeout, :headers => headers) do
     |response, request, tmpresult|
       return Response.new(response) if response.code == 200
       exception_class = case response.code
diff --git a/lib/cloudinary/search.rb b/lib/cloudinary/search.rb
new file mode 100644
index 00000000..56cf4328
--- /dev/null
+++ b/lib/cloudinary/search.rb
@@ -0,0 +1,55 @@
+class Cloudinary::Search
+  def initialize
+    @query_hash = {
+      :sort_by    => [],
+      :aggregate  => [],
+      :with_field   => []
+    }
+  end
+
+  ## implicitly generate an instance delegate the method
+  def self.method_missing(method_name, *arguments)
+    instance = new
+    instance.send(method_name, *arguments)
+  end
+
+  def expression(value)
+    @query_hash[:expression] = value
+    self
+  end
+
+  def max_results(value)
+    @query_hash[:max_results] = value
+    self
+  end
+
+  def next_cursor(value)
+    @query_hash[:next_cursor] = value
+    self
+  end
+
+  def sort_by(field_name, dir = 'desc')
+    @query_hash[:sort_by].push(field_name => dir)
+    self
+  end
+
+  def aggregate(value)
+    @query_hash[:aggregate].push(value)
+    self
+  end
+
+  def with_field(value)
+    @query_hash[:with_field].push(value)
+    self
+  end
+
+  def to_h
+    @query_hash.select { |_, value| !value.nil? && !(value.is_a?(Array) && value.empty?) }
+  end
+
+  def execute(options = {})
+    options[:content_type] = :json
+    uri = 'resources/search'
+    Cloudinary::Api.call_api(:post, uri, to_h, options)
+  end
+end
diff --git a/spec/api_spec.rb b/spec/api_spec.rb
index e8eee981..c77b4a07 100644
--- a/spec/api_spec.rb
+++ b/spec/api_spec.rb
@@ -10,14 +10,15 @@
   test_id_1 = "#{prefix}_1"
   test_id_2   = "#{prefix}_2"
   test_id_3   = "#{prefix}_3"
+  test_key = "test_key_#{SUFFIX}"
   before(:all) do
 
     @api = Cloudinary::Api
     Cloudinary::Uploader.upload(TEST_IMG, :public_id => test_id_1, :tags => [TEST_TAG, TIMESTAMP_TAG], :context => "key=value", :eager =>[:width =>TEST_WIDTH, :crop =>:scale])
     Cloudinary::Uploader.upload(TEST_IMG, :public_id => test_id_2, :tags => [TEST_TAG, TIMESTAMP_TAG], :context => "key=value", :eager =>[:width =>TEST_WIDTH, :crop =>:scale])
     Cloudinary::Uploader.upload(TEST_IMG, :public_id => test_id_3, :tags => [TEST_TAG, TIMESTAMP_TAG], :context => "key=value", :eager =>[:width =>TEST_WIDTH, :crop =>:scale])
-    Cloudinary::Uploader.upload(TEST_IMG, :public_id => test_id_1, :tags => [TEST_TAG, TIMESTAMP_TAG], :context => "test-key=test", :eager =>[:width =>TEST_WIDTH, :crop =>:scale])
-    Cloudinary::Uploader.upload(TEST_IMG, :public_id => test_id_3, :tags => [TEST_TAG, TIMESTAMP_TAG], :context => "test-key=tasty", :eager =>[:width =>TEST_WIDTH, :crop =>:scale])
+    Cloudinary::Uploader.upload(TEST_IMG, :public_id => test_id_1, :tags => [TEST_TAG, TIMESTAMP_TAG], :context => "#{test_key}=test", :eager =>[:width =>TEST_WIDTH, :crop =>:scale])
+    Cloudinary::Uploader.upload(TEST_IMG, :public_id => test_id_3, :tags => [TEST_TAG, TIMESTAMP_TAG], :context => "#{test_key}=tasty", :eager =>[:width =>TEST_WIDTH, :crop =>:scale])
   end
 
   after(:all) do
@@ -79,9 +80,9 @@
   end
   
   it "should allow listing resources by context" do
-    resources = @api.resources_by_context('test-key')["resources"]
+    resources = @api.resources_by_context(test_key)["resources"]
     expect(resources.count).to eq(2)
-    resources = @api.resources_by_context('test-key','test')["resources"]
+    resources = @api.resources_by_context(test_key,'test')["resources"]
     expect(resources.count).to eq(1)
   end
 
diff --git a/spec/search_spec.rb b/spec/search_spec.rb
new file mode 100644
index 00000000..c421ae84
--- /dev/null
+++ b/spec/search_spec.rb
@@ -0,0 +1,109 @@
+require 'spec_helper'
+require 'cloudinary'
+
+describe Cloudinary::Search do
+  context 'unit' do
+    it 'should create empty json' do
+      query_hash = Cloudinary::Search.to_h
+      expect(query_hash).to eq({})
+    end
+
+    it 'should always return same object in fluent interface' do
+      instance = Cloudinary::Search.new
+      %w(expression sort_by max_results next_cursor aggregate with_field).each do |method|
+        same_instance = instance.send(method, 'emptyarg')
+        expect(instance).to eq(same_instance)
+      end
+    end
+
+    it 'should add expression to query' do
+      query = Cloudinary::Search.expression('format:jpg').to_h
+      expect(query).to eq(expression: 'format:jpg')
+    end
+
+    it 'should add sort_by to query' do
+      query = Cloudinary::Search.sort_by('created_at', 'asc').sort_by('updated_at', 'desc').to_h
+      expect(query).to eq(sort_by: [{ 'created_at' => 'asc' }, { 'updated_at' => 'desc' }])
+    end
+
+    it 'should add max_results to query' do
+      query = Cloudinary::Search.max_results('format:jpg').to_h
+      expect(query).to eq(max_results: 'format:jpg')
+    end
+
+    it 'should add next_cursor to query' do
+      query = Cloudinary::Search.next_cursor('format:jpg').to_h
+      expect(query).to eq(next_cursor: 'format:jpg')
+    end
+
+    it 'should add aggregations arguments as array to query' do
+      query = Cloudinary::Search.aggregate('format').aggregate('size_category').to_h
+      expect(query).to eq(aggregate: %w(format size_category))
+    end
+
+    it 'should add with_field to query' do
+      query = Cloudinary::Search.with_field('context').with_field('tags').to_h
+      expect(query).to eq(with_field: %w(context tags))
+    end
+  end
+
+  context 'integration' do
+    SEARCH_TAG = TIMESTAMP_TAG + "_search"
+    include_context 'cleanup', SEARCH_TAG
+    prefix = "api_test_#{SUFFIX}"
+    test_id_1 = "#{prefix}_1"
+    test_id_2   = "#{prefix}_2"
+    test_id_3   = "#{prefix}_3"
+    before(:all) do
+      Cloudinary::Uploader.upload(TEST_IMG, public_id: test_id_1, tags: [TEST_TAG, TIMESTAMP_TAG, SEARCH_TAG], context: 'stage=in_review')
+      Cloudinary::Uploader.upload(TEST_IMG, public_id: test_id_2, tags: [TEST_TAG, TIMESTAMP_TAG, SEARCH_TAG], context: 'stage=new')
+      Cloudinary::Uploader.upload(TEST_IMG, public_id: test_id_3, tags: [TEST_TAG, TIMESTAMP_TAG, SEARCH_TAG], context: 'stage=validated')
+      sleep(3)
+    end
+
+    it "should return all images tagged with #{SEARCH_TAG}" do
+      results = Cloudinary::Search.expression("tags:#{SEARCH_TAG}").execute
+      expect(results['resources'].count).to eq 3
+    end
+
+    it "should return resource #{test_id_1}" do
+      results = Cloudinary::Search.expression("public_id:#{test_id_1}").execute
+      expect(results['resources'].count).to eq 1
+    end
+
+    it 'should paginate resources limited by tag and ordered by ascending public_id' do
+      results = Cloudinary::Search.max_results(1).expression("tags:#{SEARCH_TAG}").sort_by('public_id', 'asc').execute
+      expect(results['resources'].count).to eq 1
+      expect(results['resources'][0]['public_id']).to eq test_id_1
+      expect(results['total_count']).to eq 3
+
+      results = Cloudinary::Search.max_results(1).expression("tags:#{SEARCH_TAG}").sort_by('public_id', 'asc').next_cursor(results['next_cursor']).execute
+      expect(results['resources'].count).to eq 1
+      expect(results['resources'][0]['public_id']).to eq test_id_2
+      expect(results['total_count']).to eq 3
+
+      results = Cloudinary::Search.max_results(1).expression("tags:#{SEARCH_TAG}").sort_by('public_id', 'asc').next_cursor(results['next_cursor']).execute
+      expect(results['resources'].count).to eq 1
+      expect(results['resources'][0]['public_id']).to eq test_id_3
+      expect(results['total_count']).to eq 3
+      expect(results['next_cursor']).to be_nil
+    end
+
+    it 'should include context' do
+      results = Cloudinary::Search.expression("tags:#{SEARCH_TAG}").with_field('context').execute
+      expect(results['resources'].count).to eq 3
+      results['resources'].each do |res|
+        expect(res['context'].keys).to eq ['stage']
+      end
+    end
+    it 'should include context, tags and image_metadata' do
+      results = Cloudinary::Search.expression("tags:#{SEARCH_TAG}").with_field('context').with_field('tags').with_field('image_metadata').execute
+      expect(results['resources'].count).to eq 3
+      results['resources'].each do |res|
+        expect(res['context'].keys).to eq ['stage']
+        expect(res.key?('image_metadata')).to eq true
+        expect(res['tags'].count).to eq 3
+      end
+    end
+  end
+end