SeaTunnel MCP 开发者指南

本指南为 SeaTunnel MCP 项目的开发者提供指导，包括如何设置开发环境、代码贡献流程和测试方法。

开发环境设置

1. 克隆仓库并设置开发环境：

# 克隆仓库
git clone <repository_url>
cd seatunnel-mcp

# 创建并激活虚拟环境
python -m venv .venv
source .venv/bin/activate  # Windows: .venv\Scripts\activate

# 安装开发依赖
pip install -e ".[dev]"

2. 安装预提交钩子（可选但推荐）：

pip install pre-commit
pre-commit install

项目结构

seatunnel-mcp/
├── docs/                 # 文档
├── examples/             # 示例配置文件
├── src/
│   └── seatunnel_mcp/    # 主要源代码
│       ├── __init__.py
│       ├── __main__.py   # 入口点
│       ├── client.py     # SeaTunnel API 客户端
│       ├── tools.py      # MCP 工具定义
│       └── schema.py     # 数据模型定义
├── tests/                # 测试
├── .env.example          # 环境变量示例
├── pyproject.toml        # 项目配置
└── README.md             # 项目说明

开发流程

添加新的工具

1. 首先在 client.py 中为 SeaTunnel API 添加相应的方法
2. 在 tools.py 中创建相应的 MCP 工具
3. 更新 get_all_tools() 函数以包含新工具
4. 添加单元测试

示例：添加一个新的工具来获取日志信息

# 在 client.py 中
def get_job_logs(self, jobId: Union[str, int]) -> Dict[str, Any]:
    """获取作业日志信息。"""
    response = self._make_request("GET", f"/job-logs/{jobId}")
    return response.json()

# 在 tools.py 中
def get_job_logs_tool(client: SeaTunnelClient) -> Tool:
    """获取作业日志的工具。"""
    async def get_job_logs(jobId: Union[str, int]) -> Dict[str, Any]:
        return client.get_job_logs(jobId=jobId)

    return Tool(
        name="get-job-logs",
        description="获取指定作业的日志信息",
        fn=AsyncToolFn(get_job_logs),
        parameters=models.ParametersSchema(
            properties={
                "jobId": models.ParameterSchema(
                    type="string",
                    description="作业 ID",
                ),
            },
            required=["jobId"],
        ),
    )

# 更新 get_all_tools 函数
def get_all_tools(client: SeaTunnelClient) -> List[Tool]:
    return [
        # ... 现有工具
        get_job_logs_tool(client),
    ]

代码风格

• 使用 Black 进行代码格式化
• 使用 isort 排序导入
• 使用 mypy 进行类型检查
• 使用 flake8 进行代码风格检查

你可以使用以下命令执行这些检查：

# 代码格式化
black src tests

# 导入排序
isort src tests

# 类型检查
mypy src

# 代码风格检查
flake8 src tests

测试

单元测试

使用 pytest 运行单元测试：

pytest -xvs tests/

带覆盖率报告：

pytest --cov=src tests/

手动测试

使用 MCP Inspector 进行手动测试：

npx @modelcontextprotocol/inspector python -m src.seatunnel_mcp

集成测试

对于集成测试，你需要一个运行中的 SeaTunnel 实例。可以使用 Docker 启动一个测试实例：

docker run -d --name seatunnel -p 8090:8090 apache/seatunnel:latest

然后运行集成测试：

pytest -xvs tests/integration/

文档

• 所有公共函数、类和方法应有清晰的 docstring
• 遵循 Google Python 风格指南
• 保持 README.md 和用户指南的最新状态

提交 Pull Request

1. 确保所有测试通过
2. 更新相关文档
3. 如果添加了新功能，请同时添加相应的测试
4. 提交 PR 并在描述中详细说明变更内容

发布流程

1. 更新版本号（在 __init__.py 和 pyproject.toml 中）
2. 更新 CHANGELOG.md
3. 创建一个新的 git tag
4. 构建并上传到 PyPI：

python -m build
twine upload dist/*

问题与支持

如有问题或需要支持，可以：

• 提交 GitHub Issue
• 在 PR 中讨论
• 联系项目维护者