diff --git a/.gitlab-ci.yml b/.gitlab-ci.yml
index 4dcf6078669..fa9120acf94 100644
--- a/.gitlab-ci.yml
+++ b/.gitlab-ci.yml
@@ -293,7 +293,7 @@ specs:
     - cp -a keys.example keys
     - cp -a certs.example certs
     - cp pwned_passwords/pwned_passwords.txt.sample pwned_passwords/pwned_passwords.txt
-    - "echo -e \"test:\n  redis_url: 'redis://db-redis:6379/0'\n  redis_throttle_url: 'redis://db-redis:6379/1'\" > config/application.yml"
+    - "echo -e \"test:\n  redis_url: 'redis://db-redis:6379/0'\n  redis_throttle_url: 'redis://db-redis:6379/1'\n  redis_attempts_api_url: 'redis://db-redis:6379/2'\" > config/application.yml"
     - bundle exec rake db:create db:migrate --trace
     - bundle exec rake db:seed
     - bundle exec rake knapsack:rspec["--format documentation --format RspecJunitFormatter --out rspec.xml --format json --out rspec_json/${CI_NODE_INDEX}.json"]
diff --git a/app/services/attempts_api/attempt_event.rb b/app/services/attempts_api/attempt_event.rb
new file mode 100644
index 00000000000..bd5821d5721
--- /dev/null
+++ b/app/services/attempts_api/attempt_event.rb
@@ -0,0 +1,86 @@
+# frozen_string_literal: true
+
+module AttemptsApi
+  class AttemptEvent
+    attr_reader :jti, :iat, :event_type, :session_id, :occurred_at, :event_metadata
+
+    def initialize(
+      event_type:,
+      session_id:,
+      occurred_at:,
+      event_metadata:,
+      jti: SecureRandom.uuid,
+      iat: Time.zone.now.to_i
+    )
+      @jti = jti
+      @iat = iat
+      @event_type = event_type
+      @session_id = session_id
+      @occurred_at = occurred_at
+      @event_metadata = event_metadata
+    end
+
+    def to_jwe(public_key:, issuer:)
+      jwk = JWT::JWK.new(public_key)
+
+      JWE.encrypt(
+        payload_json(issuer: issuer),
+        public_key,
+        typ: 'secevent+jwe',
+        zip: 'DEF',
+        alg: 'RSA-OAEP',
+        enc: 'A256GCM',
+        kid: jwk.kid,
+      )
+    end
+
+    def self.from_jwe(jwe, private_key)
+      decrypted_event = JWE.decrypt(jwe, private_key)
+      parsed_event = JSON.parse(decrypted_event)
+      event_type = parsed_event['events'].keys.first.split('/').last
+      event_data = parsed_event['events'].values.first
+      jti = parsed_event['jti'].split(':').last
+      AttemptEvent.new(
+        jti: jti,
+        iat: parsed_event['iat'],
+        event_type: event_type,
+        session_id: event_data['subject']['session_id'],
+        occurred_at: Time.zone.at(event_data['occurred_at']),
+        event_metadata: event_data.symbolize_keys.except(:subject, :occurred_at),
+      )
+    end
+
+    def payload(issuer:)
+      {
+        jti: jti,
+        iat: iat,
+        iss: Rails.application.routes.url_helpers.root_url,
+        aud: issuer,
+        events: {
+          long_event_type => event_data,
+        },
+      }
+    end
+
+    def payload_json(issuer:)
+      @payload_json ||= payload(issuer:).to_json
+    end
+
+    private
+
+    def event_data
+      {
+        'subject' => {
+          'subject_type' => 'session',
+          'session_id' => session_id,
+        },
+        'occurred_at' => occurred_at.to_f,
+      }.merge(event_metadata || {})
+    end
+
+    def long_event_type
+      dasherized_name = event_type.to_s.dasherize
+      "https://schemas.login.gov/secevent/attempts-api/event-type/#{dasherized_name}"
+    end
+  end
+end
diff --git a/app/services/attempts_api/redis_client.rb b/app/services/attempts_api/redis_client.rb
new file mode 100644
index 00000000000..2f9fd809382
--- /dev/null
+++ b/app/services/attempts_api/redis_client.rb
@@ -0,0 +1,29 @@
+# frozen_string_literal: true
+
+module AttemptsApi
+  class RedisClient
+    def write_event(event_key:, jwe:, timestamp:, issuer:)
+      key = key(timestamp, issuer)
+      REDIS_ATTEMPTS_API_POOL.with do |client|
+        client.hset(key, event_key, jwe)
+        client.expire(key, IdentityConfig.store.attempts_api_event_ttl_seconds)
+      end
+    end
+
+    def read_events(timestamp:, issuer:, batch_size: 5000)
+      key = key(timestamp, issuer)
+      events = {}
+      REDIS_ATTEMPTS_API_POOL.with do |client|
+        client.hscan_each(key, count: batch_size) do |k, v|
+          events[k] = v
+        end
+      end
+      events
+    end
+
+    def key(timestamp, issuer)
+      formatted_time = timestamp.in_time_zone('UTC').change(min: 0, sec: 0).iso8601
+      "attempts-api-events:#{issuer}:#{formatted_time}"
+    end
+  end
+end
diff --git a/config/application.yml.default b/config/application.yml.default
index dcd7ef6c452..17cd144a0a6 100644
--- a/config/application.yml.default
+++ b/config/application.yml.default
@@ -40,6 +40,7 @@ allowed_verified_within_providers: '[]'
 asset_host: ''
 async_stale_job_timeout_seconds: 300
 async_wait_timeout_seconds: 60
