1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
|
#!/usr/bin/env python3
"""Common CLI functionality for the GitLab CLI MCP server."""
import argparse
import asyncio
import logging
import os
import sys
from .server import main
logger = logging.getLogger("mcp_gitlab_glab_server")
def parse_args() -> argparse.Namespace:
"""Parse command-line arguments.
Returns:
The parsed command-line arguments.
"""
parser = argparse.ArgumentParser(description="GitLab CLI 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.
Args:
args: The command-line arguments to validate.
Returns:
The validated 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
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)
|
