diff --git a/app/assets/stylesheets/components/page-footer.scss b/app/assets/stylesheets/components/page-footer.scss
index 04b6a31b1..e40c42a35 100644
--- a/app/assets/stylesheets/components/page-footer.scss
+++ b/app/assets/stylesheets/components/page-footer.scss
@@ -59,3 +59,21 @@
   }
 
 }
+
+.align-button-with-textbox {
+
+  .button {
+
+    @include media(desktop) {
+      position: relative;
+      top: 32px;
+      left: -30px;
+      width: 100%;
+      margin-right: -30px;
+      padding-top: 8px;
+      box-sizing: content-box;
+    }
+
+  }
+
+}
diff --git a/app/main/forms.py b/app/main/forms.py
index bf95901b3..f8db6e577 100644
--- a/app/main/forms.py
+++ b/app/main/forms.py
@@ -641,6 +641,11 @@ class SearchTemplatesForm(Form):
     search = SearchField('Search by name')
 
 
+class SearchNotificationsForm(Form):
+
+    to = SearchField('Search by phone number or email address')
+
+
 class PlaceholderForm(Form):
 
     pass
diff --git a/app/main/views/jobs.py b/app/main/views/jobs.py
index a44c41581..1e85faf0b 100644
--- a/app/main/views/jobs.py
+++ b/app/main/views/jobs.py
@@ -26,6 +26,7 @@ from app import (
     current_service,
     format_datetime_short)
 from app.main import main
+from app.main.forms import SearchNotificationsForm
 from app.utils import (
     get_page_from_request,
     generate_next_dict,
@@ -197,8 +198,10 @@ def view_notifications(service_id, message_type):
         'views/notifications.html',
         partials=get_notifications(service_id, message_type),
         message_type=message_type,
-        status=request.args.get('status'),
-        page=request.args.get('page', 1)
+        status=request.args.get('status') or 'sending,delivered,failed',
+        page=request.args.get('page', 1),
+        to=request.args.get('to'),
+        search_form=SearchNotificationsForm(to=request.args.get('to')),
     )
 
 
@@ -241,7 +244,9 @@ def get_notifications(service_id, message_type, status_override=None):
         page=page,
         template_type=[message_type],
         status=filter_args.get('status'),
-        limit_days=current_app.config['ACTIVITY_STATS_LIMIT_DAYS'])
+        limit_days=current_app.config['ACTIVITY_STATS_LIMIT_DAYS'],
+        to=request.args.get('to'),
+    )
 
     url_args = {
         'message_type': message_type,
@@ -249,11 +254,11 @@ def get_notifications(service_id, message_type, status_override=None):
     }
     prev_page = None
 
-    if notifications['links'].get('prev', None):
+    if 'links' in notifications and notifications['links'].get('prev', None):
         prev_page = generate_previous_dict('main.view_notifications', service_id, page, url_args=url_args)
     next_page = None
 
-    if notifications['links'].get('next', None):
+    if 'links' in notifications and notifications['links'].get('next', None):
         next_page = generate_next_dict('main.view_notifications', service_id, page, url_args)
 
     return {
diff --git a/app/notify_client/notification_api_client.py b/app/notify_client/notification_api_client.py
index 7ea6831e9..4a4399723 100644
--- a/app/notify_client/notification_api_client.py
+++ b/app/notify_client/notification_api_client.py
@@ -21,7 +21,8 @@ class NotificationApiClient(NotifyAdminAPIClient):
         limit_days=None,
         include_jobs=None,
         include_from_test_key=None,
-        format_for_csv=None
+        format_for_csv=None,
+        to=None,
     ):
         params = {}
         if page is not None:
@@ -38,6 +39,8 @@ class NotificationApiClient(NotifyAdminAPIClient):
             params['include_from_test_key'] = include_from_test_key
         if format_for_csv is not None:
             params['format_for_csv'] = format_for_csv
+        if to is not None:
+            params['to'] = to
         if job_id:
             return self.get(
                 url='/service/{}/job/{}/notifications'.format(service_id, job_id),
diff --git a/app/templates/views/notifications.html b/app/templates/views/notifications.html
index 61f16aff7..3190b4fa0 100644
--- a/app/templates/views/notifications.html
+++ b/app/templates/views/notifications.html
@@ -1,6 +1,8 @@
 {% extends "withnav_template.html" %}
 {% from "components/ajax-block.html" import ajax_block %}
 {% from "components/message-count-label.html" import message_count_label, recipient_count_label %}
+{% from "components/page-footer.html" import page_footer %}
+{% from "components/textbox.html" import textbox %}
 
 {% block service_page_title %}
   {{ message_count_label(99, message_type, suffix='') | capitalize }}
@@ -18,9 +20,27 @@
     'counts'
   ) }}
 
+  <form
+    method="get"
+    action="{{ url_for('.view_notifications', service_id=current_service.id, message_type=message_type) }}"
+    class="grid-row"
+  >
+    <div class="column-three-quarters">
+      <input type="hidden" name="status" value="{{ status }}">
+      {{ textbox(
+        search_form.to,
+        width='1-1',
+        label='Search by {}'.format('email address' if message_type == 'email' else 'phone number')
+      ) }}
+    </div>
+    <div class="column-one-quarter align-button-with-textbox">
+      <input type="submit" class="button" value="Search">
+    </div>
+  </form>
+
   {{ ajax_block(
     partials,
-    url_for('.get_notifications_as_json', service_id=current_service.id, message_type=message_type, status=status, page=page),
+    url_for('.get_notifications_as_json', service_id=current_service.id, message_type=message_type, status=status, page=page, to=to),
     'notifications'
   ) }}
 
