Action 插件¶

"""
title: My Custom Action
author: Your Name
version: 1.0.0
description: Description of your action plugin
"""

from pydantic import BaseModel, Field
from typing import Optional, Dict, Any

class Action:
    class Valves(BaseModel):
        # 在这里添加你的配置项
        show_status: bool = Field(
            default=True,
            description="Show status updates during processing"
        )

    def __init__(self):
        self.valves = self.Valves()

    async def action(
        self,
        body: dict,
        __user__: Optional[Dict[str, Any]] = None,
        __event_emitter__: Optional[Any] = None,
        __request__: Optional[Any] = None,
    ) -> Optional[dict]:
        """
        当用户点击 Action 按钮时触发的主方法。

        Args:
            body: 包含会话数据的消息体
            __user__: 当前用户信息
            __event_emitter__: 用于发送通知或状态更新
            __request__: FastAPI 请求对象

        Returns:
            修改后的 body 字典或 None
        """
        # 发送状态更新
        if __event_emitter__ and self.valves.show_status:
            await __event_emitter__({
                "type": "status",
                "data": {"description": "Processing...", "done": False}
            })

        # 插件逻辑
        messages = body.get("messages", [])
        if messages:
            last_message = messages[-1].get("content", "")
            # 在这里处理消息...

        # 完成状态
        if __event_emitter__ and self.valves.show_status:
            await __event_emitter__({
                "type": "status",
                "data": {"description": "Done!", "done": True}
            })

        return body