# Directory Structure
```
├── .gitignore
├── .python-version
├── check_mcp.py
├── CLAUDE.md
├── pyproject.toml
├── README.md
├── signal_mcp
│ ├── __init__.py
│ └── main.py
├── tests
│ └── test_parse.py
└── uv.lock
```
# Files
--------------------------------------------------------------------------------
/.python-version:
--------------------------------------------------------------------------------
```
3.13
```
--------------------------------------------------------------------------------
/.gitignore:
--------------------------------------------------------------------------------
```
# Python-generated files
__pycache__/
*.py[oc]
build/
dist/
wheels/
*.egg-info
# Virtual environments
.venv
.history
.vscode
.env
```
--------------------------------------------------------------------------------
/README.md:
--------------------------------------------------------------------------------
```markdown
# Signal MCP
An [MCP](https://github.com/mcp-signal/mcp) integration for [signal-cli](https://github.com/AsamK/signal-cli) that allows AI agents to send and receive Signal messages.
## Features
- Send messages to Signal users
- Send messages to Signal groups
- Receive and parse incoming messages
- Async support with timeout handling
- Detailed logging
## Prerequisites
This project requires [signal-cli](https://github.com/AsamK/signal-cli) to be installed and configured on your system.
### Installing signal-cli
1. **Install signal-cli**: Follow the [official installation instructions](https://github.com/AsamK/signal-cli/blob/master/README.md#installation)
2. **Register your Signal account**:
```bash
signal-cli -u YOUR_PHONE_NUMBER register
```
3. **Verify your account** with the code received via SMS:
```bash
signal-cli -u YOUR_PHONE_NUMBER verify CODE_RECEIVED
```
For more detailed setup instructions, see the [signal-cli documentation](https://github.com/AsamK/signal-cli/wiki).
## Installation
```bash
pip install -e .
# or use uv for faster installation
uv pip install -e .
```
## Usage
Run the MCP server:
```bash
./main.py --user-id YOUR_PHONE_NUMBER [--transport {sse|stdio}]
```
## API
### Tools Available
- `send_message_to_user`: Send a direct message to a Signal user
- `send_message_to_group`: Send a message to a Signal group
- `receive_message`: Wait for and receive messages with timeout support
## Development
This project uses:
- [MCP](https://github.com/mcp-signal/mcp) for agent-API integration
- Modern Python async patterns
- Type annotations throughout
```
--------------------------------------------------------------------------------
/CLAUDE.md:
--------------------------------------------------------------------------------
```markdown
# CLAUDE.md
This file provides guidance to Claude Code (claude.ai/code) when working with code in this repository.
## Commands
- Run: `python main.py --user-id YOUR_PHONE_NUMBER [--transport {sse|stdio}]`
- Lint: `ruff check .`
- Type check: `mypy .`
- Format code: `ruff format .`
## Code Style Guidelines
- **Imports**: Standard library first, then third-party, then local. Group imports by type.
- **Formatting**: Use ruff formatter (Black-compatible).
- **Types**: Use strict type annotations. Define custom types for complex structures.
- **Naming**:
- Functions/variables: snake_case
- Classes: PascalCase
- Constants: UPPER_CASE
- **Error Handling**: Use custom exception classes. Log errors before raising or returning.
- **Logging**: Use the established logger pattern with appropriate log levels.
- **Async**: Use asyncio for all I/O operations.
- **Security**: Always use shlex.quote for shell command arguments.
```
--------------------------------------------------------------------------------
/signal_mcp/__init__.py:
--------------------------------------------------------------------------------
```python
```
--------------------------------------------------------------------------------
/pyproject.toml:
--------------------------------------------------------------------------------
```toml
[project]
name = "signal-mcp"
version = "0.1.0"
description = "Add your description here"
readme = "README.md"
requires-python = ">=3.13"
dependencies = [
"mcp>=1.6.0",
]
[build-system]
requires = ["hatchling"]
build-backend = "hatchling.build"
[project.scripts]
server = "signal_mcp.main:main"
[project.optional-dependencies]
test = [
"pytest>=7.0.0",
"python-dotenv>=1.0.0",
]
[dependency-groups]
dev = [
"mypy>=1.15.0",
"ruff>=0.11.2",
]
[tool.pytest.ini_options]
testpaths = ["tests"]
python_files = "test_*.py"
[tool.hatch.build.targets.wheel]
packages = ["signal_mcp"]
```
--------------------------------------------------------------------------------
/check_mcp.py:
--------------------------------------------------------------------------------
```python
import asyncio
import os
import mcp
from mcp.client.stdio import StdioServerParameters
from mcp import ClientSession
from dotenv import load_dotenv
# Load environment variables from .env file
load_dotenv()
async def main():
# Configure the stdio client
server_params = StdioServerParameters(
command="python",
args=[
"signal_mcp/main.py",
"--user-id",
os.environ["SENDER_NUMBER"],
"--transport",
"stdio",
],
)
async with mcp.stdio_client(server_params) as transport:
stdio, write = transport
async with ClientSession(stdio, write) as session:
await session.initialize()
# List available tools
response = await session.list_tools()
print(response.tools)
# Call a tool to send a message
send_result = await session.call_tool(
"send_message_to_user",
{
"message": "Hello from MCP stdio client!",
"user_id": os.environ["RECEIVER_NUMBER"],
},
)
print(f"Send result: {send_result}")
# Receive a message with timeout
print("Waiting for message...")
receive_result = await session.call_tool(
"receive_message",
{"timeout": 10}, # 5 second timeout
)
print(f"Receive result: {receive_result}")
# Check if we received a message (might be None if timeout)
if isinstance(receive_result, tuple) and len(receive_result) >= 2:
message, sender, group = receive_result
if message and sender:
print(f"Received message from {sender}: {message}")
if group:
print(f"In group: {group}")
if __name__ == "__main__":
asyncio.run(main())
```
--------------------------------------------------------------------------------
/tests/test_parse.py:
--------------------------------------------------------------------------------
```python
import asyncio
from signal_mcp.main import _parse_receive_output
recv = """
Envelope from: “Bob Sagat” +11234567890 (device: 4) to +15551234567
Timestamp: 1744185564802 (2025-04-09T07:59:24.802Z)
Server timestamps: received: 1744185564847 (2025-04-09T07:59:24.847Z) delivered: 1744185565739 (2025-04-09T07:59:25.739Z)
Sent by unidentified/sealed sender
Received a receipt message
When: 1744185564802 (2025-04-09T07:59:24.802Z)
Is read receipt
Timestamps:
- 1744185570322 (2025-04-09T07:59:30.322Z)
Envelope from: “Bob Sagat” +11234567890 (device: 4) to +15551234567
Timestamp: 1744185565138 (2025-04-09T07:59:25.138Z)
Server timestamps: received: 1744185565194 (2025-04-09T07:59:25.194Z) delivered: 1744185565739 (2025-04-09T07:59:25.739Z)
Sent by unidentified/sealed sender
Received a typing message
Action: STARTED
Timestamp: 1744185565138 (2025-04-09T07:59:25.138Z)
Envelope from: “Bob Sagat” +11234567890 (device: 4) to +15551234567
Timestamp: 1744185565192 (2025-04-09T07:59:25.192Z)
Server timestamps: received: 1744185565302 (2025-04-09T07:59:25.302Z) delivered: 1744185565740 (2025-04-09T07:59:25.740Z)
Sent by unidentified/sealed sender
Received a receipt message
When: 1744185565192 (2025-04-09T07:59:25.192Z)
Is delivery receipt
Timestamps:
- 1744185570322 (2025-04-09T07:59:30.322Z)
Envelope from: “Bob Sagat” +11234567890 (device: 4) to +15551234567
Timestamp: 1744185565466 (2025-04-09T07:59:25.466Z)
Server timestamps: received: 1744185565586 (2025-04-09T07:59:25.586Z) delivered: 1744185565740 (2025-04-09T07:59:25.740Z)
Sent by unidentified/sealed sender
Message timestamp: 1744185565466 (2025-04-09T07:59:25.466Z)
Body: yo
With profile key
Envelope from: “Bob Sagat” +11234567890 (device: 3) to +15551234567
Timestamp: 1744185572206 (2025-04-09T07:59:32.206Z)
Server timestamps: received: 1744185566038 (2025-04-09T07:59:26.038Z) delivered: 1744185566039 (2025-04-09T07:59:26.039Z)
Sent by unidentified/sealed sender
Received a receipt message
When: 1744185572206 (2025-04-09T07:59:32.206Z)
Is delivery receipt
Timestamps:
- 1744185570322 (2025-04-09T07:59:30.322Z)
Envelope from: “Bob Sagat” +11234567890 (device: 1) to +15551234567
Timestamp: 1744185569713 (2025-04-09T07:59:29.713Z)
Server timestamps: received: 1744185567862 (2025-04-09T07:59:27.862Z) delivered: 1744185567863 (2025-04-09T07:59:27.863Z)
Sent by unidentified/sealed sender
Received a receipt message
When: 1744185569713 (2025-04-09T07:59:29.713Z)
Is delivery receipt
Timestamps:
- 1744185570322 (2025-04-09T07:59:30.322Z)
"""
def test_parse_direct_message():
expected_result = ("yo", "+11234567890", None)
result = asyncio.run(_parse_receive_output(recv))
assert result == expected_result, f"Expected {expected_result}, but got {result}"
```
--------------------------------------------------------------------------------
/signal_mcp/main.py:
--------------------------------------------------------------------------------
```python
#!/usr/bin/env -S uv run
# /// script
# requires-python = ">=3.13"
# dependencies = [
# "mcp",
# ]
# ///
from mcp.server.fastmcp import FastMCP
from typing import Optional, Tuple, Dict, Union, Any
import asyncio
import subprocess
import shlex
import argparse
from dataclasses import dataclass
import logging
# Set up logging with more detailed format
logging.basicConfig(
level=logging.DEBUG, format="%(asctime)s - %(name)s - %(levelname)s - %(message)s"
)
logger = logging.getLogger(__name__)
# Initialize MCP server
mcp = FastMCP(name="signal-cli")
logger.info("Initialized FastMCP server for signal-cli")
@dataclass
class SignalConfig:
"""Configuration for Signal CLI."""
user_id: str = "" # The user's Signal phone number
transport: str = "sse"
@dataclass
class MessageResponse:
"""Structured result for received messages."""
message: Optional[str] = None
sender_id: Optional[str] = None
group_name: Optional[str] = None
error: Optional[str] = None
class SignalError(Exception):
"""Base exception for Signal-related errors."""
pass
class SignalCLIError(SignalError):
"""Exception raised when signal-cli command fails."""
pass
SuccessResponse = Dict[str, str]
ErrorResponse = Dict[str, str]
# Global config instance
config = SignalConfig()
async def _run_signal_cli(cmd: str) -> Tuple[str, str, int | None]:
"""Helper method to run a signal-cli command."""
logger.debug(f"Executing signal-cli command: {cmd}")
try:
process = await asyncio.create_subprocess_shell(
cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE
)
stdout, stderr = await process.communicate()
stdout_str, stderr_str = stdout.decode(), stderr.decode()
if process.returncode != 0:
logger.warning(
f"signal-cli command failed with return code {process.returncode}"
)
logger.warning(f"stderr: {stderr_str}")
else:
logger.debug("signal-cli command completed successfully")
return stdout_str, stderr_str, process.returncode
except Exception as e:
logger.error(f"Error running signal-cli command: {str(e)}", exc_info=True)
raise SignalCLIError(f"Failed to run signal-cli: {str(e)}")
async def _get_group_id(group_name: str) -> Optional[str]:
"""Get the group name for a given group name."""
logger.info(f"Looking up group with name: {group_name}")
list_cmd = f"signal-cli -u {shlex.quote(config.user_id)} listGroups"
stdout, stderr, return_code = await _run_signal_cli(list_cmd)
if return_code != 0:
logger.error(f"Error listing groups: {stderr}")
return None
# Parse the output to find the group name
for line in stdout.split("\n"):
if "Name: " in line and group_name in line:
logger.info(f"Found group: {group_name}")
return group_name
logger.error(f"Could not find group with name: {group_name}")
return None
async def _send_message(message: str, target: str, is_group: bool = False) -> bool:
"""Send a message to either a user or group."""
target_type = "group" if is_group else "user"
logger.info(f"Sending message to {target_type}: {target}")
flag = "-g" if is_group else ""
cmd = f"signal-cli -u {shlex.quote(config.user_id)} send {flag} {shlex.quote(target)} -m {shlex.quote(message)}"
try:
_, stderr, return_code = await _run_signal_cli(cmd)
if return_code == 0:
logger.info(f"Successfully sent message to {target_type}: {target}")
return True
else:
logger.error(f"Error sending message to {target_type}: {stderr}")
return False
except SignalCLIError as e:
logger.error(f"Failed to send message to {target_type}: {str(e)}")
return False
async def _parse_receive_output(
stdout: str,
) -> Optional[MessageResponse]:
"""Parse the output of signal-cli receive command."""
logger.debug("Parsing received message output")
lines = stdout.split("\n")
# Process each envelope section separately
current_envelope: Dict[str, Any] = {}
current_sender: Optional[str] = None
for i, line in enumerate(lines):
line = line.strip()
if not line:
continue
if line.startswith("Envelope from:"):
# Start of a new envelope block
current_envelope = {}
# Extract phone number using a straightforward approach
# Format: Envelope from: "Bob Sagat" +11234567890 (device: 4) to +15551234567
parts = line.split("+")
if len(parts) > 1:
# Get the phone number part
phone_part = parts[1].split()[0]
current_sender = "+" + phone_part
current_envelope["sender"] = current_sender
logger.debug(f"Found sender: {current_sender}")
elif line.startswith("Body:"):
# Found a message body
if current_envelope:
message_body = line[5:].strip()
current_envelope["message"] = message_body
current_envelope["has_body"] = True
logger.debug(f"Found message body: {message_body}")
# If we have a valid message with body, return it
if current_envelope.get("has_body") and "sender" in current_envelope:
sender = current_envelope["sender"]
msg = current_envelope["message"]
group = current_envelope.get("group")
if isinstance(sender, str) and isinstance(msg, str):
logger.info(
f"Successfully parsed message from {sender}"
+ (f" in group {group}" if group else "")
)
return MessageResponse(
message=msg, sender_id=sender, group_name=group
)
elif line.startswith("Group info:"):
if current_envelope:
current_envelope["in_group"] = True
elif (
line.startswith("Name:")
and current_envelope
and current_envelope.get("in_group")
):
group_name = line[5:].strip()
current_envelope["group"] = group_name
logger.debug(f"Found group name: {group_name}")
logger.warning("Failed to parse message from output")
return None
@mcp.tool()
async def send_message_to_user(
message: str, user_id: str
) -> Union[SuccessResponse, ErrorResponse]:
"""Send a message to a specific user using signal-cli."""
logger.info(f"Tool called: send_message_to_user for user {user_id}")
try:
success = await _send_message(message, user_id, is_group=False)
if success:
logger.info(f"Successfully sent message to user {user_id}")
return {"message": "Message sent successfully"}
logger.error(f"Failed to send message to user {user_id}")
return {"error": "Failed to send message"}
except Exception as e:
logger.error(f"Error in send_message_to_user: {str(e)}", exc_info=True)
return {"error": str(e)}
@mcp.tool()
async def send_message_to_group(
message: str, group_id: str
) -> Union[SuccessResponse, ErrorResponse]:
"""Send a message to a group using signal-cli."""
logger.info(f"Tool called: send_message_to_group for group {group_id}")
try:
group_name = await _get_group_id(group_id)
if not group_name:
logger.error(f"Could not find group: {group_id}")
return {"error": f"Could not find group: {group_id}"}
success = await _send_message(message, group_name, is_group=True)
if success:
logger.info(f"Successfully sent message to group {group_id}")
return {"message": "Message sent successfully"}
logger.error(f"Failed to send message to group {group_id}")
return {"error": "Failed to send message"}
except Exception as e:
logger.error(f"Error in send_message_to_group: {str(e)}", exc_info=True)
return {"error": str(e)}
@mcp.tool()
async def receive_message(timeout: float) -> MessageResponse:
"""Wait for and receive a message using signal-cli."""
logger.info(f"Tool called: receive_message with timeout {timeout}s")
try:
cmd = f"signal-cli -u {shlex.quote(config.user_id)} receive --timeout {int(timeout)}"
stdout, stderr, return_code = await _run_signal_cli(cmd)
if return_code != 0:
if "timeout" in stderr.lower():
logger.info("Receive timeout reached with no messages")
return MessageResponse()
else:
logger.error(f"Error receiving message: {stderr}")
return MessageResponse(error=f"Failed to receive message: {stderr}")
if not stdout.strip():
logger.info("No message received within timeout")
return MessageResponse()
result = await _parse_receive_output(stdout)
if result:
logger.info(
f"Successfully received message from {result.sender_id}"
+ (f" in group {result.group_name}" if result.group_name else "")
)
return result
else:
logger.error("Received message but couldn't parse the output format")
return MessageResponse(error="Failed to parse message output")
except Exception as e:
logger.error(f"Error in receive_message: {str(e)}", exc_info=True)
return MessageResponse(error=str(e))
def initialize_server() -> SignalConfig:
"""Initialize the Signal server with configuration."""
logger.info("Initializing Signal server")
parser = argparse.ArgumentParser(description="Run the Signal MCP server")
parser.add_argument(
"--user-id", required=True, help="Signal phone number for the user"
)
parser.add_argument(
"--transport",
choices=["sse", "stdio"],
default="sse",
help="Transport to use for communication with the client. (default: sse)",
)
args = parser.parse_args()
logger.info(f"Parsed arguments: user_id={args.user_id}, transport={args.transport}")
# Set global config
config.user_id = args.user_id
config.transport = args.transport
logger.info(f"Initialized Signal server for user: {config.user_id}")
return config
def run_mcp_server():
"""Run the MCP server in the current event loop."""
config = initialize_server()
transport = config.transport
logger.info(f"Starting MCP server with transport: {transport}")
return transport
def main():
"""Main function to run the Signal MCP server."""
logger.info("Starting Signal MCP server")
try:
transport = run_mcp_server()
mcp.run(transport)
except Exception as e:
logger.error(f"Error running Signal MCP server: {str(e)}", exc_info=True)
raise
finally:
logger.info("Signal MCP server shutting down")
if __name__ == "__main__":
main()
```