长文 mcp_server

app/api/v1/endpoints/__init__.py

@@ -3,6 +3,7 @@ from .health import create_health_bp
 from .tasks import create_tasks_bp
 from .tokens import create_tokens_bp
 from .monitor import create_monitor_bp
+from .mcp import create_mcp_bp
 
 __all__ = [
     "create_abtest_bp",
@@ -10,4 +11,5 @@ __all__ = [
     "create_tasks_bp",
     "create_tokens_bp",
     "create_monitor_bp",
+    "create_mcp_bp",
 ]

app/api/v1/endpoints/mcp.py

@@ -0,0 +1,52 @@
+from __future__ import annotations
+
+from pydantic import ValidationError
+from quart import Blueprint, jsonify
+
+from app.api.v1.utils import ApiDependencies, LongArticlesMcpRequest
+from app.api.v1.utils import parse_json, validation_error_response
+from app.domains.mcp import LongArticleMCP
+
+
+def create_mcp_bp(deps: ApiDependencies) -> Blueprint:
+    bp = Blueprint("mcp", __name__)
+
+    @bp.route("/long_articles_mcp/<string:task_name>", methods=["POST"])
+    async def long_articles_mcp(task_name: str):
+        """
+        MCP 交互接口
+
+        - POST /api/long_articles_mcp/<task_name>
+        - 通过 task_name 分发到 LongArticleMCP 内对应的方法
+        - 请求体为 JSON，可选字段如 limit、offset 等，由各 task 按需使用
+        """
+        try:
+            req, _ = await parse_json(LongArticlesMcpRequest)
+        except ValidationError as e:
+            payload, status = validation_error_response(e)
+            return jsonify(payload), status
+
+        # 使用校验后的模型转 dict，保证 limit/offset 等类型正确，且含 extra 透传字段
+        params = req.model_dump(exclude_none=True)
+
+        mcp = LongArticleMCP(pool=deps.db, log_service=deps.log)
+        handler_map = {
+            "get_decode_response": mcp.get_decode_response,
+        }
+        handler = handler_map.get(task_name)
+        if handler is None:
+            return (
+                jsonify(
+                    {
+                        "code": 404,
+                        "message": f"unknown task_name: {task_name}",
+                        "data": None,
+                    }
+                ),
+                404,
+            )
+
+        result = await handler(params=params)
+        return jsonify({"code": 0, "message": "success", "data": result})
+
+    return bp

app/api/v1/routes/routes.py

@@ -9,6 +9,7 @@ from app.api.v1.endpoints import (
     create_tasks_bp,
     create_tokens_bp,
     create_monitor_bp,
+    create_mcp_bp
 )
 from app.core.config import GlobalConfigSettings
 from app.core.database import DatabaseManager
@@ -32,6 +33,7 @@ def register_v1_blueprints(deps: ApiDependencies) -> Blueprint:
     api.register_blueprint(create_tokens_bp(deps))
     api.register_blueprint(create_abtest_bp(deps))
     api.register_blueprint(create_monitor_bp(deps))
+    api.register_blueprint(create_mcp_bp(deps))
 
     return api
 

app/api/v1/utils/__init__.py

@@ -1,6 +1,12 @@
 from ._utils import parse_json, validation_error_response
 from .deps import ApiDependencies
-from .schemas import RunTaskRequest, TaskListRequest, SaveTokenRequest, GetCoverRequest
+from .schemas import (
+    RunTaskRequest,
+    TaskListRequest,
+    SaveTokenRequest,
+    GetCoverRequest,
+    LongArticlesMcpRequest,
+)
 
 
 __all__ = [
@@ -10,5 +16,6 @@ __all__ = [
     "TaskListRequest",
     "SaveTokenRequest",
     "GetCoverRequest",
+    "LongArticlesMcpRequest",
     "ApiDependencies",
 ]

app/api/v1/utils/schemas.py

@@ -1,6 +1,6 @@
 from __future__ import annotations
 
-from typing import Any, Dict, Optional
+from typing import Any, Dict, Literal, Optional
 
 from pydantic import BaseModel, ConfigDict, Field
 
@@ -39,3 +39,12 @@ class SaveTokenRequest(BaseRequest):
     """GzhCookieManager 的请求体字段不固定，先保持兼容。"""
 
     token: Optional[str] = None
+
+
+class LongArticlesMcpRequest(BaseRequest):
+    """MCP 交互接口的请求体，按 task_name 可传不同参数。"""
+
+    page: int = Field(default=1, ge=1, description="页码，从 1 开始")
+    page_size: int = Field(default=20, ge=1, le=100, description="每页条数")
+    sort_by: Optional[str] = Field(default=None, description="排序字段")
+    sort_order: Optional[Literal["asc", "desc"]] = Field(default=None, description="排序方向: asc / desc")

app/domains/mcp/__init__.py

@@ -0,0 +1,4 @@
+from .long_articles_mcp import LongArticleMCP
+
+__all__ = ["LongArticleMCP"]
+

app/domains/mcp/long_articles_mcp.py

@@ -0,0 +1,70 @@
+import math
+from typing import Any, Dict, Optional
+
+from app.core.database import DatabaseManager
+from app.core.observability import LogService
+
+
+class LongArticleMCP:
+    def __init__(self, pool: DatabaseManager, log_service: LogService):
+        self.pool = pool
+        self.log_service = log_service
+
+    SORTABLE_FIELDS = {
+        "read_cnt",
+        "read_median_multiplier",
+        "publish_time",
+    }
+
+    BASE_WHERE = "WHERE t1.status = 2"
+
+    BASE_FROM = """
+        FROM long_articles_decode_tasks t1
+        JOIN ad_platform_accounts_daily_detail t2
+        ON t1.wx_sn = t2.wx_sn
+    """
+
+    async def get_decode_response(self, params: Optional[Dict[str, Any]] = None):
+        params = params or {}
+        page = params.get("page", 1)
+        page_size = params.get("page_size", 20)
+        sort_by = params.get("sort_by")
+        sort_order = params.get("sort_order", "asc")
+
+        count_query = f"SELECT COUNT(*) AS total {self.BASE_FROM} {self.BASE_WHERE}"
+        count_result = await self.pool.async_fetch(query=count_query)
+        total = count_result[0]["total"] if count_result else 0
+        total_pages = math.ceil(total / page_size) if total > 0 else 0
+
+        data_query = f"""
+            SELECT t2.gh_id,
+                 t2.account_name,
+                 t2.article_title,
+                 t2.article_link,
+                 t2.article_cover,
+                 t2.article_text,
+                 t2.read_cnt,
+                 t2.read_median_multiplier,
+                 from_unixtime(t2.publish_timestamp, '%%Y-%%m-%%d %%H:%%i:%%s') as publish_time,
+                 t1.result
+            {self.BASE_FROM}
+            {self.BASE_WHERE}
+        """
+
+        if sort_by and sort_by in self.SORTABLE_FIELDS:
+            direction = "ASC" if sort_order == "asc" else "DESC"
+            order_column = "t2.publish_timestamp" if sort_by == "publish_time" else f"t2.{sort_by}"
+            data_query += f" ORDER BY {order_column} {direction}"
+
+        offset = (page - 1) * page_size
+        data_query += " LIMIT %s OFFSET %s"
+        rows = await self.pool.async_fetch(query=data_query, params=(page_size, offset))
+
+        return {
+            "total": total,
+            "page": page,
+            "page_size": page_size,
+            "total_pages": total_pages,
+            "items": rows,
+        }
+

app/recommend/offline_recommend/core.py → app/domains/recommend/offline_recommend/core.py

@@ -8,10 +8,10 @@ from app.infra.mapper import LongArticleDatabaseMapper
 from app.infra.mapper import PiaoquanCrawlerDatabaseMapper
 from app.infra.external import OdpsService
 
-from app.recommend.offline_recommend.strategy import I2I
-from app.recommend.offline_recommend.strategy import GetTopArticleStrategy
-from app.recommend.offline_recommend.utils import RecommendApolloClient
-from app.recommend.offline_recommend.utils import ProduceBaseData
+from app.domains.recommend.offline_recommend.strategy import I2I
+from app.domains.recommend.offline_recommend.strategy import GetTopArticleStrategy
+from app.domains.recommend.offline_recommend.utils import RecommendApolloClient
+from app.domains.recommend.offline_recommend.utils import ProduceBaseData
 
 
 class BaseOffRecommendUtils: