diff --git a/commcare_connect/connect_id_client/__init__.py b/commcare_connect/connect_id_client/__init__.py
new file mode 100644
index 00000000..fc3376b9
--- /dev/null
+++ b/commcare_connect/connect_id_client/__init__.py
@@ -0,0 +1 @@
+from .main import fetch_users, send_message, send_message_bulk  # noqa: F401
diff --git a/commcare_connect/connect_id_client/main.py b/commcare_connect/connect_id_client/main.py
new file mode 100644
index 00000000..5e36d0bf
--- /dev/null
+++ b/commcare_connect/connect_id_client/main.py
@@ -0,0 +1,45 @@
+import httpx
+from django.conf import settings
+from httpx import BasicAuth, Response
+
+from commcare_connect.connect_id_client.models import ConnectIdUser, Message, MessagingBulkResponse, MessagingResponse
+
+GET = "GET"
+POST = "POST"
+
+
+def fetch_users(phone_number_list: list[str]) -> list[ConnectIdUser]:
+    response = _make_request(GET, "/users/fetch_users", params={"phone_numbers": phone_number_list})
+    data = response.json()
+    return [ConnectIdUser(**user_dict) for user_dict in data["found_users"]]
+
+
+def send_message(message: Message):
+    """Send a push notification to a user."""
+    response = _make_request(POST, "/messaging/send/", json=message.asdict())
+    data = response.json()
+    return MessagingResponse.build(**data)
+
+
+def send_message_bulk(messages: list[Message]) -> MessagingBulkResponse:
+    """Send a push notification to multiple users."""
+    json = {"messages": [message.asdict() for message in messages]}
+    response = _make_request(POST, "/messaging/send_bulk/", json=json)
+    data = response.json()
+    return MessagingBulkResponse.build(**data)
+
+
+def _make_request(method, path, params=None, json=None) -> Response:
+    if json and not method == "POST":
+        raise ValueError("json can only be used with POST requests")
+
+    auth = BasicAuth(settings.CONNECTID_CLIENT_ID, settings.CONNECTID_CLIENT_SECRET)
+    response = httpx.request(
+        method,
+        f"{settings.CONNECTID_URL}{path}",
+        params=params,
+        json=json,
+        auth=auth,
+    )
+    response.raise_for_status()
+    return response
diff --git a/commcare_connect/connect_id_client/models.py b/commcare_connect/connect_id_client/models.py
new file mode 100644
index 00000000..915bb282
--- /dev/null
+++ b/commcare_connect/connect_id_client/models.py
@@ -0,0 +1,73 @@
+import dataclasses
+from enum import StrEnum
+
+
+@dataclasses.dataclass
+class ConnectIdUser:
+    name: str
+    username: str
+    phone_number: str
+
+    def __str__(self) -> str:
+        return f"{self.name} ({self.username})"
+
+
+class MessageStatus(StrEnum):
+    success = "success"
+    error = "error"
+    deactivated = "deactivated"
+
+
+@dataclasses.dataclass
+class UserMessageStatus:
+    username: str
+    status: MessageStatus
+
+    @classmethod
+    def build(cls, username: str, status: str):
+        return cls(username, MessageStatus(status))
+
+
+@dataclasses.dataclass
+class MessagingResponse:
+    all_success: bool
+    responses: list[UserMessageStatus]
+
+    @classmethod
+    def build(cls, all_success: bool, responses: list[dict]):
+        return cls(all_success, [UserMessageStatus.build(**response) for response in responses])
+
+    def get_failures(self) -> list[UserMessageStatus]:
+        return [response for response in self.responses if response.status != MessageStatus.success]
+
+
+@dataclasses.dataclass
+class MessagingBulkResponse:
+    all_success: bool
+    messages: list[MessagingResponse]
+
+    @classmethod
+    def build(cls, all_success: bool, messages: list[dict]):
+        return cls(all_success, [MessagingResponse.build(**message) for message in messages])
+
+    def get_failures(self) -> list[list[UserMessageStatus]]:
+        """Return a list of lists of UserMessageStatus objects, where each
+        inner list is the list of failures for a single message.
+
+        Usage:
+            for message, failures in zip(messages, response.get_failures()):
+                if failures:
+                    # do something with the failures
+        """
+        return [message.get_failures() for message in self.messages]
+
+
+@dataclasses.dataclass
+class Message:
+    usernames: list[str]
+    title: str = None
+    body: str = None
+    data: dict = None
+
+    def asdict(self):
+        return {k: v for k, v in dataclasses.asdict(self).items() if v is not None}
diff --git a/commcare_connect/connect_id_client/tests.py b/commcare_connect/connect_id_client/tests.py
new file mode 100644
index 00000000..13b0d26b
--- /dev/null
+++ b/commcare_connect/connect_id_client/tests.py
@@ -0,0 +1,109 @@
+from httpx import URL
+
+from .main import fetch_users, send_message, send_message_bulk
+from .models import Message, MessageStatus
+
+
+def test_fetch_users(httpx_mock):
+    httpx_mock.add_response(
+        method="GET",
+        json={
+            "found_users": [
+                {
+                    "username": "user_name1",
+                    "name": "name1",
+                    "phone_number": "phone_number1",
+                },
+                {
+                    "name": "name2",
+                    "username": "user_name2",
+                    "phone_number": "phone_number2",
+                },
+            ]
+        },
+    )
+
+    users = fetch_users(["phone_number1", "phone_number2"])
+    assert len(users) == 2
+    assert users[0].name == "name1"
+    assert users[1].name == "name2"
+
+    request = httpx_mock.get_request()
+    assert URL(request.url).params.get_list("phone_numbers") == ["phone_number1", "phone_number2"]
+
+
+def test_send_message(httpx_mock):
+    httpx_mock.add_response(
+        method="POST",
+        match_json={"usernames": ["user_name1"], "body": "test message"},
+        json={"all_success": True, "responses": [{"username": "user_name1", "status": "success"}]},
+    )
+    print(Message(usernames=["user_name1"], body="test message").asdict())
+    result = send_message(Message(usernames=["user_name1"], body="test message"))
+    assert result.all_success is True
+    assert len(result.responses) == 1
+    assert result.responses[0].username == "user_name1"
+    assert result.responses[0].status == MessageStatus.success
+
+
+def test_send_message_bulk(httpx_mock):
+    httpx_mock.add_response(
+        match_json={
+            "messages": [
+                {
+                    "usernames": ["user_name1", "user_name2"],
+                    "body": "test message1",
+                },
+                {
+                    "usernames": ["user_name3", "user_name4"],
+                    "body": "test message2",
+                },
+            ]
+        },
+        json={
+            "all_success": False,
+            "messages": [
+                {
+                    "all_success": True,
+                    "responses": [
+                        {"status": "success", "username": "user_name1"},
+                        {"status": "success", "username": "user_name2"},
+                    ],
+                },
+                {
+                    "all_success": False,
+                    "responses": [
+                        {"status": "error", "username": "user_name3"},
+                        {"status": "deactivated", "username": "user_name4"},
+                    ],
+                },
+            ],
+        },
+    )
+
+    result = send_message_bulk(
+        [
+            Message(usernames=["user_name1", "user_name2"], body="test message1"),
+            Message(usernames=["user_name3", "user_name4"], body="test message2"),
+        ]
+    )
+    assert result.all_success is False
+    assert len(result.messages) == 2
+
+    assert result.messages[0].all_success is True
+    assert [(resp.username, resp.status) for resp in result.messages[0].responses] == [
+        ("user_name1", MessageStatus.success),
+        ("user_name2", MessageStatus.success),
+    ]
+    assert result.messages[0].responses[0].username == "user_name1"
+    assert result.messages[0].responses[0].status == MessageStatus.success
+    assert result.messages[0].responses[1].username == "user_name2"
+    assert result.messages[0].responses[1].status == MessageStatus.success
+
+    assert result.messages[1].all_success is False
+    assert [(resp.username, resp.status) for resp in result.messages[1].responses] == [
+        ("user_name3", MessageStatus.error),
+        ("user_name4", MessageStatus.deactivated),
+    ]
+
+    assert result.get_failures() == [[], list(result.messages[1].responses)]
diff --git a/commcare_connect/opportunity/tasks.py b/commcare_connect/opportunity/tasks.py
index e09523dc..31627141 100644
--- a/commcare_connect/opportunity/tasks.py
+++ b/commcare_connect/opportunity/tasks.py
@@ -1,9 +1,8 @@
-import requests
-from django.conf import settings
 from django.core.files.base import ContentFile
 from django.core.files.storage import default_storage
 from django.utils.timezone import now
 
+from commcare_connect.connect_id_client import fetch_users
 from commcare_connect.opportunity.app_xml import get_connect_blocks_for_app, get_deliver_units_for_app
 from commcare_connect.opportunity.export import export_empty_payment_table, export_user_visit_data
 from commcare_connect.opportunity.forms import DateRanges
@@ -44,15 +43,9 @@ def create_learn_modules_and_deliver_units(opportunity_id):
 
 @celery_app.task()
 def add_connect_users(user_list: list[str], opportunity_id: str):
-    result = requests.get(
-        f"{settings.CONNECTID_URL}/users/fetch_users",
-        auth=(settings.CONNECTID_CLIENT_ID, settings.CONNECTID_CLIENT_SECRET),
-        params={"phone_numbers": user_list},
-    )
-    data = result.json()
-    for user in data["found_users"]:
+    for user in fetch_users(user_list):
         u, _ = User.objects.update_or_create(
-            username=user["username"], defaults={"phone_number": user["phone_number"], "name": user["name"]}
+            username=user.username, defaults={"phone_number": user.phone_number, "name": user.name}
         )
         opportunity_access, _ = OpportunityAccess.objects.get_or_create(user=u, opportunity_id=opportunity_id)
         invite_user(u, opportunity_access)
diff --git a/commcare_connect/opportunity/tests/test_tasks.py b/commcare_connect/opportunity/tests/test_tasks.py
index 1f7a267c..94735c10 100644
--- a/commcare_connect/opportunity/tests/test_tasks.py
+++ b/commcare_connect/opportunity/tests/test_tasks.py
@@ -2,6 +2,7 @@
 
 import pytest
 
