pythonwebsocketslack-api

How to send Slack api request to websocket without Bolt or other SDK?


I have the code below. It connects correctly to Slack, and authenticate correctly.

I want to sent a conversations.list request to list the channels. How can I do that without using Bolt or some other SDK? The system where this runs is locked down, so I'm not getting to install anything further than websockets.

I think the only missing part is figuring what to send on the websocket to request a channel list. Currently it outputs:

Opened connection
{"type":"hello","num_connections":1,"debug_info":{"host":"applink-11","build_number":105,"approximate_connection_time":18060},"connection_info":{"app_id":"Redacted"}}

The API I'm after is https://api.slack.com/methods/conversations.list

Code is

#!/usr/bin/env python2.6

import httplib
import requests
import websocket
import argparse

def on_message(ws, message):
    print(message)
    ws.send("list")

def on_error(ws, error):
    print(error)

def on_close(ws, close_status_code, close_msg):
    print("### closed ###")

def on_open(ws):
    print("Opened connection")

def run_with(url):
# websocket.enableTrace(True)
    ws = websocket.WebSocketApp(url,
    on_open=on_open,
    on_message=on_message,
    on_error=on_error,
    on_close=on_close)

    ws.run_forever()

def main():

    parser = argparse.ArgumentParser()
    parser.add_argument("token")
    args = parser.parse_args()

    url = 'https://slack.com/api/apps.connections.open'
    headers = {
'content-type': 'application/x-www-form-urlencoded',
'Authorization': 'Bearer ' + args.token}
    r = requests.post(url, headers=headers)

    if r.json()['ok']:
        url = r.json()['url']
        run_with(url)
    else:
        print(r.content)

if __name__ == '__main__':
    main()

Solution

  • You can’t get the channel list through the same WebSocket. Slack needs a normal HTTP request for conversations.list.

    Usually, you'd use WebSocket for real-time events, but you can open the socket and still use normal HTTP calls to list the channels.

    
    
    import requests
    import websocket
    import argparse
    
    def on_message(ws, message):
        print("Got message:", message)
    
    def on_error(ws, error):
        print("Error:", error)
    
    def on_close(ws, close_status_code, close_msg):
        print("Socket closed")
    
    def on_open(ws):
        print("Socket opened")
    
    def run_socket(url):
        ws = websocket.WebSocketApp(
            url,
            on_open=on_open,
            on_message=on_message,
            on_error=on_error,
            on_close=on_close
        )
        ws.run_forever()
    
    def list_channels(token):
        url = "https://slack.com/api/conversations.list"
        headers = {
            "Authorization": "Bearer " + token,
            "Content-Type": "application/x-www-form-urlencoded"
        }
        data = {
            "limit": "100"  # You can change this if you want more channels
        }
        response = requests.post(url, headers=headers, data=data)
        return response.json()
    
    def main():
        parser = argparse.ArgumentParser()
        parser.add_argument("token")
        args = parser.parse_args()
        token = args.token
    
        # 1. Open the Slack socket URL
        open_url = "https://slack.com/api/apps.connections.open"
        headers = {
            "Authorization": "Bearer " + token,
            "Content-Type": "application/x-www-form-urlencoded"
        }
        r = requests.post(open_url, headers=headers)
        data = r.json()
    
        # 2. If the socket opened okay, list channels using HTTP
        if data.get("ok"):
            channels = list_channels(token)
            print("Channels:", channels)
    
            # 3. Then connect to the socket
            socket_url = data["url"]
            run_socket(socket_url)
        else:
            print("Could not open socket:", r.text)
    
    if __name__ == "__main__":
        main()