MCP Israeli Price Comparison

统一价格比较MCP服务器

一个具有HTTP SSE（服务器发送事件）传输的综合模型上下文协议（MCP）服务器，提供价格比较应用程序所需的所有工具。使用FastAPI和Python 3.11+构建。

特性

• MCP协议支持：完整的JSON-RPC 2.0实现，支持SSE流
• 免费搜索提供商：使用DuckDuckGo、谷歌抓取和Bing进行智能回退
• 10专用工具：网络搜索、抓取、价格情报和数据存储
• 实时流媒体：服务器发送实时进度更新事件
• SQLite存储：持久的价格历史记录和缓存
• 速率限制：防止API滥用的内置保护
• Docker就绪：生产就绪的集装箱化

快速开始

先决条件

• Python 3.11+

安装

# Create virtual environment
python -m venv venv
source venv/bin/activate  # On Windows: venv\Scripts\activate

# Install dependencies
pip install -r requirements.txt

# Install Playwright browsers (for JavaScript rendering)
playwright install chromium

# Configure environment
cp .env.example .env

运行服务器

python main.py

服务器将在以下时间启动 http://localhost:8000.

使用Docker

# Build the image
docker build -t price-comparison-mcp .

# Run the container
docker run -p 8000:8000 price-comparison-mcp

API终点

可用工具

网络搜索工具（免费提供商）

1. 网络搜索 -具有自动回退功能的网络搜索（DuckDuckGo、谷歌、必应）
2. 购物搜索 -使用定价数据进行产品搜索
3. 图像搜索 -视觉产品识别的图像搜索

Web剪贴工具

1. fetch_page_content -从URL获取HTML（静态或JS渲染）
2. extract_structured_data -提取JSON-LD、微数据、打开图
3. extract_prices from html -多策略价格提取

价格情报工具

1. parse_price -将价格字符串解析为结构化格式
2. normalize_product_name -通过品牌/型号检测规范名称
3. 检测产品规格 -提取技术规格（内存、存储等）
4. 计算总成本 -计算运费、税费和折扣总计

存储工具

1. save_search_result -将价格调查结果保存到数据库
2. get_price_history -检索历史价格数据
3. 获取_平均_市场_价格 -计算价格统计

使用示例

列出工具

curl http://localhost:8000/mcp/tools

MCP初始化

curl -X POST http://localhost:8000/mcp \
  -H "Content-Type: application/json" \
  -d '{
    "jsonrpc": "2.0",
    "method": "initialize",
    "id": 1
  }'

调用工具（使用SSE流媒体）

curl -X POST http://localhost:8000/mcp \
  -H "Content-Type: application/json" \
  -d '{
    "jsonrpc": "2.0",
    "method": "tools/call",
    "params": {
      "name": "web_search",
      "arguments": {
        "query": "iPhone 15 Pro price Israel"
      }
    },
    "id": 2
  }'

Python客户端示例

import httpx
import json

async def test_mcp_server():
    async with httpx.AsyncClient() as client:
        # List tools
        response = await client.get("http://localhost:8000/mcp/tools")
        print(response.json())

        # Call tool with SSE streaming
        async with client.stream(
            "POST",
            "http://localhost:8000/mcp",
            json={
                "jsonrpc": "2.0",
                "method": "tools/call",
                "params": {
                    "name": "shopping_search",
                    "arguments": {"query": "Samsung Galaxy S24"}
                },
                "id": 1
            }
        ) as response:
            async for line in response.aiter_lines():
                if line.startswith("data: "):
                    data = json.loads(line[6:])
                    print(data)

# Run with: asyncio.run(test_mcp_server())

配置

配置是通过环境变量进行管理的。看 .env.example 对于所有选项。

项目结构

.
├── src/
│   ├── server/
│   │   ├── main.py          # FastAPI + MCP server
│   │   ├── sse_handler.py   # SSE streaming logic
│   │   └── middleware.py    # CORS, logging, rate limiting
│   ├── tools/
│   │   ├── search_tools.py  # Search with free providers
│   │   ├── search_providers.py # DuckDuckGo, Google, Bing providers
│   │   ├── scraping_tools.py # Web scraping tools
│   │   ├── price_tools.py   # Price intelligence tools
│   │   └── storage_tools.py # SQLite storage tools
│   ├── models/
│   │   └── schemas.py       # Pydantic schemas
│   ├── utils/
│   │   ├── parser.py        # Price parsing
│   │   ├── normalizer.py    # Text normalization
│   │   └── database.py      # DB utilities
│   └── config/
│       └── settings.py      # Configuration
├── tests/
│   ├── test_tools.py        # Tool unit tests
│   ├── test_server.py       # API tests
│   └── test_integration.py  # Integration tests
├── database/
│   └── init.sql             # Database schema
├── main.py                  # Entry point
├── requirements.txt
├── Dockerfile
└── README.md

测试

# Run all tests
pytest

# Run with coverage
pytest --cov=src --cov-report=html

# Run specific test file
pytest tests/test_tools.py

# Run with verbose output
pytest -v

发展

# Install dev dependencies
pip install -r requirements.txt

# Format code
black src tests
isort src tests

# Type checking
mypy src

# Linting
ruff check src tests

SSE事件格式

工具执行以以下格式流式传输事件：

// Start event
data: {"type": "start", "tool": "web_search"}

// Progress event
data: {"type": "progress", "message": "Executing tool..."}

// Result event
data: {"type": "result", "data": {...}}

// Error event (if failed)
data: {"type": "error", "message": "Error description"}

// Complete event
data: {"type": "complete"}

数据库模式

SQLite数据库包括以下表：

• search_results -包含产品信息的价格搜索结果
• user_preferences -用户设置和首选项
• price_alerts -价格警报配置
• search_cache -API响应缓存
• api_usage -速率限制跟踪

许可证

MIT许可证

贡献

1. 克隆该仓库
2. 创建要素分支
3. 进行更改
4. 运行测试
5. 提交拉取请求