CSV到PostgreSQL MCP服务器
一个MCP(模型上下文协议)服务器,将CSV文件加载到PostgreSQL数据库中,并进行验证和进度跟踪。
特性
- CSV验证:验证CSV结构并提供详细的错误消息
- 有效装载:使用PostgreSQL COPY命令进行快速批量加载
- 进度跟踪:显示长时间运行的导入的进度条
- 灵活的配置:可选数据库名称(默认为
csvimports) - 错误报告:验证错误的确切行号和列信息
安装
来自PyPI
pip install mcp-csv-postgres或使用 uv:
uv pip install mcp-csv-postgres来源
克隆存储库并安装 uv:
git clone https://github.com/raviramadoss/mcp-csv-postgres.git
cd mcp-csv-postgres
uv sync --all-extras用法
作为MCP服务器
选项1:从PyPI安装后
添加到您的Claude Desktop配置文件(~/Library/Application Support/Claude/claude_desktop_config.json 在macOS上):
{
"mcpServers": {
"mcp-csv-postgres": {
"command": "uv",
"args": [
"tool",
"run",
"--from",
"mcp-csv-postgres",
"mcp-csv-postgres"
]
}
}
}方案2:促进地方发展
{
"mcpServers": {
"mcp-csv-postgres": {
"command": "uv",
"args": [
"tool",
"run",
"--from",
"/path/to/mcp-csv-postgres",
"--python",
"3.10",
"mcp-csv-postgres"
]
}
}
}或者直接从命令行运行:
mcp-csv-postgres工具: load_csv_to_postgres
将CSV文件加载到PostgreSQL数据库中。
参数:
file_path(必填):CSV文件的路径dbname(可选):数据库名称(默认值:csvimports)host(可选):PostgreSQL主机(默认:localhost)port(可选):PostgreSQL端口(默认:5432)user(可选):PostgreSQL用户(默认值:postgres)password(可选):PostgreSQL密码table_name(可选):表名(如果未提供,则从文件名派生)
退货:
- 成功消息,其中包含数据库名称、表名称和已加载的行
- 带有详细验证或数据库错误的错误消息
测试服务器
快速测试
运行测试脚本以查看服务器的运行情况:
uv run python test_server.py这将:
- 创建示例CSV文件
- 测试将它们加载到PostgreSQL中
- 测试验证错误报告
- 显示详细输出
使用样本数据进行手动测试
CSV文件示例见 sample_data.csv.手动测试:
# In Python/IPython
import asyncio
from mcp_csv_postgres.server import call_tool
result = asyncio.run(call_tool(
"load_csv_to_postgres",
{"file_path": "sample_data.csv"}
))
print(result[0].text)与MCP检查器一起使用
使用MCP检查器工具进行测试:
npx @modelcontextprotocol/inspector uv run mcp-csv-postgresVS代码启动配置
这 .vscode/launch.json 包括:
- 运行MCP服务器:以调试模式启动服务器
- 使用CSV示例测试MCP服务器:运行测试脚本
- 运行所有测试:执行完整的测试套件
- 运行特定测试:调试单个测试文件
发展
运行测试
uv run pytest代码覆盖率
uv run pytest --cov-report=html保险范围报告将于 htmlcov/index.html.
当前覆盖率:>90%
预提交钩子
预提交挂钩被配置为在每次提交之前运行测试:
uv run pre-commit install
uv run pre-commit run --all-files建筑
模块
- 验证器.py:CSV验证,并提供详细的错误报告
- 文件存在性和可读性检查 - CSV结构验证 - 行一致性验证 - 自动方言检测
- 数据库.py:PostgreSQL数据库操作
- 连接管理 - 数据库创建 - 从CSV标题创建表格 - 行计数
- loader.py:使用COPY命令加载CSV
- 使用tqdm进行进度跟踪 - 高效散装装载 - 错误处理和回滚
- 服务器.py:MCP服务器实现
- stdio传输 - 工具注册 - 错误处理
错误处理
服务器提供详细的错误消息:
- 验证错误:CSV问题的确切行号和描述
- 数据库错误:连接、权限和SQL错误
- 文件错误:权限、编码和文件未找到错误
示例
加载CSV文件
{
"tool": "load_csv_to_postgres",
"arguments": {
"file_path": "/path/to/data.csv"
}
}使用自定义数据库
{
"tool": "load_csv_to_postgres",
"arguments": {
"file_path": "/path/to/data.csv",
"dbname": "mydb",
"table_name": "my_table"
}
}许可证
麻省理工学院
