Which integrations are available for this server?

Enables MCP communication over MQTT, using MQTT brokers as the transport layer for server discovery, tool calls, and resource access.

How do I use McpMqttServer?

1. Click on "Deploy Server". 2. Wait a few minutes for the server to deploy. Once ready, it will show a "Started" state. 3. In the chat, type @ followed by the MCP server name and your instructions, e.g., "@McpMqttServer discover MCP servers over MQTT and list their tools" That's it! The server will respond to your query, and you can continue using it as needed. Here is a step-by-step guide with screenshots.

McpMqttServer

by hCattaneo08048

Overview Schema Source Related Servers Discussions Admin

TypeScript

Remote

@emqx-ai/mcp-mqtt-sdk

A TypeScript SDK for implementing Model Context Protocol (MCP) over MQTT, supporting both browser and Node.js environments with full type safety and automatic environment detection.

🎯 Live Demo

See this SDK in action! Check out our MCP AI Companion Demo - a real-world implementation using both Python and TypeScript with this SDK to create an AI agent that can:

🎤 Control browser audio/video functionality
😊 Switch facial expressions and emotions
💬 Power interactive conversational experiences
🌐 Demonstrate cross-platform MCP communication

This demo showcases how to build sophisticated AI agents using MCP over MQTT for seamless browser control and interaction.

Related MCP server: yourco-mcp

Features

🚀 Universal: Works seamlessly in browser (WebSocket) and Node.js (TCP) environments
🔒 Type Safe: Full TypeScript support with Zod schema validation for MCP protocol
🌐 MQTT Transport: Uses MQTT as the transport layer for reliable MCP communication
🏗️ Constructor-Based: Clean object-oriented API with proper TypeScript classes
📋 Standards Compliant: Follows MCP specification v2024-11-05
🔧 Tool & Resource Support: Complete support for MCP tools and resources
🔍 Auto Discovery: Automatic server discovery over MQTT topics
🌍 Environment Detection: Automatic browser/Node.js detection with appropriate defaults

Requirements

Node.js >= 18

Installation

npm install @emqx-ai/mcp-mqtt-sdk

Quick Start

Creating an MCP Server

import { McpMqttServer } from '@emqx-ai/mcp-mqtt-sdk'

// Create server instance
const server = new McpMqttServer({
  // MQTT connection
  host: 'mqtt://localhost:1883',

  // Server identification
  serverId: 'unique-server-id-123',
  serverName: 'myapp/greeting-server',  // Hierarchical naming

  // Server information
  name: 'My MCP Server',
  version: '1.0.0',

  // Optional configuration
  description: 'A sample MCP server providing greeting tools',
  capabilities: {
    tools: { listChanged: true },
    resources: { listChanged: true, subscribe: false },
  },
})

// Add a tool
server.tool(
  'greet',
  'Greet someone with a personalized message',
  {
    type: 'object',
    properties: {
      name: {
        type: 'string',
        description: 'Name of the person to greet'
      },
      language: {
        type: 'string',
        enum: ['en', 'es', 'fr'],
        description: 'Language for greeting'
      }
    },
    required: ['name'],
  },
  async ({ name, language = 'en' }) => {
    const greetings = {
      en: `Hello, ${name}!`,
      es: `¡Hola, ${name}!`,
      fr: `Bonjour, ${name}!`
    }

    return {
      content: [{
        type: 'text',
        text: greetings[language] || greetings.en,
      }],
    }
  }
)

// Add a resource
server.resource(
  'config://app-settings',
  'Application Settings',
  async () => ({
    contents: [{
      uri: 'config://app-settings',
      mimeType: 'application/json',
      text: JSON.stringify({
        theme: 'dark',
        language: 'en',
        notifications: true
      }, null, 2),
    }],
  }),
  {
    description: 'Current application configuration',
    mimeType: 'application/json',
  }
)

