FastMCP调试测试与生产级部署实战

# 创建项目目录
mkdir mcp-file-analyzer && cd mcp-file-analyzer

# 使用uv初始化（推荐）
uv init
uv add fastmcp

# 创建标准项目布局
mkdir -p src/server tests
touch src/server/__init__.py
touch src/server/analyzer.py
touch tests/__init__.py
touch tests/test_analyzer.py

mcp-file-analyzer/
├── pyproject.toml
├── README.md
├── src/
│   └── server/
│       ├── __init__.py
│       └── analyzer.py      # MCP服务器主文件
├── tests/
│   ├── __init__.py
│   └── test_analyzer.py     # 测试套件
└── .claude/
    └── settings.json         # Claude Code项目配置

"""MCP File Analyzer Server - 分析文件内容并提供统计信息"""

import os
import stat
from datetime import datetime
from pathlib import Path
from fastmcp import FastMCP, Context

# 创建FastMCP实例
mcp = FastMCP(
    name="file-analyzer",
    version="1.0.0",
    description="分析文件内容、统计代码行数、检测文件类型的MCP服务器"
)


@mcp.tool()
def analyze_file(file_path: str) -> dict:
    """分析指定文件的详细信息，包括大小、类型、权限等。

    Args:
        file_path: 要分析的文件路径（绝对路径）

    Returns:
        包含文件详细信息的字典
    """
    path = Path(file_path).resolve()

    if not path.exists():
        raise ValueError(f"文件不存在: {file_path}")

    if not path.is_file():
        raise ValueError(f"路径不是文件: {file_path}")

    stat_info = path.stat()

    return {
        "name": path.name,
        "extension": path.suffix.lower(),
        "size_bytes": stat_info.st_size,
        "size_human": _human_readable_size(stat_info.st_size),
        "mime_type": _guess_mime(path),
        "permissions": stat.filemode(stat_info.st_mode),
        "modified": datetime.fromtimestamp(stat_info.st_mtime).isoformat(),
        "created": datetime.fromtimestamp(stat_info.st_ctime).isoformat(),
        "parent_dir": str(path.parent),
    }


@mcp.tool()
def count_lines(file_path: str, include_blank: bool = False) -> dict:
    """统计文件的行数信息，包括总行数、空行数、非空行数。

    Args:
        file_path: 文件路径
        include_blank: 是否在统计中包含空行

    Returns:
        行数统计信息
    """
    path = Path(file_path).resolve()

    if not path.exists():
        raise ValueError(f"文件不存在: {file_path}")

    try:
        content = path.read_text(encoding="utf-8")
    except UnicodeDecodeError:
        content = path.read_text(encoding="latin-1")

    lines = content.splitlines()
    blank_lines = sum(1 for line in lines if not line.strip())
    code_lines = len(lines) - blank_lines

    return {
        "total_lines": len(lines),
        "blank_lines": blank_lines,
        "code_lines": code_lines,
        "chars": len(content),
        "avg_line_length": round(len(content) / max(len(lines), 1), 1),
    }


@mcp.tool()
async def search_in_files(
    directory: str, 
    pattern: str, 
    file_extensions: list[str] | None = None,
    ctx: Context = None
) -> dict:
    """在目录中递归搜索包含指定模式的文件。

    Args:
        directory: 搜索的根目录
        pattern: 搜索的文本模式
        file_extensions: 限制搜索的文件扩展名列表（如 [".py", ".js"]）
        ctx: FastMCP上下文对象

    Returns:
        匹配结果摘要
    """
    dir_path = Path(directory).resolve()

    if not dir_path.is_dir():
        raise ValueError(f"目录不存在: {directory}")

    if ctx:
        await ctx.info(f"开始在 {directory} 中搜索 '{pattern}'...")

    matches = []
    files_scanned = 0
    extensions = set(file_extensions) if file_extensions else None

    for file_path in dir_path.rglob("*"):
        if not file_path.is_file():
            continue
        if extensions and file_path.suffix.lower() not in extensions:
            continue
        if file_path.stat().st_size > 10 * 1024 * 1024:  # 跳过 >10MB 文件
            continue

        files_scanned += 1

        try:
            content = file_path.read_text(encoding="utf-8", errors="ignore")
            if pattern.lower() in content.lower():
                line_numbers = [
                    i + 1 for i, line in enumerate(content.splitlines())
                    if pattern.lower() in line.lower()
                ]
                matches.append({
                    "file": str(file_path.relative_to(dir_path)),
                    "match_count": len(line_numbers),
                    "first_matches": line_numbers[:5],
                })
        except Exception:
            continue

        # 进度报告（每50个文件报告一次）
        if ctx and files_scanned % 50 == 0:
            await ctx.info(f"已扫描 {files_scanned} 个文件，找到 {len(matches)} 个匹配...")

    return {
        "pattern": pattern,
        "directory": str(dir_path),
        "files_scanned": files_scanned,
        "files_matched": len(matches),
        "matches": sorted(matches, key=lambda x: x["match_count"], reverse=True)[:20],
    }


