Skip to navigation
WebSockets

WebSockets

Real-time, low-latency email event streaming

WebSockets provide a persistent, bidirectional connection to PostNuvia for receiving email events in real-time. Unlike webhooks, WebSockets don’t require a public URL or external tools like ngrok.

Why Use WebSockets?

FeatureWebhookWebSocket
SetupRequires public URL + ngrokNo external tools needed
ConnectionHTTP request per eventPersistent connection
DirectionPostNuvia → Your serverBidirectional
FirewallMust expose portOutbound only
LatencyHTTP round-tripInstant streaming

Python SDK

The Python SDK provides both synchronous and asynchronous WebSocket clients.

Async Usage

import asyncio
from postnuvia import AsyncPostNuvia, Subscribe, Subscribed, MessageReceivedEvent
client = AsyncPostNuvia(api_key="YOUR_API_KEY")
async def main():
    async with client.websockets.connect() as socket:
        # Subscribe to inboxes
        await socket.send_subscribe(Subscribe(inbox_ids=["agent@postnuvia.com"]))
        # Process events as they arrive
        async for event in socket:
            if isinstance(event, Subscribed):
                print(f"Subscribed to: {event.inbox_ids}")
            elif isinstance(event, MessageReceivedEvent):
                print(f"New email from: {event.message.from_}")
                print(f"Subject: {event.message.subject}")
asyncio.run(main())

Sync Usage

from postnuvia import PostNuvia, Subscribe, Subscribed, MessageReceivedEvent
client = PostNuvia(api_key="YOUR_API_KEY")
with client.websockets.connect() as socket:
    # Subscribe to inboxes
    socket.send_subscribe(Subscribe(inbox_ids=["agent@postnuvia.com"]))
    # Process events as they arrive
    for event in socket:
        if isinstance(event, Subscribed):
            print(f"Subscribed to: {event.inbox_ids}")
        elif isinstance(event, MessageReceivedEvent):
            print(f"New email from: {event.message.from_}")
            print(f"Subject: {event.message.subject}")

Event Handler Pattern

You can also use event handlers instead of iterating:

import asyncio
from postnuvia import AsyncPostNuvia, Subscribe, EventType
client = AsyncPostNuvia(api_key="YOUR_API_KEY")
async def main():
    async with client.websockets.connect() as socket:
        # Register event handlers
        socket.on(EventType.OPEN, lambda _: print("Connected"))
        socket.on(EventType.MESSAGE, lambda msg: print("Received:", msg))
        socket.on(EventType.CLOSE, lambda _: print("Disconnected"))
        socket.on(EventType.ERROR, lambda err: print("Error:", err))
        # Subscribe and start listening
        await socket.send_subscribe(Subscribe(inbox_ids=["agent@postnuvia.com"]))
        await socket.start_listening()
asyncio.run(main())

For sync usage with event handlers, run the listener in a background thread:

import threading
from postnuvia import PostNuvia, Subscribe, EventType
client = PostNuvia(api_key="YOUR_API_KEY")
with client.websockets.connect() as socket:
    socket.on(EventType.OPEN, lambda _: print("Connected"))
    socket.on(EventType.MESSAGE, lambda msg: print("Received:", msg))
    socket.on(EventType.CLOSE, lambda _: print("Disconnected"))
    socket.on(EventType.ERROR, lambda err: print("Error:", err))
    socket.send_subscribe(Subscribe(inbox_ids=["agent@postnuvia.com"]))
    # Start listening in background thread
    listener = threading.Thread(target=socket.start_listening, daemon=True)
    listener.start()
    listener.join()

TypeScript SDK

The TypeScript SDK provides a WebSocket client with automatic reconnection.

Basic Usage

import { PostNuviaClient, PostNuvia } from "postnuvia";
const client = new PostNuviaClient({
  apiKey: process.env.POSTNUVIA_API_KEY,
});
async function main() {
  const socket = await client.websockets.connect();
  // Handle events
  socket.on("open", () => {
    console.log("Connected");
    // Subscribe to inboxes after connection is open
    socket.sendSubscribe({
      type: "subscribe",
      inboxIds: ["agent@postnuvia.com"],
    });
  });
  socket.on("message", (event: PostNuvia.Subscribed | PostNuvia.MessageReceivedEvent) => {
    if (event.type === "subscribed") {
      console.log("Subscribed to:", event.inboxIds);
    } else if (event.type === "message_received") {
      console.log("New email from:", event.message.from_);
      console.log("Subject:", event.message.subject);
    }
  });
  socket.on("close", (event) => {
    console.log("Disconnected:", event.code, event.reason);
  });
  socket.on("error", (error) => {
    console.error("Error:", error);
  });
}
main();

