Communications¶

                        ┌──────────────────────────┐
                        │  CommunicationDispatcher  │
                        └────────┬─────────────────┘
                                 │
              ┌──────────────────┼──────────────────┐
              │                  │                   │
     ┌────────▼───────┐  ┌──────▼────────┐  ┌───────▼───────┐
     │    Telegram     │  │     Email     │  │    Custom     │
     │   Communicator  │  │  Communicator │  │  Communicator │
     └────────┬────────┘  └───────────────┘  └───────────────┘
              │
   ┌──────────┴──────────┐
   │  Outgoing           │  Incoming
   │  label routing      │  eventspype events
   │  → send to chats    │  → publish IncomingMessage
   └─────────────────────┘

communications:
  enabled: true
  backends:
    my_telegram:
      type: telegram
      api_id: ${TELEGRAM_API_ID}
      api_hash: ${TELEGRAM_API_HASH}
      token: ${TELEGRAM_BOT_TOKEN}
      labels:
        - default
        - alerts
      listen_to_commands: true
      chats:
        default:
          chat_id: "-1001234567890"
          active: true
        alerts:
          chat_id: "-1009876543210"
          topic_id: 42
          command_authorized: true
          active: true

    my_email:
      type: email
      host: smtp.example.com
      port: 587
      username: ${SMTP_USER}
      password: ${SMTP_PASS}
      from_address: noreply@example.com
      start_tls: true
      default_recipients:
        - ops@example.com
      labels:
        - alerts

from processpype.communications import OutgoingMessage, emit_message

msg = OutgoingMessage(
    message="Deployment completed successfully",
    label="default",
    source="deploy-service",
    severity="info",
)
await emit_message(msg)

# Both my_telegram and my_email are registered for "alerts"
await emit_message(OutgoingMessage(
    message="Database connection lost",
    label="alerts",
    severity="critical",
    subject="DB Alert",  # used by email backend
))

from processpype.communications import get_dispatcher

dispatcher = get_dispatcher()
await dispatcher.emit(message)

from eventspype.subscribers import FunctionalEventSubscriber
from processpype.communications import get_dispatcher, IncomingMessage

dispatcher = get_dispatcher()

# Simple functional subscriber
subscriber = FunctionalEventSubscriber(
    name="my_handler",
    publication=dispatcher.incoming_publication,
)

@subscriber.on_event
async def handle_message(message: IncomingMessage) -> None:
    print(f"[{message.backend_name}] {message.sender}: {message.text}")

subscriber.subscribe(dispatcher.incoming_publisher)

from eventspype.subscribers import TrackingEventSubscriber

tracker = TrackingEventSubscriber(
    name="message_tracker",
    publication=dispatcher.incoming_publication,
)
tracker.subscribe(dispatcher.incoming_publisher)

# Later, inspect received messages
for event in tracker.events:
    print(event.data.text)

from eventspype.subscribers import QueueEventSubscriber

queue_sub = QueueEventSubscriber(
    name="message_queue",
    publication=dispatcher.incoming_publication,
)
queue_sub.subscribe(dispatcher.incoming_publisher)

# Consume from the queue in an async task
async def process_messages():
    while True:
        event = await queue_sub.get()
        await handle(event.data)

pip install processpype[telegram]

backends:
  my_telegram:
    type: telegram
    api_id: ${TELEGRAM_API_ID}
    api_hash: ${TELEGRAM_API_HASH}
    token: ${TELEGRAM_BOT_TOKEN}
    session_string: ""
    listen_to_commands: false
    labels:
      - default
    chats:
      default:
        chat_id: "-1001234567890"
        active: true
      alerts:
        chat_id: "-1001234567890"
        topic_id: 42
        command_authorized: true
        active: true

pip install processpype[email]

backends:
  my_email:
    type: email
    host: smtp.example.com
    port: 587
    username: ${SMTP_USER}
    password: ${SMTP_PASS}
    from_address: noreply@example.com
    use_tls: true
    start_tls: false
    default_recipients:
      - ops@example.com
    labels:
      - alerts

await emit_message(OutgoingMessage(
    message="Disk usage at 95%",
    label="alerts",
    severity="warning",
    subject="Disk Alert",
    metadata={"recipients": ["admin@example.com"]},
))

from processpype.communications.base import Communicator
from processpype.communications.models import OutgoingMessage
from processpype.config.models import CommunicatorBackendConfig


class SlackCommunicator(Communicator):
    def __init__(self, name: str, config: CommunicatorBackendConfig) -> None:
        super().__init__(name, config)
        self._webhook_url = config.extra_fields.get("webhook_url", "")

    async def start(self) -> None:
        # Initialize client, open connections
        self._started = True

    async def stop(self) -> None:
        # Clean up resources
        self._started = False

    async def send(self, message: OutgoingMessage) -> None:
        # Send message via Slack webhook
        ...

    @property
    def supports_receiving(self) -> bool:
        return True

    async def start(self) -> None:
        self._started = True
        # Start listening for messages, then for each:
        # self._on_incoming(IncomingMessage(text=..., backend_name=self._name))

from processpype.communications.backends import register_backend

register_backend("slack", lambda name, config: SlackCommunicator(name, config))

communications:
  enabled: true
  backends:
    my_slack:
      type: slack
      labels:
        - default
      webhook_url: ${SLACK_WEBHOOK_URL}

from processpype.communications import init_communications

await init_communications(config.communications)