@mcp.resource("file://{file_path}")
def read_file_resource(file_path: str) -> str:
    """将文件内容作为MCP Resource暴露给客户端读取。"""
    path = Path(file_path).resolve()
    if not path.exists():
        raise ValueError(f"文件不存在: {file_path}")
    return path.read_text(encoding="utf-8", errors="replace")


@mcp.prompt()
def analyze_code_prompt(file_path: str) -> str:
    """生成代码分析提示词。"""
    return f"""请分析以下代码文件，指出潜在问题和改进建议：

文件路径: {file_path}

请从以下角度分析：
1. 代码质量与可读性
2. 潜在的Bug或安全问题
3. 性能优化建议
4. 最佳实践建议"""


# 辅助函数
def _human_readable_size(size_bytes: int) -> str:
    for unit in ["B", "KB", "MB", "GB", "TB"]:
        if size_bytes < 1024:
            return f"{size_bytes:.1f} {unit}"
        size_bytes /= 1024
    return f"{size_bytes:.1f} PB"


def _guess_mime(path: Path) -> str:
    import mimetypes
    mime, _ = mimetypes.guess_type(str(path))
    return mime or "application/octet-stream"


if __name__ == "__main__":
    mcp.run(transport="stdio")

[project]
name = "mcp-file-analyzer"
version = "1.0.0"
description = "MCP服务器：文件分析与代码统计工具集"
requires-python = ">=3.10"
dependencies = [
    "fastmcp>=2.0",
]

[project.optional-dependencies]
dev = [
    "pytest>=8.0",
    "pytest-asyncio>=0.24",
]

[project.scripts]
mcp-file-analyzer = "server.analyzer:main"

[build-system]
requires = ["hatchling"]
build-backend = "hatchling.build"

# 方法一：直接使用npx运行（推荐）
npx @modelcontextprotocol/inspector uv run src/server/analyzer.py

# 方法二：全局安装后运行
npm install -g @modelcontextprotocol/inspector
mcp-inspector uv run src/server/analyzer.py

# 方法三：指定自定义端口
npx @modelcontextprotocol/inspector \
  --port 6274 \
  uv run src/server/analyzer.py

🔍 MCP Inspector is running
   Client: http://localhost:6274
   Server proxy: http://localhost:6275

{
  "file_path": "/home/user/example.py"
}

{
  "file_path": "/nonexistent/file.txt"
}

# 通过Inspector传递环境变量给服务器
npx @modelcontextprotocol/inspector \
  -e LOG_LEVEL=debug \
  -e MAX_FILE_SIZE=52428800 \
  uv run src/server/analyzer.py

┌─────────────────────────────────────┐
│        MCP Client Logging           │  ← 通过Context发送到客户端
│   (ctx.info/debug/warning/error)    │
├─────────────────────────────────────┤
│     Server-side Python Logging      │  ← 标准logging模块
│   (logging.getLogger("fastmcp"))    │
└─────────────────────────────────────┘

import logging

# 配置服务器端日志
logger = logging.getLogger("mcp-file-analyzer")
logger.setLevel(logging.DEBUG)

