diff --git a/docs/messages/scheduled_messages.md b/docs/messages/scheduled_messages.md
new file mode 100644
index 0000000..ce01e8d
--- /dev/null
+++ b/docs/messages/scheduled_messages.md
@@ -0,0 +1,89 @@
+Scheduled messages let a user write a message now and have it sent to a channel later. Until it is sent, a scheduled message is visible only to the user who scheduled it and to server-side callers. Server-side, you can schedule, update, cancel, send and list the scheduled messages of any user.
+
+Scheduled messages must be enabled for your app. Requests return an error with the `scheduled_messages_disabled` reason otherwise.
+
+## Limits
+
+- The send time must be at least 1 minute and at most the app's maximum lead time (120 days by default) from now
+- A user can have at most 100 scheduled messages, and at most 10 in one channel (app defaults)
+- Slash commands cannot be scheduled
+
+## Scheduling a Message
+
+Pass `scheduled_at` to `send_message`. It accepts a `datetime` (a naive one is taken to be in UTC) or an ISO 8601 string. The message is stored instead of sent, and the response carries the scheduled message.
+
+```python
+from datetime import datetime, timedelta, timezone
+
+response = channel.send_message(
+    {"text": "See you tomorrow"},
+    "user-id",
+    scheduled_at=datetime.now(timezone.utc) + timedelta(days=1),
+)
+
+scheduled = response["scheduled_message"]
+print(scheduled["id"], scheduled["status"], scheduled["scheduled_at"])
+```
+
+`response["message"]` holds the content as it will be sent; it is not in the channel yet, and its ID is the scheduled message ID. The ID of the sent message is only known when it is sent (see the `scheduled_message.sent` event below).
+
+## Updating a Scheduled Message
+
+Change the message, the send time, or both. A new message replaces the scheduled one as a whole.
+
+```python
+client.update_scheduled_message(
+    scheduled["id"],
+    message={"text": "See you the day after"},
+    scheduled_at=datetime.now(timezone.utc) + timedelta(days=2),
+)
+
+# Only move the send time
+client.update_scheduled_message(
+    scheduled["id"], scheduled_at=datetime.now(timezone.utc) + timedelta(days=2)
+)
+```
+
+## Sending a Scheduled Message Now
+
+```python
+client.send_scheduled_message_now(scheduled["id"])
+```
+
+## Deleting a Scheduled Message
+
+Cancels a scheduled message so it is never sent, or removes one that failed to send.
+
+```python
+client.delete_scheduled_message(scheduled["id"])
+```
+
+Update, send now and delete return an error while the message is being sent. A message that failed to send can only be deleted.
+
+## Querying Scheduled Messages
+
+Lists the pending (`scheduled`, `sending`) and `failed` messages of one user across channels: pending ones first, then failed ones, each by send time.
+
+```python
+response = client.query_scheduled_messages(
+    "user-id",
+    filter={"cid": "messaging:general", "status": ["failed"]},
+    options={"limit": 20},
+)
+
+# Next page
+if response.get("next"):
+    response = client.query_scheduled_messages(
+        "user-id", options={"next": response["next"]}
+    )
+```
+
+## Events
+
+| Event | Webhook | Description |
+| --- | --- | --- |
+| `scheduled_message.created` | No | A message was scheduled |
+| `scheduled_message.updated` | No | A scheduled message was updated or sent now |
+| `scheduled_message.deleted` | No | A scheduled message was canceled |
+| `scheduled_message.sent` | Yes | A scheduled message was sent. `message_id` holds the ID of the sent message |
+| `scheduled_message.failed` | Yes | A scheduled message failed to send. `scheduled_message.failure_reason` says why |
diff --git a/stream_chat/async_chat/channel.py b/stream_chat/async_chat/channel.py
index 5b2708d..a4f70be 100644
--- a/stream_chat/async_chat/channel.py
+++ b/stream_chat/async_chat/channel.py
@@ -4,6 +4,7 @@
 
 from stream_chat.base.channel import ChannelInterface, add_user_id
 from stream_chat.base.exceptions import StreamChannelException
+from stream_chat.types.scheduled_messages import format_scheduled_at
 from stream_chat.types.stream_response import StreamResponse
 
 
