diff --git a/Gemfile b/Gemfile
index a1ac6a19fbf..6b671ec27e0 100644
--- a/Gemfile
+++ b/Gemfile
@@ -65,6 +65,7 @@ gem 'sprockets-rails'
 gem 'stringex', require: false
 gem 'strong_migrations', '>= 0.4.2'
 gem 'subprocess', require: false
+gem 'terminal-table', require: false
 gem 'uglifier', '~> 4.2'
 gem 'valid_email', '>= 0.1.3'
 gem 'view_component', '~> 2.82.0'
@@ -124,6 +125,7 @@ group :test do
   gem 'rspec-retry'
   gem 'rspec_junit_formatter'
   gem 'shoulda-matchers', '~> 4.0', require: false
+  gem 'tableparser', require: false
   gem 'webdrivers', '~> 5.2.0'
   gem 'webmock'
   gem 'zonebie'
diff --git a/Gemfile.lock b/Gemfile.lock
index 09c160a7175..4e7644ded2b 100644
--- a/Gemfile.lock
+++ b/Gemfile.lock
@@ -645,6 +645,7 @@ GEM
     strong_migrations (0.8.0)
       activerecord (>= 5.2)
     subprocess (1.5.5)
+    tableparser (1.0.1)
     terminal-table (3.0.2)
       unicode-display_width (>= 1.1.1, < 3)
     thor (1.2.1)
@@ -821,6 +822,8 @@ DEPENDENCIES
   stringex
   strong_migrations (>= 0.4.2)
   subprocess
+  tableparser
+  terminal-table
   uglifier (~> 4.2)
   valid_email (>= 0.1.3)
   view_component (~> 2.82.0)
