The setup instructions for the signal_messenger integration deploy Signal as a notification provider, and allow you to optionally create automations for what to do if Signal receives a message.
I think it would fit the model much better if you could use Signal to talk to the conversation backend and leverage all the advancements the team have made in their “year of voice.” Something like this:
I’ve been using the following script to pipe incoming Signal messages into an Assist conversation, and send the response back over signal.
#!/usr/bin/python
import os
import json
import requests
from websockets.exceptions import ConnectionClosed
from websockets.sync.client import connect
HOMEASSISTANT_URI = os.environ.get("HOMEASSISTANT_URI", "http://localhost:8123")
SIGNAL_URI = os.environ.get("SIGNAL_URI", "http://localhost:8080")
SIGNAL_SENDER = os.environ.get("SIGNAL_SENDER", "+12345")
SIGNAL_RECIPIENTS = [x for x in os.environ.get("SIGNAL_RECIPIENTS", "").split(" ") if len(x) > 0]
SIGNAL_API_TOKENS = [x for x in os.environ.get("SIGNAL_API_TOKENS", "").split(" ") if len(x) > 0]
class Conversation:
def __init__(self, number, token):
self.number = number
self.token = token
self.session = requests.Session()
self.session.headers.update({"Authorization": f"Bearer {self.token}"})
print(self.get("/api/"))
def say(self, text):
resp = self.post("/api/conversation/process", jsondata={
"text": text,
})
print(resp)
return resp.get("response",{}).get("speech",{}).get("plain",{}).get("speech","")
def get(self, url, **kwargs):
print(HOMEASSISTANT_URI + url)
resp = self.session.get(HOMEASSISTANT_URI + url, **kwargs)
return json.loads(resp.text)
def post(self, url, data=None, jsondata=None, **kwargs):
resp = self.session.post(HOMEASSISTANT_URI + url, data, jsondata, **kwargs)
return json.loads(resp.text)
s = requests.Session()
accounts = json.loads(s.get(SIGNAL_URI + "/v1/accounts").text)
if not isinstance(accounts, list):
raise ValueError("Error getting account list - is your signal cli running?")
if SIGNAL_SENDER not in accounts:
raise ValueError("Sender not listed as a linked account")
print(f"{len(SIGNAL_RECIPIENTS)} recipients")
convos = {}
for (i,recipient) in enumerate(SIGNAL_RECIPIENTS):
convos[recipient] = Conversation(recipient, SIGNAL_API_TOKENS[i%len(SIGNAL_API_TOKENS)])
with connect("ws" + SIGNAL_URI[4:] + "/v1/receive/" + SIGNAL_SENDER) as sock:
while True:
try:
msg = sock.recv()
msg = json.loads(msg)
if "envelope" not in msg:
continue
number = msg["envelope"].get("source","")
if number == SIGNAL_SENDER:
continue
if number not in convos:
print(f'Message from {number} ignored.')
continue
if "dataMessage" not in msg["envelope"]:
continue
response = convos[number].say(msg["envelope"]["dataMessage"].get("message", None))
if isinstance(response,str) and len(response) > 0:
s.post(SIGNAL_URI + "/v2/send", json={
"message": response,
"number": SIGNAL_SENDER,
"recipients": [ number ],
})
except ConnectionClosed:
break
This script only allows incoming messages from certain numbers, and uses a long-lived access token to speak to the Assist backend in the user’s context.
Deploying Signal in this way has the added benefit of running signal-cli-rest-api in json-rpc mode, meaning you’ll not only get a much quicker response on your messages, but you can also receive more than one message every 30 seconds.
