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
|
#!/usr/bin/env python3
"""
Common CLI functionality for the GitLab Python MCP server.
"""
import argparse
import asyncio
import logging
import os
import sys
from .server import main
logger = logging.getLogger("mcp_gitlab_python_server")
def parse_args() -> argparse.Namespace:
parser = argparse.ArgumentParser(description="GitLab Python 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:
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:
os.makedirs("logs", exist_ok=True)
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"
))
console_handler = logging.StreamHandler()
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"
))
root_logger = logging.getLogger()
root_logger.setLevel(getattr(logging, level))
root_logger.handlers = []
root_logger.addHandler(file_handler)
root_logger.addHandler(console_handler)
def run_server() -> None:
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)
|
