merged with main

kick/client.py

@@ -10,8 +10,8 @@
 from .http import HTTPClient
 from .livestream import PartialLivestream
 from .message import Message
-from .users import ClientUser, PartialUser, User
 from .categories import CategorySearch
+from .users import ClientUser, PartialUser, User, DestinationInfo
 from .utils import MISSING, decorator, setup_logging
 
 if TYPE_CHECKING:
@@ -224,6 +224,16 @@ async def search_categories(self, query: str, /) -> CategorySearch:
         -----------
         query: str
             The search query string
+        """
+        data = await self.http.search_categories(query)
+        return CategorySearch(data=data)
+
+    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
         -----------
@@ -234,10 +244,18 @@ async def search_categories(self, query: str, /) -> CategorySearch:
         -----------
         SearchResponse
             The search results containing matching categories
+
+            Fetching Failed
+        Forbidden
+            You are not authenticated
+
+        Returns
+        -----------
+        str
         """
 
-        data = await self.http.search_categories(query)
-        return CategorySearch(data=data)
+        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
     from .types.search import CategorySearchResponse
 
@@ -61,10 +66,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
 
@@ -271,7 +275,11 @@ async def request(self, route: Route, **kwargs) -> Any:
             try:
                 res = await self.__session.request(
                     route.method,
-                    final_url,
+                    (
+                        url
+                        if self.whitelisted is True
+                        else f"{self.bypass_host}:{self.bypass_port}/request?url={url}"
+                    ),
                     headers=headers,
                     cookies=cookies,
                     **kwargs,
@@ -532,6 +540,17 @@ def search_categories(self, query: str) -> Response[CategorySearchResponse]:
             params=params,
             _bypass_params=True  # Flag to prevent param duplication
         )
+
+    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"]):