feat: create task warrior mcp server

1 files changed, 107 insertions, 0 deletions

diff --git a/servers/taskwarrior/src/mcp_server_taskwarrior/cli.py b/servers/taskwarrior/src/mcp_server_taskwarrior/cli.py
new file mode 100644
index 0000000..1c556ff
--- /dev/null
+++ b/servers/taskwarrior/src/mcp_server_taskwarrior/cli.py
@@ -0,0 +1,107 @@
+#!/usr/bin/env python3
+"""
+Common CLI functionality for the TaskWarrior MCP server.
+"""
+
+import argparse
+import asyncio
+import logging
+import sys
+
+from .server import main
+
+logger = logging.getLogger("mcp_taskwarrior_server")
+
+
+def parse_args() -> argparse.Namespace:
+    """Parse command-line arguments."""
+    parser = argparse.ArgumentParser(description="TaskWarrior MCP Server")
+    parser.add_argument(
+        "--transport",
+        choices=["stdio", "remote"],
+        default="stdio",
+        help="Transport type (stdio or remote)",
+    )
+    parser.add_argument(
+        "--host",
+        default="127.0.0.1",
+        help="Host to bind to for remote transport (default: 127.0.0.1)",
+    )
+    parser.add_argument(
+        "--port",
+        type=int,
+        default=8080,
+        help="Port to bind to for remote transport (default: 8080)",
+    )
+    parser.add_argument(
+        "--log-level",
+        choices=["DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"],
+        default="INFO",
+        help="Set the logging level (default: INFO)",
+    )
+    return parser.parse_args()
+
+
+def validate_args(args: argparse.Namespace) -> argparse.Namespace:
+    """Validate command-line arguments."""
+    if (
+        args.transport == "remote"
+        and args.port < 1024
+        and not sys.platform.startswith("win")
+    ):
+        logger.warning(
+            "Using a port below 1024 may require root privileges on Unix-like systems."
+        )
+    return args
+
+
+def setup_logging(level: str, transport: str) -> None:
+    """Set up logging configuration.
+
+    Args:
+        level: The base logging level from command line args
+        transport: The transport type being used (stdio or remote)
+    """
+    # Create logs directory if it doesn't exist
+    import os
+
+    os.makedirs("logs", exist_ok=True)
+
+    # Configure file handler for all logs
+    file_handler = logging.FileHandler("logs/mcp_server.log")
+    file_handler.setLevel(getattr(logging, level))
+    file_handler.setFormatter(
+        logging.Formatter("%(asctime)s - %(name)s - %(levelname)s - %(message)s")
+    )
+
+    # Configure console handler with appropriate level
+    console_handler = logging.StreamHandler()
+    # Use WARNING level for stdio transport to reduce interference
+    if transport == "stdio":
+        console_handler.setLevel(logging.WARNING)
+    else:
+        console_handler.setLevel(getattr(logging, level))
+    console_handler.setFormatter(
+        logging.Formatter("%(asctime)s - %(name)s - %(levelname)s - %(message)s")
+    )
+
+    # Configure root logger
+    root_logger = logging.getLogger()
+    root_logger.setLevel(getattr(logging, level))  # Base level for all handlers
+    root_logger.handlers = []  # Remove any existing handlers
+    root_logger.addHandler(file_handler)
+    root_logger.addHandler(console_handler)
+
+
+def run_server() -> None:
+    """Run the server with CLI arguments."""
+    args = validate_args(parse_args())
+    setup_logging(args.log_level, args.transport)
+
+    try:
+        asyncio.run(main(args.transport, args.host, args.port))
+    except KeyboardInterrupt:
+        logger.info("Server stopped by user")
+    except Exception as e:
+        logger.error(f"Error running server: {e}")
+        sys.exit(1)