Added functionality to get stream destination info (stream url and key)

Author: gub-7

kick/client.py

@@ -10,7 +10,7 @@
 from .http import HTTPClient
 from .livestream import PartialLivestream
 from .message import Message
-from .users import ClientUser, PartialUser, User
+from .users import ClientUser, PartialUser, User, DestinationInfo
 from .utils import MISSING, decorator, setup_logging
 
 if TYPE_CHECKING:
@@ -213,6 +213,28 @@ async def fetch_user(self, name: str, /) -> User:
         user = User(data=data, http=self.http)
         return user
 
+    async def fetch_stream_url_and_key(self) -> DestinationInfo:
+        """
+        |coro|
+
+        Fetches your stream URL and stream key from the API.
+        You must be authenticated to use this endpoint.
+
+        Raises
+        -----------
+        HTTPException
+            Fetching Failed
+        Forbidden
+            You are not authenticated
+
+        Returns
+        -----------
+        str
+        """
+
+        data = await self.http.get_stream_destination_url_and_key()
+        return DestinationInfo(data=data)
+
     def dispatch(self, event_name: str, *args, **kwargs) -> None:
         event_name = f"on_{event_name}"
 

kick/http.py

@@ -43,7 +43,12 @@
         ReplyOriginalSender,
         V1MessageSentPayload,
     )
-    from .types.user import ChatterPayload, ClientUserPayload, UserPayload
+    from .types.user import (
+        ChatterPayload,
+        ClientUserPayload,
+        UserPayload,
+        DestinationInfoPayload,
+    )
     from .types.videos import GetVideosPayload
 
     T = TypeVar("T")
@@ -60,10 +65,9 @@
 async def json_or_text(response: ClientResponse, /) -> Union[dict[str, Any], str]:
     text = await response.text()
     try:
-        try:
-            return json.loads(text)
-        except json.JSONDecodeError:
-            pass
+        return json.loads(text)
+    except json.JSONDecodeError:
+        pass
     except KeyError:
         pass
 
@@ -238,9 +242,11 @@ async def request(self, route: Route, **kwargs) -> Any:
             try:
                 res = await self.__session.request(
                     route.method,
-                    url
-                    if self.whitelisted is True
-                    else f"{self.bypass_host}:{self.bypass_port}/request?url={url}",
+                    (
+                        url
+                        if self.whitelisted is True
+                        else f"{self.bypass_host}:{self.bypass_port}/request?url={url}"
+                    ),
                     headers=headers,
                     cookies=cookies,
                     **kwargs,
@@ -488,6 +494,16 @@ def reply_to_message(
     def get_me(self) -> Response[ClientUserPayload]:
         return self.request(Route.root("GET", "/api/v1/user"))
 
+    def get_stream_destination_url_and_key(self) -> Response[DestinationInfoPayload]:
+        """Gets the authenticated user's stream URL and key.
+
+        Returns
+        -------
+        StreamURLKeyPayload
+            The stream URL and key information containing the publish URL and token
+        """
+        return self.request(Route.root("GET", "/stream/publish_token"))
+
     async def get_asset(self, url: str) -> bytes:
         if self.__session is MISSING:
             self.__session = ClientSession()

kick/types/user.py

@@ -140,3 +140,8 @@ class ClientUserPayload(TypedDict):
     streamer_channel: ClientUserStreamerChannelsPayload
     roles: list  # Unknown
     profilepic: str | None
+
+
+class DestinationInfoPayload(TypedDict):
+    rtmp_publish_path: str
+    rtmp_stream_token: str

kick/users.py

@@ -16,9 +16,34 @@
 if TYPE_CHECKING:
     from .chatroom import Chatroom
     from .http import HTTPClient
-    from .types.user import ClientUserPayload, InnerUser, UserPayload
+    from .types.user import ClientUserPayload, InnerUser, UserPayload, DestinationInfoPayload, StreamInfoPayload
+
+__all__ = ("DestinationInfo", "StreamInfo", "Socials", "PartialUser", "User", "ClientUser")
+
+
+class DestinationInfo(BaseDataclass["DestinationInfoPayload"]):
+    """
+    Information about a user's stream destination
+
+    Attributes
+    -----------
+    stream_url: str
+        The URL for streaming
+    stream_key: str
+        The stream key
+    """
+    
+    @property
+    def stream_url(self) -> str:
+        """The URL for streaming"""
+        return self._data["rtmp_publish_path"]
+    
+    @property
+    def stream_key(self) -> str:
+        """The stream key"""
+        return self._data["rtmp_stream_token"]
+
 
-__all__ = ("User", "Socials", "PartialUser", "ClientUser")
 
 
 class Socials(BaseDataclass["InnerUser | ClientUserPayload"]):