Deepspringai Parquet MCP Server

🚀 Parquet MCP 服务器

Parquet MCP 服务器是一个用于处理 Parquet 文件的工具集，它提供了嵌入、信息获取、格式转换等功能，能帮助用户高效地处理和分析 Parquet 数据。

🚀 快速开始

项目结构

parquet-mcp-server/
├── README.md                # 项目说明文件
├── src/                    # 源代码目录
│   ├── __init__.py         # 包初始化文件
│   ├── client.py           # MCP客户端接口
│   └── tools/              # 工具模块
│       ├── embedding.py    # 嵌入工具
│       ├── parquet_info.py  # Parquet 文件信息工具
│       ├── duckdb_conversion.py  # DuckDB 转换工具
│       ├── postgres_conversion.py  # PostgreSQL 转换工具
│       └── markdown_processing.py  # Markdown 处理工具
├── tests/                  # 测试用例目录
│   ├── __init__.py         # 测试包初始化文件
│   ├── test_embedding.py    # 嵌入功能测试
│   ├── test_parquet_info.py  # Parquet 文件信息工具测试
│   ├── test_duckdb_conversion.py  # DuckDB 转换测试
│   ├── test_postgres_conversion.py  # PostgreSQL 转换测试
│   └── test_markdown_processing.py  # Markdown 处理测试
└── requirements.txt        # 项目依赖管理文件

环境变量

在 .env 文件中添加以下环境变量：

# 必要配置
OPENAI_API_KEY=your_openai_api_key  # OpenAI API 密钥

# 可选配置（根据需要调整）
OLLAMA_SERVER_URL=http://localhost:11434  # Ollama 服务器地址，默认为本地

安装与运行

使用命令行工具安装

pip install -r requirements.txt  # 安装依赖
python src/run_server.py  # 启动 MCP 服务器

使用测试客户端

from parquet_mcp_server.client import (
    convert_to_duckdb, 
    embed_parquet, 
    get_parquet_info, 
    convert_to_postgres,
    process_markdown_file
)

# 测试 DuckDB 转换
result = convert_to_duckdb(
    parquet_path="input.parquet",
    output_dir="db_output"
)

# 测试嵌入功能
result = embed_parquet(
    input_path="input.parquet",
    output_path="output.parquet", 
    column_name="text",
    embedding_column="embeddings",
    batch_size=2
)

# 测试 Parquet 信息工具
result = get_parquet_info("input.parquet")

# 测试 PostgreSQL 转换
result = convert_to_postgres(
    parquet_path="input.parquet", 
    table_name="my_table"
)

# 测试 Markdown 处理
result = process_markdown_file(
    file_path="input.md",
    output_path="output.parquet"
)

✨ 主要特性

本项目提供了一系列处理 Parquet 文件的工具，涵盖嵌入、信息获取、格式转换等功能，具体如下：

• 嵌入工具：为指定列生成嵌入向量并保存到新列。
• Parquet 文件信息工具：获取 Parquet 文件的基本信息和统计信息。
• DuckDB 转换工具：将 Parquet 文件转换为 DuckDB 表格文件。
• PostgreSQL 转换工具：将 Parquet 文件数据导入 PostgreSQL 数据库表中。
• Markdown 处理工具：将 Markdown 文件转换为结构化的 Parquet 文件。

📦 安装指南

依赖安装

pip install -r requirements.txt

环境变量配置

在 .env 文件中添加必要的环境变量，如 OPENAI_API_KEY 等。

启动服务器

python src/run_server.py

💻 使用示例

基础用法

from parquet_mcp_server.client import (
    convert_to_duckdb, 
    embed_parquet, 
    get_parquet_info, 
    convert_to_postgres,
    process_markdown_file
)

# 测试 DuckDB 转换
result = convert_to_duckdb(
    parquet_path="input.parquet",
    output_dir="db_output"
)

# 测试嵌入功能
result = embed_parquet(
    input_path="input.parquet",
    output_path="output.parquet", 
    column_name="text",
    embedding_column="embeddings",
    batch_size=2
)

# 测试 Parquet 信息工具
result = get_parquet_info("input.parquet")

# 测试 PostgreSQL 转换
result = convert_to_postgres(
    parquet_path="input.parquet", 
    table_name="my_table"
)