# 文件日志处理器
file_handler = logging.FileHandler("mcp-analyzer.log")
file_handler.setLevel(logging.DEBUG)
file_handler.setFormatter(
    logging.Formatter("%(asctime)s [%(levelname)s] %(name)s: %(message)s")
)
logger.addHandler(file_handler)

# 控制台日志处理器（调试用）
console_handler = logging.StreamHandler()
console_handler.setLevel(logging.INFO)
console_handler.setFormatter(
    logging.Formatter("[%(levelname)s] %(message)s")
)
logger.addHandler(console_handler)

@mcp.tool()
async def search_in_files(
    directory: str,
    pattern: str,
    file_extensions: list[str] | None = None,
    max_results: int = 50,
    ctx: Context = None
) -> dict:
    """在目录中递归搜索包含指定模式的文件。

    Args:
        directory: 搜索的根目录
        pattern: 搜索的文本模式
        file_extensions: 限制搜索的文件扩展名列表
        max_results: 最大返回结果数（默认50，上限200）
        ctx: FastMCP上下文

    Returns:
        匹配结果摘要
    """
    # 参数校验
    max_results = min(max(1, max_results), 200)
    if not pattern.strip():
        raise ValueError("搜索模式不能为空")

    dir_path = Path(directory).resolve()

    if not dir_path.is_dir():
        raise ValueError(f"目录不存在: {directory}")

    # 服务端日志（写入文件）
    logger.info(
        "搜索请求: dir=%s, pattern='%s', exts=%s",
        dir_path, pattern, file_extensions
    )

    # 客户端日志（发送给Claude Code）
    if ctx:
        await ctx.info(f"🔍 开始搜索 '{pattern}'...")

    matches = []
    files_scanned = 0
    errors = []
    extensions = set(file_extensions) if file_extensions else None

    for file_path in dir_path.rglob("*"):
        if not file_path.is_file():
            continue
        if extensions and file_path.suffix.lower() not in extensions:
            continue

        try:
            file_size = file_path.stat().st_size
        except OSError as e:
            errors.append(f"无法读取文件信息 {file_path}: {e}")
            continue

        if file_size > 10 * 1024 * 1024:
            continue

        files_scanned += 1

        try:
            content = file_path.read_text(encoding="utf-8", errors="ignore")
            if pattern.lower() in content.lower():
                line_numbers = [
                    i + 1 for i, line in enumerate(content.splitlines())
                    if pattern.lower() in line.lower()
                ]
                matches.append({
                    "file": str(file_path.relative_to(dir_path)),
                    "match_count": len(line_numbers),
                    "first_matches": line_numbers[:5],
                })
        except PermissionError:
            errors.append(f"权限不足: {file_path}")
            logger.warning("权限不足: %s", file_path)
        except Exception as e:
            errors.append(f"读取错误 {file_path}: {e}")
            logger.debug("读取文件失败: %s - %s", file_path, e)
            continue

        # 进度报告
        if ctx and files_scanned % 100 == 0:
            await ctx.info(
                f"📊 已扫描 {files_scanned} 个文件，"
                f"匹配 {len(matches)} 个..."
            )

    result = {
        "pattern": pattern,
        "directory": str(dir_path),
        "files_scanned": files_scanned,
        "files_matched": len(matches),
        "matches": sorted(
            matches, key=lambda x: x["match_count"], reverse=True
        )[:max_results],
        "errors_count": len(errors),
        "errors_sample": errors[:5],
    }

    # 服务端日志
    logger.info(
        "搜索完成: 扫描 %d 文件，匹配 %d，错误 %d",
        files_scanned, len(matches), len(errors)
    )

    # 客户端通知
    if ctx:
        await ctx.info(
            f"✅ 搜索完成！找到 {len(matches)} 个匹配文件"
        )

    return result

from functools import wraps

