MCP E-commerce Load Server

MCP E-commerce Load Server

概述

MCP E-commerce Load Server 是一套集成解决方案，用于发现、验证并将联盟商店及其商品登记到电子商务系统中。该系统通过 CrewAI 框架运用人工智能，自动完成数据的搜索、验证和优先级排序。

工作流程

系统遵循以下工作流程：

1. 使用 Serper API 发现商店
2. 使用 CrewAI 智能验证商店
3. 将商店登记到数据库中（PostgreSQL/SQLAlchemy）
4. 通过 Serper MCP Server 收集商品
5. 使用 CrewAI 智能排列商品优先级
6. 可选的人工整理（通过 CSV/仪表板界面）
7. 最终加载到数据库中

要求

• Python 3.12 或更高版本
• PostgreSQL
• 外部服务的 API 密钥（OpenAI、Serper 等）

安装

1. 克隆仓库：

git clone https://github.com/brazeiro63/mcp_ecommerce_load_server.git
cd mcp_ecommerce_load_server

2. 安装依赖项：

pip install -e .

3. 配置环境变量：

cp .env.example .env
# Edite o arquivo .env com suas chaves de API e configurações

4. 运行服务：

python main.py

项目结构

project_root/
│
├── main.py                 # Orquestra a pipeline de povoamento
├── requirements.txt        # Dependências do projeto
├── pyproject.toml          # Configuração do projeto
├── crew_agents/            # Agentes de IA para descoberta e pontuação
├── app/                    # Aplicação principal
│   ├── api/                # Endpoints da API
│   ├── db/                 # Configurações de banco de dados
│   ├── models/             # Modelos de dados
│   ├── repositories/       # Camada de acesso a dados
│   ├── schemas/            # Esquemas de validação
│   └── main.py             # Ponto de entrada da aplicação
├── config/                 # Arquivos de configuração
├── review_interface/       # Interface para revisão humana
└── utils/                  # Utilitários

使用方法

系统可以通过两种方式运行：

1. 完整模式：运行完整的发现、验证和加载流程

python main.py

2. 选择模式：仅运行流程中的特定部分

python main.py --only-discover  # Apenas descoberta de lojas
python main.py --only-products  # Apenas coleta de produtos

贡献

欢迎贡献！请按以下步骤操作：

1. Fork 项目
2. 创建一个用于存放新功能的分支（git checkout -b feature/amazing-feature）
3. 提交更改（git commit -m 'Add some amazing feature'）
4. 推送到该分支（git push origin feature/amazing-feature）
5. 创建 Pull Request

许可证

本项目采用 MIT 许可证授权——详情请参阅 LICENSE 文件。

---

✅ 2. 创建测试

🧪 建议路径：tests/test_db_tools.py

创建该文件并添加：

# tests/test_db_tools.py
from src.app.db.insert_affiliate_stores import insert_affiliate_stores
from src.app.db.insert_products import insert_products

def test_insert_affiliate_store():
    sample_store = [{
        "name": "Test Store",
        "platform": "TestPlatform",
        "active": True,
        "api_credentials": {"token": "123abc"}
    }]
    result = insert_affiliate_stores(sample_store)
    assert len(result) == 1
    assert result[0].name == "Test Store"

def test_insert_product():
    sample_product = [{
        "external_id": "abc123",
        "platform": "TestPlatform",
        "title": "Test Product",
        "description": "A test product",
        "price": 19.99,
        "product_url": "http://example.com/product",
        "category": "Testing"
    }]
    result = insert_products(sample_product, affiliate_store_name="Test Store")
    assert len(result) == 1
    assert result[0].title == "Test Product"