Merge pull request #14 from sonnhfit/feature/chat-web-hook

add webhook and api chat

sonagent/commands/run_commands.py

@@ -122,6 +122,13 @@ def create_user_data_dir(args: Dict[str, Any]) -> None:
         "repo_name": "SonAgent",
         "token": "",
         "local_repo_path": ""
+    },
+    "webhook": {
+        "enabled": false,
+        "url": "",
+        "chat": {
+            "message": "{message}"
+        }
     }
 }
         """

sonagent/enums/rpcmessagetype.py

@@ -6,6 +6,7 @@ class RPCMessageType(str, Enum):
     WARNING = 'warning'
     EXCEPTION = 'exception'
     STARTUP = 'startup'
+    CHAT = 'chat'
 
     def __repr__(self):
         return self.value

sonagent/rpc/api_server/api_models.py

@@ -17,4 +17,7 @@ class AccessAndRefreshToken(AccessToken):
 
 
 class Version(BaseModel):
-    version: str
+    version: str
+
+class ChatMsg(BaseModel):
+    message: str

sonagent/rpc/api_server/api_v1.py

@@ -11,7 +11,9 @@
 from sonagent.rpc import RPC
 
 from sonagent.rpc.rpc import RPCException
-from sonagent.rpc.api_server.api_models import (Ping, Version)
+from sonagent.rpc.api_server.api_models import (Ping, Version, ChatMsg)
+from sonagent.rpc.api_server.utils import get_rpc
+
 
 logger = logging.getLogger(__name__)
 
@@ -30,8 +32,14 @@ def ping():
     return {"status": "pong"}
 
 
-@router.get('/version', response_model=Version, tags=['info'])
+@router_public.get('/version', response_model=Version, tags=['info'])
 def version():
     """ Bot Version info"""
     return {"version": __version__}
 
+
+@router_public.post('/chat', response_model=ChatMsg, tags=['chat'])
+async def chat(msg: str, rpc: RPC = Depends(get_rpc)):
+    message = await rpc.chat(msg)
+    return {"message": message}
+

sonagent/rpc/api_server/utils.py

@@ -0,0 +1,26 @@
+from typing import Any, AsyncIterator, Dict, Optional
+from uuid import uuid4
+
+from fastapi import Depends, HTTPException
+from sonagent.rpc.rpc import RPC, RPCException
+
+from .webserver import ApiServer
+
+
+def get_rpc_optional() -> Optional[RPC]:
+    if ApiServer._has_rpc:
+        return ApiServer._rpc
+    return None
+
+
+async def get_rpc() -> Optional[AsyncIterator[RPC]]:
+
+    _rpc = get_rpc_optional()
+    if _rpc:
+        try:
+            yield _rpc
+        finally:
+            pass
+    else:
+        raise RPCException('Bot is not in the correct state')
+

sonagent/rpc/rpc_manager.py

@@ -25,7 +25,6 @@ def __init__(self, sonagent) -> None:
             from sonagent.rpc.telegram import Telegram
             self.registered_modules.append(Telegram(self._rpc, config))
 
-
         # Enable local rest api server for cmd line control
         if config.get('api_server', {}).get('enabled', False):
             logger.info('Enabling rpc.api_server')
@@ -34,6 +33,15 @@ def __init__(self, sonagent) -> None:
             apiserver.add_rpc_handler(self._rpc)
             self.registered_modules.append(apiserver)
 
+        if config.get('webhook', {}).get('enabled', False):
+            logger.info('Enabling webhook ...')
+            from sonagent.rpc.webhook import Webhook
+            self.registered_modules.append(Webhook(self._rpc, config))
+
+
+
+
+
 
 
     def send_msg(self, msg: RPCSendMsg) -> None:

sonagent/rpc/rpc_types.py

@@ -10,7 +10,7 @@ class RPCSendMsgBase(TypedDict):
 
 class RPCStatusMsg(RPCSendMsgBase):
     """Used for Status, Startup and Warning messages"""
-    type: Literal[RPCMessageType.STATUS, RPCMessageType.STARTUP, RPCMessageType.WARNING]
+    type: Literal[RPCMessageType.CHAT, RPCMessageType.STATUS, RPCMessageType.STARTUP, RPCMessageType.WARNING]
     status: str
 
 

sonagent/rpc/webhook.py

@@ -0,0 +1,99 @@
+import logging
+import time
+from typing import Any, Dict, Optional
+from requests import RequestException, post
+from sonagent.enums import RPCMessageType
+from sonagent.rpc import RPC, RPCHandler
+from sonagent.rpc.rpc_types import RPCSendMsg
+
+
+logger = logging.getLogger(__name__)
+
+logger.debug('import module rpc.webhook ...')
+
+
+class Webhook(RPCHandler):
+
+    def __init__(self, rpc: RPC, config: dict) -> None:
+        """
+        Init the Webhook class, and init the super class RPCHandler
+        :param rpc: instance of RPC Helper class
+        :param config: Configuration object
+        :return: None
+        """
+        super().__init__(rpc, config)
+
+        self._url = self._config['webhook']['url']
+        self._format = self._config['webhook'].get('format', 'json')
+        self._retries = self._config['webhook'].get('retries', 0)
+        self._retry_delay = self._config['webhook'].get('retry_delay', 0.1)
+        self._timeout = self._config['webhook'].get('timeout', 10)
+
+    def cleanup(self) -> None:
+        """
+        Cleanup pending module resources.
+        This will do nothing for webhooks, they will simply not be called anymore
+        """
+        pass
+
+    def _get_value_dict(self, msg: RPCSendMsg) -> Optional[Dict[str, Any]]:
+        whconfig = self._config['webhook']
+        if msg['type'].value in whconfig:
+            # Explicit types should have priority
+            valuedict = whconfig.get(msg['type'].value)
+        # Deprecated 2022.10 - only keep generic method.
+        elif msg['type'] in [RPCMessageType.CHAT]:
+            valuedict = whconfig.get('chat')
+        elif msg['type'] in (RPCMessageType.STATUS,
+                             RPCMessageType.STARTUP,
+                             RPCMessageType.EXCEPTION,
+                             RPCMessageType.WARNING):
+            valuedict = whconfig.get('webhookstatus')
+
+        return valuedict
+
+    def send_msg(self, msg: RPCSendMsg) -> None:
+        """ Send a message to telegram channel """
+        if msg['type'] == RPCMessageType.CHAT:
+            try:
+
+                valuedict = self._get_value_dict(msg)
+
+                if not valuedict:
+                    logger.debug("Message type '%s' not configured for webhooks", msg['type'])
+                    return
+
+                payload = {key: value.format(**msg) for (key, value) in valuedict.items()}
+                self._send_msg(payload)
+            except KeyError as exc:
+                logger.exception("Problem calling Webhook. Please check your webhook configuration. "
+                                "Exception: %s", exc)
+
+    def _send_msg(self, payload: dict) -> None:
+        """do the actual call to the webhook"""
+
+        success = False
+        attempts = 0
+        while not success and attempts <= self._retries:
+            if attempts:
+                if self._retry_delay:
+                    time.sleep(self._retry_delay)
+                logger.info("Retrying webhook...")
+
+            attempts += 1
+
+            try:
+                if self._format == 'json':
+                    response = post(self._url, json=payload, timeout=self._timeout)
+                elif self._format == 'raw':
+                    response = post(self._url, data=payload['data'],
+                                    headers={'Content-Type': 'text/plain'},
+                                    timeout=self._timeout)
+                else:
+                    raise NotImplementedError(f'Unknown format: {self._format}')
+
+                response.raise_for_status()
+                success = True
+
+            except RequestException as exc:
+                logger.warning("Could not call webhook url. Exception: %s", exc)

sonagent/sonbot.py

@@ -70,7 +70,13 @@ def __init__(self, config: dict, args: Any = None) -> None:
 
     async def chat(self, input: str) -> str:
         if self.agent_mode == "chat":
-            return await self.agent.chat(input)
+
+            chat = await self.agent.chat(input)
+            try:
+                self.notify_chat_event(chat)
+            except Exception as e:
+                logger.error(f"Error notifying chat event: {e}")
+            return chat
         else:
             return await self.agent.chat_code(input)
 
@@ -152,6 +158,13 @@ def process_stopped(self) -> None:
         """
         pass
 
+    def notify_chat_event(self, msg: str, msg_type=RPCMessageType.CHAT) -> None:
+        self.rpc.send_msg({
+            'type': msg_type,
+            'message': msg
+        })
+
+
     def notify_status(self, msg: str, msg_type=RPCMessageType.STATUS) -> None:
         """
         Public method for users of this class (worker, etc.) to send notifications
@@ -162,3 +175,6 @@ def notify_status(self, msg: str, msg_type=RPCMessageType.STATUS) -> None:
             'status': msg
         })
 
+
+
+

user_data/config.json

@@ -32,5 +32,12 @@
         "repo_name": "SonAgent",
         "token": "",
         "local_repo_path": ""
+    },
+    "webhook": {
+        "enabled": false,
+        "url": "",
+        "chat": {
+            "message": "{message}"
+        }
     }
 }