def handle_tool_errors(func):
    """统一错误处理装饰器"""
    @wraps(func)
    async def wrapper(*args, **kwargs):
        try:
            return await func(*args, **kwargs)
        except ValueError as e:
            logger.warning("参数错误 in %s: %s", func.__name__, e)
            return {"error": str(e), "type": "validation_error"}
        except PermissionError as e:
            logger.error("权限错误 in %s: %s", func.__name__, e)
            return {"error": str(e), "type": "permission_error"}
        except FileNotFoundError as e:
            logger.error("文件不存在 in %s: %s", func.__name__, e)
            return {"error": str(e), "type": "not_found_error"}
        except Exception as e:
            logger.exception("未预期错误 in %s: %s", func.__name__, e)
            return {"error": "内部服务器错误", "type": "internal_error"}
    return wrapper

graph TD
    A[测试金字塔] --> B[单元测试<br/>测试单个函数逻辑]
    A --> C[工具集成测试<br/>通过FastMCP Client调用]
    A --> D[端到端测试<br/>通过Claude Code实际调用]

    B --> B1["pytest + 直接函数调用"]
    C --> C1["FastMCP Client + stdio"]
    D --> D1["Claude Code -p 模式"]

    style B fill:#4CAF50,color:white
    style C fill:#2196F3,color:white
    style D fill:#FF9800,color:white

"""MCP File Analyzer 单元测试"""

import pytest
import tempfile
from pathlib import Path
from server.analyzer import (
    analyze_file,
    count_lines,
    _human_readable_size,
    _guess_mime,
)


@pytest.fixture
def sample_file(tmp_path):
    """创建临时测试文件"""
    file = tmp_path / "test.py"
    file.write_text(
        '"""模块文档"""\n\n'
        'def hello():\n'
        '    print("Hello")\n'
        '\n'
        '# 注释行\n'
        'class Foo:\n'
        '    pass\n',
        encoding="utf-8"
    )
    return file


@pytest.fixture
def large_file(tmp_path):
    """创建大文件"""
    file = tmp_path / "large.txt"
    lines = [f"Line {i}\n" for i in range(1000)]
    file.write_text("".join(lines), encoding="utf-8")
    return file


class TestAnalyzeFile:
    def test_basic_analysis(self, sample_file):
        result = analyze_file(str(sample_file))

        assert result["name"] == "test.py"
        assert result["extension"] == ".py"
        assert result["size_bytes"] > 0
        assert "modified" in result
        assert "created" in result
        assert "permissions" in result

    def test_nonexistent_file(self):
        with pytest.raises(ValueError, match="文件不存在"):
            analyze_file("/nonexistent/path.txt")

    def test_directory_instead_of_file(self, tmp_path):
        with pytest.raises(ValueError, match="不是文件"):
            analyze_file(str(tmp_path))


class TestCountLines:
    def test_line_counting(self, sample_file):
        result = count_lines(str(sample_file))

        assert result["total_lines"] == 7
        assert result["blank_lines"] == 1
        assert result["code_lines"] == 6
        assert result["chars"] > 0

    def test_large_file(self, large_file):
        result = count_lines(str(large_file))

        assert result["total_lines"] == 1000
        assert result["code_lines"] == 1000
        assert result["blank_lines"] == 0

    def test_include_blank_flag(self, sample_file):
        result = count_lines(str(sample_file), include_blank=True)
        assert "total_lines" in result


class TestHelperFunctions:
    @pytest.mark.parametrize("size,expected", [
        (0, "0.0 B"),
        (1023, "1023.0 B"),
        (1024, "1.0 KB"),
        (1048576, "1.0 MB"),
        (1073741824, "1.0 GB"),
    ])
    def test_human_readable_size(self, size, expected):
        assert _human_readable_size(size) == expected

    def test_guess_mime_python(self):
        assert _guess_mime(Path("test.py")) == "text/x-python"

    def test_guess_mime_unknown(self):
        result = _guess_mime(Path("test.xyz123"))
        assert result == "application/octet-stream"

"""MCP集成测试 - 通过FastMCP Client与服务器交互"""

import pytest
import tempfile
from pathlib import Path
from fastmcp import Client


# 获取MCP服务器实例
from server.analyzer import mcp


