44 *
55 * Thin stdio MCP server that proxies tool calls to a remote
66 * Cloudflare Worker + Durable Object backend via Streamable HTTP.
7- * Optionally listens to SSE for real-time channel notifications.
7+ * Optionally listens via WebSocket for real-time channel notifications
8+ * (enables DO hibernation on the Worker side).
89 * Authenticates via OAuth 2.1 with PKCE (localhost callback).
910 *
1011 * Discord MCP pattern: data lives in the cloud, local MCP is a thin bridge.
@@ -547,67 +548,75 @@ server.setRequestHandler(CallToolRequestSchema, async (req) => {
547548 }
548549} ) ;
549550
550- // ── SSE Listener → Channel Notifications ─────── ──────────────────────────────
551+ // ── WebSocket Listener → Channel Notifications ──────────────────────────────
551552
552553/** Track whether OAuth has been established (first successful tool call). */
553554let _oauthEstablished = false ;
554555
555556function markOAuthEstablished ( ) {
556557 if ( ! _oauthEstablished ) {
557558 _oauthEstablished = true ;
558- if ( CHANNEL_ENABLED && ! _sseConnected ) {
559- process . stderr . write ( "[github-webhook-mcp] OAuth established, starting SSE connection\n" ) ;
560- connectSSE ( ) ;
559+ if ( CHANNEL_ENABLED && ! _wsConnected ) {
560+ process . stderr . write ( "[github-webhook-mcp] OAuth established, starting WebSocket connection\n" ) ;
561+ connectWebSocket ( ) ;
561562 }
562563 }
563564}
564565
565- let _sseConnected = false ;
566+ let _wsConnected = false ;
566567
567- async function connectSSE ( ) {
568- let EventSourceImpl ;
569- try {
570- EventSourceImpl = ( await import ( "eventsource" ) ) . default ;
571- } catch {
572- // eventsource not installed — skip SSE
573- return ;
574- }
568+ async function connectWebSocket ( ) {
569+ const wsUrl = WORKER_URL . replace ( / ^ h t t p / , "ws" ) + "/events" ;
575570
576- _sseConnected = true ;
571+ _wsConnected = true ;
577572 let retryCount = 0 ;
578573 const MAX_RETRY_DELAY = 60_000 ; // 60 seconds
579574 const BASE_RETRY_DELAY = 1_000 ; // 1 second
580575
581- async function attemptConnection ( ) {
576+ async function connect ( ) {
582577 let token ;
583578 try {
584579 token = await getAccessToken ( ) ;
585580 } catch ( err ) {
586- process . stderr . write ( `[github-webhook-mcp] SSE : failed to get access token: ${ err } ) ;
581+ process . stderr . write ( `[github-webhook-mcp] WebSocket : failed to get access token: ${ err } ) ;
587582 scheduleRetry ( ) ;
588583 return ;
589584 }
590585
591586 if ( ! token ) {
592- process . stderr . write ( "[github-webhook-mcp] SSE : no access token available, will retry\n" ) ;
587+ process . stderr . write ( "[github-webhook-mcp] WebSocket : no access token available, will retry\n" ) ;
593588 scheduleRetry ( ) ;
594589 return ;
595590 }
596591
597- const sseUrl = `${ WORKER_URL } ;
598- const es = new EventSourceImpl ( sseUrl , {
599- headers : { Authorization : `Bearer ${ token } } ,
600- } ) ;
592+ let ws ;
593+ let pingTimer = null ;
594+
595+ try {
596+ ws = new WebSocket ( wsUrl , { headers : { Authorization : `Bearer ${ token } } } ) ;
597+ } catch ( err ) {
598+ process . stderr . write ( `[github-webhook-mcp] WebSocket: failed to create connection: ${ err } ) ;
599+ scheduleRetry ( ) ;
600+ return ;
601+ }
601602
602- es . onopen = ( ) => {
603+ ws . addEventListener ( "open" , ( ) => {
603604 retryCount = 0 ; // Reset backoff on successful connection
604- process . stderr . write ( "[github-webhook-mcp] SSE: connected\n" ) ;
605- } ;
605+ process . stderr . write ( "[github-webhook-mcp] WebSocket: connected\n" ) ;
606+ // Send periodic pings to keep connection alive (25s keepalive)
607+ pingTimer = setInterval ( ( ) => {
608+ if ( ws . readyState === WebSocket . OPEN ) {
609+ ws . send ( "ping" ) ;
610+ }
611+ } , 25_000 ) ;
612+ } ) ;
606613
607- es . onmessage = ( event ) => {
614+ ws . addEventListener ( "message" , ( event ) => {
608615 try {
609- const data = JSON . parse ( event . data ) ;
610- if ( "heartbeat" in data || "status" in data ) return ;
616+ const data = JSON . parse ( typeof event . data === "string" ? event . data : event . data . toString ( ) ) ;
617+
618+ // Skip status, pong, heartbeat messages
619+ if ( "status" in data || "pong" in data || "heartbeat" in data ) return ;
611620 if ( ! data . summary ) return ;
612621
613622 const s = data . summary ;
@@ -634,29 +643,36 @@ async function connectSSE() {
634643 } catch {
635644 // Ignore parse errors
636645 }
637- } ;
646+ } ) ;
638647
639- es . onerror = ( err ) => {
640- const status = err && err . status ;
641- if ( status === 401 ) {
642- process . stderr . write ( "[github-webhook-mcp] SSE: 401 unauthorized, closing and retrying with fresh token\n" ) ;
643- _cachedTokens = null ; // Force token refresh on next attempt
648+ ws . addEventListener ( "close" , ( event ) => {
649+ if ( pingTimer ) clearInterval ( pingTimer ) ;
650+ pingTimer = null ;
651+ const code = event . code ;
652+ if ( code === 1008 || code === 4401 ) {
653+ // Policy violation or unauthorized — refresh token
654+ process . stderr . write ( `[github-webhook-mcp] WebSocket: closed with code ${ code } ) ;
655+ _cachedTokens = null ;
644656 } else {
645- process . stderr . write ( `[github-webhook-mcp] SSE: connection error ${ status ? ` (status ${ status } )` : "" } ) ;
657+ process . stderr . write ( `[github-webhook-mcp] WebSocket: closed (code ${ code } ) \n` ) ;
646658 }
647- es . close ( ) ;
648659 scheduleRetry ( ) ;
649- } ;
660+ } ) ;
661+
662+ ws . addEventListener ( "error" , ( ) => {
663+ process . stderr . write ( "[github-webhook-mcp] WebSocket: connection error\n" ) ;
664+ // Will trigger close event, which handles reconnect
665+ } ) ;
650666 }
651667
652668 function scheduleRetry ( ) {
653669 const delay = Math . min ( BASE_RETRY_DELAY * Math . pow ( 2 , retryCount ) , MAX_RETRY_DELAY ) ;
654670 retryCount ++ ;
655- process . stderr . write ( `[github-webhook-mcp] SSE : retrying in ${ Math . round ( delay / 1000 ) } ${ retryCount } ) ;
656- setTimeout ( ( ) => attemptConnection ( ) , delay ) ;
671+ process . stderr . write ( `[github-webhook-mcp] WebSocket : retrying in ${ Math . round ( delay / 1000 ) } ${ retryCount } ) ;
672+ setTimeout ( ( ) => void connect ( ) , delay ) ;
657673 }
658674
659- attemptConnection ( ) ;
675+ await connect ( ) ;
660676}
661677
662678// ── Start ────────────────────────────────────────────────────────────────────
@@ -666,15 +682,15 @@ await server.connect(transport);
666682
667683if ( CHANNEL_ENABLED ) {
668684 // Check if tokens already exist from a previous session.
669- // If so, start SSE immediately. Otherwise, defer until after the first
685+ // If so, start WebSocket immediately. Otherwise, defer until after the first
670686 // successful tool call establishes OAuth.
671687 loadTokens ( ) . then ( ( tokens ) => {
672688 if ( tokens && tokens . access_token ) {
673689 _cachedTokens = tokens ;
674690 markOAuthEstablished ( ) ;
675691 }
676- // If no tokens, SSE will start after the first successful tool call
692+ // If no tokens, WebSocket will start after the first successful tool call
677693 } ) . catch ( ( ) => {
678- // Token load failed, SSE will start after first tool call
694+ // Token load failed, WebSocket will start after first tool call
679695 } ) ;
680696}
0 commit comments