Rivian Slack/NATS Bridge

The Adventure

Activate deezwatts and engineering technical writer and please supply me with a way to implement communications with my Rivian (deezwatts), nats, and slack, knowing that i have an onboard raspberry pi and its connected to my tailnet. Provide instructions for setup under "The Implementation" in rivian-slack-nats.md. Also include a way I can test it in "the attestation" section.

The ultimate goal of this workload is to bridge the “dial tone” of the vehicle (NATS) with the collaboration tools I use every day (Slack). By leveraging an onboard Raspberry Pi 5 (“The Cheat Code”) connected to my Tailnet, I can receive real-time vehicle alerts—like Gear Guard triggers, climate updates, or low battery warnings—directly in a private Slack channel, regardless of where the truck is currently exploring.

The Implementation

The architecture consists of a persistent Python bridge running on the Raspberry Pi. This bridge acts as an intelligent router between the NATS message bus and the Shttps://www.deezwatts.com/architectures/deezwatts-agentic-ai/lack API.

1. Prerequisites

• Tailscale: The Raspberry Pi must be authenticated to your Tailnet (tailscale up).
• NATS Server: A reachable NATS server (e.g., nats.deezwatts.com:42222).
• Slack Webhook: Create a Slack App and activate an Incoming Webhook URL.

2. The Bridge Script (rivian_slack_bridge.py)

This script subscribes to a specific NATS subject and forwards formatted blocks to Slack.

rivian_slack_bridge.py

import asyncio
import json
import httpx
from nats.aio.client import Client as NATS

# Configuration
NATS_URL = "nats.deezwatts.com:42222"
SLACK_WEBHOOK = "https://hooks.slack.com/services/T000.../B000.../XXXX..."
NATS_SUBJECT = "rivian.telemetry"

async def forward_to_slack(data):
    payload = {
        "text": f"*Deezwatts Alert:* {data.get('alert', 'Status Update')}",
        "blocks": [
            {
                "type": "section",
                "text": {"type": "mrkdwn", "text": f"*Vehicle Alert:* {data.get('alert', 'Unknown')}"}
            },
            {
                "type": "context",
                "elements": [{"type": "mrkdwn", "text": f"📍 Battery: {data.get('battery')}% | Location: {data.get('location')}"}]
            }
        ]
    }
    async with httpx.AsyncClient() as client:
        await client.post(SLACK_WEBHOOK, json=payload)

async def run():
    nc = NATS()
    await nc.connect(NATS_URL)
    print(f"Connected to NATS at {NATS_URL}")

    async def message_handler(msg):
        data = json.loads(msg.data.decode())
        print(f"Received event: {data}")
        # Only forward significant alerts
        if "alert" in data:
            await forward_to_slack(data)

    await nc.subscribe(NATS_SUBJECT, cb=message_handler)

    while True:
        await asyncio.sleep(1)

if __name__ == '__main__':
    asyncio.run(run())

3. Automation via Systemd

To ensure the bridge starts on boot and survives crashes, deploy it as a systemd service on the Pi:

/etc/systemd/system/rivian-bridge.service

/etc/systemd/system/rivian-bridge.service

[Unit]
Description=Rivian NATS to Slack Bridge
After=network.target tailscaled.service

[Service]
ExecStart=/usr/bin/python3 /home/pi/deezwatts/rivian_slack_bridge.py
Restart=always
User=pi

[Install]
WantedBy=multi-user.target

The Attestation

To verify the “Dial Tone” is reaching Slack through the Tailnet, we can mock a vehicle event using the nats CLI.

1. Connect to your Tailnet from your workstation.
2. Publish a Mock Alert:

   nats pub rivian.telemetry '{"alert": "Gear Guard Proximity Triggered", "battery": 84, "location": "Silver Lake Sand Dunes"}'

3. Verify the Notification: You should see a formatted message appearing in your designated Slack channel within milliseconds.

[!TIP] Since the Pi is on the Tailnet, you can also SSH into it remotely to check the logs: journalctl -u rivian-bridge -f.