Signal AI Chat Bot with AppDaemon + Claude (Anthropic) — Full Guide
Hey everyone,
I wanted to share my setup for a Signal AI Chat Bot that lets me chat with Home Assistant through Signal Messenger, powered by Claude (Anthropic) as the conversation agent. It can answer questions about my home AND control devices — lights, heating, blinds, you name it.
Architecture Overview
The setup uses three components that are already available as HA add-ons:
- Signal Messenger REST API add-on (bbernhard) — exposes a REST + WebSocket API on port 8080
- AppDaemon add-on — runs a custom Python app that bridges Signal and HA
- Anthropic (Claude) integration — configured as a conversation agent with Assist API enabled
The flow is simple:
Signal message → WebSocket → AppDaemon → HA Conversation API (Claude) → Signal reply
Prerequisites
- Signal Messenger REST API add-on installed and registered with a phone number (I used a landline number — Signal supports voice verification for those)
- AppDaemon add-on installed, with
websockets and aiohttp added to the Python packages in the add-on configuration
- Anthropic integration set up with a conversation agent (or any other conversation agent — Google Generative AI works too, just change the
agent_id)
- A long-lived access token for HA (Profile → Security → Long-Lived Access Tokens)
Key Features
- Multi-turn conversations — conversation ID is stored per sender, so context is preserved across messages
- Group message support — works in Signal groups, not just 1:1 chats
- Allowed sender filtering — only whitelisted numbers get AI responses
- Markdown stripping — Claude loves markdown, Signal doesn’t render it, so we strip it
- Auto-reconnect — if the WebSocket drops, it reconnects after 10 seconds
- HA event firing — every incoming message fires a
signal_message_received event, so you can build additional automations on top
The Tricky Part: Group Messages
This one cost me some time. The Signal REST API WebSocket delivers group messages with an internal_id as the groupId. But the /v2/send endpoint expects the id field (which looks different and has a group. prefix). The solution: on startup, the app loads all groups via /v1/groups/NUMBER and builds a mapping from internal_id to id. Problem solved.
Setup
1. Add Python packages to AppDaemon
In the AppDaemon add-on configuration, add to “Python packages”:
websockets
aiohttp
2. Create the app file
Place this as /addon_configs/a0d7b954_appdaemon/apps/signal_listener.py:
import appdaemon.plugins.hass.hassapi as hass
import asyncio
import json
import re
import aiohttp
try:
import websockets
except ImportError:
websockets = None
def strip_markdown(text):
"""Convert markdown to plain text for Signal."""
text = re.sub(r'\*\*(.+?)\*\*', r'\1', text)
text = re.sub(r'__(.+?)__', r'\1', text)
text = re.sub(r'\*(.+?)\*', r'\1', text)
text = re.sub(r'(?<!\w)_(.+?)_(?!\w)', r'\1', text)
text = re.sub(r'`(.+?)`', r'\1', text)
text = re.sub(r'^#{1,6}\s+', '', text, flags=re.MULTILINE)
text = re.sub(r'\[(.+?)\]\(.+?\)', r'\1', text)
text = re.sub(r'^[\-\*]\s+', '• ', text, flags=re.MULTILINE)
return text.strip()
class SignalReceiver(hass.Hass):
async def initialize(self):
if websockets is None:
self.log("websockets package not installed!", level="ERROR")
return
self.uri = self.args.get("signal_ws_uri", "ws://HA_IP:8080/v1/receive/+YOUR_NUMBER")
self.own_number = self.args.get("own_number", "+YOUR_NUMBER")
self.signal_api = self.args.get("signal_api", "http://HA_IP:8080")
self.ha_url = self.args.get("ha_url", "http://supervisor/core")
self.ha_token = self.args.get("ha_token", "")
self.allowed_numbers = self.args.get("allowed_numbers", [])
self.conversations = {}
self.group_map = {}
self.log(f"Starting Signal AI Bot on {self.uri}")
self.log(f"Allowed senders: {self.allowed_numbers or 'ALL'}")
await self.load_group_map()
self.create_task(self.listen_to_signal())
async def load_group_map(self):
url = f"{self.signal_api}/v1/groups/{self.own_number}"
try:
async with aiohttp.ClientSession() as session:
async with session.get(url) as resp:
if resp.status == 200:
groups = await resp.json()
for g in groups:
internal_id = g.get("internal_id", "")
api_id = g.get("id", "")
if internal_id and api_id:
self.group_map[internal_id] = api_id
self.log(f"Loaded {len(self.group_map)} group mappings")
except Exception as e:
self.log(f"Error loading groups: {e}", level="WARNING")
async def listen_to_signal(self):
while True:
try:
async with websockets.connect(self.uri) as ws:
self.log("Connected to Signal WebSocket")
while True:
raw = await ws.recv()
try:
data = json.loads(raw)
except json.JSONDecodeError:
continue
envelope = data.get("envelope", {})
source = envelope.get("sourceNumber") or envelope.get("source", "")
data_msg = envelope.get("dataMessage", {})
message = data_msg.get("message", "")
group_info = data_msg.get("groupInfo", {})
group_id = group_info.get("groupId", "")
if source == self.own_number or not message:
continue
if self.allowed_numbers and source not in self.allowed_numbers:
continue
self.fire_event("signal_message_received", source=source, message=message, group_id=group_id)
try:
response = await self.process_with_ai(source, message)
if response:
await self.send_signal_reply(source, response, group_id)
except Exception as e:
self.log(f"AI error: {e}", level="ERROR")
except Exception as e:
self.log(f"WebSocket error: {e}", level="ERROR")
await asyncio.sleep(10)
async def process_with_ai(self, source, message):
conv_id = self.conversations.get(source)
payload = {"text": message, "agent_id": "conversation.claude_conversation", "language": "de"}
if conv_id:
payload["conversation_id"] = conv_id
headers = {"Authorization": f"Bearer {self.ha_token}", "Content-Type": "application/json"}
try:
async with aiohttp.ClientSession() as session:
async with session.post(f"{self.ha_url}/api/conversation/process", json=payload, headers=headers, timeout=aiohttp.ClientTimeout(total=30)) as resp:
if resp.status == 200:
result = await resp.json()
new_conv_id = result.get("conversation_id")
if new_conv_id:
self.conversations[source] = new_conv_id
speech = result.get("response", {}).get("speech", {}).get("plain", {}).get("speech", "")
return strip_markdown(speech)
else:
body = await resp.text()
self.log(f"API failed ({resp.status}): {body}", level="ERROR")
return None
except Exception as e:
self.log(f"HA API error: {e}", level="ERROR")
return None
async def send_signal_reply(self, source, text, group_id=""):
url = f"{self.signal_api}/v2/send"
if group_id:
api_id = self.group_map.get(group_id)
if not api_id:
await self.load_group_map()
api_id = self.group_map.get(group_id)
if not api_id:
self.log(f"Unknown group: {group_id}", level="ERROR")
return
payload = {"message": text, "number": self.own_number, "recipients": [api_id]}
else:
payload = {"message": text, "number": self.own_number, "recipients": [source]}
try:
async with aiohttp.ClientSession() as session:
async with session.post(url, json=payload) as resp:
if resp.status not in (200, 201):
body = await resp.text()
self.log(f"Send failed ({resp.status}): {body}", level="ERROR")
except Exception as e:
self.log(f"Send error: {e}", level="ERROR")
3. Configure the app
Add to /addon_configs/a0d7b954_appdaemon/apps/apps.yaml:
signal_receiver:
module: signal_listener
class: SignalReceiver
signal_ws_uri: "ws://YOUR_HA_IP:8080/v1/receive/+YOUR_SIGNAL_NUMBER"
own_number: "+YOUR_SIGNAL_NUMBER"
signal_api: "http://YOUR_HA_IP:8080"
ha_url: "http://supervisor/core"
ha_token: "YOUR_LONG_LIVED_ACCESS_TOKEN"
allowed_numbers:
- "+YOUR_PHONE_NUMBER"
4. Restart AppDaemon and check the logs.
Swapping the AI Agent
The agent_id in process_with_ai can point to any HA conversation agent. Want to use Google Generative AI instead? Just change it to conversation.google_generative_ai. Want the built-in HA Assist? Use conversation.home_assistant. The rest of the code stays the same.
Bonus: HA Events for Automations
Every incoming message fires a signal_message_received event with source, message, and group_id. You can use this to trigger automations — for example, if someone texts “alarm” to the group, you could arm your alarm system regardless of the AI response.
Tips
- Set
allowed_numbers to restrict who can talk to your bot. Leave the list empty to allow everyone (not recommended).
- The conversation ID is stored in memory per sender. If AppDaemon restarts, conversation context resets — this is fine in practice since Claude handles standalone questions well.
- If you are running HA OS,
http://supervisor/core works as the HA URL from within add-ons. For other setups, use your actual HA URL.
Hope this helps someone. Happy to answer questions!
