30 lines
666 B
Python
30 lines
666 B
Python
"""Agent OS message bus runtime."""
|
|
|
|
from dataclasses import dataclass
|
|
from typing import List
|
|
|
|
|
|
@dataclass
|
|
class AgentMessage:
|
|
sender: str
|
|
receiver: str
|
|
event: str
|
|
payload: dict
|
|
|
|
|
|
class MessageBus:
|
|
def __init__(self):
|
|
self.queue: List[AgentMessage] = []
|
|
|
|
def send(self, message: AgentMessage):
|
|
self.queue.append(message)
|
|
|
|
def broadcast(self, sender: str, event: str, payload: dict, receivers: List[str]):
|
|
for receiver in receivers:
|
|
self.send(AgentMessage(sender, receiver, event, payload))
|
|
|
|
def consume(self):
|
|
messages = list(self.queue)
|
|
self.queue.clear()
|
|
return messages
|