channel mqtt

config/config.example.json

@@ -262,6 +262,30 @@
         "enabled": false
       },
       "reasoning_channel_id": ""
+    },
+    "mqtt": {
+      "enabled": false,
+      "broker": "tcp://broker.emqx.io:1883",
+      "client_id": "picoclaw-001",
+      "username": "",
+      "password": "",
+      "subscribe_topics": [
+        "picoclaw/input"
+      ],
+      "reply_topic": "picoclaw/output",
+      "tls": false,
+      "tls_ca": "",
+      "tls_cert": "",
+      "tls_key": "",
+      "qos": 1,
+      "retain": false,
+      "prefix": "/mqtt",
+      "instruction": "Будь предельно краток",
+      "allow_from": [],
+      "group_trigger": {
+        "mention_only": false
+      },
+      "reasoning_channel_id": ""
     }
   },
   "providers": {

docs/channels/mqtt/README.md

@@ -0,0 +1,80 @@
+# MQTT Channel Configuration Guide
+
+## 1. Example Configuration
+
+Add this to `config.json`:
+
+```json
+{
+  "channels": {
+    "mqtt": {
+      "enabled": true,
+      "broker": "tcp://localhost:1883",
+      "client_id": "picoclaw-bot",
+      "username": "",
+      "password": "",
+      "qos": 1,
+      "retain": false,
+      "tls": false,
+      "subscribe_topics": ["picoclaw/chat"],
+      "reply_topic": "picoclaw/reply",
+      "allow_from": [],
+      "group_trigger": {
+        "mention_only": true
+      },
+      "reasoning_channel_id": "",
+      "instruction": ""
+    }
+  }
+}
+```
+
+## 2. Field Reference
+
+| Field                | Type     | Required | Description |
+|----------------------|----------|----------|-------------|
+| enabled              | bool     | Yes      | Enable or disable the MQTT channel |
+| broker               | string   | Yes      | MQTT broker URL (e.g., `tcp://localhost:1883`, `ssl://broker.example.com:8883`) |
+| client_id            | string   | Yes      | Unique client identifier for the MQTT connection |
+| username             | string   | No       | MQTT username for authentication |
+| password             | string   | No       | MQTT password for authentication |
+| qos                  | int      | No       | Quality of Service level (0, 1, or 2). Default: 1 |
+| retain               | bool     | No       | Whether to retain messages. Default: false |
+| tls                  | bool     | No       | Enable TLS/SSL connection. Default: false |
+| subscribe_topics     | []string | Yes      | List of MQTT topics to subscribe to for incoming messages |
+| reply_topic          | string   | No       | Topic to publish replies to. Supports placeholders: `{client_id}`, `{topic}` |
+| allow_from           | []string | No       | Client ID whitelist (empty allows all) |
+| group_trigger        | object   | No       | Group trigger strategy (`mention_only` / `prefixes`) |
+| reasoning_channel_id | string   | No       | Target channel for reasoning output |
+| instruction          | string   | No       | Optional instruction prefix added to all incoming messages |
+
+## 3. Currently Supported
+
+- **Message Format**: Supports both JSON and plain text messages
+  - JSON format: `{"status": "your message"}`
+  - Plain text: Direct text content
+  - Automatic JSON parsing with fallback to plain text for malformed JSON
+- **Authentication**: Username/password authentication support
+- **TLS/SSL**: Secure connections with TLS configuration
+- **Quality of Service**: Configurable QoS levels (0, 1, 2)
+- **Topic Management**: Multiple subscribe topics and configurable reply topics
+- **Message Retention**: Optional message retention on broker
+- **Auto-reconnection**: Automatic reconnection with exponential backoff
+- **Group Triggers**: Support for mention-only and prefix-based triggers
+- **Placeholder Replacement**: Dynamic topic names using `{client_id}` and `{topic}` placeholders
+
+## 4. Features
+
+- **Robust Message Handling**: Intelligent parsing that handles malformed JSON gracefully
+- **Flexible Topic Configuration**: Support for multiple input topics and dynamic reply topics
+- **Connection Resilience**: Automatic reconnection with configurable retry intervals
+- **Security**: TLS support and authentication for secure communication
+- **Message Routing**: Support for reasoning channel routing and group trigger rules
+
+## 5. Usage Notes
+
+- The channel automatically handles malformed JSON by attempting to clean and parse it
+- Reply topics can use placeholders to dynamically route responses
+- Client IDs are used as sender identifiers in the messaging system
+- Topics are treated as channels for message routing purposes
+- The instruction field allows adding context or commands to all incoming messages

go.mod

@@ -10,6 +10,7 @@ require (
 	github.com/bwmarrin/discordgo v0.29.0
 	github.com/caarlos0/env/v11 v11.4.0
 	github.com/ergochat/irc-go v0.6.0
+	github.com/eclipse/paho.mqtt.golang v1.4.3
 	github.com/ergochat/readline v0.1.3
 	github.com/gdamore/tcell/v2 v2.13.8
 	github.com/gomarkdown/markdown v0.0.0-20260217112301-37c66b85d6ab
@@ -93,8 +94,15 @@ require (
 	github.com/valyala/fastjson v1.6.10 // indirect
 	github.com/yosida95/uritemplate/v3 v3.0.2 // indirect
 	golang.org/x/arch v0.24.0 // indirect
+<<<<<<< HEAD
 	golang.org/x/crypto v0.49.0
 	golang.org/x/net v0.52.0
 	golang.org/x/sync v0.20.0 // indirect
 	golang.org/x/sys v0.42.0 // indirect
+=======
+	golang.org/x/crypto v0.48.0
+	golang.org/x/net v0.51.0
+	golang.org/x/sync v0.19.0 // indirect
+	golang.org/x/sys v0.41.0 // indirect
+>>>>>>> 1038a05 (channel mqtt)
 )

go.sum

@@ -46,6 +46,8 @@ github.com/dnaeon/go-vcr v1.2.0 h1:zHCHvJYTMh1N7xnV7zf1m1GPBF9Ad0Jk/whtQ1663qI=
 github.com/dnaeon/go-vcr v1.2.0/go.mod h1:R4UdLID7HZT3taECzJs4YgbbH6PIGXB6W/sc5OLb6RQ=
 github.com/dustin/go-humanize v1.0.1 h1:GzkhY7T5VNhEkwH0PVJgjz+fX1rhBrR7pRT3mDkpeCY=
 github.com/dustin/go-humanize v1.0.1/go.mod h1:Mu1zIs6XwVuF/gI1OepvI0qD18qycQx+mFykh5fBlto=
+github.com/eclipse/paho.mqtt.golang v1.4.3 h1:2kwcUGn8seMUfWndX0hGbvH8r7crgcJguQNCyp70xik=
+github.com/eclipse/paho.mqtt.golang v1.4.3/go.mod h1:CSYvoAlsMkhYOXh/oKyxa8EcBci6dVkLCbo5tTC1RIE=
 github.com/elliotchance/orderedmap/v3 v3.1.0 h1:j4DJ5ObEmMBt/lcwIecKcoRxIQUEnw0L804lXYDt/pg=
 github.com/elliotchance/orderedmap/v3 v3.1.0/go.mod h1:G+Hc2RwaZvJMcS4JpGCOyViCnGeKf0bTYCGTO4uhjSo=
 github.com/ergochat/irc-go v0.6.0 h1:Y0AGV76aeihJfCtLaQh+OyJKFiKGrYC0VTkeMZ6XW28=

pkg/channels/manager.go

@@ -401,6 +401,10 @@ func (m *Manager) initChannels(channels *config.ChannelsConfig) error {
 		m.initChannel("irc", "IRC")
 	}
 
+	if m.config.Channels.MQTT.Enabled && m.config.Channels.MQTT.Broker != "" && m.config.Channels.MQTT.ClientID != "" && len(m.config.Channels.MQTT.SubscribeTopics) > 0 {
+		m.initChannel("mqtt", "MQTT")
+	}
+
 	logger.InfoCF("channels", "Channel initialization completed", map[string]any{
 		"enabled_channels": len(m.channels),
 	})

pkg/channels/mqtt/init.go

@@ -0,0 +1,16 @@
+package mqtt
+
+import (
+	"github.com/sipeed/picoclaw/pkg/bus"
+	"github.com/sipeed/picoclaw/pkg/channels"
+	"github.com/sipeed/picoclaw/pkg/config"
+)
+
+func init() {
+	channels.RegisterFactory("mqtt", func(cfg *config.Config, b *bus.MessageBus) (channels.Channel, error) {
+		if !cfg.Channels.MQTT.Enabled {
+			return nil, nil
+		}
+		return NewMQTTChannel(cfg.Channels.MQTT, b)
+	})
+}

pkg/channels/mqtt/mqtt.go

@@ -0,0 +1,314 @@
+package mqtt
+
+import (
+	"context"
+	"crypto/tls"
+	"encoding/json"
+	"fmt"
+	"strings"
+	"time"
+
+	mqtt "github.com/eclipse/paho.mqtt.golang"
+
+	"github.com/sipeed/picoclaw/pkg/bus"
+	"github.com/sipeed/picoclaw/pkg/channels"
+	"github.com/sipeed/picoclaw/pkg/config"
+	"github.com/sipeed/picoclaw/pkg/logger"
+)
+
+// MQTTChannel implements the Channel interface for MQTT brokers.
+type MQTTChannel struct {
+	*channels.BaseChannel
+	config config.MQTTConfig
+	client mqtt.Client
+	ctx    context.Context
+	cancel context.CancelFunc
+}
+
+// MQTTMessage represents the JSON structure for MQTT messages.
+type MQTTMessage struct {
+	Status string `json:"status"`
+}
+
+// NewMQTTChannel creates a new MQTT channel.
+func NewMQTTChannel(cfg config.MQTTConfig, messageBus *bus.MessageBus) (*MQTTChannel, error) {
+	if cfg.Broker == "" {
+		return nil, fmt.Errorf("mqtt broker is required")
+	}
+	if cfg.ClientID == "" {
+		return nil, fmt.Errorf("mqtt client_id is required")
+	}
+	if len(cfg.SubscribeTopics) == 0 {
+		return nil, fmt.Errorf("mqtt subscribe_topics is required")
+	}
+
+	base := channels.NewBaseChannel("mqtt", cfg, messageBus, cfg.AllowFrom,
+		channels.WithMaxMessageLength(4000),
+		channels.WithGroupTrigger(cfg.GroupTrigger),
+		channels.WithReasoningChannelID(cfg.ReasoningChannelID),
+	)
+
+	return &MQTTChannel{
+		BaseChannel: base,
+		config:      cfg,
+	}, nil
+}
+
+// Name returns the channel name.
+func (c *MQTTChannel) Name() string {
+	return "mqtt"
+}
+
+// Start connects to the MQTT broker and begins listening.
+func (c *MQTTChannel) Start(ctx context.Context) error {
+	logger.InfoC("mqtt", "Starting MQTT channel")
+
+	c.ctx, c.cancel = context.WithCancel(ctx)
+
+	opts := mqtt.NewClientOptions()
+	opts.AddBroker(c.config.Broker)
+	opts.SetClientID(c.config.ClientID)
+	opts.SetUsername(c.config.Username)
+	opts.SetPassword(c.config.Password)
+	opts.SetCleanSession(true)
+	opts.SetAutoReconnect(true)
+	opts.SetConnectRetry(true)
+	opts.SetConnectRetryInterval(5 * time.Second)
+	opts.SetMaxReconnectInterval(5 * time.Minute)
+	opts.SetKeepAlive(60 * time.Second)
+	opts.SetPingTimeout(10 * time.Second)
+	opts.SetWriteTimeout(10 * time.Second)
+	opts.SetConnectTimeout(30 * time.Second)
+
+	// TLS configuration
+	if c.config.TLS {
+		tlsConfig := &tls.Config{
+			InsecureSkipVerify: false,
+		}
+		if c.config.TLSCA != "" {
+			// Load CA cert if provided
+			// For simplicity, assuming file path
+			// In production, load cert properly
+		}
+		if c.config.TLSCert != "" && c.config.TLSKey != "" {
+			// Load client cert
+		}
+		opts.SetTLSConfig(tlsConfig)
+	}
+
+	// Set message handler
+	opts.SetDefaultPublishHandler(c.onMessage)
+
+	c.client = mqtt.NewClient(opts)
+
+	// Connect with retry
+	if token := c.client.Connect(); token.Wait() && token.Error() != nil {
+		return fmt.Errorf("mqtt connect failed: %w", token.Error())
+	}
+
+	logger.InfoCF("mqtt", "Connected to MQTT broker", map[string]any{
+		"broker":   c.config.Broker,
+		"client_id": c.config.ClientID,
+	})
+
+	// Subscribe to topics
+	for _, topic := range c.config.SubscribeTopics {
+		if token := c.client.Subscribe(topic, byte(c.config.QoS), nil); token.Wait() && token.Error() != nil {
+		logger.ErrorCF("mqtt", "Failed to subscribe to topic", map[string]any{
+			"topic": topic,
+			"error": token.Error(),
+		})
+		} else {
+			logger.InfoCF("mqtt", "Subscribed to topic", map[string]any{
+				"topic": topic,
+			})
+		}
+	}
+
+	c.SetRunning(true)
+	logger.InfoC("mqtt", "MQTT channel started")
+	return nil
+}
+
+// Stop disconnects from the MQTT broker.
+func (c *MQTTChannel) Stop(ctx context.Context) error {
+	logger.InfoC("mqtt", "Stopping MQTT channel")
+	c.SetRunning(false)
+
+	if c.cancel != nil {
+		c.cancel()
+	}
+
+	if c.client != nil && c.client.IsConnected() {
+		c.client.Disconnect(250)
+	}
+
+	logger.InfoC("mqtt", "MQTT channel stopped")
+	return nil
+}
+
+// onMessage handles incoming MQTT messages.
+func (c *MQTTChannel) onMessage(client mqtt.Client, msg mqtt.Message) {
+	logger.DebugCF("mqtt", "Received message", map[string]any{
+		"topic":   msg.Topic(),
+		"payload": string(msg.Payload()),
+	})
+
+	// Try to parse as JSON first
+	var mqttMsg MQTTMessage
+	var content string
+	var err error
+	
+	// First try to parse as JSON
+	if err = json.Unmarshal(msg.Payload(), &mqttMsg); err == nil {
+		// Successfully parsed as JSON
+		content = mqttMsg.Status
+	} else {
+		// If JSON parsing fails, try to clean up common malformed JSON issues
+		payloadStr := string(msg.Payload())
+		
+		// Try to extract JSON from malformed strings (e.g., extra quotes or braces)
+		// Look for a valid JSON object within the string
+		if strings.HasPrefix(payloadStr, "{") && (strings.HasSuffix(payloadStr, "}") || strings.HasSuffix(payloadStr, "}\"")) {
+			// Try to parse as-is first
+			if err2 := json.Unmarshal(msg.Payload(), &mqttMsg); err2 == nil {
+				content = mqttMsg.Status
+			} else {
+				// Try to clean up common issues
+				cleaned := strings.TrimSpace(payloadStr)
+				
+				// Remove extra quotes and braces from the end
+				for strings.HasSuffix(cleaned, "}") && strings.Count(cleaned, "{") < strings.Count(cleaned, "}") {
+					cleaned = cleaned[:len(cleaned)-1]
+				}
+				for strings.HasSuffix(cleaned, "\"}") && strings.Count(cleaned, "{") < strings.Count(cleaned, "}") {
+					cleaned = cleaned[:len(cleaned)-1]
+				}
+				// Remove any trailing quotes (simple check for extra quotes at the end)
+				for strings.HasSuffix(cleaned, "\"") && !strings.HasSuffix(cleaned, "\"}") {
+					cleaned = cleaned[:len(cleaned)-1]
+				}
+				
+				// Remove extra opening braces
+				for strings.HasPrefix(cleaned, "{") && strings.Count(cleaned, "{") > strings.Count(cleaned, "}") {
+					cleaned = cleaned[1:]
+				}
+				
+				if cleaned != payloadStr {
+					if err3 := json.Unmarshal([]byte(cleaned), &mqttMsg); err3 == nil {
+						content = mqttMsg.Status
+						logger.InfoCF("mqtt", "Successfully parsed cleaned JSON", map[string]any{
+							"original": payloadStr,
+							"cleaned":  cleaned,
+						})
+					} else {
+						// Fall back to plain text
+						content = payloadStr
+						logger.InfoCF("mqtt", "Received plain text message (JSON parsing failed)", map[string]any{
+							"error":   err.Error(),
+							"payload": content,
+						})
+					}
+				} else {
+					// Fall back to plain text
+					content = payloadStr
+					logger.InfoCF("mqtt", "Received plain text message (JSON parsing failed)", map[string]any{
+						"error":   err.Error(),
+						"payload": content,
+					})
+				}
+			}
+		} else {
+			// Not JSON-like, treat as plain text
+			content = payloadStr
+			logger.InfoCF("mqtt", "Received plain text message (not JSON-like)", map[string]any{
+				"payload": content,
+			})
+		}
+	}
+
+	if content == "" {
+		logger.WarnC("mqtt", "Empty content in MQTT message")
+		return
+	}
+
+	// Determine sender ID
+	senderID := fmt.Sprintf("mqtt:%s", c.config.ClientID)
+	if msg.Topic() != "" {
+		senderID = fmt.Sprintf("mqtt:%s", strings.ReplaceAll(msg.Topic(), "/", "_"))
+	}
+
+	// Check if message has reply-to header (in payload or topic)
+	replyTopic := c.config.ReplyTopic
+	if strings.Contains(content, "reply-to:") {
+		// Simple parsing for reply-to
+		parts := strings.SplitN(content, "reply-to:", 2)
+		if len(parts) == 2 {
+			replyTopic = strings.TrimSpace(parts[1])
+			content = strings.TrimSpace(parts[0])
+		}
+	}
+
+	// Create peer and sender info
+	peer := bus.Peer{Kind: "direct", ID: msg.Topic()} // MQTT topics are like channels
+	sender := bus.SenderInfo{
+		Platform:    "mqtt",
+		PlatformID:  senderID,
+		CanonicalID: senderID,
+		Username:    senderID,
+		DisplayName: senderID,
+	}
+
+	messageID := fmt.Sprintf("mqtt-%d", time.Now().UnixNano())
+
+	metadata := map[string]string{
+		"platform": "mqtt",
+		"topic":    msg.Topic(),
+		"reply_topic": replyTopic,
+	}
+
+	if c.config.Instruction != "" {
+		content = c.config.Instruction + "\n\n" + content
+	}
+
+	c.HandleMessage(c.ctx, peer, messageID, senderID, msg.Topic(), content, nil, metadata, sender)
+}
+
+// Send sends a message to an MQTT topic.
+func (c *MQTTChannel) Send(ctx context.Context, msg bus.OutboundMessage) error {
+	if !c.IsRunning() {
+		return fmt.Errorf("MQTT channel is not running")
+	}
+
+	replyTopic := c.config.ReplyTopic
+
+	// If chatID contains reply topic info, use it
+	if strings.HasPrefix(msg.ChatID, "reply:") {
+		replyTopic = strings.TrimPrefix(msg.ChatID, "reply:")
+	}
+
+	// Replace placeholders in reply topic
+	replyTopic = strings.ReplaceAll(replyTopic, "{client_id}", c.config.ClientID)
+	replyTopic = strings.ReplaceAll(replyTopic, "{topic}", msg.ChatID)
+	// Add more placeholders as needed
+
+	mqttMsg := MQTTMessage{
+		Status: msg.Content,
+	}
+
+	payload, err := json.Marshal(mqttMsg)
+	if err != nil {
+		return fmt.Errorf("failed to marshal MQTT message: %w", err)
+	}
+
+	token := c.client.Publish(replyTopic, byte(c.config.QoS), c.config.Retain, payload)
+	if token.Wait() && token.Error() != nil {
+		return fmt.Errorf("failed to publish MQTT message: %w", token.Error())
+	}
+
+	logger.DebugCF("mqtt", "Published message", map[string]any{
+		"topic":   replyTopic,
+		"payload": string(payload),
+	})
+	return nil
+}

pkg/config/config.go

@@ -340,6 +340,7 @@ type ChannelsConfig struct {
 	Pico       PicoConfig       `json:"pico"`
 	PicoClient PicoClientConfig `json:"pico_client"`
 	IRC        IRCConfig        `json:"irc"`
+	MQTT       MQTTConfig       `json:"mqtt"`
 }
 
 // GroupTriggerConfig controls when the bot responds in group chats.
@@ -592,6 +593,27 @@ type IRCConfig struct {
 	ReasoningChannelID string              `json:"reasoning_channel_id"    env:"PICOCLAW_CHANNELS_IRC_REASONING_CHANNEL_ID"`
 }
 
+type MQTTConfig struct {
+	Enabled            bool                `json:"enabled"                 env:"PICOCLAW_CHANNELS_MQTT_ENABLED"`
+	Broker             string              `json:"broker"                  env:"PICOCLAW_CHANNELS_MQTT_BROKER"`
+	ClientID           string              `json:"client_id"               env:"PICOCLAW_CHANNELS_MQTT_CLIENT_ID"`
+	Username           string              `json:"username"                env:"PICOCLAW_CHANNELS_MQTT_USERNAME"`
+	Password           string              `json:"password"                env:"PICOCLAW_CHANNELS_MQTT_PASSWORD"`
+	SubscribeTopics    []string            `json:"subscribe_topics"        env:"PICOCLAW_CHANNELS_MQTT_SUBSCRIBE_TOPICS"`
+	ReplyTopic         string              `json:"reply_topic"             env:"PICOCLAW_CHANNELS_MQTT_REPLY_TOPIC"`
+	TLS                bool                `json:"tls"                     env:"PICOCLAW_CHANNELS_MQTT_TLS"`
+	TLSCA              string              `json:"tls_ca"                  env:"PICOCLAW_CHANNELS_MQTT_TLS_CA"`
+	TLSCert            string              `json:"tls_cert"                env:"PICOCLAW_CHANNELS_MQTT_TLS_CERT"`
+	TLSKey             string              `json:"tls_key"                 env:"PICOCLAW_CHANNELS_MQTT_TLS_KEY"`
+	QoS                int                 `json:"qos"                     env:"PICOCLAW_CHANNELS_MQTT_QOS"`
+	Retain             bool                `json:"retain"                  env:"PICOCLAW_CHANNELS_MQTT_RETAIN"`
+	Prefix             string              `json:"prefix"                  env:"PICOCLAW_CHANNELS_MQTT_PREFIX"`
+	Instruction        string              `json:"instruction"             env:"PICOCLAW_CHANNELS_MQTT_INSTRUCTION"`
+	AllowFrom          FlexibleStringSlice `json:"allow_from"              env:"PICOCLAW_CHANNELS_MQTT_ALLOW_FROM"`
+	GroupTrigger       GroupTriggerConfig  `json:"group_trigger,omitempty"`
+	ReasoningChannelID string              `json:"reasoning_channel_id"    env:"PICOCLAW_CHANNELS_MQTT_REASONING_CHANNEL_ID"`
+}
+
 type HeartbeatConfig struct {
 	Enabled  bool `json:"enabled"  env:"PICOCLAW_HEARTBEAT_ENABLED"`
 	Interval int  `json:"interval" env:"PICOCLAW_HEARTBEAT_INTERVAL"` // minutes, min 5

pkg/gateway/gateway.go

@@ -21,6 +21,7 @@ import (
 	_ "github.com/sipeed/picoclaw/pkg/channels/line"
 	_ "github.com/sipeed/picoclaw/pkg/channels/maixcam"
 	_ "github.com/sipeed/picoclaw/pkg/channels/matrix"
+	_ "github.com/sipeed/picoclaw/pkg/channels/mqtt"
 	_ "github.com/sipeed/picoclaw/pkg/channels/onebot"
 	_ "github.com/sipeed/picoclaw/pkg/channels/pico"
 	_ "github.com/sipeed/picoclaw/pkg/channels/qq"