@pytest.fixture
def test_workspace(tmp_path):
    """创建完整的测试工作区"""
    # Python文件
    (tmp_path / "main.py").write_text(
        "import os\n\nprint(os.getcwd())\n"
    )
    # JavaScript文件
    (tmp_path / "app.js").write_text(
        "const x = 1;\n// comment\nconsole.log(x);\n"
    )
    # 空文件
    (tmp_path / "empty.txt").write_text("")
    # 子目录
    sub = tmp_path / "subdir"
    sub.mkdir()
    (sub / "helper.py").write_text("def help():\n    pass\n")
    return tmp_path


@pytest.mark.asyncio
async def test_tool_list():
    """测试工具列表是否正确注册"""
    async with Client(mcp) as client:
        tools = await client.list_tools()

        tool_names = [t.name for t in tools]
        assert "analyze_file" in tool_names
        assert "count_lines" in tool_names
        assert "search_in_files" in tool_names


@pytest.mark.asyncio
async def test_analyze_file_via_mcp(test_workspace):
    """通过MCP协议调用analyze_file"""
    async with Client(mcp) as client:
        result = await client.call_tool(
            "analyze_file",
            {"file_path": str(test_workspace / "main.py")}
        )

        assert result is not None
        # 验证返回结构
        data = result.data if hasattr(result, 'data') else result
        assert data["extension"] == ".py"


@pytest.mark.asyncio
async def test_count_lines_via_mcp(test_workspace):
    """通过MCP协议调用count_lines"""
    async with Client(mcp) as client:
        result = await client.call_tool(
            "count_lines",
            {"file_path": str(test_workspace / "app.js")}
        )

        data = result.data if hasattr(result, 'data') else result
        assert data["total_lines"] == 3


@pytest.mark.asyncio
async def test_search_in_files_via_mcp(test_workspace):
    """通过MCP协议调用search_in_files"""
    async with Client(mcp) as client:
        result = await client.call_tool(
            "search_in_files",
            {
                "directory": str(test_workspace),
                "pattern": "import",
                "file_extensions": [".py"]
            }
        )

        data = result.data if hasattr(result, 'data') else result
        assert data["files_matched"] >= 1
        assert data["files_scanned"] >= 2  # main.py + subdir/helper.py


@pytest.mark.asyncio
async def test_resource_reading(test_workspace):
    """测试Resource读取"""
    async with Client(mcp) as client:
        resources = await client.list_resources()
        # Resource模板应可被发现
        assert len(resources) >= 0  # 模板Resource可能不直接列出


@pytest.mark.asyncio
async def test_error_handling_via_mcp():
    """测试通过MCP的错误处理"""
    async with Client(mcp) as client:
        # 调用不存在的文件应返回错误
        result = await client.call_tool(
            "analyze_file",
            {"file_path": "/nonexistent/file.txt"}
        )
        # FastMCP会将异常转化为错误响应
        assert result is not None

# 运行所有测试
uv run pytest tests/ -v

# 只运行单元测试
uv run pytest tests/test_analyzer.py -v

# 只运行集成测试
uv run pytest tests/test_mcp_integration.py -v

# 带覆盖率报告
uv run pytest tests/ -v --cov=server --cov-report=html

# 监视模式（开发时使用）
uv run pytest-watch tests/ -v

tests/test_analyzer.py::TestAnalyzeFile::test_basic_analysis PASSED
tests/test_analyzer.py::TestAnalyzeFile::test_nonexistent_file PASSED
tests/test_analyzer.py::TestAnalyzeFile::test_directory_instead_of_file PASSED
tests/test_analyzer.py::TestCountLines::test_line_counting PASSED
tests/test_analyzer.py::TestCountLines::test_large_file PASSED
tests/test_mcp_integration.py::test_tool_list PASSED
tests/test_mcp_integration.py::test_analyze_file_via_mcp PASSED
tests/test_mcp_integration.py::test_search_in_files_via_mcp PASSED
======================== 12 passed in 2.34s ========================

