Let's make a websocket client with Python. (Access token authentication)

pip install websocket-client

#!/usr/bin/env python
# -*- coding: utf-8 -*-

import os
import sys
import urllib
import urllib2
import json
import string

import websocket
import requests

class TypeTalkSample:
    access_token = None

    def __init__(self):
    	#Get an access token from typetalk
        client_id = "xxxxxxx"
        client_secret = "xxxxxxx"
        params = {
            'client_id': client_id,
            'client_secret': client_secret,
            'grant_type': 'client_credentials',
            'scope': 'topic.read,topic.post',
        }
        data = urllib.urlencode(params)
        res = urllib2.urlopen(urllib2.Request('https://typetalk.in/oauth2/access_token', data))
        
        #Hold access token
        self.access_token = json.load(res)['access_token']

		#Define websocket, here you can set websocket using access token by registering authorization in header.
        websocket.enableTrace(True)
        ws = websocket.WebSocketApp("wss://typetalk.in/api/v1/streaming", header=["Authorization: Bearer %s" % self.access_token], on_open=self.on_open, on_message=self.on_message, on_error=self.on_error, on_close=self.on_close)
        
        #Start websocket. Ctrl+Make sure it ends with C.
        try:
            ws.run_forever()
        except KeyboardInterrupt:
            ws.close()

	#The method defined here becomes the callback function of websocket.
	#When receiving a message
    def on_message(self, ws, message):
        print message

	#When an error occurs
    def on_error(self, ws, error):
        print error

	#When closing websocket
    def on_close(self, ws):
        print 'disconnected streaming server'

	#When opening websocket
    def on_open(self, ws):
        print 'connected streaming server'


if __name__ == "__main__":
    typetalk = TypeTalkSample()

Recommended Posts