How do I handle inbound emails with my agent? | PostNuvia

PostNuvia offers two ways to process incoming emails, each suited to different use cases.

1. Webhooks (Recommended for Production)

Configure a webhook URL and PostNuvia will send a POST request to your endpoint whenever an email arrives. This is the most reliable approach for production applications.

Python

from flask import Flask, request
from postnuvia import PostNuvia
app = Flask(__name__)
client = PostNuvia()
@app.route("/webhooks", methods=["POST"])
def handle_webhook():
    payload = request.json
    
    if payload["event_type"] == "message.received":
        message = payload["message"]
        
        # Your agent processes the email here
        reply_text = your_agent.process(message)
        
        # Reply in the same thread
        client.inboxes.messages.reply(
            inbox_id=message["inbox_id"],
            message_id=message["message_id"],
            text=reply_text
        )
    
    return "OK", 200

Python

client.webhooks.create(
    url="https://your-domain.ngrok-free.app/webhooks",
    events=["message.received"],
)

Always return a 200 OK immediately and process the webhook in the background. If your endpoint takes too long to respond, PostNuvia will retry delivery. Also, filter out message.sent events to prevent your agent from replying to its own messages in a loop.

For local development, use ngrok to expose your local server. See the Webhook Setup Guide for full instructions.

2. WebSockets (Best for Real-Time, No Public URL)

Stream email events over a persistent connection. No public URL or ngrok needed, which makes this ideal for local development and desktop agents.

Python

import asyncio
from postnuvia import AsyncPostNuvia, Subscribe, Subscribed, MessageReceivedEvent
client = AsyncPostNuvia()
async def main():
    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, 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())

The SDK also provides a synchronous client if you prefer:

Python

from postnuvia import PostNuvia, Subscribe, MessageReceivedEvent
client = PostNuvia()
with client.websockets.connect() as socket:
    socket.send_subscribe(Subscribe(
        inbox_ids=["agent@postnuvia.com"]
    ))
    for event in socket:
        if isinstance(event, MessageReceivedEvent):
            print(f"New email from: {event.message.from_}")

See the WebSocket Overview for more details.

Which should I use?

Method	Best for	Requires public URL?	Real-time?
Webhooks	Production applications	Yes	Yes
WebSockets	Local dev, desktop agents	No	Yes

For most production use cases, webhooks are recommended. They are reliable, event-driven, and integrate well with serverless platforms. If you need real-time events without exposing a public URL, WebSockets are the best option.