linter-mcp/mcp_stdio_server.py at main · Daniele111222/linter-mcp · GitHub

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
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
#!/usr/bin/env python3
"""MCP STDIO 服务器入口

专门用于MCP协议的标准输入输出通信服务器。
此文件作为MCP客户端（如Trae AI）的入口点，通过stdin/stdout进行JSON-RPC 2.0通信。

配置示例：
{
  "mcpServers": {
    "linter-mcp": {
      "command": "C:\\Users\\朱炜杰\\Desktop\\linter-mcp\\venv\\Scripts\\python.exe",
      "args": [
        "C:\\Users\\朱炜杰\\Desktop\\linter-mcp\\mcp_stdio_server.py"
      ],
      "env": {
        "PYTHONIOENCODING": "utf-8",
        "PYTHONUNBUFFERED": "1"
      }
    }
  }
}
"""

import sys
import os
import asyncio
import json
from pathlib import Path

# 添加项目根目录到Python路径
project_root = Path(__file__).parent
sys.path.insert(0, str(project_root))

# 添加src目录到Python路径并切换工作目录
src_path = project_root / "src"
sys.path.insert(0, str(src_path))
os.chdir(src_path)

try:
    from src.config import config_manager
    from src.mcp_handler import McpHandler
except ImportError as e:
    # 如果导入失败，输出错误并退出
    error_response = {
        "jsonrpc": "2.0",
        "error": {
            "code": -32603,
            "message": f"模块导入失败: {str(e)}"
        },
        "id": None
    }
    print(json.dumps(error_response, ensure_ascii=True))
    sys.exit(1)


def setup_mcp_environment():
    """设置MCP环境 - 极简版本

    只检查配置文件是否存在，不进行复杂的初始化。
    """
    try:
        config_path = project_root / "config.yaml"
        if config_path.exists():
            return True
        else:
            return False
    except Exception:
        return False


async def main():
    """MCP STDIO服务器主入口

    从标准输入读取JSON-RPC请求，处理后输出到标准输出。
    严格遵循MCP协议规范，确保输出格式的纯净性。
    """
    # 设置工作目录为项目根目录
    os.chdir(project_root)

    # 设置环境
    config_exists = setup_mcp_environment()

    if not config_exists:
        # 如果配置文件不存在，输出错误响应并退出
        error_response = {
            "jsonrpc": "2.0",
            "error": {
                "code": -32603,
                "message": "配置文件不存在"
            },
            "id": None
        }
        print(json.dumps(error_response, ensure_ascii=True))
        sys.exit(1)

    # 创建MCP处理器
    mcp_handler = McpHandler()

    # 启动MCP服务器主循环
    try:
        while True:
            # 从stdin读取请求
            line = sys.stdin.readline()
            if not line:
                break

            line = line.strip()
            if not line:
                continue

            # 处理请求并输出响应
            response = mcp_handler.handle_request(line)
            print(response, flush=True)

    except KeyboardInterrupt:
        # 静默处理中断信号
        pass
    except EOFError:
        # 静默处理EOF
        pass
    except Exception:
        # 静默处理所有其他异常，避免输出到控制台
        pass


if __name__ == "__main__":
    # 确保在MCP模式下运行
    try:
        asyncio.run(main())
    except KeyboardInterrupt:
        # 静默处理中断信号，不输出任何信息
        sys.exit(0)
    except Exception:
        # 静默处理所有异常，避免输出到控制台
        sys.exit(1)