Uses environment variables for configuration management, loading settings from .env files to configure the RustFS connection and server parameters
Supports testing framework integration for running automated tests on the file upload and download functionality
Built as a Python-based MCP server providing file management capabilities through Python asyncio operations
Integrates with Ruff for code quality checking and linting during development
Integrates with RustFS storage service through the RustFS SDK, enabling file upload to RustFS endpoints and retrieval of file information
Uses TOML format for project configuration through pyproject.toml file
Click on "Install Server".
Wait a few minutes for the server to deploy. Once ready, it will show a "Started" state.
In the chat, type
@followed by the MCP server name and your instructions, e.g., "@RustFS File Management MCP Serverupload my report.pdf from the documents folder"
That's it! The server will respond to your query, and you can continue using it as needed.
Here is a step-by-step guide with screenshots.
FS MCP Server
基于FastMCP构建的文件上传/下载MCP服务,集成RustFS SDK实现AI自动调用文件存储和下载功能。
功能特性
文件上传: 支持上传本地文件到RustFS存储服务
文件下载: 支持从任意HTTP/HTTPS URL下载文件到本地
异步处理: 基于asyncio的异步文件操作
错误处理: 完善的错误处理和异常管理
配置验证: 启动时验证必需的环境变量配置
安装
1. 克隆项目
git clone <repository-url>
cd fs_mcp2. 安装依赖
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 srcMCP工具
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): 要下载的文件URLdownload_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服务端点URLFS_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来改进这个项目。