+from commcare_connect.connect_id_client.models import ConnectIdUser
 from commcare_connect.opportunity.models import OpportunityAccess
 from commcare_connect.opportunity.tasks import add_connect_users
 from commcare_connect.opportunity.tests.factories import OpportunityFactory
@@ -13,15 +14,13 @@ class TestConnectUserCreation:
     def test_add_connect_user(self):
         opportunity = OpportunityFactory()
         with (
-            mock.patch("commcare_connect.opportunity.tasks.requests.get") as request,
+            mock.patch("commcare_connect.opportunity.tasks.fetch_users") as fetch_users,
             mock.patch("commcare_connect.users.helpers.send_sms"),
         ):
-            request.return_value.json.return_value = {
-                "found_users": [
-                    {"username": "test", "phone_number": "+15555555555", "name": "a"},
-                    {"username": "test2", "phone_number": "+12222222222", "name": "b"},
-                ]
-            }
+            fetch_users.return_value = [
+                ConnectIdUser(username="test", phone_number="+15555555555", name="a"),
+                ConnectIdUser(username="test2", phone_number="+12222222222", name="b"),
+            ]
             add_connect_users(["+15555555555", "+12222222222"], opportunity.id)
 
         user_list = User.objects.filter(username="test")
diff --git a/requirements/base.txt b/requirements/base.txt
index 677f6ac7..00cf97a8 100644
--- a/requirements/base.txt
+++ b/requirements/base.txt
@@ -1,5 +1,5 @@
 #
-# This file is autogenerated by pip-compile with Python 3.10
+# This file is autogenerated by pip-compile with Python 3.11
 # by the following command:
 #
 #    inv requirements
@@ -23,9 +23,7 @@ argon2-cffi-bindings==21.2.0
 asgiref==3.7.2
     # via django
 async-timeout==4.0.3
-    # via
-    #   aiohttp
-    #   redis
+    # via aiohttp
 attrs==23.1.0
     # via
     #   aiohttp
@@ -121,8 +119,6 @@ drf-spectacular==0.26.4
     # via -r requirements/base.in
 et-xmlfile==1.1.0
     # via openpyxl
-exceptiongroup==1.1.3
-    # via anyio
 flatten-dict==0.4.2
     # via -r requirements/base.in
 frozenlist==1.4.0
@@ -241,8 +237,6 @@ text-unidecode==1.3
     # via python-slugify
 twilio==8.8.0
     # via -r requirements/base.in
-typing-extensions==4.7.1
-    # via asgiref
 tzdata==2023.3
     # via
     #   celery
diff --git a/requirements/dev.txt b/requirements/dev.txt
index 6930d0e0..54c5146b 100644
--- a/requirements/dev.txt
+++ b/requirements/dev.txt
@@ -179,7 +179,7 @@ pytest==7.4.0
     #   pytest-httpx
 pytest-django==4.5.2
     # via -r requirements/dev.in
-pytest-httpx==0.23.1
+pytest-httpx==0.24.0
     # via -r requirements/dev.in
 python-dateutil==2.8.2
     # via
diff --git a/tasks.py b/tasks.py
index fbd12ac4..c5c1d299 100644
--- a/tasks.py
+++ b/tasks.py
@@ -32,11 +32,14 @@ def down(c: Context):
 
 
 @task
-def requirements(c: Context, upgrade=False):
-    """Re-compile the pip requirements files"""
+def requirements(c: Context, upgrade=False, upgrade_package=None):
+    if upgrade and upgrade_package:
+        raise Exit("Cannot specify both upgrade and upgrade-package", -1)
     args = " -U" if upgrade else ""
     cmd_base = "pip-compile -q --resolver=backtracking"
     env = {"CUSTOM_COMPILE_COMMAND": "inv requirements"}
+    if upgrade_package:
+        cmd_base += f" --upgrade-package {upgrade_package}"
     c.run(f"{cmd_base} requirements/base.in{args}", env=env)
     c.run(f"{cmd_base} requirements/dev.in{args}", env=env)
     # can't use backtracking resolver for now: https://github.com/pypa/pip/issues/8713