#!/usr/bin/env python3
"""
Command-line interface for the Matrix Command-Line Agent.
This module provides a command-line interface for interacting with Matrix chat
using the Matrix Command-Line Agent. It allows users to connect to Matrix servers,
join rooms, send messages, and listen for incoming messages.
"""
import os
import sys
import argparse
import asyncio
import logging
import yaml
import getpass
from typing import Dict, List, Optional, Any
from .matrix_client import MatrixClient, MatrixConfig
from nio import RoomMessageText
# Configure logging
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)
# Default config path
DEFAULT_CONFIG_PATH = os.path.expanduser("~/.config/matrix-cmd/config.yaml")
def parse_args() -> argparse.Namespace:
"""
Parse command-line arguments.
Returns:
argparse.Namespace: Parsed arguments
"""
parser = argparse.ArgumentParser(
description="Matrix Command-Line Agent - Interact with Matrix chat from the command line"
)
# Add subparsers for different commands
subparsers = parser.add_subparsers(dest="command", help="Command to execute")
# Config command
config_parser = subparsers.add_parser("config", help="Configure the Matrix agent")
config_parser.add_argument(
"--path",
default=DEFAULT_CONFIG_PATH,
help=f"Path to the configuration file (default: {DEFAULT_CONFIG_PATH})"
)
config_parser.add_argument(
"--init",
action="store_true",
help="Initialize a new configuration file with default values"
)
# Connect command
connect_parser = subparsers.add_parser("connect", help="Connect to a Matrix server")
connect_parser.add_argument(
"--config",
default=DEFAULT_CONFIG_PATH,
help=f"Path to the configuration file (default: {DEFAULT_CONFIG_PATH})"
)
connect_parser.add_argument(
"--homeserver",
help="Matrix homeserver URL (overrides config)"
)
connect_parser.add_argument(
"--user",
help="Matrix user ID (overrides config)"
)
connect_parser.add_argument(
"--password",
help="Matrix password (overrides config, will prompt if not provided)"
)
connect_parser.add_argument(
"--room",
action="append",
help="Room ID to join (can be specified multiple times, overrides config)"
)
# Enhanced features
connect_parser.add_argument(
"--enable-search",
action="store_true",
help="Enable search functionality"
)
connect_parser.add_argument(
"--enable-conversation-tracking",
action="store_true",
help="Enable conversation history tracking"
)
connect_parser.add_argument(
"--history-file",
help="Path to the conversation history file"
)
connect_parser.add_argument(
"--respond-to-all-messages",
action="store_true",
help="Respond to all messages, not just mentions"
)
# Send command
send_parser = subparsers.add_parser("send", help="Send a message to a Matrix room")
send_parser.add_argument(
"--config",
default=DEFAULT_CONFIG_PATH,
help=f"Path to the configuration file (default: {DEFAULT_CONFIG_PATH})"
)
send_parser.add_argument(
"--room",
required=True,
help="Room ID to send the message to"
)
send_parser.add_argument(
"message",
help="Message to send"
)
# Listen command
listen_parser = subparsers.add_parser("listen", help="Listen for messages in joined rooms")
listen_parser.add_argument(
"--config",
default=DEFAULT_CONFIG_PATH,
help=f"Path to the configuration file (default: {DEFAULT_CONFIG_PATH})"
)
# Interactive command
interactive_parser = subparsers.add_parser("interactive", help="Start an interactive session")
interactive_parser.add_argument(
"--config",
default=DEFAULT_CONFIG_PATH,
help=f"Path to the configuration file (default: {DEFAULT_CONFIG_PATH})"
)
interactive_parser.add_argument(
"--room",
required=True,
help="Room ID for the interactive session"
)
# Version command
subparsers.add_parser("version", help="Show version information")
return parser.parse_args()
def load_config(config_path: str) -> Dict[str, Any]:
"""
Load configuration from a file.
Args:
config_path: Path to the configuration file
Returns:
Dict[str, Any]: Configuration dictionary
"""
if not os.path.exists(config_path):
logger.error(f"Configuration file not found: {config_path}")
logger.info(f"Run 'matrix-cmd config --init' to create a default configuration file")
sys.exit(1)
try:
return MatrixConfig.load_config(config_path)
except Exception as e:
logger.error(f"Error loading configuration: {e}")
sys.exit(1)
async def run_config_command(args: argparse.Namespace) -> None:
"""
Run the config command.
Args:
args: Command-line arguments
"""
config_path = args.path
if args.init:
# Create default configuration
if os.path.exists(config_path):
overwrite = input(f"Configuration file already exists at {config_path}. Overwrite? (y/n): ")
if overwrite.lower() != 'y':
logger.info("Aborted")
return
# Create directory if it doesn't exist
os.makedirs(os.path.dirname(config_path), exist_ok=True)
# Create default config
default_config = MatrixConfig.create_default_config()
# Save config
try:
MatrixConfig.save_config(default_config, config_path)
logger.info(f"Created default configuration at {config_path}")
logger.info(f"Edit this file to configure your Matrix connection")
except Exception as e:
logger.error(f"Error creating configuration: {e}")
sys.exit(1)
else:
# Show current configuration
if not os.path.exists(config_path):
logger.error(f"Configuration file not found: {config_path}")
logger.info(f"Run 'matrix-cmd config --init' to create a default configuration file")
return
try:
config = MatrixConfig.load_config(config_path)
print("\nCurrent Configuration:")
print("======================")
print(f"Matrix Homeserver: {config['MATRIX']['HOMESERVER']}")
print(f"Matrix User: {config['MATRIX']['USER']}")
print(f"Matrix Password: {'*' * 8}")
print(f"Matrix Device ID: {config['MATRIX']['DEVICE_ID']}")
print(f"Matrix Rooms: {', '.join(config['MATRIX']['ROOMS'])}")
print(f"Matrix Admin Users: {', '.join(config['MATRIX']['ADMIN_USERS'])}")
print(f"Ollama Host: {config['OLLAMA']['HOST']}")
print(f"Ollama Model: {config['OLLAMA']['MODEL']}")
print(f"Ollama Prompt: {config['OLLAMA']['PROMPT']}")
print(f"Ollama Available Models: {', '.join(config['OLLAMA']['AVAILABLE_MODELS'])}")
print("\nTo modify this configuration, edit the file directly or run 'matrix-cmd config --init' to reset to defaults")
except Exception as e:
logger.error(f"Error reading configuration: {e}")
sys.exit(1)
async def run_connect_command(args: argparse.Namespace) -> None:
"""
Run the connect command.
Args:
args: Command-line arguments
"""
# Load configuration
config = load_config(args.config)
# Override with command-line arguments
homeserver = args.homeserver or config["MATRIX"]["HOMESERVER"]
user_id = args.user or config["MATRIX"]["USER"]
password = args.password or config["MATRIX"]["PASSWORD"]
# Prompt for password if not provided
if password == "your-password" or not password:
password = getpass.getpass(f"Password for {user_id}: ")
# Get rooms to join
rooms = args.room or config["MATRIX"]["ROOMS"]
# Get enhanced features from config or command-line arguments
enable_search = args.enable_search
enable_conversation_tracking = args.enable_conversation_tracking
history_file = args.history_file
respond_to_all_messages = args.respond_to_all_messages
# If config has ENHANCED section, use those values as defaults
if "ENHANCED" in config:
if not args.enable_search and "ENABLE_SEARCH" in config["ENHANCED"]:
enable_search = config["ENHANCED"]["ENABLE_SEARCH"]
if not args.enable_conversation_tracking and "ENABLE_CONVERSATION_TRACKING" in config["ENHANCED"]:
enable_conversation_tracking = config["ENHANCED"]["ENABLE_CONVERSATION_TRACKING"]
if not args.history_file and "HISTORY_FILE" in config["ENHANCED"]:
history_file = config["ENHANCED"]["HISTORY_FILE"]
if not args.respond_to_all_messages and "RESPOND_TO_ALL_MESSAGES" in config["ENHANCED"]:
respond_to_all_messages = config["ENHANCED"]["RESPOND_TO_ALL_MESSAGES"]
# Get OpenSearch config if available
opensearch_config = None
if "ENHANCED" in config and "OPENSEARCH" in config["ENHANCED"]:
opensearch_config = config["ENHANCED"]["OPENSEARCH"]
# Create Matrix client
client = MatrixClient(
homeserver=homeserver,
user_id=user_id,
password=password,
device_id=config["MATRIX"]["DEVICE_ID"],
ollama_host=config["OLLAMA"]["HOST"],
ollama_model=config["OLLAMA"]["MODEL"],
ollama_prompt=config["OLLAMA"]["PROMPT"],
available_models=config["OLLAMA"]["AVAILABLE_MODELS"],
admin_users=config["MATRIX"]["ADMIN_USERS"],
enable_search=enable_search,
enable_conversation_tracking=enable_conversation_tracking,
opensearch_config=opensearch_config,
history_file=history_file,
respond_to_all_messages=respond_to_all_messages
)
# Connect to the server with retry capabilities
logger.info(f"Connecting to {homeserver} as {user_id}...")
success = await client.connect(max_retries=3, initial_retry_delay=1.0)
if not success:
logger.error("Failed to connect to Matrix server")
sys.exit(1)
# Join rooms
for room_id in rooms:
logger.info(f"Joining room {room_id}...")
success = await client.join_room(room_id)
if not success:
logger.error(f"Failed to join room {room_id}")
# Start listening for messages
logger.info("Connected and joined rooms. Press Ctrl+C to disconnect.")
try:
await client.listen()
except KeyboardInterrupt:
logger.info("Disconnecting...")
await client.disconnect()
logger.info("Disconnected")
async def run_send_command(args: argparse.Namespace) -> None:
"""
Run the send command.
Args:
args: Command-line arguments
"""
# Load configuration
config = load_config(args.config)
# Create Matrix client
client = MatrixClient(
homeserver=config["MATRIX"]["HOMESERVER"],
user_id=config["MATRIX"]["USER"],
password=config["MATRIX"]["PASSWORD"],
device_id=config["MATRIX"]["DEVICE_ID"]
)
# Connect to the server with retry capabilities
logger.info(f"Connecting to {config['MATRIX']['HOMESERVER']} as {config['MATRIX']['USER']}...")
success = await client.connect(max_retries=2, initial_retry_delay=1.0)
if not success:
logger.error("Failed to connect to Matrix server")
sys.exit(1)
# Send message
logger.info(f"Sending message to room {args.room}...")
success = await client.send_message(args.room, args.message)
if not success:
logger.error(f"Failed to send message to room {args.room}")
sys.exit(1)
logger.info("Message sent successfully")
# Disconnect
await client.disconnect()
async def run_listen_command(args: argparse.Namespace) -> None:
"""
Run the listen command.
Args:
args: Command-line arguments
"""
# Load configuration
config = load_config(args.config)
# Create Matrix client
client = MatrixClient(
homeserver=config["MATRIX"]["HOMESERVER"],
user_id=config["MATRIX"]["USER"],
password=config["MATRIX"]["PASSWORD"],
device_id=config["MATRIX"]["DEVICE_ID"],
ollama_host=config["OLLAMA"]["HOST"],
ollama_model=config["OLLAMA"]["MODEL"],
ollama_prompt=config["OLLAMA"]["PROMPT"],
available_models=config["OLLAMA"]["AVAILABLE_MODELS"],
admin_users=config["MATRIX"]["ADMIN_USERS"]
)
# Connect to the server with retry capabilities
logger.info(f"Connecting to {config['MATRIX']['HOMESERVER']} as {config['MATRIX']['USER']}...")
success = await client.connect(max_retries=3, initial_retry_delay=1.0)
if not success:
logger.error("Failed to connect to Matrix server")
sys.exit(1)
# Join rooms
for room_id in config["MATRIX"]["ROOMS"]:
logger.info(f"Joining room {room_id}...")
success = await client.join_room(room_id)
if not success:
logger.error(f"Failed to join room {room_id}")
# Start listening for messages
logger.info("Listening for messages. Press Ctrl+C to stop.")
try:
await client.listen()
except KeyboardInterrupt:
logger.info("Disconnecting...")
await client.disconnect()
logger.info("Disconnected")
async def run_interactive_command(args: argparse.Namespace) -> None:
"""
Run the interactive command.
Args:
args: Command-line arguments
"""
# Load configuration
config = load_config(args.config)
# Create Matrix client
client = MatrixClient(
homeserver=config["MATRIX"]["HOMESERVER"],
user_id=config["MATRIX"]["USER"],
password=config["MATRIX"]["PASSWORD"],
device_id=config["MATRIX"]["DEVICE_ID"],
ollama_host=config["OLLAMA"]["HOST"],
ollama_model=config["OLLAMA"]["MODEL"],
ollama_prompt=config["OLLAMA"]["PROMPT"],
available_models=config["OLLAMA"]["AVAILABLE_MODELS"],
admin_users=config["MATRIX"]["ADMIN_USERS"]
)
# Connect to the server with retry capabilities
logger.info(f"Connecting to {config['MATRIX']['HOMESERVER']} as {config['MATRIX']['USER']}...")
success = await client.connect(max_retries=3, initial_retry_delay=1.0)
if not success:
logger.error("Failed to connect to Matrix server")
sys.exit(1)
# Join the specified room
logger.info(f"Joining room {args.room}...")
success = await client.join_room(args.room)
if not success:
logger.error(f"Failed to join room {args.room}")
sys.exit(1)
# Start interactive session
logger.info(f"Starting interactive session in room {args.room}")
logger.info("Type your messages and press Enter to send. Type /quit to exit.")
# Start listening for messages
logger.info("Starting to listen for messages...")
# Create a task for syncing with the server
sync_task = asyncio.create_task(client.listen())
try:
# Input loop
while True:
# Use asyncio.run_in_executor to make input non-blocking
message = await asyncio.get_event_loop().run_in_executor(None, lambda: input("> "))
if message.lower() == "/quit":
break
if message:
success = await client.send_message(args.room, message)
if not success:
logger.error(f"Failed to send message to room {args.room}")
except KeyboardInterrupt:
pass
finally:
# Cancel the sync task
sync_task.cancel()
try:
await sync_task
except asyncio.CancelledError:
pass
logger.info("Disconnecting...")
await client.disconnect()
logger.info("Disconnected")
def run_version_command() -> None:
"""
Run the version command.
"""
from . import __version__
print(f"Matrix Command-Line Agent v{__version__}")
print("A command-line agent for interacting with Matrix chat")
def main() -> int:
"""
Main entry point for the CLI.
Returns:
int: Exit code
"""
args = parse_args()
if args.command == "config":
asyncio.run(run_config_command(args))
elif args.command == "connect":
asyncio.run(run_connect_command(args))
elif args.command == "send":
asyncio.run(run_send_command(args))
elif args.command == "listen":
asyncio.run(run_listen_command(args))
elif args.command == "interactive":
asyncio.run(run_interactive_command(args))
elif args.command == "version":
run_version_command()
else:
# No command specified, show help
print("No command specified. Use one of the following commands:")
print(" config - Configure the Matrix agent")
print(" connect - Connect to a Matrix server")
print(" send - Send a message to a Matrix room")
print(" listen - Listen for messages in joined rooms")
print(" interactive - Start an interactive session")
print(" version - Show version information")
print("\nRun 'matrix-cmd <command> --help' for more information on a command")
return 1
return 0
if __name__ == "__main__":
sys.exit(main())