React/Next.js Usage

Using the SDK with React:

import { useEffect, useState } from "react";
import { PostNuviaClient, PostNuvia } from "postnuvia";
function usePostNuviaWebSocket(apiKey: string, inboxIds: string[]) {
  const [lastMessage, setLastMessage] = useState<PostNuvia.MessageReceivedEvent | null>(null);
  const [isConnected, setIsConnected] = useState(false);
  useEffect(() => {
    const client = new PostNuviaClient({ apiKey });
    let socket: Awaited<ReturnType<typeof client.websockets.connect>>;
    async function connect() {
      socket = await client.websockets.connect();
      socket.on("open", () => {
        setIsConnected(true);
        socket.sendSubscribe({
          type: "subscribe",
          inboxIds,
        });
      });
      socket.on("message", (event) => {
        if (event.type === "message_received") {
          setLastMessage(event);
        }
      });
      socket.on("close", () => setIsConnected(false));
    }
    connect();
    return () => socket?.close();
  }, [apiKey, inboxIds.join(",")]);
  return { lastMessage, isConnected };
}

Subscribe Options

When subscribing to events, you can filter by inbox, pod, or event type:

Python:

from postnuvia import Subscribe
# Subscribe to specific inboxes
Subscribe(inbox_ids=["inbox1@postnuvia.com", "inbox2@postnuvia.com"])
# Subscribe to pods
Subscribe(pod_ids=["pod-id-1", "pod-id-2"])
# Subscribe to specific event types
Subscribe(
    inbox_ids=["agent@postnuvia.com"],
    event_types=["message.received", "message.sent"]
)
# Subscribe to filtered inbound events
Subscribe(
    inbox_ids=["agent@postnuvia.com"],
    event_types=[
        "message.received",
        "message.received.spam",
        "message.received.blocked",
        "message.received.unauthenticated",
    ]
)

TypeScript:

// Subscribe to specific inboxes
socket.sendSubscribe({
  type: "subscribe",
  inboxIds: ["inbox1@postnuvia.com", "inbox2@postnuvia.com"],
});
// Subscribe to pods
socket.sendSubscribe({
  type: "subscribe",
  podIds: ["pod-id-1", "pod-id-2"],
});
// Subscribe to specific event types
socket.sendSubscribe({
  type: "subscribe",
  inboxIds: ["agent@postnuvia.com"],
  eventTypes: ["message.received", "message.sent"],
});
// Subscribe to filtered inbound events
socket.sendSubscribe({
  type: "subscribe",
  inboxIds: ["agent@postnuvia.com"],
  eventTypes: [
    "message.received",
    "message.received.spam",
    "message.received.blocked",
    "message.received.unauthenticated",
  ],
});

A subscription without event_types receives every event type except calendar events. It receives spam, blocked, and unauthenticated events only if the API key had the matching label visibility permission (label_spam_read, label_blocked_read, or label_unauthenticated_read) when the connection opened. Naming one of them in event_types without that permission makes the subscribe fail. Calendar events reach only subscriptions that name them in event_types. Each subscribe replaces the event types of any earlier subscription to the same inboxes or pods on that connection, so list every event type you want.

Event Types

Connection Events

EventPythonTypeScriptDescription
subscribedSubscribedPostNuvia.SubscribedSubscription confirmed

Message Events

EventPythonTypeScriptDescription
message_receivedMessageReceivedEventPostNuvia.MessageReceivedEventEmail received. Check event_type for message.received, message.received.spam, message.received.blocked, or message.received.unauthenticated
message_sentMessageSentEventPostNuvia.MessageSentEventEmail was sent
message_deliveredMessageDeliveredEventPostNuvia.MessageDeliveredEventEmail was delivered
message_bouncedMessageBouncedEventPostNuvia.MessageBouncedEventEmail bounced
message_complainedMessageComplainedEventPostNuvia.MessageComplainedEventEmail marked as spam
message_rejectedMessageRejectedEventPostNuvia.MessageRejectedEventEmail was rejected
message_openedMessageOpenedEventPostNuvia.MessageOpenedEventTracked email was opened for the first time

Domain Events

EventPythonTypeScriptDescription
domain_verifiedDomainVerifiedEventPostNuvia.DomainVerifiedEventDomain verification completed

Calendar Events

Calendar events require the calendar_event_read permission, and reach only subscriptions that name them in event_types. See Calendar Webhooks for payloads and timing.