{
  "mcpServers": {
    "file-analyzer": {
      "command": "uv",
      "args": [
        "--directory",
        "/path/to/mcp-file-analyzer",
        "run",
        "src/server/analyzer.py"
      ],
      "env": {
        "LOG_LEVEL": "INFO"
      }
    }
  }
}

claude mcp add file-analyzer \
  uv --directory /path/to/mcp-file-analyzer run src/server/analyzer.py \
  -e LOG_LEVEL=INFO

# 列出已配置的MCP服务器
claude mcp list

# 查看特定服务器的工具列表
claude mcp tools file-analyzer

你: 分析一下 /home/user/project/main.py 的文件信息

Claude Code 会自动调用 file-analyzer 的 analyze_file 工具，
返回文件的详细分析结果。

你: 帮我在 /home/user/project 目录下搜索所有包含 "TODO" 的 Python 文件

Claude Code 会调用 search_in_files 工具，限定 .py 扩展名。

# 脱离Claude Code，单独用Inspector测试
npx @modelcontextprotocol/inspector uv run src/server/analyzer.py

graph LR
    subgraph "本地开发"
        A[Claude Code CLI] -->|stdio| B[MCP Server<br/>analyzer.py]
    end

    subgraph "团队共享"
        C[Claude Code] -->|stdio| D[uvx 启动<br/>从PyPI安装]
        D --> B2[MCP Server]
    end

    subgraph "远程服务"
        E[Claude Code] -->|SSE/HTTP| F[Nginx 反向代理]
        F --> G[MCP Server<br/>SSE模式]
        G --> H[(共享存储)]
    end

    style A fill:#4CAF50,color:white
    style C fill:#2196F3,color:white
    style E fill:#FF9800,color:white

# 打包发布到PyPI（团队内部分发）
uv build
uv publish --index internal

# 团队成员安装使用
claude mcp add file-analyzer -- uvx mcp-file-analyzer

# src/server/analyzer_sse.py
from server.analyzer import mcp

if __name__ == "__main__":
    # SSE模式，监听指定端口
    mcp.run(
        transport="sse",
        host="0.0.0.0",
        port=8080,
    )

# Dockerfile
FROM python:3.12-slim

WORKDIR /app
COPY pyproject.toml .
COPY src/ src/

RUN pip install --no-cache-dir .

EXPOSE 8080

CMD ["python", "-m", "server.analyzer_sse"]

# 构建和运行
docker build -t mcp-file-analyzer .
docker run -d -p 8080:8080 \
  -v /shared/data:/data \
  --name file-analyzer \
  mcp-file-analyzer

{
  "mcpServers": {
    "file-analyzer-remote": {
      "type": "sse",
      "url": "http://file-analyzer.internal:8080/sse"
    }
  }
}

# 1. 路径遍历防护
ALLOWED_DIRS = [Path("/data/workspace").resolve()]

def validate_path(file_path: str) -> Path:
    """防止路径遍历攻击"""
    path = Path(file_path).resolve()
    if not any(path.is_relative_to(allowed) for allowed in ALLOWED_DIRS):
        raise ValueError(f"路径不在允许范围内: {file_path}")
    return path

# 2. 资源限制
MAX_FILE_SIZE = 50 * 1024 * 1024  # 50MB
MAX_SEARCH_RESULTS = 200
MAX_CONCURRENT_REQUESTS = 10

# 3. 输入校验
import re

def validate_pattern(pattern: str) -> str:
    """防止ReDoS攻击"""
    if len(pattern) > 500:
        raise ValueError("搜索模式过长")
    return pattern

# 4. 速率限制（SSE模式）
from time import time
_request_times: list[float] = []

def check_rate_limit():
    """简单的滑动窗口速率限制"""
    now = time()
    _request_times[:] = [t for t in _request_times if now - t < 60]
    if len(_request_times) >= MAX_CONCURRENT_REQUESTS * 6:
        raise ValueError("请求过于频繁，请稍后再试")
    _request_times.append(now)

