diff --git a/pyrightconfig.stricter.json b/pyrightconfig.stricter.json
index 4242a2ba3280..c518c2ad2b02 100644
--- a/pyrightconfig.stricter.json
+++ b/pyrightconfig.stricter.json
@@ -33,6 +33,7 @@
         "stubs/braintree",
         "stubs/caldav",
         "stubs/cffi",
+        "stubs/channels",
         "stubs/click-web",
         "stubs/corus",
         "stubs/dateparser",
diff --git a/stubs/channels/METADATA.toml b/stubs/channels/METADATA.toml
new file mode 100644
index 000000000000..cf12e31aa2eb
--- /dev/null
+++ b/stubs/channels/METADATA.toml
@@ -0,0 +1,7 @@
+version = "4.*"
+upstream_repository = "https://github.com/django/channels"
+requires = ["django-stubs>=4.2,<5.3", "asgiref"]
+
+[tool.stubtest]
+skip = true  # TODO: enable stubtest once Django mypy plugin config is supported
+stubtest_requirements = ["daphne"]
diff --git a/stubs/channels/channels/__init__.pyi b/stubs/channels/channels/__init__.pyi
new file mode 100644
index 000000000000..ae22f453163e
--- /dev/null
+++ b/stubs/channels/channels/__init__.pyi
@@ -0,0 +1,2 @@
+__version__: str
+DEFAULT_CHANNEL_LAYER: str
diff --git a/stubs/channels/channels/apps.pyi b/stubs/channels/channels/apps.pyi
new file mode 100644
index 000000000000..72007d91886c
--- /dev/null
+++ b/stubs/channels/channels/apps.pyi
@@ -0,0 +1,5 @@
+from django.apps import AppConfig
+
+class ChannelsConfig(AppConfig):
+    name: str = ...
+    verbose_name: str = ...
diff --git a/stubs/channels/channels/auth.pyi b/stubs/channels/channels/auth.pyi
new file mode 100644
index 000000000000..5f3b7b5e20fe
--- /dev/null
+++ b/stubs/channels/channels/auth.pyi
@@ -0,0 +1,32 @@
+from typing import Any
+
+from asgiref.typing import ASGIReceiveCallable, ASGISendCallable
+from channels.middleware import BaseMiddleware
+from django.contrib.auth.backends import BaseBackend
+from django.contrib.auth.base_user import AbstractBaseUser
+from django.contrib.auth.models import AnonymousUser
+from django.utils.functional import LazyObject
+
+from .consumer import _ChannelScope, _LazySession
+from .db import database_sync_to_async
+from .utils import _ChannelApplication
+
+@database_sync_to_async
+def get_user(scope: _ChannelScope) -> AbstractBaseUser | AnonymousUser: ...
+@database_sync_to_async
+def login(scope: _ChannelScope, user: AbstractBaseUser, backend: BaseBackend | None = ...) -> None: ...
+@database_sync_to_async
+def logout(scope: _ChannelScope) -> None: ...
+def _get_user_session_key(session: _LazySession) -> Any: ...
+
+class UserLazyObject(AbstractBaseUser, LazyObject):
+    def _setup(self) -> None: ...
+
+class AuthMiddleware(BaseMiddleware):
+    def populate_scope(self, scope: _ChannelScope) -> None: ...
+    async def resolve_scope(self, scope: _ChannelScope) -> None: ...
+    async def __call__(
+        self, scope: _ChannelScope, receive: ASGIReceiveCallable, send: ASGISendCallable
+    ) -> _ChannelApplication: ...
+
+def AuthMiddlewareStack(inner: _ChannelApplication) -> _ChannelApplication: ...
diff --git a/stubs/channels/channels/consumer.pyi b/stubs/channels/channels/consumer.pyi
new file mode 100644
index 000000000000..d975b70735e1
--- /dev/null
+++ b/stubs/channels/channels/consumer.pyi
@@ -0,0 +1,57 @@
+from collections.abc import Awaitable
+from typing import Any, ClassVar, Protocol, type_check_only
+
+from asgiref.typing import ASGIReceiveCallable, ASGISendCallable, Scope, WebSocketScope
+from channels.auth import UserLazyObject
+from channels.db import database_sync_to_async
+from channels.layers import BaseChannelLayer
+from django.contrib.sessions.backends.base import SessionBase
+from django.utils.functional import LazyObject
+
+@type_check_only
+class _LazySession(SessionBase, LazyObject):  # type: ignore[misc]
+    _wrapped: SessionBase
+
+# Base ASGI Scope definition
+@type_check_only
+class _ChannelScope(WebSocketScope, total=False):
+    # Channel specific
+    channel: str
+    url_route: dict[str, Any]
+    path_remaining: str
+
+    # Auth specific
+    cookies: dict[str, str]
+    session: _LazySession
+    user: UserLazyObject | None
+
+def get_handler_name(message: dict[str, Any]) -> str: ...
+@type_check_only
+class _ASGIApplicationProtocol(Protocol):
+    consumer_class: Any
+    consumer_initkwargs: dict[str, Any]
+
+    def __call__(self, scope: Scope, receive: ASGIReceiveCallable, send: ASGISendCallable) -> Awaitable[None]: ...
+
+class AsyncConsumer:
+    _sync: ClassVar[bool] = ...
+    channel_layer_alias: ClassVar[str] = ...
+
+    scope: _ChannelScope
+    channel_layer: BaseChannelLayer
+    channel_name: str
+    channel_receive: ASGIReceiveCallable
+    base_send: ASGISendCallable
+
+    async def __call__(self, scope: _ChannelScope, receive: ASGIReceiveCallable, send: ASGISendCallable) -> None: ...
+    async def dispatch(self, message: dict[str, Any]) -> None: ...
+    async def send(self, message: dict[str, Any]) -> None: ...
+    @classmethod
+    def as_asgi(cls, **initkwargs: Any) -> _ASGIApplicationProtocol: ...
+
+class SyncConsumer(AsyncConsumer):
+    _sync: ClassVar[bool] = ...
+
+    @database_sync_to_async
+    def dispatch(self, message: dict[str, Any]) -> None: ...  # type: ignore[override]
+    def send(self, message: dict[str, Any]) -> None: ...  # type: ignore[override]
diff --git a/stubs/channels/channels/db.pyi b/stubs/channels/channels/db.pyi
new file mode 100644
index 000000000000..98c68d1e21db
--- /dev/null
+++ b/stubs/channels/channels/db.pyi
@@ -0,0 +1,15 @@
+from asyncio import BaseEventLoop
+from collections.abc import Callable, Coroutine
+from typing import Any, TypeVar
+from typing_extensions import ParamSpec
+
+from asgiref.sync import SyncToAsync
+
+_P = ParamSpec("_P")
+_R = TypeVar("_R")
+
+class DatabaseSyncToAsync(SyncToAsync[_P, _R]):
+    def thread_handler(self, loop: BaseEventLoop, *args: Any, **kwargs: Any) -> Any: ...
+
+def database_sync_to_async(func: Callable[_P, _R]) -> Callable[_P, Coroutine[Any, Any, _R]]: ...
+async def aclose_old_connections() -> None: ...
diff --git a/stubs/channels/channels/exceptions.pyi b/stubs/channels/channels/exceptions.pyi
new file mode 100644
index 000000000000..eaba1dfaee14
--- /dev/null
+++ b/stubs/channels/channels/exceptions.pyi
@@ -0,0 +1,8 @@
+class RequestAborted(Exception): ...
+class RequestTimeout(RequestAborted): ...
+class InvalidChannelLayerError(ValueError): ...
+class AcceptConnection(Exception): ...
+class DenyConnection(Exception): ...
+class ChannelFull(Exception): ...
+class MessageTooLarge(Exception): ...
+class StopConsumer(Exception): ...
diff --git a/stubs/channels/channels/generic/__init__.pyi b/stubs/channels/channels/generic/__init__.pyi
new file mode 100644
index 000000000000..e69de29bb2d1
diff --git a/stubs/channels/channels/generic/http.pyi b/stubs/channels/channels/generic/http.pyi
new file mode 100644
index 000000000000..85d83a7fd208
--- /dev/null
+++ b/stubs/channels/channels/generic/http.pyi
@@ -0,0 +1,19 @@
+from collections.abc import Iterable
+from typing import Any
+
+from asgiref.typing import HTTPDisconnectEvent, HTTPRequestEvent, HTTPScope
+from channels.consumer import AsyncConsumer
+
+class AsyncHttpConsumer(AsyncConsumer):
+    body: list[bytes]
+    scope: HTTPScope  # type: ignore[assignment]
+
+    def __init__(self, *args: Any, **kwargs: Any) -> None: ...
+    async def send_headers(self, *, status: int = ..., headers: Iterable[tuple[bytes, bytes]] | None = ...) -> None: ...
+    async def send_body(self, body: bytes, *, more_body: bool = ...) -> None: ...
+    async def send_response(self, status: int, body: bytes, **kwargs: Any) -> None: ...
+    async def handle(self, body: bytes) -> None: ...
+    async def disconnect(self) -> None: ...
+    async def http_request(self, message: HTTPRequestEvent) -> None: ...
+    async def http_disconnect(self, message: HTTPDisconnectEvent) -> None: ...
+    async def send(self, message: dict[str, Any]) -> None: ...  # type: ignore[override]
diff --git a/stubs/channels/channels/generic/websocket.pyi b/stubs/channels/channels/generic/websocket.pyi
new file mode 100644
index 000000000000..c380d0c14dc8
--- /dev/null
+++ b/stubs/channels/channels/generic/websocket.pyi
@@ -0,0 +1,65 @@
+from typing import Any
+
+from asgiref.typing import WebSocketConnectEvent, WebSocketDisconnectEvent, WebSocketReceiveEvent
+from channels.consumer import AsyncConsumer, SyncConsumer, _ChannelScope
+from channels.layers import BaseChannelLayer
+
+class WebsocketConsumer(SyncConsumer):
+    groups: list[str] | None
+    scope: _ChannelScope
+    channel_name: str
+    channel_layer: BaseChannelLayer
+    channel_receive: Any
+    base_send: Any
+
+    def __init__(self, *args: Any, **kwargs: Any) -> None: ...
+    def websocket_connect(self, message: WebSocketConnectEvent) -> None: ...
+    def connect(self) -> None: ...
+    def accept(self, subprotocol: str | None = ..., headers: list[tuple[str, str]] | None = ...) -> None: ...
+    def websocket_receive(self, message: WebSocketReceiveEvent) -> None: ...
+    def receive(self, text_data: str | None = ..., bytes_data: bytes | None = ...) -> None: ...
+    def send(  # type: ignore[override]
+        self, text_data: str | None = ..., bytes_data: bytes | None = ..., close: bool = ...
+    ) -> None: ...
+    def close(self, code: int | bool | None = ..., reason: str | None = ...) -> None: ...
+    def websocket_disconnect(self, message: WebSocketDisconnectEvent) -> None: ...
+    def disconnect(self, code: int) -> None: ...
+
+class JsonWebsocketConsumer(WebsocketConsumer):
+    def receive(self, text_data: str | None = ..., bytes_data: bytes | None = ..., **kwargs: Any) -> None: ...
+    def receive_json(self, content: Any, **kwargs: Any) -> None: ...
+    def send_json(self, content: Any, close: bool = ...) -> None: ...
+    @classmethod
+    def decode_json(cls, text_data: str) -> Any: ...
+    @classmethod
+    def encode_json(cls, content: Any) -> str: ...
+
+class AsyncWebsocketConsumer(AsyncConsumer):
+    groups: list[str] | None
+    scope: _ChannelScope
+    channel_name: str
+    channel_layer: BaseChannelLayer
+    channel_receive: Any
+    base_send: Any
+
+    def __init__(self, *args: Any, **kwargs: Any) -> None: ...
+    async def websocket_connect(self, message: WebSocketConnectEvent) -> None: ...
+    async def connect(self) -> None: ...
+    async def accept(self, subprotocol: str | None = ..., headers: list[tuple[str, str]] | None = ...) -> None: ...
+    async def websocket_receive(self, message: WebSocketReceiveEvent) -> None: ...
+    async def receive(self, text_data: str | None = ..., bytes_data: bytes | None = ...) -> None: ...
+    async def send(  # type: ignore[override]
+        self, text_data: str | None = ..., bytes_data: bytes | None = ..., close: bool = ...
+    ) -> None: ...
+    async def close(self, code: int | bool | None = ..., reason: str | None = ...) -> None: ...
+    async def websocket_disconnect(self, message: WebSocketDisconnectEvent) -> None: ...
+    async def disconnect(self, code: int) -> None: ...
+
+class AsyncJsonWebsocketConsumer(AsyncWebsocketConsumer):
+    async def receive(self, text_data: str | None = ..., bytes_data: bytes | None = ..., **kwargs: Any) -> None: ...
+    async def receive_json(self, content: Any, **kwargs: Any) -> None: ...
+    async def send_json(self, content: Any, close: bool = ...) -> None: ...
+    @classmethod
+    async def decode_json(cls, text_data: str) -> Any: ...
+    @classmethod
+    async def encode_json(cls, content: Any) -> str: ...
diff --git a/stubs/channels/channels/layers.pyi b/stubs/channels/channels/layers.pyi
new file mode 100644
index 000000000000..022e03f38d96
--- /dev/null
+++ b/stubs/channels/channels/layers.pyi
@@ -0,0 +1,96 @@
+import asyncio
+from re import Pattern
+from typing import Any, overload
+from typing_extensions import TypeAlias, deprecated
+
+class ChannelLayerManager:
+    backends: dict[str, BaseChannelLayer]
+
+    def __init__(self) -> None: ...
+    def _reset_backends(self, setting: str, **kwargs: Any) -> None: ...
+    @property
+    def configs(self) -> dict[str, Any]: ...
+    def make_backend(self, name: str) -> BaseChannelLayer: ...
+    def make_test_backend(self, name: str) -> Any: ...
+    def _make_backend(self, name: str, config: dict[str, Any]) -> BaseChannelLayer: ...
+    def __getitem__(self, key: str) -> BaseChannelLayer: ...
+    def __contains__(self, key: str) -> bool: ...
+    def set(self, key: str, layer: BaseChannelLayer) -> BaseChannelLayer | None: ...
+
+_ChannelCapacityPattern: TypeAlias = Pattern[str] | str
+_ChannelCapacityDict: TypeAlias = dict[_ChannelCapacityPattern, int]
+_CompiledChannelCapacity: TypeAlias = list[tuple[Pattern[str], int]]
+
+class BaseChannelLayer:
+    MAX_NAME_LENGTH: int = ...
+    expiry: int
+    capacity: int
+    channel_capacity: _ChannelCapacityDict
+    channel_name_regex: Pattern[str]
+    group_name_regex: Pattern[str]
+    invalid_name_error: str
+
+    def __init__(self, expiry: int = ..., capacity: int = ..., channel_capacity: _ChannelCapacityDict | None = ...) -> None: ...
+    def compile_capacities(self, channel_capacity: _ChannelCapacityDict) -> _CompiledChannelCapacity: ...
+    def get_capacity(self, channel: str) -> int: ...
+    @overload
+    def match_type_and_length(self, name: str) -> bool: ...
+    @overload
+    def match_type_and_length(self, name: Any) -> bool: ...
+    @overload
+    def require_valid_channel_name(self, name: str, receive: bool = ...) -> bool: ...
+    @overload
+    def require_valid_channel_name(self, name: Any, receive: bool = ...) -> bool: ...
+    @overload
+    def require_valid_group_name(self, name: str) -> bool: ...
+    @overload
+    def require_valid_group_name(self, name: Any) -> bool: ...
+    @overload
+    def valid_channel_names(self, names: list[str], receive: bool = ...) -> bool: ...
+    @overload
+    def valid_channel_names(self, names: list[Any], receive: bool = ...) -> bool: ...
+    def non_local_name(self, name: str) -> str: ...
+    async def send(self, channel: str, message: dict[str, Any]) -> None: ...
+    async def receive(self, channel: str) -> dict[str, Any]: ...
+    async def new_channel(self) -> str: ...
+    async def flush(self) -> None: ...
+    async def group_add(self, group: str, channel: str) -> None: ...
+    async def group_discard(self, group: str, channel: str) -> None: ...
+    async def group_send(self, group: str, message: dict[str, Any]) -> None: ...
+    @deprecated("Use require_valid_channel_name instead.")
+    def valid_channel_name(self, channel_name: str, receive: bool = ...) -> bool: ...
+    @deprecated("Use require_valid_group_name instead.")
+    def valid_group_name(self, group_name: str) -> bool: ...
+
+_InMemoryQueueData: TypeAlias = tuple[float, dict[str, Any]]
+
+class InMemoryChannelLayer(BaseChannelLayer):
+    channels: dict[str, asyncio.Queue[_InMemoryQueueData]]
+    groups: dict[str, dict[str, float]]
+    group_expiry: int
+
+    def __init__(
+        self,
+        expiry: int = ...,
+        group_expiry: int = ...,
+        capacity: int = ...,
+        channel_capacity: _ChannelCapacityDict | None = ...,
+        **kwargs: Any,
+    ) -> None: ...
+
+    extensions: list[str]
+
+    async def send(self, channel: str, message: dict[str, Any]) -> None: ...
+    async def receive(self, channel: str) -> dict[str, Any]: ...
+    async def new_channel(self, prefix: str = ...) -> str: ...
+    def _clean_expired(self) -> None: ...
+    async def flush(self) -> None: ...
+    async def close(self) -> None: ...
+    def _remove_from_groups(self, channel: str) -> None: ...
+    async def group_add(self, group: str, channel: str) -> None: ...
+    async def group_discard(self, group: str, channel: str) -> None: ...
+    async def group_send(self, group: str, message: dict[str, Any]) -> None: ...
+
+def get_channel_layer(alias: str = ...) -> BaseChannelLayer | None: ...
+
+channel_layers: ChannelLayerManager
diff --git a/stubs/channels/channels/management/__init__.pyi b/stubs/channels/channels/management/__init__.pyi
new file mode 100644
index 000000000000..e69de29bb2d1
diff --git a/stubs/channels/channels/management/commands/__init__.pyi b/stubs/channels/channels/management/commands/__init__.pyi
new file mode 100644
index 000000000000..e69de29bb2d1
diff --git a/stubs/channels/channels/management/commands/runworker.pyi b/stubs/channels/channels/management/commands/runworker.pyi
new file mode 100644
index 000000000000..1ea4a1e6d4f8
--- /dev/null
+++ b/stubs/channels/channels/management/commands/runworker.pyi
@@ -0,0 +1,20 @@
+import logging
+from argparse import ArgumentParser
+from typing import Any
+
+from channels.layers import BaseChannelLayer
+from channels.worker import Worker
+from django.core.management.base import BaseCommand
+
+logger: logging.Logger
+
+class Command(BaseCommand):
+    leave_locale_alone: bool = ...
+    worker_class: type[Worker] = ...
+    verbosity: int
+    channel_layer: BaseChannelLayer
+
+    def add_arguments(self, parser: ArgumentParser) -> None: ...
+    def handle(
+        self, *args: Any, application_path: str | None = ..., channels: list[str] | None = ..., layer: str = ..., **options: Any
+    ) -> None: ...
diff --git a/stubs/channels/channels/middleware.pyi b/stubs/channels/channels/middleware.pyi
new file mode 100644
index 000000000000..339ae9218244
--- /dev/null
+++ b/stubs/channels/channels/middleware.pyi
@@ -0,0 +1,12 @@
+from asgiref.typing import ASGIReceiveCallable, ASGISendCallable
+
+from .consumer import _ChannelScope
+from .utils import _ChannelApplication
+
+class BaseMiddleware:
+    inner: _ChannelApplication
+
+    def __init__(self, inner: _ChannelApplication) -> None: ...
+    async def __call__(
+        self, scope: _ChannelScope, receive: ASGIReceiveCallable, send: ASGISendCallable
+    ) -> _ChannelApplication: ...
diff --git a/stubs/channels/channels/routing.pyi b/stubs/channels/channels/routing.pyi
new file mode 100644
index 000000000000..94d025e24985
--- /dev/null
+++ b/stubs/channels/channels/routing.pyi
@@ -0,0 +1,32 @@
+from typing import Any, type_check_only
+
+from asgiref.typing import ASGIReceiveCallable, ASGISendCallable
+from django.urls.resolvers import URLPattern
+
+from .consumer import _ASGIApplicationProtocol, _ChannelScope
+from .utils import _ChannelApplication
+
+def get_default_application() -> ProtocolTypeRouter: ...
+
+class ProtocolTypeRouter:
+    application_mapping: dict[str, _ChannelApplication]
+
+    def __init__(self, application_mapping: dict[str, Any]) -> None: ...
+    async def __call__(self, scope: _ChannelScope, receive: ASGIReceiveCallable, send: ASGISendCallable) -> None: ...
+
+@type_check_only
+class _ExtendedURLPattern(URLPattern):
+    callback: _ASGIApplicationProtocol | URLRouter
+
+class URLRouter:
+    _path_routing: bool = ...
+    routes: list[_ExtendedURLPattern | URLRouter]
+
+    def __init__(self, routes: list[_ExtendedURLPattern | URLRouter]) -> None: ...
+    async def __call__(self, scope: _ChannelScope, receive: ASGIReceiveCallable, send: ASGISendCallable) -> None: ...
+
+class ChannelNameRouter:
+    application_mapping: dict[str, _ChannelApplication]
+
+    def __init__(self, application_mapping: dict[str, _ChannelApplication]) -> None: ...
+    async def __call__(self, scope: _ChannelScope, receive: ASGIReceiveCallable, send: ASGISendCallable) -> None: ...
diff --git a/stubs/channels/channels/security/__init__.pyi b/stubs/channels/channels/security/__init__.pyi
new file mode 100644
index 000000000000..e69de29bb2d1
diff --git a/stubs/channels/channels/security/websocket.pyi b/stubs/channels/channels/security/websocket.pyi
new file mode 100644
index 000000000000..aa76a1b4012e
--- /dev/null
+++ b/stubs/channels/channels/security/websocket.pyi
@@ -0,0 +1,25 @@
+from collections.abc import Iterable
+from re import Pattern
+from typing import Any
+from urllib.parse import ParseResult
+
+from asgiref.typing import ASGIReceiveCallable, ASGISendCallable
+from channels.consumer import _ChannelScope
+from channels.generic.websocket import AsyncWebsocketConsumer
+from channels.utils import _ChannelApplication
+
+class OriginValidator:
+    application: _ChannelApplication
+    allowed_origins: Iterable[str | Pattern[str]]
+
+    def __init__(self, application: _ChannelApplication, allowed_origins: Iterable[str | Pattern[str]]) -> None: ...
+    async def __call__(self, scope: _ChannelScope, receive: ASGIReceiveCallable, send: ASGISendCallable) -> Any: ...
+    def valid_origin(self, parsed_origin: ParseResult | None) -> bool: ...
+    def validate_origin(self, parsed_origin: ParseResult | None) -> bool: ...
+    def match_allowed_origin(self, parsed_origin: ParseResult | None, pattern: str | Pattern[str]) -> bool: ...
+    def get_origin_port(self, origin: ParseResult | None) -> int | None: ...
+
+def AllowedHostsOriginValidator(application: _ChannelApplication) -> OriginValidator: ...
+
+class WebsocketDenier(AsyncWebsocketConsumer):
+    async def connect(self) -> None: ...
diff --git a/stubs/channels/channels/sessions.pyi b/stubs/channels/channels/sessions.pyi
new file mode 100644
index 000000000000..85b73adadf9f
--- /dev/null
+++ b/stubs/channels/channels/sessions.pyi
@@ -0,0 +1,53 @@
+import datetime
+from collections.abc import Awaitable
+from typing import Any
+
+from asgiref.typing import ASGIReceiveCallable, ASGISendCallable
+from channels.consumer import _ChannelScope
+from channels.utils import _ChannelApplication
+
+class CookieMiddleware:
+    inner: _ChannelApplication
+
+    def __init__(self, inner: _ChannelApplication) -> None: ...
+    async def __call__(self, scope: _ChannelScope, receive: ASGIReceiveCallable, send: ASGISendCallable) -> Any: ...
+    @classmethod
+    def set_cookie(
+        cls,
+        message: dict[str, Any],
+        key: str,
+        value: str = "",
+        max_age: int | None = ...,
+        expires: str | datetime.datetime | None = ...,
+        path: str = ...,
+        domain: str | None = ...,
+        secure: bool = ...,
+        httponly: bool = ...,
+        samesite: str = ...,
+    ) -> None: ...
+    @classmethod
+    def delete_cookie(cls, message: dict[str, Any], key: str, path: str = ..., domain: str | None = ...) -> None: ...
+
+class InstanceSessionWrapper:
+    save_message_types: list[str]
+    cookie_response_message_types: list[str]
+    cookie_name: str
+    session_store: Any
+    scope: _ChannelScope
+    activated: bool
+    real_send: ASGISendCallable
+
+    def __init__(self, scope: _ChannelScope, send: ASGISendCallable) -> None: ...
+    async def resolve_session(self) -> None: ...
+    async def send(self, message: dict[str, Any]) -> Awaitable[None]: ...
+    async def save_session(self) -> None: ...
+
+class SessionMiddleware:
+    inner: _ChannelApplication
+
+    def __init__(self, inner: _ChannelApplication) -> None: ...
+    async def __call__(
+        self, scope: _ChannelScope, receive: ASGIReceiveCallable, send: ASGISendCallable
+    ) -> _ChannelApplication: ...
+
+def SessionMiddlewareStack(inner: _ChannelApplication) -> _ChannelApplication: ...
diff --git a/stubs/channels/channels/testing/__init__.pyi b/stubs/channels/channels/testing/__init__.pyi
new file mode 100644
index 000000000000..1cb75a0bf6dd
--- /dev/null
+++ b/stubs/channels/channels/testing/__init__.pyi
@@ -0,0 +1,6 @@
+from .application import ApplicationCommunicator
+from .http import HttpCommunicator
+from .live import ChannelsLiveServerTestCase
+from .websocket import WebsocketCommunicator
+
+__all__ = ["ApplicationCommunicator", "HttpCommunicator", "ChannelsLiveServerTestCase", "WebsocketCommunicator"]
diff --git a/stubs/channels/channels/testing/application.pyi b/stubs/channels/channels/testing/application.pyi
new file mode 100644
index 000000000000..313d10cc52d6
--- /dev/null
+++ b/stubs/channels/channels/testing/application.pyi
@@ -0,0 +1,12 @@
+from typing import Any
+
+from asgiref.testing import ApplicationCommunicator as BaseApplicationCommunicator
+
+def no_op() -> None: ...
+
+class ApplicationCommunicator(BaseApplicationCommunicator):
+    async def send_input(self, message: dict[str, Any]) -> None: ...
+    async def receive_output(self, timeout: float = ...) -> dict[str, Any]: ...
+    async def receive_nothing(self, timeout: float = ..., interval: float = ...) -> bool: ...
+    async def wait(self, timeout: float = ...) -> None: ...
+    def stop(self, exceptions: bool = ...) -> None: ...
diff --git a/stubs/channels/channels/testing/http.pyi b/stubs/channels/channels/testing/http.pyi
new file mode 100644
index 000000000000..0e72190f3198
--- /dev/null
+++ b/stubs/channels/channels/testing/http.pyi
@@ -0,0 +1,41 @@
+from collections.abc import Iterable
+from typing import Literal, TypedDict, type_check_only
+
+from channels.testing.application import ApplicationCommunicator
+from channels.utils import _ChannelApplication
+
+# HTTP test-specific response type
+@type_check_only
+class _HTTPTestResponse(TypedDict, total=False):
+    status: int
+    headers: Iterable[tuple[bytes, bytes]]
+    body: bytes
+
+@type_check_only
+class _HTTPTestScope(TypedDict, total=False):
+    type: Literal["http"]
+    http_version: str
+    method: str
+    scheme: str
+    path: str
+    raw_path: bytes
+    query_string: bytes
+    root_path: str
+    headers: Iterable[tuple[bytes, bytes]] | None
+    client: tuple[str, int] | None
+    server: tuple[str, int | None] | None
+
+class HttpCommunicator(ApplicationCommunicator):
+    scope: _HTTPTestScope
+    body: bytes
+    sent_request: bool
+
+    def __init__(
+        self,
+        application: _ChannelApplication,
+        method: str,
+        path: str,
+        body: bytes = ...,
+        headers: Iterable[tuple[bytes, bytes]] | None = ...,
+    ) -> None: ...
+    async def get_response(self, timeout: float = ...) -> _HTTPTestResponse: ...
diff --git a/stubs/channels/channels/testing/live.pyi b/stubs/channels/channels/testing/live.pyi
new file mode 100644
index 000000000000..16322bbe59f4
--- /dev/null
+++ b/stubs/channels/channels/testing/live.pyi
@@ -0,0 +1,35 @@
+from collections.abc import Callable
+from typing import Any, ClassVar
+from typing_extensions import TypeAlias
+
+from channels.routing import ProtocolTypeRouter
+from channels.utils import _ChannelApplication
+from django.contrib.staticfiles.handlers import ASGIStaticFilesHandler
+from django.db.backends.base.base import BaseDatabaseWrapper
+from django.db.backends.sqlite3.base import DatabaseWrapper
+from django.test.testcases import TransactionTestCase
+from django.test.utils import modify_settings
+
+DaphneProcess: TypeAlias = Any  # TODO: temporary hack for daphne.testing.DaphneProcess; remove once daphne provides types
+
+_StaticWrapper: TypeAlias = Callable[[ProtocolTypeRouter], _ChannelApplication]
+
+def make_application(*, static_wrapper: _StaticWrapper | None) -> Any: ...
+
+class ChannelsLiveServerTestCase(TransactionTestCase):
+    host: ClassVar[str] = ...
+    ProtocolServerProcess: ClassVar[type[DaphneProcess]] = ...
+    static_wrapper: ClassVar[type[ASGIStaticFilesHandler]] = ...
+    serve_static: ClassVar[bool] = ...
+
+    _port: int
+    _server_process: DaphneProcess
+    _live_server_modified_settings: modify_settings
+
+    @property
+    def live_server_url(self) -> str: ...
+    @property
+    def live_server_ws_url(self) -> str: ...
+    def _pre_setup(self) -> None: ...
+    def _post_teardown(self) -> None: ...
+    def _is_in_memory_db(self, connection: BaseDatabaseWrapper | DatabaseWrapper) -> bool: ...
diff --git a/stubs/channels/channels/testing/websocket.pyi b/stubs/channels/channels/testing/websocket.pyi
new file mode 100644
index 000000000000..c181d57f8b66
--- /dev/null
+++ b/stubs/channels/channels/testing/websocket.pyi
@@ -0,0 +1,51 @@
+from collections.abc import Iterable
+from typing import Any, Literal, TypedDict, overload, type_check_only
+from typing_extensions import NotRequired, TypeAlias
+
+from asgiref.typing import ASGIVersions
+from channels.testing.application import ApplicationCommunicator
+from channels.utils import _ChannelApplication
+
+@type_check_only
+class _WebsocketTestScope(TypedDict, total=False):
+    spec_version: int
+    type: Literal["websocket"]
+    asgi: ASGIVersions
+    http_version: str
+    scheme: str
+    path: str
+    raw_path: bytes
+    query_string: bytes
+    root_path: str
+    headers: Iterable[tuple[bytes, bytes]] | None
+    client: tuple[str, int] | None
+    server: tuple[str, int | None] | None
+    subprotocols: Iterable[str] | None
+    state: NotRequired[dict[str, Any]]
+    extensions: dict[str, dict[object, object]] | None
+
+_Connected: TypeAlias = bool
+_CloseCodeOrAcceptSubProtocol: TypeAlias = int | str | None
+_WebsocketConnectResponse: TypeAlias = tuple[_Connected, _CloseCodeOrAcceptSubProtocol]
+
+class WebsocketCommunicator(ApplicationCommunicator):
+    scope: _WebsocketTestScope
+    response_headers: list[tuple[bytes, bytes]] | None
+
+    def __init__(
+        self,
+        application: _ChannelApplication,
+        path: str,
+        headers: Iterable[tuple[bytes, bytes]] | None = ...,
+        subprotocols: Iterable[str] | None = ...,
+        spec_version: int | None = ...,
+    ) -> None: ...
+    async def connect(self, timeout: float = ...) -> _WebsocketConnectResponse: ...
+    async def send_to(self, text_data: str | None = ..., bytes_data: bytes | None = ...) -> None: ...
+    @overload
+    async def send_json_to(self, data: dict[str, Any]) -> None: ...
+    @overload
+    async def send_json_to(self, data: Any) -> None: ...
+    async def receive_from(self, timeout: float = ...) -> str | bytes: ...
+    async def receive_json_from(self, timeout: float = ...) -> dict[str, Any]: ...
+    async def disconnect(self, code: int = ..., timeout: float = ...) -> None: ...
diff --git a/stubs/channels/channels/utils.pyi b/stubs/channels/channels/utils.pyi
new file mode 100644
index 000000000000..b92c892b04e3
--- /dev/null
+++ b/stubs/channels/channels/utils.pyi
@@ -0,0 +1,16 @@
+from collections.abc import Awaitable, Callable
+from typing import Any, Protocol, type_check_only
+from typing_extensions import TypeAlias
+
+from asgiref.typing import ASGIApplication, ASGIReceiveCallable
+
+def name_that_thing(thing: Any) -> str: ...
+async def await_many_dispatch(
+    consumer_callables: list[Callable[[], Awaitable[ASGIReceiveCallable]]], dispatch: Callable[[dict[str, Any]], Awaitable[None]]
+) -> None: ...
+@type_check_only
+class _MiddlewareProtocol(Protocol):
+    def __init__(self, *args: Any, **kwargs: Any) -> None: ...
+    async def __call__(self, scope: Any, receive: Any, send: Any) -> Any: ...
+
+_ChannelApplication: TypeAlias = _MiddlewareProtocol | ASGIApplication  # noqa: Y047
diff --git a/stubs/channels/channels/worker.pyi b/stubs/channels/channels/worker.pyi
new file mode 100644
index 000000000000..08d979ac76a9
--- /dev/null
+++ b/stubs/channels/channels/worker.pyi
@@ -0,0 +1,13 @@
+from asgiref.server import StatelessServer
+from channels.layers import BaseChannelLayer
+from channels.utils import _ChannelApplication
+
+class Worker(StatelessServer):
+    channels: list[str]
+    channel_layer: BaseChannelLayer
+
+    def __init__(
+        self, application: _ChannelApplication, channels: list[str], channel_layer: BaseChannelLayer, max_applications: int = ...
+    ) -> None: ...
+    async def handle(self) -> None: ...
+    async def listener(self, channel: str) -> None: ...