1from abc import ABC, abstractmethod
2
3from integrations.models import Message, SourceType
4
5
6class Manager(ABC):
7 manager_type: SourceType
8
9 @abstractmethod
10 async def receive_message(self, message: Message):
11 "Receive message from integration"
12 raise NotImplementedError
13
14 @abstractmethod
15 def send_message(self, message: Message):
16 "Send message to integration from Openhands server"
17 raise NotImplementedError
18
19 @abstractmethod
20 def start_job(self):
21 "Kick off a job with openhands agent"
22 raise NotImplementedError
23
24 def create_outgoing_message(self, msg: str | dict, ephemeral: bool = False):
25 return Message(source=SourceType.OPENHANDS, message=msg, ephemeral=ephemeral)
26