diff --git a/bin/data-pull b/bin/data-pull
new file mode 100755
index 00000000000..f4b64f46bce
--- /dev/null
+++ b/bin/data-pull
@@ -0,0 +1,6 @@
+#!/usr/bin/env ruby
+
+ENV['LOGIN_TASK_LOG_LEVEL'] ||= 'warn'
+require_relative '../config/environment.rb'
+require 'data_pull'
+DataPull.new(argv: ARGV.dup, stdout: STDOUT, stderr: STDERR).run
diff --git a/lib/data_pull.rb b/lib/data_pull.rb
new file mode 100644
index 00000000000..512eba6394d
--- /dev/null
+++ b/lib/data_pull.rb
@@ -0,0 +1,217 @@
+require 'optparse'
+
+class DataPull
+  attr_reader :argv, :stdout, :stderr
+
+  def initialize(argv:, stdout:, stderr:)
+    @argv = argv
+    @stdout = stdout
+    @stderr = stderr
+  end
+
+  Result = Struct.new(
+    :table, # tabular output, rendered as an ASCII table or as CSV
+    :log_message, # summary message used for audit logging, DO NOT PUT PII HERE
+    keyword_init: true,
+  )
+
+  Config = Struct.new(
+    :include_missing,
+    :format,
+    :show_help,
+    keyword_init: true,
+  ) do
+    alias_method :include_missing?, :include_missing
+    alias_method :show_help?, :show_help
+  end
+
+  def config
+    @config ||= Config.new(
+      include_missing: true,
+      format: :table,
+      show_help: false,
+    )
+  end
+
+  def run
+    option_parser.parse!(argv)
+    subtask_class = subtask(argv.shift)
+
+    if config.show_help? || !subtask_class
+      stdout.puts option_parser
+      return
+    end
+
+    result = subtask_class.new.run(args: argv, include_missing: config.include_missing?)
+
+    stderr.puts result.log_message
+
+    render_output(result.table)
+  end
+
+  # @param [Array<Array<String>>] rows
+  def render_output(rows)
+    return if rows.blank?
+
+    case config.format
+    when :table
+      require 'terminal-table'
+      table = Terminal::Table.new
+      header, *body = rows
+      table << header
+      table << :separator
+      body.each do |row|
+        table << row
+      end
+      stdout.puts table
+    when :csv
+      require 'csv'
+      CSV.instance(stdout) do |csv|
+        rows.each do |row|
+          csv << row
+        end
+      end
+    when :json
+      headers, *body = rows
+
+      objects = body.map do |values|
+        headers.zip(values).to_h
+      end
+
+      stdout.puts JSON.pretty_generate(objects)
+    else
+      raise "Unknown format=#{config.format}"
+    end
+  end
+
+  # @api private
+  # A subtask is a class that has a run method, the type signature should look like:
+  # +#run(args: Array<String>, include_missing: Boolean) -> Result+
+  # @return [Class,nil]
+  def subtask(name)
+    {
+      'uuid-lookup' => UuidLookup,
+      'uuid-convert' => UuidConvert,
+      'email-lookup' => EmailLookup,
+    }[name]
+  end
+
+  def option_parser
+    @option_parser ||= OptionParser.new do |opts|
+      opts.banner = <<~EOS
+        #{$PROGRAM_NAME} [subcommand] [arguments] [options]
+
+        Example usage:
+
+          * #{$PROGRAM_NAME} uuid-lookup email1@example.com email2@example.com
+
+          * #{$PROGRAM_NAME} uuid-convert partner-uuid1 partner-uuid2
+
+          * #{$PROGRAM_NAME} email-lookup uuid1 uuid2
+
+        Options:
+      EOS
+
+      opts.on('--help') do
+        config.show_help = true
+      end
+
+      opts.on('--csv') do
+        config.format = :csv
+      end
+
+      opts.on('--table', 'Output format as an ASCII table (default)') do
+        config.format = :table
+      end
+
+      opts.on('--json') do
+        config.format = :json
+      end
+
+      opts.on('--[no-]include-missing', <<~STR) do |include_missing|
+        Whether or not to add rows in the output for missing inputs, defaults to off
+      STR
+        config.include_missing = include_missing
+      end
+    end
+  end
+
+  class UuidLookup
+    def run(args:, include_missing:)
+      emails = args
+
+      table = []
+      table << %w[email uuid]
+
+      uuids = []
+
+      emails.each do |email|
+        user = User.find_with_email(email)
+        if user
+          table << [email, user.uuid]
+          uuids << user.uuid
+        elsif include_missing
+          table << [email, '[NOT FOUND]']
+        end
+      end
+
+      Result.new(
+        log_message: "uuid-lookup, uuids: #{uuids.join(', ')}",
+        table:,
+      )
+    end
+  end
+
+  class UuidConvert
+    def run(args:, include_missing:)
+      partner_uuids = args
+
+      table = []
+      table << %w[partner_uuid source internal_uuid]
+      identities = AgencyIdentity.includes(:user, :agency).where(uuid: partner_uuids).order(:uuid)
+
+      identities.each do |identity|
+        table << [identity.uuid, identity.agency.name, identity.user.uuid]
+      end
+
+      if include_missing
+        (partner_uuids - identities.map(&:uuid)).each do |missing_uuid|
+          table << [missing_uuid, '[NOT FOUND]', '[NOT FOUND]']
+        end
+      end
+
+      Result.new(
+        log_message: "uuid-convert, uuids: #{identities.map { |u| u.user.uuid }.join(', ')}",
+        table:,
+      )
+    end
+  end
+
+  class EmailLookup
+    def run(args:, include_missing:)
+      uuids = args
+
+      users = User.includes(:email_addresses).where(uuid: uuids).order(:uuid)
+
+      table = []
+      table << %w[uuid email confirmed_at]
+
+      users.each do |user|
+        user.email_addresses.sort_by(&:id).each do |email_address|
+          table << [user.uuid, email_address.email, email_address.confirmed_at.round(6)]
+        end
+      end
+
+      if include_missing
+        (uuids - users.map(&:uuid)).each do |missing_uuid|
+          table << [missing_uuid, '[NOT FOUND]', nil]
+        end
+      end
+
+      Result.new(
+        log_message: "email-lookup, uuids: #{users.map(&:uuid).join(', ')}",
+        table:,
+      )
+    end
+  end
+end
diff --git a/spec/factories/agency_identities.rb b/spec/factories/agency_identities.rb
new file mode 100644
index 00000000000..c0d26ef90cf
--- /dev/null
+++ b/spec/factories/agency_identities.rb
@@ -0,0 +1,7 @@
+FactoryBot.define do
+  factory :agency_identity do
+    association :user, factory: %i[user fully_registered]
+    association :agency
+    uuid { SecureRandom.uuid }
+  end
+end
diff --git a/spec/lib/data_pull_spec.rb b/spec/lib/data_pull_spec.rb
new file mode 100644
index 00000000000..38d2cb58e38
--- /dev/null
+++ b/spec/lib/data_pull_spec.rb
@@ -0,0 +1,163 @@
+require 'rails_helper'
+require 'tableparser'
+require 'data_pull'
+
+RSpec.describe DataPull do
+  let(:stdout) { StringIO.new }
+  let(:stderr) { StringIO.new }
+  let(:argv) { [] }
+
+  subject(:data_pull) { DataPull.new(argv:, stdout:, stderr:) }
+
+  describe 'command line flags' do
+    let(:argv) { ['uuid-lookup', user.email_addresses.first.email] }
+    let(:user) { create(:user) }
+
+    describe '--help' do
+      before { argv << '--help' }
+      it 'prints a help message' do
+        data_pull.run
+
+        expect(stdout.string).to include('Options:')
+      end
+    end
+
+    describe '--csv' do
+      before { argv << '--csv' }
+      it 'formats output as CSV' do
+        data_pull.run
+
+        expect(CSV.parse(stdout.string)).to eq(
+          [
+            ['email', 'uuid'],
+            [user.email_addresses.first.email, user.uuid],
+          ],
+        )
+      end
+    end
+
+    describe '--table' do
+      before { argv << '--table' }
+      it 'formats output as an ASCII table' do
+        data_pull.run
+
+        expect(Tableparser.parse(stdout.string)).to eq(
+          [
+            ['email', 'uuid'],
+            [user.email_addresses.first.email, user.uuid],
+          ],
+        )
+      end
+    end
+
+    describe '--json' do
+      before { argv << '--json' }
+      it 'formats output as JSON' do
+        data_pull.run
+
+        expect(JSON.parse(stdout.string)).to eq(
+          [
+            {
+              'email' => user.email_addresses.first.email,
+              'uuid' => user.uuid,
+            },
+          ],
+        )
+      end
+    end
+
+    describe '--include-missing' do
+      let(:argv) { ['uuid-lookup', 'does_not_exist@example.com', '--include-missing', '--json'] }
+      it 'adds rows for missing values' do
+        data_pull.run
+
+        expect(JSON.parse(stdout.string)).to eq(
+          [
+            {
+              'email' => 'does_not_exist@example.com',
+              'uuid' => '[NOT FOUND]',
+            },
+          ],
+        )
+      end
+    end
+
+    describe '--no-include-missing' do
+      let(:argv) { ['uuid-lookup', 'does_not_exist@example.com', '--no-include-missing', '--json'] }
+      it 'does not add rows for missing values' do
+        data_pull.run
+
+        expect(JSON.parse(stdout.string)).to be_empty
+      end
+    end
+  end
+
+  describe DataPull::UuidLookup do
+    subject(:subtask) { DataPull::UuidLookup.new }
+
+    describe '#run' do
+      let(:users) { create_list(:user, 2) }
+
+      let(:args) { [*users.map { |u| u.email_addresses.first.email }, 'missing@example.com'] }
+      let(:include_missing) { true }
+
+      subject(:result) { subtask.run(args:, include_missing:) }
+
+      it 'looks up the UUIDs for the given email addresses' do
+        expect(result.table).to eq(
+          [
+            ['email', 'uuid'],
+            *users.map { |u| [u.email_addresses.first.email, u.uuid] },
+            ['missing@example.com', '[NOT FOUND]'],
+          ],
+        )
+      end
+    end
+  end
+
+  describe DataPull::UuidConvert do
+    subject(:subtask) { DataPull::UuidConvert.new }
+
+    describe '#run' do
+      let(:agency_identities) { create_list(:agency_identity, 2).sort_by(&:uuid) }
+
+      let(:args) { [*agency_identities.map(&:uuid), 'does-not-exist'] }
+      let(:include_missing) { true }
+      subject(:result) { subtask.run(args:, include_missing:) }
+
+      it 'converts the agency agency identities to internal UUIDs' do
+        expect(result.table).to eq(
+          [
+            ['partner_uuid', 'source', 'internal_uuid'],
+            *agency_identities.map { |a| [a.uuid, a.agency.name, a.user.uuid] },
+            ['does-not-exist', '[NOT FOUND]', '[NOT FOUND]'],
+          ],
+        )
+      end
+    end
+  end
+
+  describe DataPull::EmailLookup do
+    subject(:subtask) { DataPull::EmailLookup.new }
+
+    describe '#run' do
+      let(:user) { create(:user, :with_multiple_emails) }
+
+      let(:args) { [user.uuid, 'does-not-exist'] }
+      let(:include_missing) { true }
+      subject(:result) { subtask.run(args:, include_missing:) }
+
+      it 'loads email addresses for the user' do
+        expect(result.table).to eq(
+          [
+            ['uuid', 'email', 'confirmed_at'],
+            *user.email_addresses.sort_by(&:id).map do |e|
+              [e.user.uuid, e.email, e.confirmed_at.round(6)]
+            end,
+            ['does-not-exist', '[NOT FOUND]', nil],
+          ],
+        )
+      end
+    end
+  end
+end