diff --git a/ably/__init__.py b/ably/__init__.py
index 95412484..b17e1db1 100644
--- a/ably/__init__.py
+++ b/ably/__init__.py
@@ -5,6 +5,7 @@
 from ably.rest.push import Push
 from ably.rest.rest import AblyRest
 from ably.types.annotation import Annotation, AnnotationAction
+from ably.types.apnsbroadcast import PushApnsBroadcast
 from ably.types.capability import Capability
 from ably.types.channelmode import ChannelMode
 from ably.types.channeloptions import ChannelOptions
diff --git a/ably/rest/push.py b/ably/rest/push.py
index f99b2b1d..7cbebf4f 100644
--- a/ably/rest/push.py
+++ b/ably/rest/push.py
@@ -1,12 +1,15 @@
 from typing import Optional
+from urllib import parse
 
 from ably.http.paginatedresult import PaginatedResult, format_params
+from ably.types.apnsbroadcast import PushApnsBroadcast
 from ably.types.channelsubscription import (
     PushChannelSubscription,
     channel_subscriptions_response_processor,
     channels_response_processor,
 )
 from ably.types.device import DeviceDetails, device_details_response_processor
+from ably.util.exceptions import AblyException
 
 
 class Push:
@@ -26,6 +29,7 @@ def __init__(self, ably):
         self.__ably = ably
         self.__device_registrations = PushDeviceRegistrations(ably)
         self.__channel_subscriptions = PushChannelSubscriptions(ably)
+        self.__live_activity = PushLiveActivity(ably)
 
     @property
     def ably(self):
@@ -39,6 +43,12 @@ def device_registrations(self):
     def channel_subscriptions(self):
         return self.__channel_subscriptions
 