// Event handlers
server.on('ready', () => {
  console.log('Server is ready!')
  console.log('Topics:', server.getTopics())
})

server.on('error', (error) => {
  console.error('Server error:', error)
})

// Start the server
await server.start()

Creating an MCP Client

import { McpMqttClient } from '@emqx-ai/mcp-mqtt-sdk'

// Create client instance
const client = new McpMqttClient({
  // MQTT connection
  host: 'mqtt://localhost:1883',

  // Client information
  name: 'My MCP Client',
  version: '1.0.0',
})

// Set up server discovery handler
client.on('serverDiscovered', async (server) => {
  console.log('📡 Discovered server:', server.name, '(ID:', server.serverId, ')')

  try {
    // Connect to the discovered server using serverId
    await client.initializeServer(server.serverId)
    console.log('✅ Connected to:', server.name)

    // List available tools using serverId
    const tools = await client.listTools(server.serverId)
    console.log('🔧 Available tools:', tools.map(t => t.name))

    // Call a tool using serverId
    if (tools.some(t => t.name === 'greet')) {
      const result = await client.callTool(server.serverId, 'greet', {
        name: 'World',
        language: 'es'
      })
      console.log('🎉 Tool result:', result.content[0]?.text)
    }

    // List and read resources using serverId
    const resources = await client.listResources(server.serverId)
    console.log('📚 Available resources:', resources.map(r => r.uri))

    if (resources.some(r => r.uri === 'config://app-settings')) {
      const config = await client.readResource(server.serverId, 'config://app-settings')
      console.log('📄 Config:', config.contents[0]?.text)
    }
  } catch (error) {
    console.error('❌ Error with server:', server.name, error)
  }
})

// Connect and start discovery
await client.connect()

// Graceful shutdown
process.on('SIGINT', async () => {
  await client.disconnect()
  process.exit(0)
})

API Reference

McpMqttServer

Constructor

new McpMqttServer(config: McpMqttServerConfig)

Configuration:

interface McpMqttServerConfig {
  // MQTT connection settings
  host: string
  username?: string
  password?: string

  // Server identification (required)
  serverId: string      // Unique server ID (MQTT client ID)
  serverName: string    // Hierarchical server name (e.g., "app/feature/server")

  // Server information (required)
  name: string
  version: string

  // Optional configuration
  description?: string     // Server description
  capabilities?: {
    prompts?: { listChanged?: boolean }
    resources?: { subscribe?: boolean; listChanged?: boolean }
    tools?: { listChanged?: boolean }
  }
  rbac?: {               // Optional role-based access control
    roles: Array<{
      name: string
      description: string
      allowed_methods: string[]
      allowed_tools: string[] | "all"
      allowed_resources: string[] | "all"
    }>
  }
}

Methods

`tool(name, description, inputSchema, handler)`

server.tool(
  'calculate',
  'Perform mathematical calculations',
  {
    type: 'object',
    properties: {
      expression: {
        type: 'string',
        description: 'Mathematical expression to evaluate'
      },
    },
    required: ['expression'],
  },
  async ({ expression }) => {
    try {
      // Safe evaluation - implement your own parser
      const result = evaluateExpression(expression)
      return {
        content: [{
          type: 'text',
          text: `${expression} = ${result}`,
        }],
      }
    } catch (error) {
      return {
        content: [{
          type: 'text',
          text: `Error: ${error.message}`,
        }],
        isError: true,
      }
    }
  }
)

`resource(uri, name, handler, options?)`

server.resource(
  'file://logs/app.log',
  'Application Logs',
  async () => {
    const logs = await readLogFile()
    return {
      contents: [{
        uri: 'file://logs/app.log',
        mimeType: 'text/plain',
        text: logs,
      }],
    }
  },
  {
    description: 'Current application log entries',
    mimeType: 'text/plain',
  }
)

`start()` / `stop()`

Control server lifecycle.

await server.start()  // Start listening for requests
await server.stop()   // Gracefully shutdown

