Following Azure Devops pipelines live logs with signalr websocket endpoint

azdo_live_logs.py

import asyncio
import aiohttp
import os
import base64
from urllib.parse import urlencode, urlunparse
import requests


def fetch_auth_header(azdo_pat: str) -> str:
    pat = azdo_pat
    basic_auth_header = f"Basic {base64.b64encode(f':{pat}'.encode()).decode()}"
    return basic_auth_header

def fetch_connection_token(auth_header: str, account_id: str, organization: str, project_id: str) -> str:
    query_params = {
        'transport': 'webSockets',
        'contextToken': account_id,
    }
    negotiate_url = urlunparse((
        'https',
        'dev.azure.com',
        f'{organization}/_apis/{project_id}/signalr/negotiate',
        '',
        urlencode(query_params),
        ''
    ))
    print("Negotiating token...")
    headers = {
        'Authorization': auth_header
    }
    resp = requests.get(negotiate_url, headers=headers)
    conn_token = resp.json()['ConnectionToken']
    if resp.status_code != 200:
        raise Exception(f"Failed to negotiate connection token: {resp.text}")
    return conn_token

async def connect_to_signalr(auth_header: str, connection_token: str, account_id: str, organization: str, project_id: str, build_id: int):

    context_token = account_id

    query_params = {
        'transport': 'webSockets',
        'contextToken': context_token,
        'connectionToken': connection_token,
    }

    signalr_url = urlunparse((
        'wss',
        'dev.azure.com',
        f'_signalr/{organization}/_apis/{project_id}/signalr/connect',
        '',
        urlencode(query_params),
        ''
    ))

    print("Connecting to websocket...")
    headers = {
        'Authorization': auth_header
    }
    async with aiohttp.ClientSession() as session:
        async with session.ws_connect(signalr_url, headers=headers) as ws:
            print("Connected to WebSocket")
            await ws.send_json({
                "protocol": "json",
                "version": 1
            })
            print("Sent handshake message")

            async def send_message(ws, hub_name, method_name, args):
                message = {
                    "H": hub_name,
                    "M": method_name,
                    "A": args,
                    "I": 0  # Message ID, can be incremented if needed
                }
                await ws.send_json(message)
                print(f"Sent message: {message}")

            # send a message selecting specific build to watch
            await send_message(ws, "builddetailhub", "WatchBuild", [project_id, build_id])

            async for message in ws:
                if message.type == aiohttp.WSMsgType.TEXT:
                    data = message.json()
                    # Process the incoming message data here
                    print("Received message:", data)
                elif message.type == aiohttp.WSMsgType.ERROR:
                    print("WebSocket error:", message.data)
                    break

 # account id, here is how to get it: https://medium.com/@shivapatel1102001/get-list-of-organization-from-azure-devops-microsoft-account-861ea29dae93
account_id       = "00000000-0000-0000-0000-000000000000"
organization     = "contoso" # you organization name
project_id       = "00000000-0000-0000-0000-000000000000" # project id the build belongs to
build_id         =  919 # build id you would like to follow
auth_header      = fetch_auth_header(os.getenv('AZDO_PERSONAL_ACCESS_TOKEN'))
connection_token = fetch_connection_token(auth_header, account_id, organization, project_id)

loop = asyncio.get_event_loop()
loop.run_until_complete(connect_to_signalr(auth_header, connection_token, account_id, organization, project_id, build_id))