AgentCoreService/toolkit/lark_alert_for_human_intervention.py

from typing import List

from toolkit.base import BaseToolkit
from toolkit.function_tool import FunctionTool
import requests


class LarkAlertForHumanIntervention(BaseToolkit):
    r"""A toolkit for Lark alert for human intervention."""

    def __init__(self):
        self.webhook_url = 'https://open.feishu.cn/open-apis/bot/v2/hook/c316b559-1c6a-4c4e-97c9-50b44e4c2a9d'
        super().__init__()

    def send_lark_alert_for_human_intervention(
        self, message: str
    ) -> str:
        r"""Sends a Lark alert for human intervention.

        Args:
            message (str): The message to send.
        Returns:
            str: A confirmation message.
        """
        req_body = {
            "msg_type": "text",
            "content": {
                "text": f'[Agent告警]{message}'
            }
        }
        try:
            response = requests.post(
                url=self.webhook_url,
                json=req_body
            )
            return f"Alert sent successfully: {response.status_code}"
        except Exception as e:
            return f"Failed to send alert: {e}"

    def get_tools(self) -> List[FunctionTool]:
        return [FunctionTool(self.send_lark_alert_for_human_intervention)]

if __name__ == '__main__':
    toolkit = LarkAlertForHumanIntervention()
    tools = toolkit.get_tools()
    for tool in tools:
        print(f"Tool schema: {tool.get_openai_tool_schema()}")
    resp = toolkit.send_lark_alert_for_human_intervention('测试')
    print(resp)