FS MCP Server

基于FastMCP构建的文件上传/下载MCP服务，集成RustFS SDK实现AI自动调用文件存储和下载功能。

功能特性

• 文件上传: 支持上传本地文件到RustFS存储服务
• 文件下载: 支持从任意HTTP/HTTPS URL下载文件到本地
• 异步处理: 基于asyncio的异步文件操作
• 错误处理: 完善的错误处理和异常管理
• 配置验证: 启动时验证必需的环境变量配置

安装

1. 克隆项目

git clone <repository-url>
cd fs_mcp

2. 安装依赖

pip install -e .

或安装开发依赖：

pip install -e ".[dev]"

3. 配置环境变量

复制环境变量模板并配置：

cp .env.example .env

编辑 .env 文件：

# RustFS配置
FS_URL=https://your-rustfs-endpoint.com
FS_AK=your-access-key
FS_SK=your-secret-key

# 可选配置
TIMEOUT=30

使用方法

启动服务

# 直接运行
python -m src.server

# 或使用模块方式
python -m src

MCP工具

1. upload_file

上传本地文件到RustFS存储服务。

参数:

• file_path (string): 本地文件的绝对路径

返回值:

{
  "success": true,
  "filename": "example.txt",
  "size": 1024,
  "content_type": "text/plain",
  "access_url": "https://fs.example.com/files/example.txt",
  "file_id": "example.txt",
  "message": "文件 'example.txt' 上传成功"
}

2. download_file

从指定URL下载文件到本地路径。

参数:

• url (string): 要下载的文件URL
• download_path (string): 本地保存路径

返回值:

{
  "success": true,
  "url": "https://example.com/file.pdf",
  "file_path": "/path/to/save/file.pdf",
  "filename": "file.pdf",
  "size": 2048000,
  "content_type": "application/pdf",
  "message": "文件 'file.pdf' 下载成功"
}

使用示例

文件上传示例

# 通过MCP客户端调用上传工具
result = await mcp_client.call_tool("upload_file", {
    "file_path": "/home/user/documents/report.pdf"
})

文件下载示例

# 下载文件到指定目录
result = await mcp_client.call_tool("download_file", {
    "url": "https://example.com/data.csv",
    "download_path": "/home/user/downloads/"
})

# 下载文件到指定路径
result = await mcp_client.call_tool("download_file", {
    "url": "https://example.com/image.png",
    "download_path": "/home/user/downloads/saved_image.png"
})

错误处理

服务提供详细的错误信息：

常见错误类型

• FileNotFoundError: 文件不存在
• ValueError: 参数无效或URL格式错误
• RuntimeError: 上传/下载操作失败
• ConfigurationError: 环境变量配置错误

错误示例

# 文件不存在
try:
    await mcp_client.call_tool("upload_file", {
        "file_path": "/nonexistent/file.txt"
    })
except FileNotFoundError as e:
    print(f"错误: {e}")

# URL无效
try:
    await mcp_client.call_tool("download_file", {
        "url": "invalid-url",
        "download_path": "/tmp/"
    })
except ValueError as e:
    print(f"错误: {e}")

开发

项目结构

fs_mcp/
├── src/
│   ├── __init__.py          # 包初始化
│   ├── __main__.py          # 命令行入口
│   ├── server.py            # MCP服务器主程序
│   ├── config.py            # 配置管理
│   ├── rustfs_client.py     # RustFS客户端
│   ├── upload_tool.py       # 上传工具
│   ├── download_tool.py     # 下载工具
│   ├── exceptions.py        # 自定义异常
│   └── utils.py             # 工具函数
├── pyproject.toml           # 项目配置
├── .env.example             # 环境变量模板
└── README.md               # 项目文档

运行测试

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

# 运行测试
pytest

代码格式化

# 使用black格式化代码
black src/

# 使用ruff检查代码质量
ruff check src/

配置说明

必需环境变量

• FS_URL: RustFS服务端点URL
• FS_AK: RustFS访问密钥
• FS_SK: RustFS密钥

可选环境变量

• TIMEOUT: 请求超时时间（秒），默认30

RustFS API要求

本服务假设RustFS提供以下API端点：

• POST /api/upload: 文件上传
• GET /api/files/{file_id}: 获取文件信息

上传请求格式：

• Method: POST
• Content-Type: multipart/form-data
• Headers: Authorization: Bearer {access_key}:{secret_key}
• Files: file (文件内容)
• Data: filename (文件名), size (文件大小)

许可证

MIT License

贡献

欢迎提交Issue和Pull Request来改进这个项目。