flowchart TB
    START([开始开发MCP服务器]) --> INIT[uv init + uv add fastmcp]
    INIT --> CODE[编写Tool/Resource/Prompt]
    CODE --> UNIT_TEST[编写单元测试<br/>pytest]
    UNIT_TEST --> PASS1{测试通过?}
    PASS1 -->|否| CODE
    PASS1 -->|是| INSPECTOR[MCP Inspector<br/>可视化调试]
    INSPECTOR --> INTEGRATION[编写集成测试<br/>FastMCP Client]
    INTEGRATION --> PASS2{测试通过?}
    PASS2 -->|否| CODE
    PASS2 -->|是| CONNECT[接入Claude Code<br/>claude mcp add]
    CONNECT --> VERIFY[端到端验证<br/>自然语言调用]
    VERIFY --> PROD{部署方式?}
    PROD -->|个人/小团队| LOCAL[uvx 本地部署<br/>stdio传输]
    PROD -->|企业/远程| REMOTE[Docker + SSE<br/>远程部署]
    LOCAL --> DONE([🎉 上线])
    REMOTE --> DONE

    style START fill:#4CAF50,color:white
    style DONE fill:#FF9800,color:white
    style INSPECTOR fill:#2196F3,color:white

"""Code Review MCP Server - 自动化代码质量审查"""

import ast
import re
from pathlib import Path
from dataclasses import dataclass, asdict
from fastmcp import FastMCP, Context

mcp = FastMCP(
    name="code-reviewer",
    version="1.0.0",
    description="Python代码质量审查MCP服务器"
)

# 定义问题严重级别
SEVERITY_INFO = "info"
SEVERITY_WARNING = "warning"
SEVERITY_ERROR = "error"


@dataclass
class CodeIssue:
    """代码问题"""
    file: str
    line: int
    severity: str
    rule: str
    message: str


def _check_line_length(lines: list[str], filename: str, limit: int = 120) -> list[CodeIssue]:
    """检查行长度"""
    issues = []
    for i, line in enumerate(lines, 1):
        if len(line.rstrip()) > limit:
            issues.append(CodeIssue(
                file=filename, line=i,
                severity=SEVERITY_WARNING,
                rule="line-too-long",
                message=f"行长度 {len(line.rstrip())} 超过限制 {limit}"
            ))
    return issues


def _check_todo_fixme(lines: list[str], filename: str) -> list[CodeIssue]:
    """检查TODO/FIXME标记"""
    issues = []
    pattern = re.compile(r'#\s*(TODO|FIXME|HACK|XXX)', re.IGNORECASE)
    for i, line in enumerate(lines, 1):
        match = pattern.search(line)
        if match:
            issues.append(CodeIssue(
                file=filename, line=i,
                severity=SEVERITY_INFO,
                rule=f"marker-{match.group(1).lower()}",
                message=f"发现 {match.group(1)} 标记: {line.strip()}"
            ))
    return issues


def _check_python_ast(code: str, filename: str) -> list[CodeIssue]:
    """使用AST进行语法级别检查"""
    issues = []
    try:
        tree = ast.parse(code, filename)
    except SyntaxError as e:
        issues.append(CodeIssue(
            file=filename, line=e.lineno or 0,
            severity=SEVERITY_ERROR,
            rule="syntax-error",
            message=f"语法错误: {e.msg}"
        ))
        return issues

    for node in ast.walk(tree):
        # 检查过深的嵌套
        if isinstance(node, (ast.If, ast.For, ast.While)):
            depth = 0
            parent = node
            while hasattr(parent, 'parent'):
                depth += 1
                parent = getattr(parent, 'parent', None)
            if depth > 4:
                issues.append(CodeIssue(
                    file=filename, line=node.lineno,
                    severity=SEVERITY_WARNING,
                    rule="deep-nesting",
                    message=f"嵌套深度 {depth} 超过建议值 4"
                ))

        # 检查过长的参数列表
        if isinstance(node, ast.FunctionDef):
            if len(node.args.args) > 7:
                issues.append(CodeIssue(
                    file=filename, line=node.lineno,
                    severity=SEVERITY_WARNING,
                    rule="too-many-params",
                    message=f"函数 '{node.name}' 有 {len(node.args.args)} 个参数（建议 ≤7）"
                ))

            # 检查缺失docstring
            docstring = ast.get_docstring(node)
            if not docstring and not node.name.startswith("_"):
                issues.append(CodeIssue(
                    file=filename, line=node.lineno,
                    severity=SEVERITY_INFO,
                    rule="missing-docstring",
                    message=f"公开函数 '{node.name}' 缺少docstring"
                ))

    return issues