@@ -11,6 +12,8 @@ class Channel(ChannelInterface):
     async def send_message(
         self, message: Dict, user_id: str, **options: Any
     ) -> StreamResponse:
+        if options.get("scheduled_at") is not None:
+            options["scheduled_at"] = format_scheduled_at(options["scheduled_at"])
         payload = {"message": add_user_id(message, user_id), **options}
         return await self.client.post(f"{self.url}/message", data=payload)
 
diff --git a/stream_chat/async_chat/client.py b/stream_chat/async_chat/client.py
index 74e6e4e..547a470 100644
--- a/stream_chat/async_chat/client.py
+++ b/stream_chat/async_chat/client.py
@@ -28,6 +28,11 @@
 from stream_chat.types.campaign import CampaignData, QueryCampaignsOptions
 from stream_chat.types.channel_batch import ChannelsBatchOptions
 from stream_chat.types.draft import QueryDraftsFilter, QueryDraftsOptions
+from stream_chat.types.scheduled_messages import (
+    QueryScheduledMessagesFilter,
+    QueryScheduledMessagesOptions,
+    format_scheduled_at,
+)
 from stream_chat.types.segment import (
     QuerySegmentsOptions,
     QuerySegmentTargetsOptions,
@@ -992,6 +997,47 @@ async def query_reminders(
         params["user_id"] = user_id
         return await self.post("reminders/query", data=params)
 
+    async def update_scheduled_message(
+        self,
+        scheduled_message_id: str,
+        message: Optional[Dict] = None,
+        scheduled_at: Optional[Union[str, datetime.datetime]] = None,
+    ) -> StreamResponse:
+        if message is None and scheduled_at is None:
+            raise ValueError("message or scheduled_at is required")
+        data: Dict[str, Any] = {}
+        if message is not None:
+            data["message"] = message
+        if scheduled_at is not None:
+            data["scheduled_at"] = format_scheduled_at(scheduled_at)
+        return await self.patch(f"scheduled_messages/{scheduled_message_id}", data=data)
+
+    async def delete_scheduled_message(
+        self, scheduled_message_id: str
+    ) -> StreamResponse:
+        return await self.delete(f"scheduled_messages/{scheduled_message_id}")
+
+    async def send_scheduled_message_now(
+        self, scheduled_message_id: str
+    ) -> StreamResponse:
+        return await self.post(f"scheduled_messages/{scheduled_message_id}/send_now")
+
+    async def query_scheduled_messages(
+        self,
+        user_id: str,
+        filter: Optional[QueryScheduledMessagesFilter] = None,
+        sort: Optional[List[SortParam]] = None,
+        options: Optional[QueryScheduledMessagesOptions] = None,
+    ) -> StreamResponse:
+        data: Dict[str, Any] = {"user_id": user_id}
+        if filter is not None:
+            data["filter"] = cast(dict, filter)
+        if sort is not None:
+            data["sort"] = sort
+        if options is not None:
+            data.update(cast(dict, options))
+        return await self.post("scheduled_messages/query", data=data)
+
     async def get_user_locations(self, user_id: str, **options: Any) -> StreamResponse:
         params = {"user_id": user_id, **options}
         return await self.get("users/live_locations", params=params)
diff --git a/stream_chat/base/channel.py b/stream_chat/base/channel.py
index 0ab74d6..1f0c324 100644
--- a/stream_chat/base/channel.py
+++ b/stream_chat/base/channel.py
@@ -40,6 +40,10 @@ def send_message(
 
         :param message: the Message object
         :param user_id: the ID of the user that created the message
+        :param options: additional request fields, e.g. skip_push. Pass
+            scheduled_at (a datetime, naive taken as UTC, or an ISO 8601 string)
+            to store the message and send it at that time instead of now; the
+            response then carries scheduled_message
         :return: the Server Response
         """
         pass
diff --git a/stream_chat/base/client.py b/stream_chat/base/client.py
index 4c17bbb..5e8bd4a 100644
--- a/stream_chat/base/client.py
+++ b/stream_chat/base/client.py
@@ -10,6 +10,10 @@
 from stream_chat.types.base import SortParam
 from stream_chat.types.campaign import CampaignData, QueryCampaignsOptions
 from stream_chat.types.draft import QueryDraftsFilter, QueryDraftsOptions
+from stream_chat.types.scheduled_messages import (
+    QueryScheduledMessagesFilter,
+    QueryScheduledMessagesOptions,
+)
 from stream_chat.types.segment import (
     QuerySegmentsOptions,
     QuerySegmentTargetsOptions,
@@ -1589,6 +1593,69 @@ def query_reminders(
         """
         pass
 
+    @abc.abstractmethod
+    def update_scheduled_message(
+        self,
+        scheduled_message_id: str,
+        message: Optional[Dict] = None,
+        scheduled_at: Optional[Union[str, datetime.datetime]] = None,
+    ) -> Union[StreamResponse, Awaitable[StreamResponse]]:
+        """
+        Replaces the message and/or the send time of a scheduled message.
+        At least one of message and scheduled_at must be set.
+
+        :param scheduled_message_id: The ID of the scheduled message
+        :param message: Replaces the scheduled message as a whole. The sender stays
+            the user who scheduled it
+        :param scheduled_at: The new send time. A naive datetime is taken to be in UTC
+        :return: API response with the scheduled message
+        """
+        pass
+
+    @abc.abstractmethod
+    def delete_scheduled_message(
+        self, scheduled_message_id: str
+    ) -> Union[StreamResponse, Awaitable[StreamResponse]]:
+        """
+        Cancels a scheduled message so it is never sent, or removes a failed one.
+
+        :param scheduled_message_id: The ID of the scheduled message
+        :return: API response with the scheduled message
+        """
+        pass
+
+    @abc.abstractmethod
+    def send_scheduled_message_now(
+        self, scheduled_message_id: str
+    ) -> Union[StreamResponse, Awaitable[StreamResponse]]:
+        """
+        Moves the send time of a scheduled message to now and queues the send.
+
+        :param scheduled_message_id: The ID of the scheduled message
+        :return: API response with the scheduled message
+        """
+        pass
+
+    @abc.abstractmethod
+    def query_scheduled_messages(
+        self,
+        user_id: str,
+        filter: Optional[QueryScheduledMessagesFilter] = None,
+        sort: Optional[List[SortParam]] = None,
+        options: Optional[QueryScheduledMessagesOptions] = None,
+    ) -> Union[StreamResponse, Awaitable[StreamResponse]]:
+        """
+        Lists the scheduled and failed messages of one user across channels:
+        pending ones first, then failed ones, each by scheduled_at.
+
+        :param user_id: The ID of the user whose scheduled messages to list
+        :param filter: Narrows the result by channel (cid) and status
+        :param sort: Only scheduled_at ascending is supported
+        :param options: Pagination: limit (1 to 100, default 20) and next
+        :return: API response with scheduled_messages and next
+        """
+        pass
+
     @abc.abstractmethod
     def get_user_locations(
         self, user_id: str, **options: Any
diff --git a/stream_chat/channel.py b/stream_chat/channel.py
index b3006d9..54e8a12 100644
--- a/stream_chat/channel.py
+++ b/stream_chat/channel.py
@@ -4,6 +4,7 @@
 
 from stream_chat.base.channel import ChannelInterface, add_user_id
 from stream_chat.base.exceptions import StreamChannelException
+from stream_chat.types.scheduled_messages import format_scheduled_at
 from stream_chat.types.stream_response import StreamResponse
 
 
@@ -11,6 +12,8 @@ class Channel(ChannelInterface):
     def send_message(
         self, message: Dict, user_id: str, **options: Any
     ) -> StreamResponse:
+        if options.get("scheduled_at") is not None:
+            options["scheduled_at"] = format_scheduled_at(options["scheduled_at"])
         payload = {"message": add_user_id(message, user_id), **options}
         return self.client.post(f"{self.url}/message", data=payload)
 
diff --git a/stream_chat/client.py b/stream_chat/client.py
index 9d67a20..adafd66 100644
--- a/stream_chat/client.py
+++ b/stream_chat/client.py
@@ -25,6 +25,11 @@
 from stream_chat.types.campaign import CampaignData, QueryCampaignsOptions
 from stream_chat.types.channel_batch import ChannelsBatchOptions
 from stream_chat.types.draft import QueryDraftsFilter, QueryDraftsOptions
+from stream_chat.types.scheduled_messages import (
+    QueryScheduledMessagesFilter,
+    QueryScheduledMessagesOptions,
+    format_scheduled_at,
+)
 from stream_chat.types.segment import (
     QuerySegmentsOptions,
     QuerySegmentTargetsOptions,
@@ -954,6 +959,43 @@ def query_reminders(
         params["user_id"] = user_id
         return self.post("reminders/query", data=params)
 
+    def update_scheduled_message(
+        self,
+        scheduled_message_id: str,
+        message: Optional[Dict] = None,
+        scheduled_at: Optional[Union[str, datetime.datetime]] = None,
+    ) -> StreamResponse:
+        if message is None and scheduled_at is None:
+            raise ValueError("message or scheduled_at is required")
+        data: Dict[str, Any] = {}
+        if message is not None:
+            data["message"] = message
+        if scheduled_at is not None:
+            data["scheduled_at"] = format_scheduled_at(scheduled_at)
+        return self.patch(f"scheduled_messages/{scheduled_message_id}", data=data)
+
+    def delete_scheduled_message(self, scheduled_message_id: str) -> StreamResponse:
+        return self.delete(f"scheduled_messages/{scheduled_message_id}")
+
+    def send_scheduled_message_now(self, scheduled_message_id: str) -> StreamResponse:
+        return self.post(f"scheduled_messages/{scheduled_message_id}/send_now")
+
+    def query_scheduled_messages(
+        self,
+        user_id: str,
+        filter: Optional[QueryScheduledMessagesFilter] = None,
+        sort: Optional[List[SortParam]] = None,
+        options: Optional[QueryScheduledMessagesOptions] = None,
+    ) -> StreamResponse:
+        data: Dict[str, Any] = {"user_id": user_id}
+        if filter is not None:
+            data["filter"] = cast(dict, filter)
+        if sort is not None:
+            data["sort"] = sort
+        if options is not None:
+            data.update(cast(dict, options))
+        return self.post("scheduled_messages/query", data=data)
+
     def get_user_locations(self, user_id: str, **options: Any) -> StreamResponse:
         params = {"user_id": user_id, **options}
         return self.get("users/live_locations", params=params)
diff --git a/stream_chat/tests/async_chat/test_scheduled_messages.py b/stream_chat/tests/async_chat/test_scheduled_messages.py
new file mode 100644
index 0000000..93384a1
--- /dev/null
+++ b/stream_chat/tests/async_chat/test_scheduled_messages.py
@@ -0,0 +1,60 @@
+from datetime import datetime, timedelta, timezone
+from typing import Dict
+
+import pytest
+
+from stream_chat.async_chat import StreamChatAsync
+from stream_chat.base.exceptions import StreamAPIException
+
+
+async def schedule_message(channel, user_id: str) -> Dict:
+    """Schedules a message, skipping the test when the app cannot schedule.
+
+    Scheduled messages are enabled per app by Stream, not through the app
+    settings API, so the CI app cannot be baselined into having them.
+    """
+    try:
+        response = await channel.send_message(
+            {"text": "scheduled hello"},
+            user_id,
+            scheduled_at=datetime.now(timezone.utc) + timedelta(hours=1),
+        )
+    except StreamAPIException as e:
+        if "scheduled_messages_disabled" in e.response_text:
+            pytest.skip("scheduled messages are not enabled for this app")
+        raise
+    if "scheduled_message" not in response:
+        pytest.skip("the API does not support scheduled messages yet")
+    return response["scheduled_message"]
+
+
+class TestScheduledMessages:
+    @pytest.mark.asyncio
+    async def test_scheduled_message_lifecycle(
+        self, client: StreamChatAsync, channel, random_user
+    ):
+        scheduled = await schedule_message(channel, random_user["id"])
+        assert scheduled["cid"] == channel.cid
+        assert scheduled["status"] == "scheduled"
+
+        response = await client.query_scheduled_messages(
+            random_user["id"], filter={"cid": channel.cid}
+        )
+        assert [m["id"] for m in response["scheduled_messages"]] == [scheduled["id"]]
+
+        response = await client.update_scheduled_message(
+            scheduled["id"], message={"text": "edited"}
+        )
+        assert response["scheduled_message"]["message"]["text"] == "edited"
+
+        response = await client.delete_scheduled_message(scheduled["id"])
+        assert response["scheduled_message"]["id"] == scheduled["id"]
+
+    @pytest.mark.asyncio
+    async def test_send_scheduled_message_now(
+        self, client: StreamChatAsync, channel, random_user
+    ):
+        scheduled = await schedule_message(channel, random_user["id"])
+
+        response = await client.send_scheduled_message_now(scheduled["id"])
+        assert response["scheduled_message"]["id"] == scheduled["id"]
diff --git a/stream_chat/tests/test_scheduled_messages.py b/stream_chat/tests/test_scheduled_messages.py
new file mode 100644
index 0000000..1748cc8
--- /dev/null
+++ b/stream_chat/tests/test_scheduled_messages.py
@@ -0,0 +1,66 @@
+from datetime import datetime, timedelta, timezone
+from typing import Dict
+
+import pytest
+
+from stream_chat import StreamChat
+from stream_chat.base.exceptions import StreamAPIException
+
+
+def schedule_message(channel, user_id: str) -> Dict:
+    """Schedules a message, skipping the test when the app cannot schedule.
+
+    Scheduled messages are enabled per app by Stream, not through the app
+    settings API, so the CI app cannot be baselined into having them.
+    """
+    try:
+        response = channel.send_message(
+            {"text": "scheduled hello"},
+            user_id,
+            scheduled_at=datetime.now(timezone.utc) + timedelta(hours=1),
+        )
+    except StreamAPIException as e:
+        if "scheduled_messages_disabled" in e.response_text:
+            pytest.skip("scheduled messages are not enabled for this app")
+        raise
+    if "scheduled_message" not in response:
+        pytest.skip("the API does not support scheduled messages yet")
+    return response["scheduled_message"]
+
+
+class TestScheduledMessages:
+    def test_scheduled_message_lifecycle(
+        self, client: StreamChat, channel, random_user
+    ):
+        scheduled = schedule_message(channel, random_user["id"])
+        assert scheduled["cid"] == channel.cid
+        assert scheduled["user"]["id"] == random_user["id"]
+        assert scheduled["status"] == "scheduled"
+        assert scheduled["message"]["text"] == "scheduled hello"
+
+        response = client.query_scheduled_messages(
+            random_user["id"], filter={"cid": channel.cid}
+        )
+        assert [m["id"] for m in response["scheduled_messages"]] == [scheduled["id"]]
+        assert response["scheduled_messages"][0]["channel"]["cid"] == channel.cid
+
+        response = client.update_scheduled_message(
+            scheduled["id"],
+            message={"text": "edited"},
+            scheduled_at=datetime.now(timezone.utc) + timedelta(hours=2),
+        )
+        assert response["scheduled_message"]["message"]["text"] == "edited"
+
+        response = client.delete_scheduled_message(scheduled["id"])
+        assert response["scheduled_message"]["id"] == scheduled["id"]
+
+        response = client.query_scheduled_messages(
+            random_user["id"], filter={"cid": channel.cid}
+        )
+        assert response["scheduled_messages"] == []
+
+    def test_send_scheduled_message_now(self, client: StreamChat, channel, random_user):
+        scheduled = schedule_message(channel, random_user["id"])
+
+        response = client.send_scheduled_message_now(scheduled["id"])
+        assert response["scheduled_message"]["id"] == scheduled["id"]
diff --git a/stream_chat/tests/test_scheduled_messages_unit.py b/stream_chat/tests/test_scheduled_messages_unit.py
new file mode 100644
index 0000000..26563fa
--- /dev/null
+++ b/stream_chat/tests/test_scheduled_messages_unit.py
@@ -0,0 +1,179 @@
+import datetime
+from unittest.mock import AsyncMock, MagicMock
+
+import pytest
+
+from stream_chat import StreamChat
+from stream_chat.async_chat import StreamChatAsync
+from stream_chat.async_chat.channel import Channel as AsyncChannel
+from stream_chat.channel import Channel
+from stream_chat.types.scheduled_messages import format_scheduled_at
+
+CET = datetime.timezone(datetime.timedelta(hours=1))
+
+
+def _sync_client() -> StreamChat:
+    client = StreamChat(api_key="key", api_secret="secret")
+    client.post = MagicMock()  # type: ignore[method-assign]
+    client.patch = MagicMock()  # type: ignore[method-assign]
+    client.delete = MagicMock()  # type: ignore[method-assign]
+    return client
+
+
+def test_format_scheduled_at() -> None:
+    assert (
+        format_scheduled_at(datetime.datetime(2030, 1, 2, 4, 4, 5, tzinfo=CET))
+        == "2030-01-02T03:04:05+00:00"
+    )
+    assert (
+        format_scheduled_at(datetime.datetime(2030, 1, 2, 3, 4, 5))
+        == "2030-01-02T03:04:05+00:00"
+    )
+    assert format_scheduled_at("2030-01-02T03:04:05Z") == "2030-01-02T03:04:05Z"
+
+
+def test_send_message_with_scheduled_at() -> None:
+    client = MagicMock()
+    Channel(client, "messaging", "chan").send_message(
+        {"text": "later"},
+        "user-1",
+        scheduled_at=datetime.datetime(2030, 1, 2, 4, 4, 5, tzinfo=CET),
+    )
+    client.post.assert_called_once_with(
+        "channels/messaging/chan/message",
+        data={
+            "message": {"text": "later", "user": {"id": "user-1"}},
+            "scheduled_at": "2030-01-02T03:04:05+00:00",
+        },
+    )
+
+
+def test_send_message_without_scheduled_at() -> None:
+    client = MagicMock()
+    Channel(client, "messaging", "chan").send_message({"text": "now"}, "user-1")
+    client.post.assert_called_once_with(
+        "channels/messaging/chan/message",
+        data={"message": {"text": "now", "user": {"id": "user-1"}}},
+    )
+
+
+def test_update_scheduled_message() -> None:
+    client = _sync_client()
+    client.update_scheduled_message(
+        "sm-1",
+        message={"text": "edited"},
+        scheduled_at=datetime.datetime(2030, 1, 2, 3, 4, 5),
+    )
+    client.patch.assert_called_once_with(
+        "scheduled_messages/sm-1",
+        data={
+            "message": {"text": "edited"},
+            "scheduled_at": "2030-01-02T03:04:05+00:00",
+        },
+    )
+
+
+def test_update_scheduled_message_only_time() -> None:
+    client = _sync_client()
+    client.update_scheduled_message("sm-1", scheduled_at="2030-01-02T03:04:05Z")
+    client.patch.assert_called_once_with(
+        "scheduled_messages/sm-1", data={"scheduled_at": "2030-01-02T03:04:05Z"}
+    )
+
+
+def test_update_scheduled_message_requires_a_change() -> None:
+    client = _sync_client()
+    with pytest.raises(ValueError):
+        client.update_scheduled_message("sm-1")
+    client.patch.assert_not_called()
+
+
+def test_delete_scheduled_message() -> None:
+    client = _sync_client()
+    client.delete_scheduled_message("sm-1")
+    client.delete.assert_called_once_with("scheduled_messages/sm-1")
+
+
+def test_send_scheduled_message_now() -> None:
+    client = _sync_client()
+    client.send_scheduled_message_now("sm-1")
+    client.post.assert_called_once_with("scheduled_messages/sm-1/send_now")
+
+
+def test_query_scheduled_messages() -> None:
+    client = _sync_client()
+    client.query_scheduled_messages(
+        "user-1",
+        filter={"cid": "messaging:chan", "status": ["scheduled", "failed"]},
+        sort=[{"field": "scheduled_at", "direction": 1}],
+        options={"limit": 100, "next": "cursor"},
+    )
+    client.post.assert_called_once_with(
+        "scheduled_messages/query",
+        data={
+            "user_id": "user-1",
+            "filter": {"cid": "messaging:chan", "status": ["scheduled", "failed"]},
+            "sort": [{"field": "scheduled_at", "direction": 1}],
+            "limit": 100,
+            "next": "cursor",
+        },
+    )
+
+
+def test_query_scheduled_messages_minimal() -> None:
+    client = _sync_client()
+    client.query_scheduled_messages("user-1")
+    client.post.assert_called_once_with(
+        "scheduled_messages/query", data={"user_id": "user-1"}
+    )
+
+
+@pytest.mark.asyncio
+async def test_async_send_message_with_scheduled_at() -> None:
+    client = MagicMock()
+    client.post = AsyncMock()
+    await AsyncChannel(client, "messaging", "chan").send_message(
+        {"text": "later"}, "user-1", scheduled_at="2030-01-02T03:04:05Z"
+    )
+    client.post.assert_called_once_with(
+        "channels/messaging/chan/message",
+        data={
+            "message": {"text": "later", "user": {"id": "user-1"}},
+            "scheduled_at": "2030-01-02T03:04:05Z",
+        },
+    )
+
+
+@pytest.mark.asyncio
+async def test_async_scheduled_message_calls() -> None:
+    async with StreamChatAsync(api_key="key", api_secret="secret") as client:
+        client.post = AsyncMock()  # type: ignore[method-assign]
+        client.patch = AsyncMock()  # type: ignore[method-assign]
+        client.delete = AsyncMock()  # type: ignore[method-assign]
+
+        await client.update_scheduled_message("sm-1", message={"text": "edited"})
+        client.patch.assert_called_once_with(
+            "scheduled_messages/sm-1", data={"message": {"text": "edited"}}
+        )
+
+        await client.delete_scheduled_message("sm-1")
+        client.delete.assert_called_once_with("scheduled_messages/sm-1")
+
+        await client.send_scheduled_message_now("sm-1")
+        client.post.assert_called_once_with("scheduled_messages/sm-1/send_now")
+
+        client.post.reset_mock()
+        await client.query_scheduled_messages(
+            "user-1", filter={"cid": "messaging:chan"}, options={"limit": 10}
+        )
+        client.post.assert_called_once_with(
+            "scheduled_messages/query",
+            data={
+                "user_id": "user-1",
+                "filter": {"cid": "messaging:chan"},
+                "limit": 10,
+            },
+        )
+
+        with pytest.raises(ValueError):
+            await client.update_scheduled_message("sm-1")
diff --git a/stream_chat/types/scheduled_messages.py b/stream_chat/types/scheduled_messages.py
new file mode 100644
index 0000000..6e81311
--- /dev/null
+++ b/stream_chat/types/scheduled_messages.py
@@ -0,0 +1,52 @@
+import datetime
+import sys
+from typing import List, Union
+
+if sys.version_info >= (3, 8):
+    from typing import Literal, TypedDict
+else:
+    from typing_extensions import Literal, TypedDict
+
+
+ScheduledMessageStatus = Literal["scheduled", "sending", "failed", "canceled", "sent"]
+"""
+Status of a scheduled message. ``canceled`` is reported when a scheduled message
+is deleted, ``sent`` only by the ``scheduled_message.sent`` event.
+"""
+
+
+class QueryScheduledMessagesFilter(TypedDict, total=False):
+    """
+    Narrows a scheduled messages query. Empty fields mean no restriction.
+
+    Parameters:
+        cid: Only the scheduled messages of this channel.
+        status: Only these statuses: ``scheduled``, ``sending`` and/or ``failed``.
+    """
+
+    cid: str
+    status: List[ScheduledMessageStatus]
+
+
+class QueryScheduledMessagesOptions(TypedDict, total=False):
+    """
+    Pagination of a scheduled messages query.
+
+    Parameters:
+        limit: Page size, 1 to 100. Defaults to 20.
+        next: The ``next`` value of a previous response, to continue from that point.
+    """
+
+    limit: int
+    next: str
+
+
+def format_scheduled_at(scheduled_at: Union[str, datetime.datetime]) -> str:
+    """
+    Formats a send time for the API. A naive datetime is taken to be in UTC.
+    """
+    if isinstance(scheduled_at, datetime.datetime):
+        if scheduled_at.tzinfo is None:
+            scheduled_at = scheduled_at.replace(tzinfo=datetime.timezone.utc)
+        return scheduled_at.astimezone(datetime.timezone.utc).isoformat()
+    return scheduled_at