EventPythonTypeScriptDescription
calendar_event_createdCalendarEventCreatedEventPostNuvia.CalendarEventCreatedEventA calendar event was created, or a date of a recurring event was first edited
calendar_event_updatedCalendarEventUpdatedEventPostNuvia.CalendarEventUpdatedEventA calendar event or date changed, or a date was cancelled
calendar_event_deletedCalendarEventDeletedEventPostNuvia.CalendarEventDeletedEventA calendar event was deleted
calendar_event_respondedCalendarEventRespondedEventPostNuvia.CalendarEventRespondedEventAn attendee’s response changed
calendar_event_startingCalendarEventStartingEventPostNuvia.CalendarEventStartingEventA calendar event or date started
calendar_event_endingCalendarEventEndingEventPostNuvia.CalendarEventEndingEventA calendar event or date ended

Message Properties

The event.message object contains:

PythonTypeScriptDescription
inbox_idinboxIdInbox that received the email
message_idmessageIdUnique message ID
thread_idthreadIdConversation thread ID
from_from_Sender email address
totoRecipients list
subjectsubjectSubject line
texttextPlain text body
htmlhtmlHTML body (if present)
attachmentsattachmentsList of attachments

Error Handling

Python:

from postnuvia import AsyncPostNuvia, Subscribe, MessageReceivedEvent
from postnuvia.core.api_error import ApiError
client = AsyncPostNuvia(api_key="YOUR_API_KEY")
async def main():
    try:
        async with client.websockets.connect() as socket:
            await socket.send_subscribe(Subscribe(inbox_ids=["agent@postnuvia.com"]))
            async for event in socket:
                if isinstance(event, MessageReceivedEvent):
                    await process_email(event.message)
    except ApiError as e:
        print(f"API error: {e.status_code} - {e.body}")
    except Exception as e:
        print(f"Connection error: {e}")

TypeScript:

import { PostNuviaClient, PostNuvia, PostNuviaError } from "postnuvia";
const client = new PostNuviaClient({
  apiKey: process.env.POSTNUVIA_API_KEY,
});
async function main() {
  try {
    const socket = await client.websockets.connect();
    socket.on("open", () => {
      socket.sendSubscribe({
        type: "subscribe",
        inboxIds: ["agent@postnuvia.com"],
      });
    });
    socket.on("message", (event: PostNuvia.MessageReceivedEvent) => {
      if (event.type === "message_received") {
        processEmail(event.message);
      }
    });
    socket.on("error", (error) => {
      console.error("WebSocket error:", error);
    });
    socket.on("close", (event) => {
      console.log("Disconnected:", event.code, event.reason);
    });
  } catch (err) {
    if (err instanceof PostNuviaError) {
      console.error(`API error: ${err.statusCode} - ${err.message}`);
    } else {
      console.error("Connection error:", err);
    }
  }
}
main();

Copy for Cursor / Claude

Copy one of the blocks below into Cursor or Claude for WebSockets in one shot.

"""
PostNuvia WebSockets — copy into Cursor/Claude. Real-time events, no public URL needed.
Sync: with client.websockets.connect() as socket: socket.send_subscribe(Subscribe(inbox_ids=[...])); for event in socket: ...
Async: async with client.websockets.connect() as socket: await socket.send_subscribe(...); async for event in socket: ...
Subscribe(inbox_ids=[...], pod_ids=[...], event_types=[...])
Event types: Subscribed, MessageReceivedEvent, MessageSentEvent, MessageDeliveredEvent, MessageBouncedEvent, MessageComplainedEvent, MessageRejectedEvent, MessageOpenedEvent, DomainVerifiedEvent, CalendarEventCreatedEvent, CalendarEventUpdatedEvent, CalendarEventDeletedEvent, CalendarEventRespondedEvent, CalendarEventStartingEvent, CalendarEventEndingEvent
Omitting event_types receives every event type except calendar.event.*, which must be named. Spam, blocked, and unauthenticated events also need the matching label_spam_read / label_blocked_read / label_unauthenticated_read permission.
"""
from postnuvia import PostNuvia, Subscribe, Subscribed, MessageReceivedEvent
client = PostNuvia(api_key="YOUR_API_KEY")
with client.websockets.connect() as socket:
  socket.send_subscribe(Subscribe(inbox_ids=["agent@postnuvia.com"]))
  for event in socket:
    if isinstance(event, Subscribed): print(event.inbox_ids)
    elif isinstance(event, MessageReceivedEvent): print(event.message.subject)