`getTopics()`

Get MQTT topics used by this server.

const { request, response } = server.getTopics()

Events

server.on('ready', () => console.log('Server ready'))
server.on('error', (error) => console.error('Server error:', error))
server.on('closed', () => console.log('Server closed'))

McpMqttClient

Client Constructor

new McpMqttClient(config: McpMqttClientConfig)

Configuration:

interface McpMqttClientConfig {
  // MQTT connection settings
  host: string
  clientId?: string
  username?: string
  password?: string
  clean?: boolean
  keepalive?: number
  connectTimeout?: number
  reconnectPeriod?: number

  // Client information (required)
  name: string
  version: string

  // Optional configuration
  capabilities?: {
    roots?: { listChanged?: boolean }
    sampling?: Record<string, any>
  }

  // Advanced MQTT settings (optional)
  will?: {
    topic: string
    payload: string | Buffer
    qos?: 0 | 1 | 2
    retain?: boolean
  }
  properties?: Record<string, any>
}

Client Methods

Connection Management

await client.connect()           // Connect to MQTT broker and start discovery
await client.disconnect()        // Disconnect from broker
await client.initializeServer(serverId)  // Initialize connection to a specific server

Tool Operations

const tools = await client.listTools(serverId)
const result = await client.callTool(serverId, toolName, args)

Resource Operations

const resources = await client.listResources(serverId)
const data = await client.readResource(serverId, uri)

Discovery

const discovered = client.getDiscoveredServers()
const connected = client.getConnectedServers()

Client Events

client.on('serverDiscovered', (server) => {
  console.log('Found server:', server.name, 'ID:', server.serverId)
})

client.on('serverInitialized', (server) => {
  console.log('Connected to:', server.name)
})

client.on('serverDisconnected', (serverId) => {
  console.log('Server disconnected:', serverId)
})

client.on('connected', () => console.log('Client connected'))
client.on('disconnected', () => console.log('Client disconnected'))
client.on('error', (error) => console.error('Client error:', error))

MQTT Configuration

The SDK supports comprehensive MQTT connection options:

interface MqttConnectionOptions {
  host: string               // Complete connection URL (e.g., ws://localhost:8083, wss://broker.emqx.io:8084)
  clientId?: string          // Auto-generated if not provided
  username?: string
  password?: string
  clean?: boolean            // Clean session (default: true)
  keepalive?: number         // Keep-alive interval in seconds
  connectTimeout?: number    // Connection timeout in milliseconds
  reconnectPeriod?: number   // Reconnection period in milliseconds
  will?: {                   // Last Will Testament
    topic: string
    payload: string | Buffer
    qos?: 0 | 1 | 2
    retain?: boolean
  };
}

Connection Examples

The host parameter should be a complete connection URL:

// WebSocket connections (browser)
host: 'ws://localhost:8083'
host: 'wss://broker.emqx.io:8084'

// TCP connections (Node.js)
host: 'mqtt://localhost:1883'
host: 'mqtts://broker.emqx.io:8883'

Advanced Usage

Accessing Underlying MQTT Client

Both server and client provide getMqttClient() method to access the underlying MQTT client for custom pub/sub operations:

const mqttClient = server.getMqttClient() // or client.getMqttClient()
if (mqttClient) {
  mqttClient.subscribe('custom/topic', { qos: 1 })
  mqttClient.publish('custom/topic', 'Hello World')
}

Custom Tool Validation

import { z } from 'zod'

// Define schema for tool parameters
const CalculateSchema = z.object({
  operation: z.enum(['add', 'subtract', 'multiply', 'divide']),
  a: z.number(),
  b: z.number(),
})