+attempts_api_event_ttl_seconds: 3_600
 attribute_encryption_key:
 attribute_encryption_key_queue: '[]'
 available_locales: 'en,es,fr,zh'
@@ -321,6 +322,8 @@ recaptcha_site_key: ''
 recommend_webauthn_platform_for_sms_ab_test_account_creation_percent: 0
 recommend_webauthn_platform_for_sms_ab_test_authentication_percent: 0
 recovery_code_length: 4
+redis_attempts_api_pool_size: 1
+redis_attempts_api_url: redis://localhost:6379/2
 redis_pool_size: 10
 redis_throttle_pool_size: 5
 redis_throttle_url: redis://localhost:6379/1
diff --git a/config/initializers/01_redis.rb b/config/initializers/01_redis.rb
index 39851d84299..3a88b1c43f9 100644
--- a/config/initializers/01_redis.rb
+++ b/config/initializers/01_redis.rb
@@ -9,3 +9,8 @@
 REDIS_THROTTLE_POOL = ConnectionPool.new(size: IdentityConfig.store.redis_throttle_pool_size) do
   Redis.new(url: IdentityConfig.store.redis_throttle_url)
 end.freeze
+
+REDIS_ATTEMPTS_API_POOL =
+  ConnectionPool.new(size: IdentityConfig.store.redis_attempts_api_pool_size) do
+    Redis.new(url: IdentityConfig.store.redis_attempts_api_url)
+  end.freeze
diff --git a/lib/identity_config.rb b/lib/identity_config.rb
index c8178d27958..4e085294df9 100644
--- a/lib/identity_config.rb
+++ b/lib/identity_config.rb
@@ -58,6 +58,7 @@ def self.store
     config.add(:asset_host, type: :string)
     config.add(:async_stale_job_timeout_seconds, type: :integer)
     config.add(:async_wait_timeout_seconds, type: :integer)
+    config.add(:attempts_api_event_ttl_seconds, type: :integer)
     config.add(:attribute_encryption_key, type: :string)
     config.add(:attribute_encryption_key_queue, type: :json)
     config.add(:available_locales, type: :comma_separated_string_list)
@@ -355,6 +356,8 @@ def self.store
     config.add(:recaptcha_secret_key, type: :string)
     config.add(:recaptcha_site_key, type: :string)
     config.add(:recovery_code_length, type: :integer)
