Message provider

Message - Read and send messages

TODO: Support group messaging

pythonxbox.api.provider.message.MESSAGE_MAX_LEN = 256 module-attribute

pythonxbox.api.provider.message.MessageProvider(client)

Bases: BaseProvider

Source code in src/pythonxbox/api/provider/baseprovider.py

def __init__(self, client: "XboxLiveClient") -> None:
    """
    Initialize an the BaseProvider

    Args:
        client (:class:`XboxLiveClient`): Instance of XboxLiveClient
    """
    self.client = client

MSG_URL = 'https://xblmessaging.xboxlive.com' class-attribute instance-attribute

HEADERS_MESSAGE = {'x-xbl-contract-version': '1'} class-attribute instance-attribute

HEADERS_HORIZON = {'x-xbl-contract-version': '2'} class-attribute instance-attribute

get_inbox(max_items=100, **kwargs) async

Get messages

Returns:

Type	Description
InboxResponse	class:InboxResponse: Inbox Response

Source code in src/pythonxbox/api/provider/message/__init__.py

async def get_inbox(self, max_items: int = 100, **kwargs) -> InboxResponse:
    """
    Get messages

    Returns:
        :class:`InboxResponse`: Inbox Response
    """
    url = f"{self.MSG_URL}/network/Xbox/users/me/inbox"
    params = {"maxItems": max_items}
    resp = await self.client.session.get(
        url, params=params, headers=self.HEADERS_MESSAGE, **kwargs
    )
    resp.raise_for_status()
    return InboxResponse.model_validate_json(resp.text)

get_conversation(xuid, max_items=100, **kwargs) async

Get detailed conversation info

Parameters:

Name	Type	Description	Default
xuid	str	Xuid of user having a conversation with	required

Returns:

Type	Description
ConversationResponse	class:ConversationResponse: Conversation Response

Source code in src/pythonxbox/api/provider/message/__init__.py

async def get_conversation(
    self, xuid: str, max_items: int = 100, **kwargs
) -> ConversationResponse:
    """
    Get detailed conversation info

    Args:
        xuid: Xuid of user having a conversation with

    Returns:
        :class:`ConversationResponse`: Conversation Response
    """
    url = f"{self.MSG_URL}/network/Xbox/users/me/conversations/users/xuid({xuid})"
    params = {"maxItems": max_items}
    resp = await self.client.session.get(
        url, params=params, headers=self.HEADERS_MESSAGE, **kwargs
    )
    resp.raise_for_status()
    return ConversationResponse.model_validate_json(resp.text)

delete_conversation(conversation_id, horizon, **kwargs) async

Delete message

NOTE: Returns HTTP Status Code 200 on success

Parameters:

Name	Type	Description	Default
conversation_id	str	Message Id	required
horizon	str	Delete horizon from get conversation response	required

Returns: True on success, False otherwise

Source code in src/pythonxbox/api/provider/message/__init__.py

async def delete_conversation(
    self, conversation_id: str, horizon: str, **kwargs
) -> bool:
    """
    Delete message

    **NOTE**: Returns HTTP Status Code **200** on success

    Args:
        conversation_id: Message Id
        horizon: Delete horizon from get conversation response

    Returns: True on success, False otherwise
    """
    url = f"{self.MSG_URL}/network/Xbox/users/me/conversations/horizon"
    post_data = {
        "conversations": [
            {
                "conversationId": conversation_id,
                "conversationType": "OneToOne",
                "horizonType": "Delete",
                "horizon": horizon,
            }
        ]
    }
    resp = await self.client.session.put(
        url, json=post_data, headers=self.HEADERS_HORIZON, **kwargs
    )
    return resp.status_code == HTTPStatus.OK

delete_message(conversation_id, message_id, **kwargs) async

Delete message

NOTE: Returns HTTP Status Code 200 on success

Parameters:

Name	Type	Description	Default
conversation_id	str	Conversation Id	required
message_id	str	Message Id	required

Returns: True on success, False otherwise

Source code in src/pythonxbox/api/provider/message/__init__.py

async def delete_message(
    self, conversation_id: str, message_id: str, **kwargs
) -> bool:
    """
    Delete message

    **NOTE**: Returns HTTP Status Code **200** on success

    Args:
        conversation_id: Conversation Id
        message_id: Message Id

    Returns: True on success, False otherwise
    """
    url = f"{self.MSG_URL}/network/Xbox/users/me/conversations/{conversation_id}/messages/{message_id}"
    resp = await self.client.session.delete(
        url, headers=self.HEADERS_MESSAGE, **kwargs
    )
    return resp.status_code == HTTPStatus.OK

send_message(xuid, message_text, **kwargs) async

Send message to an xuid

Parameters:

Name	Type	Description	Default
xuid	str	Xuid	required
message_text	str	Message text	required

Returns:

Type	Description
SendMessageResponse	class:SendMessageResponse: Send Message Response

Source code in src/pythonxbox/api/provider/message/__init__.py

async def send_message(
    self, xuid: str, message_text: str, **kwargs
) -> SendMessageResponse:
    """
    Send message to an xuid

    Args:
        xuid: Xuid
        message_text: Message text

    Returns:
        :class:`SendMessageResponse`: Send Message Response
    """
    if len(message_text) > MESSAGE_MAX_LEN:
        raise ValueError(
            f"Message text exceeds max length of {MESSAGE_MAX_LEN} chars"
        )

    url = f"{self.MSG_URL}/network/Xbox/users/me/conversations/users/xuid({xuid})"
    post_data = {
        "parts": [
            {
                "contentType": "text",
                "version": 0,
                "text": message_text,
            }
        ]
    }
    resp = await self.client.session.post(
        url, json=post_data, headers=self.HEADERS_MESSAGE, **kwargs
    )
    resp.raise_for_status()
    return SendMessageResponse.model_validate_json(resp.text)