server.tool(
  'calculate',
  'Perform arithmetic operations',
  {
    type: 'object',
    properties: {
      operation: { type: 'string', enum: ['add', 'subtract', 'multiply', 'divide'] },
      a: { type: 'number' },
      b: { type: 'number' },
    },
    required: ['operation', 'a', 'b'],
  },
  async (params) => {
    // Validate with Zod
    const { operation, a, b } = CalculateSchema.parse(params)

    let result: number
    switch (operation) {
      case 'add': result = a + b; break
      case 'subtract': result = a - b; break
      case 'multiply': result = a * b; break
      case 'divide':
        if (b === 0) throw new Error('Division by zero')
        result = a / b
        break
    }

    return {
      content: [{
        type: 'text',
        text: `${a} ${operation} ${b} = ${result}`,
      }],
    }
  }
)

Error Handling

// Server-side error handling
server.tool('risky-operation', 'An operation that might fail', schema, async (params) => {
  try {
    const result = await performRiskyOperation(params)
    return {
      content: [{ type: 'text', text: `Success: ${result}` }],
    }
  } catch (error) {
    return {
      content: [{ type: 'text', text: `Operation failed: ${error.message}` }],
      isError: true, // Mark as error response
    }
  }
})

// Client-side error handling
try {
  const result = await client.callTool(serverName, 'risky-operation', params)
  if (result.isError) {
    console.error('Tool returned error:', result.content[0]?.text)
  } else {
    console.log('Success:', result.content[0]?.text)
  }
} catch (error) {
  console.error('Tool call failed:', error.message)
}

Resource Streaming

// Server: Streaming resource
server.resource(
  'stream://live-data',
  'Live Data Stream',
  async () => {
    const chunks = await getLiveDataChunks()
    return {
      contents: chunks.map((chunk, index) => ({
        uri: `stream://live-data#${index}`,
        mimeType: 'application/json',
        text: JSON.stringify(chunk),
      })),
    }
  }
)

// Client: Read streaming resource
const stream = await client.readResource(serverName, 'stream://live-data')
for (const content of stream.contents) {
  const data = JSON.parse(content.text!)
  processStreamChunk(data)
}

Development

# Install dependencies
npm install

# Build
npm run build

# Build in watch mode
npm run build:watch

# Run tests
npm run test

# Type checking
npm run typecheck

# Linting
npm run lint
npm run lint:fix

Protocol Details

MQTT Topic Structure

The SDK follows the official MCP over MQTT specification topic hierarchy:

MCP over MQTT Topic Structure:

🗂️ Server Topics:
├── $mcp-server/{server-id}/{server-name}              # Control topic (initialization)
├── $mcp-server/capability/{server-id}/{server-name}   # Capability change notifications
└── $mcp-server/presence/{server-id}/{server-name}     # Server presence (online/offline)

🗂️ Client Topics:
├── $mcp-client/capability/{mcp-client-id}             # Client capability changes
└── $mcp-client/presence/{mcp-client-id}               # Client presence

🗂️ RPC Communication:
└── $mcp-rpc/{mcp-client-id}/{server-id}/{server-name} # Bidirectional RPC communication

Example Topics:
- Control: $mcp-server/server-123/myapp/greeting-server
- Capability: $mcp-server/capability/server-123/myapp/greeting-server
- Presence: $mcp-server/presence/server-123/myapp/greeting-server
- RPC: $mcp-rpc/client-456/server-123/myapp/greeting-server

Message Flow

Service Discovery: Client subscribes to $mcp-server/presence/+/#
Server Registration: Server publishes presence to $mcp-server/presence/{server-id}/{server-name}
Initialization: Client sends initialize request to $mcp-server/{server-id}/{server-name}
RPC Communication: Bidirectional communication via $mcp-rpc/{client-id}/{server-id}/{server-name}

Error Codes

The SDK follows JSON-RPC 2.0 error codes:

-32700: Parse error
-32600: Invalid request
-32601: Method not found
-32602: Invalid params
-32603: Internal error
-32000 to -32099: Implementation-defined server errors

Other Language SDKs

Looking for MCP over MQTT support in other languages?

