@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.

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

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)`

Register a tool that clients can call.

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?)`

Register a resource that clients can read.

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.

Name		Name	Last commit message	Last commit date
Latest commit History 25 Commits
.github/workflows		.github/workflows
examples		examples
src		src
test		test
.eslintrc.json		.eslintrc.json
.gitignore		.gitignore
.nvmrc		.nvmrc
.prettierrc.json		.prettierrc.json
LICENSE		LICENSE
README.md		README.md
jest.config.js		jest.config.js
package-lock.json		package-lock.json
package.json		package.json
tsconfig.json		tsconfig.json
tsconfig.prod.json		tsconfig.prod.json

License

emqx/mcp-typescript-sdk

Folders and files

Latest commit

History

Repository files navigation

@emqx-ai/mcp-mqtt-sdk

🎯 Live Demo

Features

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

About

Topics

Resources

License

Uh oh!

Stars

Watchers

Forks

Releases 4

Packages 0

Uh oh!

Languages

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

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

`start()` / `stop()`

`getTopics()`

Packages