+    config.add(:redis_attempts_api_pool_size, type: :integer)
+    config.add(:redis_attempts_api_url, type: :string)
     config.add(:redis_pool_size, type: :integer)
     config.add(:redis_throttle_pool_size, type: :integer)
     config.add(:redis_throttle_url, type: :string)
diff --git a/spec/rails_helper.rb b/spec/rails_helper.rb
index 4dc85032e9f..3c6c8ca9c6f 100644
--- a/spec/rails_helper.rb
+++ b/spec/rails_helper.rb
@@ -111,6 +111,7 @@ class Analytics
     Telephony::Test::Call.clear_calls
     PushNotification::LocalEventQueue.clear!
     REDIS_THROTTLE_POOL.with { |client| client.flushdb } if Identity::Hostdata.config
+    REDIS_ATTEMPTS_API_POOL.with { |client| client.flushdb } if Identity::Hostdata.config
   end
 
   config.before(:each) do
diff --git a/spec/services/attempts_api/attempt_event_spec.rb b/spec/services/attempts_api/attempt_event_spec.rb
new file mode 100644
index 00000000000..3ed0446151e
--- /dev/null
+++ b/spec/services/attempts_api/attempt_event_spec.rb
@@ -0,0 +1,70 @@
+require 'rails_helper'
+
+RSpec.describe AttemptsApi::AttemptEvent do
+  let(:attempts_api_private_key) { OpenSSL::PKey::RSA.new(2048) }
+  let(:attempts_api_public_key) { attempts_api_private_key.public_key }
+
+  let(:jti) { 'test-unique-id' }
+  let(:iat) { Time.zone.now.to_i }
+  let(:event_type) { 'test-event' }
+  let(:session_id) { 'test-session-id' }
+  let(:occurred_at) { Time.zone.now.round }
+  let(:event_metadata) { { 'foo' => 'bar' } }
+  let(:service_provider) { build(:service_provider) }
+
+  subject do
+    described_class.new(
+      jti: jti,
+      iat: iat,
+      event_type: event_type,
+      session_id: session_id,
+      occurred_at: occurred_at,
+      event_metadata: event_metadata,
+    )
+  end
+
+  describe '#to_jwe' do
+    it 'returns a JWE for the event' do
+      jwe = subject.to_jwe(issuer: service_provider.issuer, public_key: attempts_api_public_key)
+
+      header_str, *_rest = JWE::Serialization::Compact.decode(jwe)
+      headers = JSON.parse(header_str)
+
+      expect(headers['alg']).to eq('RSA-OAEP')
+      expect(headers['kid']).to eq(JWT::JWK.new(attempts_api_public_key).kid)
+
+      decrypted_jwe_payload = JWE.decrypt(jwe, attempts_api_private_key)
+
+      token = JSON.parse(decrypted_jwe_payload)
+
+      expect(token['iss']).to eq(Rails.application.routes.url_helpers.root_url)
+      expect(token['jti']).to eq(jti)
+      expect(token['iat']).to eq(iat)
+      expect(token['aud']).to eq(service_provider.issuer)
+
+      event_key = 'https://schemas.login.gov/secevent/attempts-api/event-type/test-event'
+      event_data = token['events'][event_key]
+
+      expect(event_data['subject']).to eq(
+        'subject_type' => 'session', 'session_id' => 'test-session-id',
+      )
+      expect(event_data['foo']).to eq('bar')
+      expect(event_data['occurred_at']).to eq(occurred_at.to_f)
+    end
+  end
+
+  describe '.from_jwe' do
+    it 'returns an event decrypted from the JWE' do
+      jwe = subject.to_jwe(issuer: service_provider.issuer, public_key: attempts_api_public_key)
+
+      decrypted_event = described_class.from_jwe(jwe, attempts_api_private_key)
+
+      expect(decrypted_event.jti).to eq(subject.jti)
+      expect(decrypted_event.iat).to eq(subject.iat)
+      expect(decrypted_event.event_type).to eq(subject.event_type)
+      expect(decrypted_event.session_id).to eq(subject.session_id)
+      expect(decrypted_event.occurred_at).to eq(subject.occurred_at)
+      expect(decrypted_event.event_metadata).to eq(subject.event_metadata.symbolize_keys)
+    end
+  end
+end
diff --git a/spec/services/attempts_api/redis_client_spec.rb b/spec/services/attempts_api/redis_client_spec.rb
new file mode 100644
index 00000000000..d768f01eeca
--- /dev/null
+++ b/spec/services/attempts_api/redis_client_spec.rb
@@ -0,0 +1,92 @@
+require 'rails_helper'
+
+RSpec.describe AttemptsApi::RedisClient do
+  let(:attempts_api_private_key) { OpenSSL::PKey::RSA.new(2048) }
+  let(:attempts_api_public_key) { attempts_api_private_key.public_key }
+  let(:issuer) { 'test' }
+
+  describe '#write_event' do
+    it 'writes the attempt data to redis with the event key as the key' do
+      freeze_time do
+        now = Time.zone.now
+        event = AttemptsApi::AttemptEvent.new(
+          event_type: 'test_event',
+          session_id: 'test-session-id',
+          occurred_at: Time.zone.now,
+          event_metadata: {
+            first_name: Idp::Constants::MOCK_IDV_APPLICANT[:first_name],
+          },
+        )
+        event_key = event.jti
+        jwe = event.to_jwe(issuer: issuer, public_key: attempts_api_public_key)
+
+        subject.write_event(event_key: event_key, jwe: jwe, timestamp: now, issuer: issuer)
+
+        result = subject.read_events(timestamp: now, issuer: issuer)
+        expect(result[event_key]).to eq(jwe)
+      end
+    end
+  end
+
+  describe '#read_events' do
+    it 'reads the event events from redis' do
+      freeze_time do
+        now = Time.zone.now
+        events = {}
+        3.times do
+          event = AttemptsApi::AttemptEvent.new(
+            event_type: 'test_event',
+            session_id: 'test-session-id',
+            occurred_at: now,
+            event_metadata: {
+              first_name: Idp::Constants::MOCK_IDV_APPLICANT[:first_name],
+            },
+          )
+          event_key = event.jti
+          jwe = event.to_jwe(issuer: issuer, public_key: attempts_api_public_key)
+          events[event_key] = jwe
+        end
+        events.each do |event_key, jwe|
+          subject.write_event(event_key: event_key, jwe: jwe, timestamp: now, issuer: issuer)
+        end
+
+        result = subject.read_events(timestamp: now, issuer: issuer)
+
+        expect(result).to eq(events)
+      end
+    end
+
+    it 'stores events in hourly buckets' do
+      time1 = Time.new(2022, 1, 1, 1, 0, 0, 'Z')
+      time2 = Time.new(2022, 1, 1, 2, 0, 0, 'Z')
+      event1 = AttemptsApi::AttemptEvent.new(
+        event_type: 'test_event',
+        session_id: 'test-session-id',
+        occurred_at: time1,
+        event_metadata: {
+          first_name: Idp::Constants::MOCK_IDV_APPLICANT[:first_name],
+        },
+      )
+      event2 = AttemptsApi::AttemptEvent.new(
+        event_type: 'test_event',
+        session_id: 'test-session-id',
+        occurred_at: time2,
+        event_metadata: {
+          first_name: Idp::Constants::MOCK_IDV_APPLICANT[:first_name],
+        },
+      )
+      jwe1 = event1.to_jwe(issuer: issuer, public_key: attempts_api_public_key)
+      jwe2 = event2.to_jwe(issuer: issuer, public_key: attempts_api_public_key)
+
+      subject.write_event(
+        event_key: event1.jti, jwe: jwe1, timestamp: event1.occurred_at, issuer: issuer,
+      )
+      subject.write_event(
+        event_key: event2.jti, jwe: jwe2, timestamp: event2.occurred_at, issuer: issuer,
+      )
+
+      expect(subject.read_events(timestamp: time1, issuer: issuer)).to eq({ event1.jti => jwe1 })
+      expect(subject.read_events(timestamp: time2, issuer: issuer)).to eq({ event2.jti => jwe2 })
+    end
+  end
+end