PubNub MCP Server

Instructions

Implementation Reference

• index.js:1246-1349 (handler)

  The handler function that subscribes to the specified PubNub channel, sets up a message listener, collects the requested number of messages or times out, then unsubscribes and returns the received messages.

  toolHandlers['pubnub_subscribe_and_receive_messages'] = async ({ channel, messageCount = 1, timeout }) => {
    try {
      return new Promise((resolve, reject) => {
        let messagesReceived = [];
        let completed = false;
        let timeoutId;
  
        // Create subscription for the channel
        const channelEntity = pubnub.channel(channel);
        const subscription = channelEntity.subscription();
  
        // Set up message listener
        const messageListener = (messageEvent) => {
          if (!completed) {
            messagesReceived.push({
              channel: messageEvent.channel,
              message: messageEvent.message,
              publisher: messageEvent.publisher,
              timetoken: messageEvent.timetoken,
              subscription: messageEvent.subscription
            });
  
            // Check if we've received the desired number of messages
            if (messagesReceived.length >= messageCount) {
              completed = true;
              
              // Clean up
              if (timeoutId) {
                clearTimeout(timeoutId);
              }
              subscription.unsubscribe();
              subscription.removeListener(messageListener);
  
              // Return all received messages
              resolve({
                content: [
                  {
                    type: 'text',
                    text: JSON.stringify({
                      channel: channel,
                      messageCount: messagesReceived.length,
                      messages: messagesReceived
                    }, null, 2)
                  }
                ]
              });
            }
          }
        };
  
        // Add listener and subscribe
        subscription.addListener({ message: messageListener });
        subscription.subscribe();
  
        // Set timeout if specified
        if (timeout && timeout > 0) {
          timeoutId = setTimeout(() => {
            if (!completed) {
              completed = true;
              subscription.unsubscribe();
              subscription.removeListener(messageListener);
              
              if (messagesReceived.length > 0) {
                // Return partial results if some messages were received
                resolve({
                  content: [
                    {
                      type: 'text',
                      text: JSON.stringify({
                        channel: channel,
                        messageCount: messagesReceived.length,
                        messages: messagesReceived,
                        note: `Timeout: Only ${messagesReceived.length} of ${messageCount} requested messages received within ${timeout}ms`
                      }, null, 2)
                    }
                  ]
                });
              } else {
                // No messages received at all
                resolve({
                  content: [
                    {
                      type: 'text',
                      text: `Timeout: No messages received on channel '${channel}' within ${timeout}ms`
                    }
                  ]
                });
              }
            }
          }, timeout);
        }
      });
    } catch (err) {
      return {
        content: [
          {
            type: 'text',
            text: `Error subscribing to channel and receiving messages: ${err.message || err}`
          }
        ],
        isError: true
      };
    }
  };

• index.js:1352-1360 (schema)

  Tool metadata including name, description, and Zod parameter schema for input validation.

  toolDefinitions['pubnub_subscribe_and_receive_messages'] = {
    name: 'pubnub_subscribe_and_receive_messages',
    description: 'Subscribes to a PubNub channel and waits to receive a specified number of messages, then automatically unsubscribes. Call this tool when you need to listen for messages on a channel. Optionally specify a timeout in milliseconds to avoid waiting indefinitely.',
    parameters: {
      channel: z.string().describe('Name of the PubNub channel to subscribe to and receive messages from'),
      messageCount: z.number().optional().default(1).describe('Number of messages to wait for before unsubscribing (default: 1)'),
      timeout: z.number().optional().describe('Optional timeout in milliseconds. If not all messages are received within this time, the subscription will end (default: no timeout)')
    }
  };

• index.js:1363-1394 (registration)

  The registerAllTools function iterates over all tool definitions and registers each tool (including pubnub_subscribe_and_receive_messages) with the MCP server using server.tool(). Called for the main server and HTTP sessions.

  function registerAllTools(serverInstance, chatSdkMode = false) {
    // Tools to exclude when in chat SDK mode
    const chatSdkExcludedTools = [
      'read_pubnub_sdk_docs',
      'write_pubnub_app', 
      'read_pubnub_resources',
      'manage_pubnub_account'
    ];
  
    for (const toolName in toolDefinitions) {
      if (toolHandlers[toolName]) {
        // Skip excluded tools when in chat SDK mode
        if (chatSdkMode && chatSdkExcludedTools.includes(toolName)) {
          continue;
        }
  
        // Special handling for chat SDK docs tool
        if (toolName === 'read_pubnub_chat_sdk_docs' && 
            (chatSdkLanguages.length === 0 || chatSdkTopics.length === 0)) {
          continue; // Skip this tool if chat SDK data isn't loaded
        }
        
        const toolDef = toolDefinitions[toolName];
        serverInstance.tool(
          toolDef.name,
          toolDef.description,
          toolDef.parameters,
          wrapToolHandler(toolHandlers[toolName], toolName)
        );
      }
    }
  }