@mcp.tool()
async def review_python_file(
    file_path: str,
    checks: list[str] | None = None,
    ctx: Context = None
) -> dict:
    """审查Python文件的代码质量。

    Args:
        file_path: Python文件路径
        checks: 要执行的检查项列表，默认全部。
                可选: line-length, todo-markers, ast-checks
        ctx: MCP上下文

    Returns:
        审查结果摘要
    """
    path = Path(file_path).resolve()
    if not path.exists():
        raise ValueError(f"文件不存在: {file_path}")
    if path.suffix != ".py":
        raise ValueError("仅支持Python文件审查")

    if ctx:
        await ctx.info(f"📋 开始审查: {path.name}")

    code = path.read_text(encoding="utf-8", errors="replace")
    lines = code.splitlines()

    all_checks = checks or ["line-length", "todo-markers", "ast-checks"]
    issues: list[CodeIssue] = []

    if "line-length" in all_checks:
        issues.extend(_check_line_length(lines, path.name))
    if "todo-markers" in all_checks:
        issues.extend(_check_todo_fixme(lines, path.name))
    if "ast-checks" in all_checks:
        issues.extend(_check_python_ast(code, path.name))

    # 统计
    summary = {
        "file": str(path),
        "total_lines": len(lines),
        "total_issues": len(issues),
        "by_severity": {
            "error": sum(1 for i in issues if i.severity == SEVERITY_ERROR),
            "warning": sum(1 for i in issues if i.severity == SEVERITY_WARNING),
            "info": sum(1 for i in issues if i.severity == SEVERITY_INFO),
        },
        "issues": [asdict(i) for i in sorted(issues, key=lambda x: (
            {"error": 0, "warning": 1, "info": 2}[x.severity],
            x.line
        ))],
    }

    if ctx:
        await ctx.info(
            f"✅ 审查完成: {summary['total_issues']} 个问题 "
            f"(🔴{summary['by_severity']['error']} "
            f"🟡{summary['by_severity']['warning']} "
            f"ℹ️{summary['by_severity']['info']})"
        )

    return summary


@mcp.tool()
async def review_directory(
    directory: str,
    ctx: Context = None
) -> dict:
    """批量审查目录下所有Python文件的代码质量。

    Args:
        directory: 目标目录路径
        ctx: MCP上下文

    Returns:
        批量审查结果
    """
    dir_path = Path(directory).resolve()
    if not dir_path.is_dir():
        raise ValueError(f"目录不存在: {directory}")

    py_files = list(dir_path.rglob("*.py"))
    if ctx:
        await ctx.info(f"📂 发现 {len(py_files)} 个Python文件")

    results = []
    for i, py_file in enumerate(py_files):
        try:
            result = await review_python_file(str(py_file), ctx=ctx)
            results.append(result)
        except Exception as e:
            results.append({
                "file": str(py_file),
                "error": str(e)
            })

        if ctx and (i + 1) % 10 == 0:
            await ctx.info(f"进度: {i + 1}/{len(py_files)}")

    total_issues = sum(r.get("total_issues", 0) for r in results)
    return {
        "directory": str(dir_path),
        "files_reviewed": len(results),
        "total_issues": total_issues,
        "file_results": results,
    }


if __name__ == "__main__":
    mcp.run(transport="stdio")

# 添加到Claude Code
claude mcp add code-reviewer \
  uv --directory /path/to/project run src/server/code_reviewer.py

# 在Claude Code中使用
# 你: "帮我审查一下 main.py 的代码质量"
# 你: "审查整个 src 目录的Python代码"