+    @property
+    def live_activity(self):
+        """Controls iOS Live Activities over an APNs broadcast channel created
+        with `create_apns_broadcast`. Experimental."""
+        return self.__live_activity
+
     async def publish(self, recipient: dict, data: dict, timeout: Optional[float] = None):
         """Publish a push notification to a single device.
 
@@ -62,6 +72,96 @@ async def publish(self, recipient: dict, data: dict, timeout: Optional[float] =
         body.update({'recipient': recipient})
         await self.ably.http.post('/push/publish', body=body, timeout=timeout)
 
+    async def create_apns_broadcast(self, message_storage_policy: int) -> PushApnsBroadcast:
+        """Creates an APNs broadcast channel for use with an iOS Live Activity.
+        Call once before starting the Live Activity and persist the returned
+        ids for the session.
+
+        Experimental: this is a preview feature and may change in a future
+        non-major release.
+
+        :Parameters:
+        - `message_storage_policy`: 1 to cache the last update so that devices
+          subscribing late receive the current content state, 0 to disable caching
+        """
+        body = {'messageStoragePolicy': message_storage_policy}
+        response = await self.ably.http.post('/push/apnsBroadcastChannels', body=body)
+        return PushApnsBroadcast.from_dict(response.to_native())
+
+
+class PushLiveActivity:
+    """Controls the lifecycle of an iOS Live Activity over an APNs broadcast
+    channel created with `PushAdmin.create_apns_broadcast`.
+
+    Experimental: this is a preview feature and may change in a future
+    non-major release.
+    """
+
+    def __init__(self, ably):
+        self.__ably = ably
+
+    @property
+    def ably(self):
+        return self.__ably
+
+    async def start(self, recipient: dict, apns_broadcast: str, apns: dict, headers: Optional[dict] = None):
+        """Sends a push-to-start notification, starting a new Live Activity on
+        each targeted device.
+
+        :Parameters:
+        - `recipient`: either `{'channels': [...]}`, targeting the devices
+          subscribed to any of those channels, or `{'deviceId': ...}`,
+          targeting a single device
+        - `apns_broadcast`: the id returned by `create_apns_broadcast`
+        - `apns`: the APNs Live Activity start payload, passed through as-is
+        - `headers`: optional APNs delivery headers, such as `apns-priority`
+          and `apns-expiration`
+        """
+        channels = recipient.get('channels')
+        device_id = recipient.get('deviceId')
+        if bool(channels) == bool(device_id):
+            raise AblyException(
+                'PushLiveActivity.start() requires exactly one of recipient channels or deviceId', 400, 40000)
+
+        body = {'apns': apns}
+        if channels:
+            body['channels'] = channels
+        else:
+            body['deviceId'] = device_id
+        await self._post(apns_broadcast, 'start', body, headers)
+
+    async def update(self, apns_broadcast: str, apns: dict, headers: Optional[dict] = None):
+        """Sends a content-state update to every device with an active Live
+        Activity on the broadcast channel. APNs fans the single push out to
+        the subscribed devices.
+
+        :Parameters:
+        - `apns_broadcast`: the id returned by `create_apns_broadcast`
+        - `apns`: the APNs Live Activity update payload, passed through as-is
+        - `headers`: optional APNs delivery headers, such as `apns-priority`
+          and `apns-expiration`
+        """
+        await self._post(apns_broadcast, 'broadcast', {'apns': apns}, headers)
+
+    async def end(self, apns_broadcast: str, apns: dict, headers: Optional[dict] = None):
+        """Ends the Live Activity on every subscribed device and cleans up the
+        APNs channel. The broadcast id is no longer valid afterwards.
+
+        :Parameters:
+        - `apns_broadcast`: the id returned by `create_apns_broadcast`
+        - `apns`: the APNs Live Activity end payload, passed through as-is
+        - `headers`: optional APNs delivery headers, such as `apns-priority`
+          and `apns-expiration`
+        """
+        await self._post(apns_broadcast, 'end', {'apns': apns}, headers)
+
+    async def _post(self, apns_broadcast: str, action: str, body: dict, headers: Optional[dict]):
+        # The APNs delivery headers travel in the request body, not as HTTP headers
+        if headers:
+            body['headers'] = headers
+        path = f'/push/apnsBroadcastChannels/{parse.quote(apns_broadcast, safe="")}/{action}'
+        await self.ably.http.post(path, body=body)
+
 
 class PushDeviceRegistrations:
 
diff --git a/ably/types/apnsbroadcast.py b/ably/types/apnsbroadcast.py
new file mode 100644
index 00000000..442958fd
--- /dev/null
+++ b/ably/types/apnsbroadcast.py
@@ -0,0 +1,23 @@
+class PushApnsBroadcast:
+    """An APNs broadcast channel created for an iOS Live Activity.
+
+    Experimental: this is a preview feature and may change in a future non-major release.
+    """
+
+    def __init__(self, id, apns_channel_id):
+        self.__id = id
+        self.__apns_channel_id = apns_channel_id
+
+    @property
+    def id(self):
+        """The opaque Ably broadcast id, passed to the `push.admin.live_activity` methods."""
+        return self.__id
+
+    @property
+    def apns_channel_id(self):
+        """The Apple-assigned channel id."""
+        return self.__apns_channel_id
+
+    @classmethod
+    def from_dict(cls, obj):
+        return cls(id=obj.get('id'), apns_channel_id=obj.get('apnsChannelId'))
diff --git a/test/unit/push_live_activity_test.py b/test/unit/push_live_activity_test.py
new file mode 100644
index 00000000..49b7a89c
--- /dev/null
+++ b/test/unit/push_live_activity_test.py
@@ -0,0 +1,40 @@
+import httpx
+import pytest
+
+from ably import AblyRest
+from ably.types.testoptions import TestOptions
+from ably.util.exceptions import AblyException
+
+
+class RecordingTransport(httpx.AsyncBaseTransport):
+    def __init__(self):
+        self.requests = []
+
+    async def handle_async_request(self, request):
+        self.requests.append(request)
+        return httpx.Response(200, json={})
+
+
+# RSH1e1a: the recipient must name exactly one of channels and deviceId,
+# and is rejected before any request is made
+@pytest.mark.parametrize('recipient', [
+    {},
+    {'channels': []},
+    {'channels': ['a'], 'deviceId': 'device-1'},
+])
+async def test_live_activity_start_rejects_invalid_recipient(recipient):
+    transport = RecordingTransport()
+    ably = AblyRest(key='appId.keyId:keySecret', _test_options=TestOptions(http_transport=transport))
+    try:
+        with pytest.raises(AblyException) as excinfo:
+            await ably.push.admin.live_activity.start(
+                recipient=recipient,
+                apns_broadcast='broadcast-1',
+                apns={'aps': {'event': 'start'}},
+            )
+    finally:
+        await ably.close()
+
+    assert excinfo.value.code == 40000
+    assert excinfo.value.status_code == 400
+    assert transport.requests == []
diff --git a/test/uts/rest/unit/push/push_admin_apns_broadcast_test.py b/test/uts/rest/unit/push/push_admin_apns_broadcast_test.py
new file mode 100644
index 00000000..2a437e01
--- /dev/null
+++ b/test/uts/rest/unit/push/push_admin_apns_broadcast_test.py
@@ -0,0 +1,110 @@
+"""Derived from uts/rest/unit/push/push_admin_apns_broadcast.md in ably/specification.
+
+Spec points: RSH1, RSH1d
+"""
+
+import msgpack
+import pytest
+
+from ably.types.apnsbroadcast import PushApnsBroadcast
+from ably.util.exceptions import AblyException
+from test.uts.helpers.client import rest_client
+from test.uts.helpers.mock_http import MockHttpClient
+
+BROADCAST_RESPONSE = {'id': 'broadcast-1', 'apnsChannelId': 'apns-1'}
+
+
+def capture_and_respond(captured_requests, status=201, body=None):
+    def on_request(request):
+        captured_requests.append(request)
+        request.respond_with(status, body if body is not None else BROADCAST_RESPONSE)
+
+    return on_request
+
+
+# UTS: rest/unit/RSH1d/create-apns-broadcast-post-0
+async def test_rsh1d_create_apns_broadcast_post():
+    captured_requests = []
+    mock_http = MockHttpClient(
+        on_connection_attempt=lambda conn: conn.respond_with_success(),
+        on_request=capture_and_respond(captured_requests),
+    )
+    client = rest_client(mock_http)
+
+    await client.push.admin.create_apns_broadcast(message_storage_policy=1)
+
+    assert len(captured_requests) == 1
+
+    request = captured_requests[0]
+    assert request.method == 'POST'
+    assert request.url.path == '/push/apnsBroadcastChannels'
+
+
+# UTS: rest/unit/RSH1d/message-storage-policy-1
+async def test_rsh1d_message_storage_policy():
+    captured_requests = []
+    mock_http = MockHttpClient(
+        on_connection_attempt=lambda conn: conn.respond_with_success(),
+        on_request=capture_and_respond(captured_requests),
+    )
+    client = rest_client(mock_http)
+
+    await client.push.admin.create_apns_broadcast(message_storage_policy=0)
+
+    assert len(captured_requests) == 1
+
+    body = msgpack.unpackb(captured_requests[0].body)
+    assert body['messageStoragePolicy'] == 0
+
+
+# UTS: rest/unit/RSH1d/returns-ids-2
+async def test_rsh1d_returns_ids():
+    mock_http = MockHttpClient(
+        on_connection_attempt=lambda conn: conn.respond_with_success(),
+        on_request=lambda request: request.respond_with(201, {
+            'id': 'broadcast-xyz',
+            'apnsChannelId': 'apple-channel-abc',
+        }),
+    )
+    client = rest_client(mock_http)
+
+    result = await client.push.admin.create_apns_broadcast(message_storage_policy=1)
+
+    assert isinstance(result, PushApnsBroadcast)
+    assert result.id == 'broadcast-xyz'
+    assert result.apns_channel_id == 'apple-channel-abc'
+
+
+# UTS: rest/unit/RSH1d/auth-header-3
+async def test_rsh1d_auth_header():
+    captured_requests = []
+    mock_http = MockHttpClient(
+        on_connection_attempt=lambda conn: conn.respond_with_success(),
+        on_request=capture_and_respond(captured_requests),
+    )
+    client = rest_client(mock_http)
+
+    await client.push.admin.create_apns_broadcast(message_storage_policy=1)
+
+    assert len(captured_requests) == 1
+    assert captured_requests[0].headers['Authorization'].startswith('Basic ')
+
+
+# UTS: rest/unit/RSH1d/server-error-4
+async def test_rsh1d_server_error():
+    mock_http = MockHttpClient(
+        on_connection_attempt=lambda conn: conn.respond_with_success(),
+        on_request=lambda request: request.respond_with(400, {
+            'error': {
+                'code': 40000,
+                'statusCode': 400,
+                'message': 'Invalid request',
+            },
+        }),
+    )
+    client = rest_client(mock_http)
+
+    with pytest.raises(AblyException) as excinfo:
+        await client.push.admin.create_apns_broadcast(message_storage_policy=1)
+
+    assert excinfo.value.code == 40000
diff --git a/test/uts/rest/unit/push/push_admin_live_activity_test.py b/test/uts/rest/unit/push/push_admin_live_activity_test.py
new file mode 100644
index 00000000..bfbaeb9d
--- /dev/null
+++ b/test/uts/rest/unit/push/push_admin_live_activity_test.py
@@ -0,0 +1,193 @@
+"""Derived from uts/rest/unit/push/push_admin_live_activity.md in ably/specification.
+
+Spec points: RSH1, RSH1e, RSH1e1, RSH1e2, RSH1e3
+"""
+
+from urllib.parse import quote
+
+import msgpack
+import pytest
+
+from ably.util.exceptions import AblyException
+from test.uts.helpers.client import rest_client
+from test.uts.helpers.mock_http import MockHttpClient
+
+APNS_HEADERS = {'apns-priority': '10', 'apns-expiration': '1782948701'}
+
+SERVER_ERROR = {
+    'error': {
+        'code': 40000,
+        'statusCode': 400,
+        'message': 'Invalid request',
+    },
+}
+
+
+def capture_and_respond(captured_requests, status=200, body=None):
+    def on_request(request):
+        captured_requests.append(request)
+        request.respond_with(status, body if body is not None else {})
+
+    return on_request
+
+
+def capturing_client(captured_requests):
+    mock_http = MockHttpClient(
+        on_connection_attempt=lambda conn: conn.respond_with_success(),
+        on_request=capture_and_respond(captured_requests),
+    )
+    return rest_client(mock_http)
+
+
+def failing_client():
+    mock_http = MockHttpClient(
+        on_connection_attempt=lambda conn: conn.respond_with_success(),
+        on_request=lambda request: request.respond_with(400, SERVER_ERROR),
+    )
+    return rest_client(mock_http)
+
+
+# UTS: rest/unit/RSH1e1/start-post-0
+async def test_rsh1e1_start_post():
+    captured_requests = []
+    client = capturing_client(captured_requests)
+
+    await client.push.admin.live_activity.start(
+        recipient={'channels': ['nba:lakers', 'nba:celtics']},
+        apns_broadcast='broadcast-1',
+        apns={'aps': {'event': 'start', 'attributes-type': 'GameAttributes'}},
+        headers=APNS_HEADERS,
+    )
+
+    assert len(captured_requests) == 1
+
+    request = captured_requests[0]
+    assert request.method == 'POST'
+    assert request.url.path == '/push/apnsBroadcastChannels/broadcast-1/start'
+
+    body = msgpack.unpackb(request.body)
+    assert body['channels'] == ['nba:lakers', 'nba:celtics']
+    assert body['apns']['aps']['event'] == 'start'
+    assert 'deviceId' not in body
+    assert body['headers'] == APNS_HEADERS
+    assert request.headers['Authorization'].startswith('Basic ')
+
+
+# UTS: rest/unit/RSH1e1/start-deviceid-encode-1
+async def test_rsh1e1_start_deviceid_encode():
+    captured_requests = []
+    client = capturing_client(captured_requests)
+
+    await client.push.admin.live_activity.start(
+        recipient={'deviceId': 'device-7'},
+        apns_broadcast='broadcast/with space',
+        apns={'aps': {'event': 'start'}},
+    )
+
+    assert len(captured_requests) == 1
+
+    request = captured_requests[0]
+    # url.path is percent-decoded, so the encoding the spec asserts on is read off raw_path
+    encoded_id = quote('broadcast/with space', safe='')
+    assert request.url.raw_path == '/push/apnsBroadcastChannels/' + encoded_id + '/start'
+
+    body = msgpack.unpackb(request.body)
+    assert body['deviceId'] == 'device-7'
+    assert 'channels' not in body
+
+
+# UTS: rest/unit/RSH1e2/update-post-0
+async def test_rsh1e2_update_post():
+    captured_requests = []
+    client = capturing_client(captured_requests)
+
+    await client.push.admin.live_activity.update(
+        apns_broadcast='broadcast-1',
+        apns={'aps': {'event': 'update', 'content-state': {'homeScore': 14}}},
+        headers=APNS_HEADERS,
+    )
+
+    assert len(captured_requests) == 1
+
+    request = captured_requests[0]
+    assert request.method == 'POST'
+    assert request.url.path == '/push/apnsBroadcastChannels/broadcast-1/broadcast'
+
+    body = msgpack.unpackb(request.body)
+    assert body['apns']['aps']['event'] == 'update'
+    # The optional APNs delivery headers are sent in the request body under a "headers" key
+    assert body['headers'] == APNS_HEADERS
+    assert request.headers['Authorization'].startswith('Basic ')
+
+
+# UTS: rest/unit/RSH1e2/update-no-headers-1
+async def test_rsh1e2_update_no_headers():
+    captured_requests = []
+    client = capturing_client(captured_requests)
+
+    await client.push.admin.live_activity.update(
+        apns_broadcast='broadcast-1',
+        apns={'aps': {'event': 'update', 'content-state': {}}},
+    )
+
+    assert len(captured_requests) == 1
+
+    body = msgpack.unpackb(captured_requests[0].body)
+    assert body['apns']['aps']['event'] == 'update'
+    assert 'headers' not in body
+
+
+# UTS: rest/unit/RSH1e3/end-post-0
+async def test_rsh1e3_end_post():
+    captured_requests = []
+    client = capturing_client(captured_requests)
+
+    await client.push.admin.live_activity.end(
+        apns_broadcast='broadcast-1',
+        apns={'aps': {'event': 'end', 'content-state': {'homeScore': 112}, 'dismissal-date': 1700000000}},
+    )
+
+    assert len(captured_requests) == 1
+
+    request = captured_requests[0]
+    assert request.method == 'POST'
+    assert request.url.path == '/push/apnsBroadcastChannels/broadcast-1/end'
+
+    body = msgpack.unpackb(request.body)
+    assert body['apns']['aps']['event'] == 'end'
+    assert body['apns']['aps']['dismissal-date'] == 1700000000
+    assert request.headers['Authorization'].startswith('Basic ')
+
+
+# UTS: rest/unit/RSH1e1/server-error-2
+async def test_rsh1e1_server_error():
+    client = failing_client()
+
+    with pytest.raises(AblyException) as excinfo:
+        await client.push.admin.live_activity.start(
+            recipient={'channels': ['nba:lakers']},
+            apns_broadcast='broadcast-1',
+            apns={},
+        )
+
+    assert excinfo.value.code == 40000
+
+
+# UTS: rest/unit/RSH1e2/server-error-2
+async def test_rsh1e2_server_error():
+    client = failing_client()
+
+    with pytest.raises(AblyException) as excinfo:
+        await client.push.admin.live_activity.update(apns_broadcast='broadcast-1', apns={})
+
+    assert excinfo.value.code == 40000
+
+
+# UTS: rest/unit/RSH1e3/server-error-1
+async def test_rsh1e3_server_error():
+    client = failing_client()
+
+    with pytest.raises(AblyException) as excinfo:
+        await client.push.admin.live_activity.end(apns_broadcast='broadcast-1', apns={})
+
+    assert excinfo.value.code == 40000