diff --git a/.github/workflows/ci.yaml b/.github/workflows/ci.yaml
index 45628d00d33..4042319cd0b 100644
--- a/.github/workflows/ci.yaml
+++ b/.github/workflows/ci.yaml
@@ -35,7 +35,7 @@ on:
 env:
   CACHE_VERSION: 5
   PIP_CACHE_VERSION: 4
-  MYPY_CACHE_VERSION: 7
+  MYPY_CACHE_VERSION: 8
   HA_SHORT_VERSION: "2024.4"
   DEFAULT_PYTHON: "3.11"
   ALL_PYTHON_VERSIONS: "['3.11', '3.12']"
diff --git a/homeassistant/components/telegram_bot/__init__.py b/homeassistant/components/telegram_bot/__init__.py
index 2ba7752a85f..1aeed6a25bb 100644
--- a/homeassistant/components/telegram_bot/__init__.py
+++ b/homeassistant/components/telegram_bot/__init__.py
@@ -1,15 +1,14 @@
 """Support to send and receive Telegram messages."""
 from __future__ import annotations
 
-from functools import partial
+import asyncio
 import importlib
 import io
 from ipaddress import ip_network
 import logging
 from typing import Any
 
-import requests
-from requests.auth import HTTPBasicAuth, HTTPDigestAuth
+import httpx
 from telegram import (
     Bot,
     CallbackQuery,
@@ -21,10 +20,10 @@ from telegram import (
     Update,
     User,
 )
+from telegram.constants import ParseMode
 from telegram.error import TelegramError
-from telegram.ext import CallbackContext, Filters
-from telegram.parsemode import ParseMode
-from telegram.utils.request import Request
+from telegram.ext import CallbackContext, filters
+from telegram.request import HTTPXRequest
 import voluptuous as vol
 
 from homeassistant.const import (
@@ -283,7 +282,7 @@ SERVICE_MAP = {
 }
 
 
-def load_data(
+async def load_data(
     hass,
     url=None,
     filepath=None,
@@ -297,35 +296,48 @@ def load_data(
     try:
         if url is not None:
             # Load data from URL
-            params = {"timeout": 15}
+            params = {}
+            headers = {}
             if authentication == HTTP_BEARER_AUTHENTICATION and password is not None:
-                params["headers"] = {"Authorization": f"Bearer {password}"}
+                headers = {"Authorization": f"Bearer {password}"}
             elif username is not None and password is not None:
                 if authentication == HTTP_DIGEST_AUTHENTICATION:
-                    params["auth"] = HTTPDigestAuth(username, password)
+                    params["auth"] = httpx.DigestAuth(username, password)
                 else:
-                    params["auth"] = HTTPBasicAuth(username, password)
+                    params["auth"] = httpx.BasicAuth(username, password)
             if verify_ssl is not None:
                 params["verify"] = verify_ssl
+
             retry_num = 0
-            while retry_num < num_retries:
-                req = requests.get(url, **params)
-                if not req.ok:
-                    _LOGGER.warning(
-                        "Status code %s (retry #%s) loading %s",
-                        req.status_code,
-                        retry_num + 1,
-                        url,
-                    )
-                else:
-                    data = io.BytesIO(req.content)
-                    if data.read():
-                        data.seek(0)
-                        data.name = url
-                        return data
-                    _LOGGER.warning("Empty data (retry #%s) in %s)", retry_num + 1, url)
-                retry_num += 1
-            _LOGGER.warning("Can't load data in %s after %s retries", url, retry_num)
+            async with httpx.AsyncClient(
+                timeout=15, headers=headers, **params
+            ) as client:
+                while retry_num < num_retries:
+                    req = await client.get(url)
+                    if req.status_code != 200:
+                        _LOGGER.warning(
+                            "Status code %s (retry #%s) loading %s",
+                            req.status_code,
+                            retry_num + 1,
+                            url,
+                        )
+                    else:
+                        data = io.BytesIO(req.content)
+                        if data.read():
+                            data.seek(0)
+                            data.name = url
+                            return data
+                        _LOGGER.warning(
+                            "Empty data (retry #%s) in %s)", retry_num + 1, url
+                        )
+                    retry_num += 1
+                    if retry_num < num_retries:
+                        await asyncio.sleep(
+                            1
+                        )  # Add a sleep to allow other async operations to proceed
+                _LOGGER.warning(
+                    "Can't load data in %s after %s retries", url, retry_num
+                )
         elif filepath is not None:
             if hass.config.is_allowed_path(filepath):
                 return open(filepath, "rb")
@@ -406,9 +418,7 @@ async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
         _LOGGER.debug("New telegram message %s: %s", msgtype, kwargs)
 
         if msgtype == SERVICE_SEND_MESSAGE:
-            await hass.async_add_executor_job(
-                partial(notify_service.send_message, **kwargs)
-            )
+            await notify_service.send_message(**kwargs)
         elif msgtype in [
             SERVICE_SEND_PHOTO,
             SERVICE_SEND_ANIMATION,
@@ -416,33 +426,19 @@ async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
             SERVICE_SEND_VOICE,
             SERVICE_SEND_DOCUMENT,
         ]:
-            await hass.async_add_executor_job(
-                partial(notify_service.send_file, msgtype, **kwargs)
-            )
+            await notify_service.send_file(msgtype, **kwargs)
         elif msgtype == SERVICE_SEND_STICKER:
-            await hass.async_add_executor_job(
-                partial(notify_service.send_sticker, **kwargs)
-            )
+            await notify_service.send_sticker(**kwargs)
         elif msgtype == SERVICE_SEND_LOCATION:
-            await hass.async_add_executor_job(
-                partial(notify_service.send_location, **kwargs)
-            )
+            await notify_service.send_location(**kwargs)
         elif msgtype == SERVICE_SEND_POLL:
-            await hass.async_add_executor_job(
-                partial(notify_service.send_poll, **kwargs)
-            )
+            await notify_service.send_poll(**kwargs)
         elif msgtype == SERVICE_ANSWER_CALLBACK_QUERY:
-            await hass.async_add_executor_job(
-                partial(notify_service.answer_callback_query, **kwargs)
-            )
+            await notify_service.answer_callback_query(**kwargs)
         elif msgtype == SERVICE_DELETE_MESSAGE:
-            await hass.async_add_executor_job(
-                partial(notify_service.delete_message, **kwargs)
-            )
+            await notify_service.delete_message(**kwargs)
         else:
-            await hass.async_add_executor_job(
-                partial(notify_service.edit_message, msgtype, **kwargs)
-            )
+            await notify_service.edit_message(msgtype, **kwargs)
 
     # Register notification services
     for service_notif, schema in SERVICE_MAP.items():
@@ -460,11 +456,13 @@ def initialize_bot(p_config):
     proxy_params = p_config.get(CONF_PROXY_PARAMS)
 
     if proxy_url is not None:
-        request = Request(
-            con_pool_size=8, proxy_url=proxy_url, urllib3_proxy_kwargs=proxy_params
-        )
+        # These have been kept for backwards compatibility, they can actually be stuffed into the URL.
+        # Side note: In the future we should deprecate these and raise a repair issue if we find them here.
+        auth = proxy_params.pop("username"), proxy_params.pop("password")
+        proxy = httpx.Proxy(proxy_url, auth=auth, **proxy_params)
+        request = HTTPXRequest(connection_pool_size=8, proxy=proxy)
     else:
-        request = Request(con_pool_size=8)
+        request = HTTPXRequest(connection_pool_size=8)
     return Bot(token=api_key, request=request)
 
 
@@ -616,10 +614,12 @@ class TelegramNotificationService:
                 )
         return params
 
-    def _send_msg(self, func_send, msg_error, message_tag, *args_msg, **kwargs_msg):
+    async def _send_msg(
+        self, func_send, msg_error, message_tag, *args_msg, **kwargs_msg
+    ):
         """Send one message."""
         try:
-            out = func_send(*args_msg, **kwargs_msg)
+            out = await func_send(*args_msg, **kwargs_msg)
             if not isinstance(out, bool) and hasattr(out, ATTR_MESSAGEID):
                 chat_id = out.chat_id
                 message_id = out[ATTR_MESSAGEID]
@@ -636,7 +636,7 @@ class TelegramNotificationService:
                 }
                 if message_tag is not None:
                     event_data[ATTR_MESSAGE_TAG] = message_tag
-                self.hass.bus.fire(EVENT_TELEGRAM_SENT, event_data)
+                self.hass.bus.async_fire(EVENT_TELEGRAM_SENT, event_data)
             elif not isinstance(out, bool):
                 _LOGGER.warning(
                     "Update last message: out_type:%s, out=%s", type(out), out
@@ -647,14 +647,14 @@ class TelegramNotificationService:
                 "%s: %s. Args: %s, kwargs: %s", msg_error, exc, args_msg, kwargs_msg
             )
 
-    def send_message(self, message="", target=None, **kwargs):
+    async def send_message(self, message="", target=None, **kwargs):
         """Send a message to one or multiple pre-allowed chat IDs."""
         title = kwargs.get(ATTR_TITLE)
         text = f"{title}\n{message}" if title else message
         params = self._get_msg_kwargs(kwargs)
         for chat_id in self._get_target_chat_ids(target):
             _LOGGER.debug("Send message in chat ID %s with params: %s", chat_id, params)
-            self._send_msg(
+            await self._send_msg(
                 self.bot.send_message,
                 "Error sending message",
                 params[ATTR_MESSAGE_TAG],
@@ -665,15 +665,15 @@ class TelegramNotificationService:
                 disable_notification=params[ATTR_DISABLE_NOTIF],
                 reply_to_message_id=params[ATTR_REPLY_TO_MSGID],
                 reply_markup=params[ATTR_REPLYMARKUP],
-                timeout=params[ATTR_TIMEOUT],
+                read_timeout=params[ATTR_TIMEOUT],
             )
 
-    def delete_message(self, chat_id=None, **kwargs):
+    async def delete_message(self, chat_id=None, **kwargs):
         """Delete a previously sent message."""
         chat_id = self._get_target_chat_ids(chat_id)[0]
         message_id, _ = self._get_msg_ids(kwargs, chat_id)
         _LOGGER.debug("Delete message %s in chat ID %s", message_id, chat_id)
-        deleted = self._send_msg(
+        deleted = await self._send_msg(
             self.bot.delete_message, "Error deleting message", None, chat_id, message_id
         )
         # reduce message_id anyway:
@@ -682,7 +682,7 @@ class TelegramNotificationService:
             self._last_message_id[chat_id] -= 1
         return deleted
 
-    def edit_message(self, type_edit, chat_id=None, **kwargs):
+    async def edit_message(self, type_edit, chat_id=None, **kwargs):
         """Edit a previously sent message."""
         chat_id = self._get_target_chat_ids(chat_id)[0]
         message_id, inline_message_id = self._get_msg_ids(kwargs, chat_id)
@@ -698,7 +698,7 @@ class TelegramNotificationService:
             title = kwargs.get(ATTR_TITLE)
             text = f"{title}\n{message}" if title else message
             _LOGGER.debug("Editing message with ID %s", message_id or inline_message_id)
-            return self._send_msg(
+            return await self._send_msg(
                 self.bot.edit_message_text,
                 "Error editing text message",
                 params[ATTR_MESSAGE_TAG],
@@ -709,10 +709,10 @@ class TelegramNotificationService:
                 parse_mode=params[ATTR_PARSER],
                 disable_web_page_preview=params[ATTR_DISABLE_WEB_PREV],
                 reply_markup=params[ATTR_REPLYMARKUP],
-                timeout=params[ATTR_TIMEOUT],
+                read_timeout=params[ATTR_TIMEOUT],
             )
         if type_edit == SERVICE_EDIT_CAPTION:
-            return self._send_msg(
+            return await self._send_msg(
                 self.bot.edit_message_caption,
                 "Error editing message attributes",
                 params[ATTR_MESSAGE_TAG],
@@ -721,11 +721,11 @@ class TelegramNotificationService:
                 inline_message_id=inline_message_id,
                 caption=kwargs.get(ATTR_CAPTION),
                 reply_markup=params[ATTR_REPLYMARKUP],
-                timeout=params[ATTR_TIMEOUT],
+                read_timeout=params[ATTR_TIMEOUT],
                 parse_mode=params[ATTR_PARSER],
             )
 
-        return self._send_msg(
+        return await self._send_msg(
             self.bot.edit_message_reply_markup,
             "Error editing message attributes",
             params[ATTR_MESSAGE_TAG],
@@ -733,10 +733,10 @@ class TelegramNotificationService:
             message_id=message_id,
             inline_message_id=inline_message_id,
             reply_markup=params[ATTR_REPLYMARKUP],
-            timeout=params[ATTR_TIMEOUT],
+            read_timeout=params[ATTR_TIMEOUT],
         )
 
-    def answer_callback_query(
+    async def answer_callback_query(
         self, message, callback_query_id, show_alert=False, **kwargs
     ):
         """Answer a callback originated with a press in an inline keyboard."""
@@ -747,20 +747,20 @@ class TelegramNotificationService:
             message,
             show_alert,
         )
-        self._send_msg(
+        await self._send_msg(
             self.bot.answer_callback_query,
             "Error sending answer callback query",
             params[ATTR_MESSAGE_TAG],
             callback_query_id,
             text=message,
             show_alert=show_alert,
-            timeout=params[ATTR_TIMEOUT],
+            read_timeout=params[ATTR_TIMEOUT],
         )
 
-    def send_file(self, file_type=SERVICE_SEND_PHOTO, target=None, **kwargs):
+    async def send_file(self, file_type=SERVICE_SEND_PHOTO, target=None, **kwargs):
         """Send a photo, sticker, video, or document."""
         params = self._get_msg_kwargs(kwargs)
-        file_content = load_data(
+        file_content = await load_data(
             self.hass,
             url=kwargs.get(ATTR_URL),
             filepath=kwargs.get(ATTR_FILE),
@@ -775,7 +775,7 @@ class TelegramNotificationService:
                 _LOGGER.debug("Sending file to chat ID %s", chat_id)
 
                 if file_type == SERVICE_SEND_PHOTO:
-                    self._send_msg(
+                    await self._send_msg(
                         self.bot.send_photo,
                         "Error sending photo",
                         params[ATTR_MESSAGE_TAG],
@@ -785,12 +785,12 @@ class TelegramNotificationService:
                         disable_notification=params[ATTR_DISABLE_NOTIF],
                         reply_to_message_id=params[ATTR_REPLY_TO_MSGID],
                         reply_markup=params[ATTR_REPLYMARKUP],
-                        timeout=params[ATTR_TIMEOUT],
+                        read_timeout=params[ATTR_TIMEOUT],
                         parse_mode=params[ATTR_PARSER],
                     )
 
                 elif file_type == SERVICE_SEND_STICKER:
-                    self._send_msg(
+                    await self._send_msg(
                         self.bot.send_sticker,
                         "Error sending sticker",
                         params[ATTR_MESSAGE_TAG],
@@ -799,11 +799,11 @@ class TelegramNotificationService:
                         disable_notification=params[ATTR_DISABLE_NOTIF],
                         reply_to_message_id=params[ATTR_REPLY_TO_MSGID],
                         reply_markup=params[ATTR_REPLYMARKUP],
-                        timeout=params[ATTR_TIMEOUT],
+                        read_timeout=params[ATTR_TIMEOUT],
                     )
 
                 elif file_type == SERVICE_SEND_VIDEO:
-                    self._send_msg(
+                    await self._send_msg(
                         self.bot.send_video,
                         "Error sending video",
                         params[ATTR_MESSAGE_TAG],
@@ -813,11 +813,11 @@ class TelegramNotificationService:
                         disable_notification=params[ATTR_DISABLE_NOTIF],
                         reply_to_message_id=params[ATTR_REPLY_TO_MSGID],
                         reply_markup=params[ATTR_REPLYMARKUP],
-                        timeout=params[ATTR_TIMEOUT],
+                        read_timeout=params[ATTR_TIMEOUT],
                         parse_mode=params[ATTR_PARSER],
                     )
                 elif file_type == SERVICE_SEND_DOCUMENT:
-                    self._send_msg(
+                    await self._send_msg(
                         self.bot.send_document,
                         "Error sending document",
                         params[ATTR_MESSAGE_TAG],
@@ -827,11 +827,11 @@ class TelegramNotificationService:
                         disable_notification=params[ATTR_DISABLE_NOTIF],
                         reply_to_message_id=params[ATTR_REPLY_TO_MSGID],
                         reply_markup=params[ATTR_REPLYMARKUP],
-                        timeout=params[ATTR_TIMEOUT],
+                        read_timeout=params[ATTR_TIMEOUT],
                         parse_mode=params[ATTR_PARSER],
                     )
                 elif file_type == SERVICE_SEND_VOICE:
-                    self._send_msg(
+                    await self._send_msg(
                         self.bot.send_voice,
                         "Error sending voice",
                         params[ATTR_MESSAGE_TAG],
@@ -841,10 +841,10 @@ class TelegramNotificationService:
                         disable_notification=params[ATTR_DISABLE_NOTIF],
                         reply_to_message_id=params[ATTR_REPLY_TO_MSGID],
                         reply_markup=params[ATTR_REPLYMARKUP],
-                        timeout=params[ATTR_TIMEOUT],
+                        read_timeout=params[ATTR_TIMEOUT],
                     )
                 elif file_type == SERVICE_SEND_ANIMATION:
-                    self._send_msg(
+                    await self._send_msg(
                         self.bot.send_animation,
                         "Error sending animation",
                         params[ATTR_MESSAGE_TAG],
@@ -854,7 +854,7 @@ class TelegramNotificationService:
                         disable_notification=params[ATTR_DISABLE_NOTIF],
                         reply_to_message_id=params[ATTR_REPLY_TO_MSGID],
                         reply_markup=params[ATTR_REPLYMARKUP],
-                        timeout=params[ATTR_TIMEOUT],
+                        read_timeout=params[ATTR_TIMEOUT],
                         parse_mode=params[ATTR_PARSER],
                     )
 
@@ -862,13 +862,13 @@ class TelegramNotificationService:
         else:
             _LOGGER.error("Can't send file with kwargs: %s", kwargs)
 
-    def send_sticker(self, target=None, **kwargs):
+    async def send_sticker(self, target=None, **kwargs):
         """Send a sticker from a telegram sticker pack."""
         params = self._get_msg_kwargs(kwargs)
         stickerid = kwargs.get(ATTR_STICKER_ID)
         if stickerid:
             for chat_id in self._get_target_chat_ids(target):
-                self._send_msg(
+                await self._send_msg(
                     self.bot.send_sticker,
                     "Error sending sticker",
                     params[ATTR_MESSAGE_TAG],
@@ -877,12 +877,12 @@ class TelegramNotificationService:
                     disable_notification=params[ATTR_DISABLE_NOTIF],
                     reply_to_message_id=params[ATTR_REPLY_TO_MSGID],
                     reply_markup=params[ATTR_REPLYMARKUP],
-                    timeout=params[ATTR_TIMEOUT],
+                    read_timeout=params[ATTR_TIMEOUT],
                 )
         else:
-            self.send_file(SERVICE_SEND_STICKER, target, **kwargs)
+            await self.send_file(SERVICE_SEND_STICKER, target, **kwargs)
 
-    def send_location(self, latitude, longitude, target=None, **kwargs):
+    async def send_location(self, latitude, longitude, target=None, **kwargs):
         """Send a location."""
         latitude = float(latitude)
         longitude = float(longitude)
@@ -891,7 +891,7 @@ class TelegramNotificationService:
             _LOGGER.debug(
                 "Send location %s/%s to chat ID %s", latitude, longitude, chat_id
             )
-            self._send_msg(
+            await self._send_msg(
                 self.bot.send_location,
                 "Error sending location",
                 params[ATTR_MESSAGE_TAG],
@@ -900,10 +900,10 @@ class TelegramNotificationService:
                 longitude=longitude,
                 disable_notification=params[ATTR_DISABLE_NOTIF],
                 reply_to_message_id=params[ATTR_REPLY_TO_MSGID],
-                timeout=params[ATTR_TIMEOUT],
+                read_timeout=params[ATTR_TIMEOUT],
             )
 
-    def send_poll(
+    async def send_poll(
         self,
         question,
         options,
@@ -917,7 +917,7 @@ class TelegramNotificationService:
         openperiod = kwargs.get(ATTR_OPEN_PERIOD)
         for chat_id in self._get_target_chat_ids(target):
             _LOGGER.debug("Send poll '%s' to chat ID %s", question, chat_id)
-            self._send_msg(
+            await self._send_msg(
                 self.bot.send_poll,
                 "Error sending poll",
                 params[ATTR_MESSAGE_TAG],
@@ -929,14 +929,14 @@ class TelegramNotificationService:
                 open_period=openperiod,
                 disable_notification=params[ATTR_DISABLE_NOTIF],
                 reply_to_message_id=params[ATTR_REPLY_TO_MSGID],
-                timeout=params[ATTR_TIMEOUT],
+                read_timeout=params[ATTR_TIMEOUT],
             )
 
-    def leave_chat(self, chat_id=None):
+    async def leave_chat(self, chat_id=None):
         """Remove bot from chat."""
         chat_id = self._get_target_chat_ids(chat_id)[0]
         _LOGGER.debug("Leave from chat ID %s", chat_id)
-        leaved = self._send_msg(
+        leaved = await self._send_msg(
             self.bot.leave_chat, "Error leaving chat", None, chat_id
         )
         return leaved
@@ -950,8 +950,8 @@ class BaseTelegramBotEntity:
         self.allowed_chat_ids = config[CONF_ALLOWED_CHAT_IDS]
         self.hass = hass
 
-    def handle_update(self, update: Update, context: CallbackContext) -> bool:
-        """Handle updates from bot dispatcher set up by the respective platform."""
+    async def handle_update(self, update: Update, context: CallbackContext) -> bool:
+        """Handle updates from bot application set up by the respective platform."""
         _LOGGER.debug("Handling update %s", update)
         if not self.authorize_update(update):
             return False
@@ -972,12 +972,12 @@ class BaseTelegramBotEntity:
             return True
 
         _LOGGER.debug("Firing event %s: %s", event_type, event_data)
-        self.hass.bus.fire(event_type, event_data)
+        self.hass.bus.async_fire(event_type, event_data)
         return True
 
     @staticmethod
-    def _get_command_event_data(command_text: str) -> dict[str, str | list]:
-        if not command_text.startswith("/"):
+    def _get_command_event_data(command_text: str | None) -> dict[str, str | list]:
+        if not command_text or not command_text.startswith("/"):
             return {}
         command_parts = command_text.split()
         command = command_parts[0]
@@ -990,7 +990,7 @@ class BaseTelegramBotEntity:
             ATTR_CHAT_ID: message.chat.id,
             ATTR_DATE: message.date,
         }
-        if Filters.command.filter(message):
+        if filters.COMMAND.filter(message):
             # This is a command message - set event type to command and split data into command and args
             event_type = EVENT_TELEGRAM_COMMAND
             event_data.update(self._get_command_event_data(message.text))
diff --git a/homeassistant/components/telegram_bot/manifest.json b/homeassistant/components/telegram_bot/manifest.json
index a4964638242..c176e6c2cdf 100644
--- a/homeassistant/components/telegram_bot/manifest.json
+++ b/homeassistant/components/telegram_bot/manifest.json
@@ -6,5 +6,5 @@
   "documentation": "https://www.home-assistant.io/integrations/telegram_bot",
   "iot_class": "cloud_push",
   "loggers": ["telegram"],
-  "requirements": ["python-telegram-bot==13.1", "PySocks==1.7.1"]
+  "requirements": ["python-telegram-bot[socks]==21.0.1"]
 }
diff --git a/homeassistant/components/telegram_bot/polling.py b/homeassistant/components/telegram_bot/polling.py
index 69bb4dc4963..bac6262cc6a 100644
--- a/homeassistant/components/telegram_bot/polling.py
+++ b/homeassistant/components/telegram_bot/polling.py
@@ -3,7 +3,7 @@ import logging
 
 from telegram import Update
 from telegram.error import NetworkError, RetryAfter, TelegramError, TimedOut
-from telegram.ext import CallbackContext, TypeHandler, Updater
+from telegram.ext import ApplicationBuilder, CallbackContext, TypeHandler
 
 from homeassistant.const import EVENT_HOMEASSISTANT_START, EVENT_HOMEASSISTANT_STOP
 
@@ -22,7 +22,7 @@ async def async_setup_platform(hass, bot, config):
     return True
 
 
-def process_error(update: Update, context: CallbackContext) -> None:
+async def process_error(update: Update, context: CallbackContext) -> None:
     """Telegram bot error handler."""
     try:
         if context.error:
@@ -35,26 +35,29 @@ def process_error(update: Update, context: CallbackContext) -> None:
 
 
 class PollBot(BaseTelegramBotEntity):
-    """Controls the Updater object that holds the bot and a dispatcher.
+    """Controls the Application object that holds the bot and an updater.
 
-    The dispatcher is set up by the super class to pass telegram updates to `self.handle_update`
+    The application is set up to pass telegram updates to `self.handle_update`
     """
 
     def __init__(self, hass, bot, config):
-        """Create Updater and Dispatcher before calling super()."""
-        self.bot = bot
-        self.updater = Updater(bot=bot, workers=4)
-        self.dispatcher = self.updater.dispatcher
-        self.dispatcher.add_handler(TypeHandler(Update, self.handle_update))
-        self.dispatcher.add_error_handler(process_error)
+        """Create Application to poll for updates."""
         super().__init__(hass, config)
+        self.bot = bot
+        self.application = ApplicationBuilder().bot(self.bot).build()
+        self.application.add_handler(TypeHandler(Update, self.handle_update))
+        self.application.add_error_handler(process_error)
 
-    def start_polling(self, event=None):
+    async def start_polling(self, event=None):
         """Start the polling task."""
         _LOGGER.debug("Starting polling")
-        self.updater.start_polling()
+        await self.application.initialize()
+        await self.application.updater.start_polling()
+        await self.application.start()
 
-    def stop_polling(self, event=None):
+    async def stop_polling(self, event=None):
         """Stop the polling task."""
         _LOGGER.debug("Stopping polling")
-        self.updater.stop()
+        await self.application.updater.stop()
+        await self.application.stop()
+        await self.application.shutdown()
diff --git a/homeassistant/components/telegram_bot/strings.json b/homeassistant/components/telegram_bot/strings.json
index de5de685409..2e1d51f31e8 100644
--- a/homeassistant/components/telegram_bot/strings.json
+++ b/homeassistant/components/telegram_bot/strings.json
@@ -29,8 +29,8 @@
           "description": "Disables link previews for links in the message."
         },
         "timeout": {
-          "name": "Timeout",
-          "description": "Timeout for send message. Will help with timeout errors (poor internet connection, etc)s."
+          "name": "Read timeout",
+          "description": "Read timeout for send message. Will help with timeout errors (poor internet connection, etc)s."
         },
         "keyboard": {
           "name": "Keyboard",
@@ -95,8 +95,8 @@
           "description": "Enable or disable SSL certificate verification. Set to false if you're downloading the file from a URL and you don't want to validate the SSL certificate of the server."
         },
         "timeout": {
-          "name": "Timeout",
-          "description": "Timeout for send photo. Will help with timeout errors (poor internet connection, etc)."
+          "name": "Read timeout",
+          "description": "Read timeout for send photo."
         },
         "keyboard": {
           "name": "[%key:component::telegram_bot::services::send_message::fields::keyboard::name%]",
@@ -157,8 +157,8 @@
           "description": "[%key:component::telegram_bot::services::send_photo::fields::verify_ssl::description%]"
         },
         "timeout": {
-          "name": "Timeout",
-          "description": "Timeout for send sticker. Will help with timeout errors (poor internet connection, etc)."
+          "name": "Read timeout",
+          "description": "Read timeout for send sticker."
         },
         "keyboard": {
           "name": "[%key:component::telegram_bot::services::send_message::fields::keyboard::name%]",
@@ -223,7 +223,7 @@
           "description": "[%key:component::telegram_bot::services::send_photo::fields::verify_ssl::description%]"
         },
         "timeout": {
-          "name": "Timeout",
+          "name": "Read timeout",
           "description": "[%key:component::telegram_bot::services::send_sticker::fields::timeout::description%]"
         },
         "keyboard": {
@@ -289,8 +289,8 @@
           "description": "[%key:component::telegram_bot::services::send_photo::fields::verify_ssl::description%]"
         },
         "timeout": {
-          "name": "Timeout",
-          "description": "Timeout for send video. Will help with timeout errors (poor internet connection, etc)."
+          "name": "Read timeout",
+          "description": "Read timeout for send video."
         },
         "keyboard": {
           "name": "[%key:component::telegram_bot::services::send_message::fields::keyboard::name%]",
@@ -351,8 +351,8 @@
           "description": "[%key:component::telegram_bot::services::send_photo::fields::verify_ssl::description%]"
         },
         "timeout": {
-          "name": "Timeout",
-          "description": "Timeout for send voice. Will help with timeout errors (poor internet connection, etc)."
+          "name": "Read timeout",
+          "description": "Read timeout for send voice."
         },
         "keyboard": {
           "name": "[%key:component::telegram_bot::services::send_message::fields::keyboard::name%]",
@@ -417,8 +417,8 @@
           "description": "[%key:component::telegram_bot::services::send_photo::fields::verify_ssl::description%]"
         },
         "timeout": {
-          "name": "Timeout",
-          "description": "Timeout for send document. Will help with timeout errors (poor internet connection, etc)."
+          "name": "Read timeout",
+          "description": "Read timeout for send document."
         },
         "keyboard": {
           "name": "[%key:component::telegram_bot::services::send_message::fields::keyboard::name%]",
@@ -459,7 +459,7 @@
           "description": "[%key:component::telegram_bot::services::send_message::fields::disable_notification::description%]"
         },
         "timeout": {
-          "name": "Timeout",
+          "name": "Read timeout",
           "description": "[%key:component::telegram_bot::services::send_photo::fields::timeout::description%]"
         },
         "keyboard": {
@@ -513,8 +513,8 @@
           "description": "[%key:component::telegram_bot::services::send_message::fields::disable_notification::description%]"
         },
         "timeout": {
-          "name": "Timeout",
-          "description": "Timeout for send poll. Will help with timeout errors (poor internet connection, etc)."
+          "name": "Read timeout",
+          "description": "Read timeout for send poll."
         },
         "message_tag": {
           "name": "[%key:component::telegram_bot::services::send_message::fields::message_tag::name%]",
@@ -617,8 +617,8 @@
           "description": "Show a permanent notification."
         },
         "timeout": {
-          "name": "Timeout",
-          "description": "Timeout for sending the answer. Will help with timeout errors (poor internet connection, etc)."
+          "name": "Read timeout",
+          "description": "Read timeout for sending the answer."
         }
       }
     },
diff --git a/homeassistant/components/telegram_bot/webhooks.py b/homeassistant/components/telegram_bot/webhooks.py
index c21cffa84b1..50fd7bc8427 100644
--- a/homeassistant/components/telegram_bot/webhooks.py
+++ b/homeassistant/components/telegram_bot/webhooks.py
@@ -8,7 +8,7 @@ import string
 
 from telegram import Update
 from telegram.error import TimedOut
-from telegram.ext import Dispatcher, TypeHandler
+from telegram.ext import Application, TypeHandler
 
 from homeassistant.components.http import HomeAssistantView
 from homeassistant.const import EVENT_HOMEASSISTANT_STOP
@@ -36,16 +36,17 @@ async def async_setup_platform(hass, bot, config):
         _LOGGER.error("Invalid telegram webhook %s must be https", pushbot.webhook_url)
         return False
 
+    await pushbot.start_application()
     webhook_registered = await pushbot.register_webhook()
     if not webhook_registered:
         return False
 
-    hass.bus.async_listen_once(EVENT_HOMEASSISTANT_STOP, pushbot.deregister_webhook)
+    hass.bus.async_listen_once(EVENT_HOMEASSISTANT_STOP, pushbot.stop_application)
     hass.http.register_view(
         PushBotView(
             hass,
             bot,
-            pushbot.dispatcher,
+            pushbot.application,
             config[CONF_TRUSTED_NETWORKS],
             secret_token,
         )
@@ -57,13 +58,13 @@ class PushBot(BaseTelegramBotEntity):
     """Handles all the push/webhook logic and passes telegram updates to `self.handle_update`."""
 
     def __init__(self, hass, bot, config, secret_token):
-        """Create Dispatcher before calling super()."""
+        """Create Application before calling super()."""
         self.bot = bot
         self.trusted_networks = config[CONF_TRUSTED_NETWORKS]
         self.secret_token = secret_token
-        # Dumb dispatcher that just gets our updates to our handler callback (self.handle_update)
-        self.dispatcher = Dispatcher(bot, None)
-        self.dispatcher.add_handler(TypeHandler(Update, self.handle_update))
+        # Dumb Application that just gets our updates to our handler callback (self.handle_update)
+        self.application = Application.builder().bot(bot).updater(None).build()
+        self.application.add_handler(TypeHandler(Update, self.handle_update))
         super().__init__(hass, config)
 
         self.base_url = config.get(CONF_URL) or get_url(
@@ -71,15 +72,15 @@ class PushBot(BaseTelegramBotEntity):
         )
         self.webhook_url = f"{self.base_url}{TELEGRAM_WEBHOOK_URL}"
 
-    def _try_to_set_webhook(self):
+    async def _try_to_set_webhook(self):
         _LOGGER.debug("Registering webhook URL: %s", self.webhook_url)
         retry_num = 0
         while retry_num < 3:
             try:
-                return self.bot.set_webhook(
+                return await self.bot.set_webhook(
                     self.webhook_url,
                     api_kwargs={"secret_token": self.secret_token},
-                    timeout=5,
+                    connect_timeout=5,
                 )
             except TimedOut:
                 retry_num += 1
@@ -87,11 +88,14 @@ class PushBot(BaseTelegramBotEntity):
 
         return False
 
+    async def start_application(self):
+        """Handle starting the Application object."""
+        await self.application.initialize()
+        await self.application.start()
+
     async def register_webhook(self):
         """Query telegram and register the URL for our webhook."""
-        current_status = await self.hass.async_add_executor_job(
-            self.bot.get_webhook_info
-        )
+        current_status = await self.bot.get_webhook_info()
         # Some logging of Bot current status:
         last_error_date = getattr(current_status, "last_error_date", None)
         if (last_error_date is not None) and (isinstance(last_error_date, int)):
@@ -105,7 +109,7 @@ class PushBot(BaseTelegramBotEntity):
             _LOGGER.debug("telegram webhook status: %s", current_status)
 
         if current_status and current_status["url"] != self.webhook_url:
-            result = await self.hass.async_add_executor_job(self._try_to_set_webhook)
+            result = await self._try_to_set_webhook()
             if result:
                 _LOGGER.info("Set new telegram webhook %s", self.webhook_url)
             else:
@@ -114,10 +118,16 @@ class PushBot(BaseTelegramBotEntity):
 
         return True
 
-    def deregister_webhook(self, event=None):
+    async def stop_application(self, event=None):
+        """Handle gracefully stopping the Application object."""
+        await self.deregister_webhook()
+        await self.application.stop()
+        await self.application.shutdown()
+
+    async def deregister_webhook(self):
         """Query telegram and deregister the URL for our webhook."""
         _LOGGER.debug("Deregistering webhook URL")
-        return self.bot.delete_webhook()
+        await self.bot.delete_webhook()
 
 
 class PushBotView(HomeAssistantView):
@@ -127,11 +137,11 @@ class PushBotView(HomeAssistantView):
     url = TELEGRAM_WEBHOOK_URL
     name = "telegram_webhooks"
 
-    def __init__(self, hass, bot, dispatcher, trusted_networks, secret_token):
+    def __init__(self, hass, bot, application, trusted_networks, secret_token):
         """Initialize by storing stuff needed for setting up our webhook endpoint."""
         self.hass = hass
         self.bot = bot
-        self.dispatcher = dispatcher
+        self.application = application
         self.trusted_networks = trusted_networks
         self.secret_token = secret_token
 
@@ -153,6 +163,6 @@ class PushBotView(HomeAssistantView):
 
         update = Update.de_json(update_data, self.bot)
         _LOGGER.debug("Received Update on %s: %s", self.url, update)
-        await self.hass.async_add_executor_job(self.dispatcher.process_update, update)
+        await self.application.process_update(update)
 
         return None
diff --git a/pyproject.toml b/pyproject.toml
index e03e0fda88e..e27562d1a8d 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -508,8 +508,6 @@ filterwarnings = [
     "ignore:datetime.*utcnow\\(\\) is deprecated and scheduled for removal:DeprecationWarning:miio.miioprotocol",
     # https://github.com/hunterjm/python-onvif-zeep-async/pull/51 - >3.1.12
     "ignore:datetime.*utcnow\\(\\) is deprecated and scheduled for removal:DeprecationWarning:onvif.client",
-    # Fixed upstream in python-telegram-bot - >=20.0
-    "ignore:python-telegram-bot is using upstream urllib3:UserWarning:telegram.utils.request",
     # https://github.com/xeniter/romy/pull/1 - >0.0.7
     "ignore:with timeout\\(\\) is deprecated, use async with timeout\\(\\) instead:DeprecationWarning:romy.utils",
     # https://github.com/grahamwetzler/smart-meter-texas/pull/143 - >0.5.3
diff --git a/requirements_all.txt b/requirements_all.txt
index 91404852ece..bc00afbfeab 100644
--- a/requirements_all.txt
+++ b/requirements_all.txt
@@ -92,9 +92,6 @@ PyQRCode==1.2.1
 # homeassistant.components.rmvtransport
 PyRMVtransport==0.3.3
 
-# homeassistant.components.telegram_bot
-PySocks==1.7.1
-
 # homeassistant.components.switchbot
 PySwitchbot==0.45.0
 
@@ -2300,7 +2297,7 @@ python-tado==0.17.4
 python-technove==1.2.2
 
 # homeassistant.components.telegram_bot
-python-telegram-bot==13.1
+python-telegram-bot[socks]==21.0.1
 
 # homeassistant.components.vlc
 python-vlc==3.0.18122
diff --git a/requirements_test_all.txt b/requirements_test_all.txt
index 0cb0228e201..63b063badd9 100644
--- a/requirements_test_all.txt
+++ b/requirements_test_all.txt
@@ -80,9 +80,6 @@ PyQRCode==1.2.1
 # homeassistant.components.rmvtransport
 PyRMVtransport==0.3.3
 
-# homeassistant.components.telegram_bot
-PySocks==1.7.1
-
 # homeassistant.components.switchbot
 PySwitchbot==0.45.0
 
@@ -1773,7 +1770,7 @@ python-tado==0.17.4
 python-technove==1.2.2
 
 # homeassistant.components.telegram_bot
-python-telegram-bot==13.1
+python-telegram-bot[socks]==21.0.1
 
 # homeassistant.components.tile
 pytile==2023.04.0
diff --git a/tests/components/telegram_bot/conftest.py b/tests/components/telegram_bot/conftest.py
index af23efc1afc..f607ed468c5 100644
--- a/tests/components/telegram_bot/conftest.py
+++ b/tests/components/telegram_bot/conftest.py
@@ -2,13 +2,19 @@
 from unittest.mock import patch
 
 import pytest
+from telegram import User
 
 from homeassistant.components.telegram_bot import (
     CONF_ALLOWED_CHAT_IDS,
     CONF_TRUSTED_NETWORKS,
     DOMAIN,
 )
-from homeassistant.const import CONF_API_KEY, CONF_PLATFORM, CONF_URL
+from homeassistant.const import (
+    CONF_API_KEY,
+    CONF_PLATFORM,
+    CONF_URL,
+    EVENT_HOMEASSISTANT_START,
+)
 from homeassistant.setup import async_setup_component
 
 
@@ -65,6 +71,23 @@ def mock_register_webhook():
         yield
 
 
+@pytest.fixture
+def mock_external_calls():
+    """Mock calls that make calls to the live Telegram API."""
+    test_user = User(123456, "Testbot", True)
+    with patch(
+        "telegram.Bot.get_me",
+        return_value=test_user,
+    ), patch(
+        "telegram.Bot._bot_user",
+        test_user,
+    ), patch(
+        "telegram.Bot.bot",
+        test_user,
+    ), patch("telegram.ext.Updater._bootstrap"):
+        yield
+
+
 @pytest.fixture
 def mock_generate_secret_token():
     """Mock secret token generated for webhook."""
@@ -174,7 +197,11 @@ def update_callback_query():
 
 @pytest.fixture
 async def webhook_platform(
-    hass, config_webhooks, mock_register_webhook, mock_generate_secret_token
+    hass,
+    config_webhooks,
+    mock_register_webhook,
+    mock_external_calls,
+    mock_generate_secret_token,
 ):
     """Fixture for setting up the webhooks platform using appropriate config and mocks."""
     await async_setup_component(
@@ -183,14 +210,18 @@ async def webhook_platform(
         config_webhooks,
     )
     await hass.async_block_till_done()
+    yield
+    await hass.async_stop()
 
 
 @pytest.fixture
-async def polling_platform(hass, config_polling):
+async def polling_platform(hass, config_polling, mock_external_calls):
     """Fixture for setting up the polling platform using appropriate config and mocks."""
     await async_setup_component(
         hass,
         DOMAIN,
         config_polling,
     )
+    # Fire this event to start polling
+    hass.bus.fire(EVENT_HOMEASSISTANT_START)
     await hass.async_block_till_done()
diff --git a/tests/components/telegram_bot/test_telegram_bot.py b/tests/components/telegram_bot/test_telegram_bot.py
index be28f7be636..f86f70e77c1 100644
--- a/tests/components/telegram_bot/test_telegram_bot.py
+++ b/tests/components/telegram_bot/test_telegram_bot.py
@@ -1,25 +1,17 @@
 """Tests for the telegram_bot component."""
-import pytest
+from unittest.mock import AsyncMock, patch
+
 from telegram import Update
-from telegram.ext.dispatcher import Dispatcher
 
 from homeassistant.components.telegram_bot import DOMAIN, SERVICE_SEND_MESSAGE
 from homeassistant.components.telegram_bot.webhooks import TELEGRAM_WEBHOOK_URL
 from homeassistant.core import HomeAssistant
+from homeassistant.setup import async_setup_component
 
 from tests.common import async_capture_events
 from tests.typing import ClientSessionGenerator
 
 
-@pytest.fixture(autouse=True)
-def clear_dispatcher():
-    """Clear the singleton that telegram.ext.dispatcher.Dispatcher sets on itself."""
-    yield
-    Dispatcher._set_singleton(None)
-    # This is how python-telegram-bot resets the dispatcher in their test suite
-    Dispatcher._Dispatcher__singleton_semaphore.release()
-
-
 async def test_webhook_platform_init(hass: HomeAssistant, webhook_platform) -> None:
     """Test initialization of the webhooks platform."""
     assert hass.services.has_service(DOMAIN, SERVICE_SEND_MESSAGE) is True
@@ -109,18 +101,38 @@ async def test_webhook_endpoint_generates_telegram_callback_event(
 
 
 async def test_polling_platform_message_text_update(
-    hass: HomeAssistant, polling_platform, update_message_text
+    hass: HomeAssistant, config_polling, update_message_text
 ) -> None:
-    """Provide the `PollBot`s `Dispatcher` with an `Update` and assert fired `telegram_text` event."""
+    """Provide the `BaseTelegramBotEntity.update_handler` with an `Update` and assert fired `telegram_text` event."""
     events = async_capture_events(hass, "telegram_text")
 
-    def telegram_dispatcher_callback():
-        dispatcher = Dispatcher.get_instance()
-        update = Update.de_json(update_message_text, dispatcher.bot)
-        dispatcher.process_update(update)
+    with patch(
+        "homeassistant.components.telegram_bot.polling.ApplicationBuilder"
+    ) as application_builder_class:
+        await async_setup_component(
+            hass,
+            DOMAIN,
+            config_polling,
+        )
+        await hass.async_block_till_done()
+        # Set up the integration with the polling platform inside the patch context manager.
+        application = (
+            application_builder_class.return_value.bot.return_value.build.return_value
+        )
+        # Then call the callback and assert events fired.
+        handler = application.add_handler.call_args[0][0]
+        handle_update_callback = handler.callback
 
-    # python-telegram-bots `Updater` uses threading, so we need to schedule its callback in a sync context.
-    await hass.async_add_executor_job(telegram_dispatcher_callback)
+        # Create Update object using library API.
+        application.bot.defaults.tzinfo = None
+        update = Update.de_json(update_message_text, application.bot)
+
+        # handle_update_callback == BaseTelegramBotEntity.update_handler
+        await handle_update_callback(update, None)
+
+        application.updater.stop = AsyncMock()
+        application.stop = AsyncMock()
+        application.shutdown = AsyncMock()
 
     # Make sure event has fired
     await hass.async_block_till_done()