Action Plugins¶

"""
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):
        # Add your configuration options here
        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]:
        """
        Main action method triggered when user clicks the action button.

        Args:
            body: Message body containing conversation data
            __user__: Current user information
            __event_emitter__: For sending notifications and status updates
            __request__: FastAPI request object

        Returns:
            Modified body dict or None
        """
        # Send status update
        if __event_emitter__ and self.valves.show_status:
            await __event_emitter__({
                "type": "status",
                "data": {"description": "Processing...", "done": False}
            })

        # Your plugin logic here
        messages = body.get("messages", [])
        if messages:
            last_message = messages[-1].get("content", "")
            # Process the message...

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

        return body