From 5c5311f9305d1acce88aa0e977b914bc4f42a744 Mon Sep 17 00:00:00 2001 From: avaksru Date: Tue, 17 Mar 2026 18:24:50 +0300 Subject: [PATCH] JSON format --- config/config.example.json | 2 + docs/channels/mqtt/README.md | 68 ++++++++++++++++++++- pkg/channels/mqtt/mqtt.go | 114 +++++++++++++---------------------- pkg/config/config.go | 2 + 4 files changed, 112 insertions(+), 74 deletions(-) diff --git a/config/config.example.json b/config/config.example.json index 51b19110d..4eae0b69e 100644 --- a/config/config.example.json +++ b/config/config.example.json @@ -272,7 +272,9 @@ "subscribe_topics": [ "picoclaw/input" ], + "subscribe_json_key": null, "reply_topic": "picoclaw/output", + "reply_json_key": null, "tls": false, "tls_ca": "", "tls_cert": "", diff --git a/docs/channels/mqtt/README.md b/docs/channels/mqtt/README.md index 370c44ffa..a54eeeeba 100644 --- a/docs/channels/mqtt/README.md +++ b/docs/channels/mqtt/README.md @@ -16,8 +16,12 @@ Add this to `config.json`: "qos": 1, "retain": false, "tls": false, - "subscribe_topics": ["picoclaw/chat"], - "reply_topic": "picoclaw/reply", + "subscribe_topics": [ + "picoclaw/input" + ], + "subscribe_json_key": null, + "reply_topic": "picoclaw/output", + "reply_json_key": null, "allow_from": [], "group_trigger": { "mention_only": true @@ -42,7 +46,9 @@ Add this to `config.json`: | 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 | +| subscribe_json_key | string | No | JSON key to extract from incoming messages. If null, treats message as plain text | | reply_topic | string | No | Topic to publish replies to. Supports placeholders: `{client_id}`, `{topic}` | +| reply_json_key | string | No | JSON key to use when sending replies. If null, sends as plain text | | 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 | @@ -54,6 +60,8 @@ Add this to `config.json`: - JSON format: `{"status": "your message"}` - Plain text: Direct text content - Automatic JSON parsing with fallback to plain text for malformed JSON +- **JSON Key Extraction**: When `subscribe_json_key` is set, extracts specific field from JSON messages +- **JSON Response Formatting**: When `reply_json_key` is set, sends responses as JSON with specified key - **Authentication**: Username/password authentication support - **TLS/SSL**: Secure connections with TLS configuration - **Quality of Service**: Configurable QoS levels (0, 1, 2) @@ -67,6 +75,7 @@ Add this to `config.json`: - **Robust Message Handling**: Intelligent parsing that handles malformed JSON gracefully - **Flexible Topic Configuration**: Support for multiple input topics and dynamic reply topics +- **JSON Message Processing**: Configurable JSON key extraction and response formatting - **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 @@ -77,4 +86,57 @@ Add this to `config.json`: - 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 \ No newline at end of file +- The instruction field allows adding context or commands to all incoming messages + +## 6. JSON Configuration Examples + +### Plain Text Mode (Default) +```json +{ + "subscribe_json_key": null, + "reply_json_key": null +} +``` +- Incoming messages are treated as plain text +- Outgoing messages are sent as plain text + +### JSON Input Mode +```json +{ + "subscribe_json_key": "message", + "reply_json_key": null +} +``` +- Incoming JSON: `{"message": "Hello world", "timestamp": 1234567890}` +- Extracted content: `"Hello world"` +- Outgoing messages are sent as plain text + +### JSON Output Mode +```json +{ + "subscribe_json_key": null, + "reply_json_key": "response" +} +``` +- Incoming messages are treated as plain text +- Outgoing JSON: `{"response": "Bot reply"}` + +### Full JSON Mode +```json +{ + "subscribe_json_key": "input", + "reply_json_key": "output" +} +``` +- Incoming JSON: `{"input": "What's the weather?", "location": "Moscow"}` +- Extracted content: `"What's the weather?"` +- Outgoing JSON: `{"output": "The weather is sunny"}` + +## 7. Troubleshooting + +### Common Issues + +1. **JSON parsing fails**: Ensure your JSON messages are valid +2. **Key not found**: Verify the JSON key exists in your messages +3. **Connection issues**: Check broker URL, credentials, and TLS settings +4. **Permission denied**: Verify client ID is in the `allow_from` list if configured diff --git a/pkg/channels/mqtt/mqtt.go b/pkg/channels/mqtt/mqtt.go index d6efe040d..77a131e8f 100644 --- a/pkg/channels/mqtt/mqtt.go +++ b/pkg/channels/mqtt/mqtt.go @@ -154,77 +154,40 @@ func (c *MQTTChannel) onMessage(client mqtt.Client, msg mqtt.Message) { "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 + // Check if subscribe_json_key is configured + if c.config.SubscribeJSONKey != nil && *c.config.SubscribeJSONKey != "" { + // Parse as JSON and extract the specified key + var jsonMsg map[string]interface{} + if err := json.Unmarshal(msg.Payload(), &jsonMsg); err == nil { + // Successfully parsed as JSON + if value, exists := jsonMsg[*c.config.SubscribeJSONKey]; exists { + content = fmt.Sprintf("%v", value) + logger.InfoCF("mqtt", "Extracted JSON value", map[string]any{ + "key": *c.config.SubscribeJSONKey, + "content": content, + }) } 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, - }) - } + logger.WarnCF("mqtt", "JSON key not found in message", map[string]any{ + "key": *c.config.SubscribeJSONKey, + }) + content = string(msg.Payload()) // Fall back to raw payload } } 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, + // JSON parsing failed, treat as plain text + logger.InfoCF("mqtt", "JSON parsing failed, treating as plain text", map[string]any{ + "error": err.Error(), + "payload": string(msg.Payload()), }) + content = string(msg.Payload()) } + } else { + // No JSON key configured, treat as plain text + content = string(msg.Payload()) + logger.InfoCF("mqtt", "Received plain text message", map[string]any{ + "payload": content, + }) } if content == "" { @@ -292,13 +255,22 @@ func (c *MQTTChannel) Send(ctx context.Context, msg bus.OutboundMessage) error { replyTopic = strings.ReplaceAll(replyTopic, "{topic}", msg.ChatID) // Add more placeholders as needed - mqttMsg := MQTTMessage{ - Status: msg.Content, - } + var payload []byte + var err error - payload, err := json.Marshal(mqttMsg) - if err != nil { - return fmt.Errorf("failed to marshal MQTT message: %w", err) + // Check if reply_json_key is configured + if c.config.ReplyJSONKey != nil && *c.config.ReplyJSONKey != "" { + // Send as JSON with the specified key + jsonMsg := map[string]string{ + *c.config.ReplyJSONKey: msg.Content, + } + payload, err = json.Marshal(jsonMsg) + if err != nil { + return fmt.Errorf("failed to marshal MQTT JSON message: %w", err) + } + } else { + // Send as plain text + payload = []byte(msg.Content) } token := c.client.Publish(replyTopic, byte(c.config.QoS), c.config.Retain, payload) @@ -311,4 +283,4 @@ func (c *MQTTChannel) Send(ctx context.Context, msg bus.OutboundMessage) error { "payload": string(payload), }) return nil -} \ No newline at end of file +} diff --git a/pkg/config/config.go b/pkg/config/config.go index 95b827a91..b4138eb45 100644 --- a/pkg/config/config.go +++ b/pkg/config/config.go @@ -600,7 +600,9 @@ type MQTTConfig struct { 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"` + SubscribeJSONKey *string `json:"subscribe_json_key,omitempty"` ReplyTopic string `json:"reply_topic" env:"PICOCLAW_CHANNELS_MQTT_REPLY_TOPIC"` + ReplyJSONKey *string `json:"reply_json_key,omitempty"` 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"`