development

MCP Server Trending 开发规范

1. 开发前准备（必须执行）

在开始任何新功能开发之前，必须执行以下步骤：

1.1 切换到 main 分支并更新

# 切换到 main 分支
git checkout main

# 拉取最新的 main 分支代码
git pull origin main

1.2 基于用户意图创建新分支

根据用户的开发需求，创建语义化的分支名称：

# 分支命名规范
# - feat/xxx: 新功能
# - fix/xxx: Bug 修复
# - docs/xxx: 文档更新
# - refactor/xxx: 代码重构
# - test/xxx: 测试相关

git checkout -b feat/add-xxx-fetcher  # 示例：添加新的 fetcher
git checkout -b fix/xxx-serialization  # 示例：修复序列化问题
git checkout -b docs/update-readme     # 示例：更新文档

2. 新 Fetcher 开发规范

2.1 目录结构

新的 fetcher 必须遵循以下目录结构：

src/mcp_server_trending/
├── fetchers/
│   └── {platform_name}/        # 新平台目录
│       ├── __init__.py         # 导出 Fetcher 类
│       └── fetcher.py          # Fetcher 实现
└── models/
    └── {platform_name}.py      # 数据模型定义

2.2 Model 定义规范

所有数据模型必须继承自 BaseModel：

from dataclasses import dataclass, field
from datetime import datetime
from typing import Any

from .base import BaseModel


@dataclass
class YourModel(BaseModel):
    """数据模型描述"""

    rank: int
    id: str
    name: str
    url: str
    # ... 其他字段

    # 可选字段使用默认值
    description: str | None = None
    created_at: datetime | None = None
    tags: list[str] = field(default_factory=list)

2.3 Fetcher 实现规范

from typing import Any

from ..base import BaseFetcher
from ...models.base import TrendingResponse
from ...models.{platform} import YourModel


class YourFetcher(BaseFetcher):
    """Fetcher 描述"""

    PLATFORM = "{platform_name}"
    BASE_URL = "https://api.example.com"

    async def fetch_xxx(
        self,
        limit: int = 20,
        use_cache: bool = True,
    ) -> TrendingResponse:
        """获取数据的方法描述"""

        cache_key = f"{self.PLATFORM}:xxx:limit={limit}"

        # 检查缓存
        if use_cache:
            cached = self._get_from_cache(cache_key)
            if cached:
                return cached

        try:
            # 实现数据获取逻辑
            items = []
            # ... 获取和解析数据

            response = TrendingResponse(
                success=True,
                platform=self.PLATFORM,
                data=items,
                metadata={"limit": limit},
            )

            # 保存到缓存
            self._save_to_cache(cache_key, response)
            return response

        except Exception as e:
            return TrendingResponse(
                success=False,
                platform=self.PLATFORM,
                error=str(e),
            )