Python SDK - MCP over MQTT implementation for Python
Erlang SDK - MCP over MQTT implementation for Erlang
Paho MCP over MQTT - MCP over MQTT implementation in C
ESP MCP over MQTT - MCP over MQTT for ESP32/embedded devices in C

Contributing

Fork the repository
Create a feature branch (git checkout -b feature/amazing-feature)
Make your changes with tests
Run the test suite (npm test)
Submit a pull request

License

Apache License 2.0. See LICENSE file for details.

This server cannot be deployed

Shareable, User-Owned Context BackendPeople, applications, and agents read, write, and coordinate together in one context lake. Reach it over MCP, a CLI, a Python SDK, or REST.fulcradynamics.comAd

Maintenance

ActivityActive

ResponsivenessNo issues

Resources

Need Help?

Related Servers

Related MCP Connectors

AgentsJunction B
io.github.JakubTrousil
Discover MCP servers and A2A agents; verify, message, post, follow, react, and receive webhooks.
Agent Orchestration Search Knowledge & Memory
291
DataGroutOAuth
ai.datagrout
Discover, plan and run tools across your connected business systems through one OAuth MCP endpoint
App Automation Autonomous Agents
1
Tempr MCP ConnectorOAuth
io.temprhq
One OAuth connector for your team's remote MCP servers, with Tool Packs, guardrails and logs.
Security Observability Developer Tools
1
OmegaAIOAuth
com.queryomega
Query, browse, and automate OmegaAI workspaces from any MCP client. Streamable HTTP with OAuth 2.0.
Workplace & Productivity App Automation
2

Related MCP Servers

MQTTX SSE Server
Home Automation & IoT Communication
ysfscream
A
license
Not graded
quality
D
maintenance
Enables MQTT broker operations like connect, subscribe, and publish through MCP over SSE transport.
Updated 2 years ago (2025-03-20 05:55 UTC)
5
MIT
yourco-mcp
Developer Tools Security
amitmohapatra
F
license
Not graded
quality
B
maintenance
Enables developers to build MCP servers with registry-managed tool metadata, runtime hot-reloading, pluggable authentication and authorization, per-audience tool views, and resilient stateless operation.
Updated a month ago (2026-09-21 08:11 UTC)
-
Attenza
Autonomous Agents App Automation
maxhillaert
A
license
Not graded
quality
B
maintenance
Provides OAuth-based MCP integration and client-side tooling for connecting agents to a remote service, with CLI and plugin support.
Updated a month ago (2026-09-13 22:42 UTC)
MIT
McpHub
Agent Orchestration Developer Tools
John0King
A
license
Not graded
quality
A
maintenance
Enables AI tools to uniformly discover, inspect, and call tools, prompts, and resources from multiple upstream MCP servers through a small set of fixed MCP tools, over stdio or HTTP.
Updated 2 months ago (2026-08-10 11:40 UTC)
2
MIT

@emqx-ai/mcp-mqtt-sdk

🎯 Live Demo

Features

Requirements

Installation

Quick Start

Creating an MCP Server

Creating an MCP Client

API Reference

McpMqttServer

Constructor

Methods

tool(name, description, inputSchema, handler)

resource(uri, name, handler, options?)

start() / stop()

getTopics()

Events

McpMqttClient

Client Constructor

Client Methods

Connection Management

Tool Operations

Resource Operations

Discovery

Client Events

MQTT Configuration

Connection Examples

Advanced Usage

Accessing Underlying MQTT Client

Custom Tool Validation

Error Handling

Resource Streaming

Development

Protocol Details

MQTT Topic Structure

Message Flow

Error Codes

Other Language SDKs

Contributing

License

Maintenance

Resources

Related MCP Connectors

Related MCP Servers

MQTTX SSE Server

yourco-mcp

Attenza

McpHub

`tool(name, description, inputSchema, handler)`

`resource(uri, name, handler, options?)`

`start()` / `stop()`

`getTopics()`