"""FastMCP server for iMessage."""
import argparse
import logging
import os
import signal
import sys
from typing import Any
from fastmcp import FastMCP
from .tools import (
hello_world,
check_permissions,
search_contacts,
lookup_contact,
list_conversations,
get_conversation_messages,
get_recent_messages,
get_message_context,
search_messages,
send_message,
search_index_status,
rebuild_search_index,
get_rebuild_progress,
)
# Configure logging
logging.basicConfig(
level=os.environ.get("LOG_LEVEL", "INFO"),
format="%(asctime)s - %(name)s - %(levelname)s - %(message)s",
)
logger = logging.getLogger(__name__)
# Create FastMCP server instance
mcp = FastMCP(
name="jons-mcp-imessage",
instructions="""
# iMessage MCP Server
A local MCP server for querying and sending iMessages on macOS.
## Permissions Required
**Full Disk Access**: Required to read ~/Library/Messages/chat.db
- Grant to Terminal.app (if running from terminal)
- Or grant to Claude Desktop.app (if using Claude Desktop)
- Location: System Settings > Privacy & Security > Full Disk Access
**Automation Permission**: Required for sending messages
- Prompted automatically on first send
- Or grant in: System Settings > Privacy & Security > Automation
Use `check_permissions` to diagnose permission issues.
## Available Tools
### Reading Messages
| Tool | Description |
|------|-------------|
| `check_permissions` | Verify database access and diagnose permission issues |
| `list_conversations` | List all conversations with metadata |
| `get_conversation_messages` | Get messages from a specific conversation |
| `get_recent_messages` | Get recent messages across all conversations |
| `get_message_context` | Get messages before/after a specific message in the same thread |
| `search_messages` | Search messages by text with optional filters |
| `search_contacts` | Search for contacts by phone/email |
### Sending Messages
| Tool | Description |
|------|-------------|
| `send_message` | Send a message to an existing conversation |
**send_message Limitations:**
- Can ONLY send to contacts with existing conversations
- No delivery confirmation - messages may silently fail
- Messages.app must be running
## Example Workflows
**Check if permissions are configured:**
```
check_permissions()
```
**List recent conversations:**
```
list_conversations(limit=10)
```
**Get messages from a conversation:**
```
get_conversation_messages(contact="+15551234567", limit=20)
```
**Search for messages:**
```
search_messages(query="dinner", sender="+15551234567")
```
**Send a message:**
```
send_message(recipient="+15551234567", message="Hello!")
```
""",
)
# Register tools
mcp.tool(hello_world)
mcp.tool(check_permissions)
mcp.tool(search_contacts)
mcp.tool(lookup_contact)
mcp.tool(list_conversations)
mcp.tool(get_conversation_messages)
mcp.tool(get_recent_messages)
mcp.tool(get_message_context)
mcp.tool(search_messages)
mcp.tool(send_message)
mcp.tool(search_index_status)
mcp.tool(rebuild_search_index)
mcp.tool(get_rebuild_progress)
# Signal handling for graceful shutdown
def signal_handler(signum: int, frame: Any) -> None:
"""Handle shutdown signals."""
logger.info(f"Received signal {signum}, shutting down...")
sys.exit(0)
def main() -> None:
parser = argparse.ArgumentParser(description="iMessage MCP Server")
parser.add_argument(
"project_path",
nargs="?",
help="Path to the project (defaults to current directory)",
)
args = parser.parse_args()
if args.project_path:
logger.info(f"Starting server with project path: {args.project_path}")
else:
logger.info("Starting server with current directory")
signal.signal(signal.SIGINT, signal_handler)
signal.signal(signal.SIGTERM, signal_handler)
mcp.run()
if __name__ == "__main__":
main()