diff --git a/tests/app/main/views/test_jobs.py b/tests/app/main/views/test_jobs.py
index 0980bfcaf..0625f37e7 100644
--- a/tests/app/main/views/test_jobs.py
+++ b/tests/app/main/views/test_jobs.py
@@ -8,6 +8,7 @@ from bs4 import BeautifulSoup
 
 from app.main.views.jobs import get_time_left, get_status_filters
 from tests import notification_json
+from tests.conftest import SERVICE_ONE_ID
 from freezegun import freeze_time
 
 
@@ -311,6 +312,13 @@ def test_should_show_updates_for_one_job_as_json(
         (None, 1)
     ]
 )
+@pytest.mark.parametrize(
+    "to_argument, expected_to_argument", [
+        ('', ''),
+        ('+447900900123', '+447900900123'),
+        ('test@example.com', 'test@example.com'),
+    ]
+)
 def test_can_show_notifications(
     logged_in_client,
     service_one,
@@ -322,13 +330,17 @@ def test_can_show_notifications(
     expected_api_call,
     page_argument,
     expected_page_argument,
+    to_argument,
+    expected_to_argument,
 ):
     response = logged_in_client.get(url_for(
         'main.view_notifications',
         service_id=service_one['id'],
         message_type=message_type,
         status=status_argument,
-        page=page_argument))
+        page=page_argument,
+        to=to_argument,
+    ))
     assert response.status_code == 200
     content = response.get_data(as_text=True)
     notifications = notification_json(service_one['id'])
@@ -348,13 +360,16 @@ def test_can_show_notifications(
         assert query_dict['status'] == [status_argument]
     if expected_page_argument:
         assert query_dict['page'] == [str(expected_page_argument)]
+    if to_argument:
+        assert query_dict['to'] == [to_argument]
 
     mock_get_notifications.assert_called_with(
         limit_days=7,
         page=expected_page_argument,
         service_id=service_one['id'],
         status=expected_api_call,
-        template_type=[message_type]
+        template_type=[message_type],
+        to=expected_to_argument,
     )
 
     json_response = logged_in_client.get(url_for(
@@ -367,6 +382,62 @@ def test_can_show_notifications(
     assert json_content.keys() == {'counts', 'notifications'}
 
 
+@pytest.mark.parametrize("initial_query_arguments, expected_status_field_value, expected_search_box_contents", [
+    (
+        {
+            'message_type': 'sms',
+        },
+        'sending,delivered,failed',
+        '',
+    ),
+    (
+        {
+            'message_type': 'sms',
+            'to': '+33(0)5-12-34-56-78',
+        },
+        'sending,delivered,failed',
+        '+33(0)5-12-34-56-78',
+    ),
+    (
+        {
+            'status': 'failed',
+            'message_type': 'email',
+            'page': '99',
+            'to': 'test@example.com',
+        },
+        'failed',
+        'test@example.com',
+    ),
+])
+def test_search_recipient_form(
+    logged_in_client,
+    mock_get_notifications,
+    mock_get_detailed_service,
+    initial_query_arguments,
+    expected_status_field_value,
+    expected_search_box_contents,
+):
+    response = logged_in_client.get(url_for(
+        'main.view_notifications',
+        service_id=SERVICE_ONE_ID,
+        **initial_query_arguments
+    ))
+    assert response.status_code == 200
+    page = BeautifulSoup(response.data.decode('utf-8'), 'html.parser')
+
+    action_url = page.find("form")['action']
+    url = urlparse(action_url)
+    assert url.path == '/services/{}/notifications/{}'.format(
+        SERVICE_ONE_ID,
+        initial_query_arguments['message_type']
+    )
+    query_dict = parse_qs(url.query)
+    assert query_dict == {}
+
+    assert page.find("input", {'name': 'status'})['value'] == expected_status_field_value
+    assert page.find("input", {'name': 'to'})['value'] == expected_search_box_contents
+
+
 def test_should_show_notifications_for_a_service_with_next_previous(
     logged_in_client,
     service_one,
diff --git a/tests/conftest.py b/tests/conftest.py
index 37dda3ed3..e9171cb8e 100644
--- a/tests/conftest.py
+++ b/tests/conftest.py
@@ -1069,6 +1069,7 @@ def mock_get_notifications(mocker, api_user_active):
         rows=5,
         include_jobs=None,
         include_from_test_key=None,
+        to=None,
     ):
         job = None
         if job_id is not None:
@@ -1101,7 +1102,9 @@ def mock_get_notifications_with_previous_next(mocker):
                            status=None,
                            limit_days=None,
                            include_jobs=None,
-                           include_from_test_key=None):
+                           include_from_test_key=None,
+                           to=None,
+                           ):
         return notification_json(service_id, with_links=True)
 
     return mocker.patch(