# 测试 Markdown 处理
result = process_markdown_file(
    file_path="input.md",
    output_path="output.parquet"
)

📚 详细文档

功能模块说明

嵌入工具

def embed_parquet(
    input_path: str,
    output_path: str, 
    column_name: str = "text",
    embedding_column: str = "embeddings", 
    batch_size: int = 16
) -> None:
    # 功能：为指定列生成嵌入向量并保存到新列
    # 输入：
    #   input_path: 输入 Parquet 文件路径
    #   output_path: 输出 Parquet 文件路径
    #   column_name: 需要生成嵌入的列名，默认为 "text"
    #   embedding_column: 嵌入结果保存的列名，默认为 "embeddings"
    #   batch_size: 处理批次大小，默认为 16

Parquet 文件信息工具

def get_parquet_info(parquet_path: str) -> Dict:
    # 功能：获取 Parquet 文件的基本信息和统计信息
    # 输入：
    #   parquet_path: Parquet 文件路径
    # 输出：
    #   Dict: 包含文件大小、行数、列信息等的字典

DuckDB 转换工具

def convert_to_duckdb(
    parquet_path: str, 
    output_path: str,
    table_name: str = "parquet_table"
) -> None:
    # 功能：将 Parquet 文件转换为 DuckDB 表格文件
    # 输入：
    #   parquet_path: 输入 Parquet 文件路径
    #   output_path: 输出 .db 文件路径
    #   table_name: 目标表名，默认为 "parquet_table"

PostgreSQL 转换工具

def convert_to_postgres(
    parquet_path: str,
    host: str,
    database: str,
    user: str,
    password: str,
    table_name: str = "parquet_table",
    chunk_size: int = 1000
) -> None:
    # 功能：将 Parquet 文件数据导入 PostgreSQL 数据库表中
    # 输入：
    #   parquet_path: 输入 Parquet 文件路径
    #   host: 数据库主机地址
    #   database: 数据库名称
    #   user: 数据库用户名
    #   password: 数据库密码
    #   table_name: 目标表名，默认为 "parquet_table"
    #   chunk_size: 每次插入的数据量，默认为 1000

Markdown 处理工具

def process_markdown_file(
    file_path: str,
    output_path: str,
    encoding: str = "utf-8",
    split_by: str = None,
    keep_separators: bool = False
) -> None:
    # 功能：将 Markdown 文件转换为结构化的 Parquet 文件
    # 输入：
    #   file_path: 输入 Markdown 文件路径
    #   output_path: 输出 Parquet 文件路径
    #   encoding: 字符编码，默认为 "utf-8"
    #   split_by: 按分隔符分割内容（可选）
    #   keep_separators: 是否保留分隔符，默认为 False

测试

启动测试服务器

python src/run_server.py  # 启动 MCP 服务端

执行测试用例

python -m pytest tests/  # 运行所有测试用例
python -m pytest tests/test_embedding.py  # 测试嵌入功能
python -m pytest tests/test_parquet_info.py  # 测试 Parquet 文件信息工具
python -m pytest tests/test_duckdb_conversion.py  # 测试 DuckDB 转换工具
python -m pytest tests/test_postgres_conversion.py  # 测试 PostgreSQL 转换工具
python -m pytest tests/test_markdown_processing.py  # 测试 Markdown 处理工具

API 响应格式

嵌入响应

{
    "status": "success",
    "message": "embedding completed successfully",
    "result": {
        "input_count": 100,
        "output_file": "output.parquet"
    }
}

文件信息响应

{
    "file_size": "12.5 MB",
    "num_rows": 1000,
    "columns": [
        {"name": "text", "type": "string"},
        {"name": "label", "type": "int"}
    ]
}

🔧 技术细节

本项目使用 Python 语言开发，依赖于多个第三方库，如 llama-cpp-python、duckdb 等。在处理 Parquet 文件时，采用了分批处理的方式，以提高处理大文件时的性能。同时，通过设置环境变量的方式来管理 API 密钥等敏感信息，增强了项目的安全性。

📄 注意事项

⚠️ 重要提示

• 确保安装了所有依赖项，特别是 llama-cpp-python 和 duckdb。
• 设置正确的环境变量以避免 API 调用错误。
• 处理大文件时建议分批处理以保证性能。