> For clean Markdown content of this page, append .md to this URL. For the complete documentation index, see https://docs.postnuvia.com/llms.txt.

# Sales Agent with WebSocket

> A step-by-step guide to building an AI-powered sales agent that uses WebSocket for real-time email processing without polling or webhooks.

## Overview

Learn how to build a real-time sales agent that processes emails instantly using WebSocket connections. Unlike webhook-based agents that require ngrok and public URLs, this WebSocket approach connects directly to PostNuvia for true real-time processing with minimal setup.

This agent demonstrates a practical sales workflow: a manager delegates customer outreach to the agent, which then handles the entire conversation autonomously while keeping the manager informed of key signals.

## What You'll Build

By the end of this guide, you'll have a working sales agent that:

1. **Connects via WebSocket** for instant, real-time email processing
2. **Handles manager emails** by extracting customer info and sending personalized outreach
3. **Processes customer replies** with AI-powered, context-aware responses
4. **Notifies the manager** when customers show strong buying signals

Here's the workflow:

```
Manager sends email with customer info
         ↓
    Agent extracts customer email
         ↓
    Agent generates AI sales pitch → Sends to customer
         ↓
    Agent confirms to manager
         ↓
    [Customer replies]
         ↓
    Agent detects intent + generates AI response
         ↓
    If interested → Notifies manager
```

## WebSocket vs Webhook: Why WebSocket?

| Feature      | Webhook Approach            | WebSocket Approach       |
| ------------ | --------------------------- | ------------------------ |
| Setup        | Requires ngrok + public URL | No external tools needed |
| Architecture | Flask server + HTTP         | Pure async Python        |
| Latency      | HTTP round-trip             | Instant streaming        |
| Firewall     | Must expose port            | Outbound only            |

For more details, see the [WebSocket API Reference](/api-reference/websockets) and the [Python SDK WebSocket documentation](https://github.com/postnuvia-to/postnuvia-python?tab=readme-ov-file#websockets).

## Prerequisites

Before you begin, make sure you have:

> **Info**
>
> **Required:**
>
> * Python 3.11 or higher installed
> * An [PostNuvia account](https://console.postnuvia.com) and API key
> * An [OpenAI account](https://platform.openai.com) and API key

## Project Setup

### Step 1: Create Project Directory

Create a new directory for your agent:

```bash
mkdir sales-agent-websocket
cd sales-agent-websocket
```

### Step 2: Create the Agent Code

Create a file named `main.py` and paste the following code:

#### Click to view full main.py code

```python
"""
Sales Agent using PostNuvia WebSocket

This is a simple example showing how to:
- Connect to PostNuvia via WebSocket for real-time email processing
- Use OpenAI to handle sales conversations
- Send emails to customers and respond to replies
"""

import asyncio
import os
import re
from dotenv import load_dotenv
from postnuvia import AsyncPostNuvia, Subscribe, Subscribed, MessageReceivedEvent
from openai import AsyncOpenAI

# Load environment variables
load_dotenv()

# Initialize clients
postnuvia = AsyncPostNuvia(api_key=os.getenv("POSTNUVIA_API_KEY"))
openai = AsyncOpenAI(api_key=os.getenv("OPENAI_API_KEY"))

# Simple conversation history (thread_id -> messages)
conversations = {}

# Store manager email for notifications
manager_email = None


def extract_email(from_field):
    """Extract email address from 'Name <email@example.com>' format"""
    match = re.search(r'<(.+?)>', from_field)
    return match.group(1) if match else from_field


def is_from_manager(email_body):
    """Simple check if email is from sales manager (contains customer info)"""
    keywords = ['customer', 'lead', 'contact', 'reach out', 'email']
    return any(keyword in email_body.lower() for keyword in keywords)


def extract_customer_info(email_body):
    """Extract customer email from manager's message"""
    email_pattern = r'\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,}\b'
    emails = re.findall(email_pattern, email_body)

    # Return the first email found (should be the customer's email in the message body)
    if emails:
        return emails[0]
    return None


async def get_ai_response(messages, system_prompt):
    """Get response from OpenAI"""
    try:
        response = await openai.chat.completions.create(
            model="gpt-4o-mini",
            messages=[
                {"role": "system", "content": system_prompt},
                *messages
            ],
            temperature=0.7,
        )
        return response.choices[0].message.content
    except Exception as e:
        print(f"Error getting AI response: {e}")
        return "I apologize, but I encountered an error. Please try again."


async def send_email(inbox_id, to_email, subject, body):
    """Send a new email"""
    try:
        await postnuvia.inboxes.messages.send(
            inbox_id=inbox_id,
            to=[to_email],
            subject=subject,
            text=body
        )
        print(f"✓ Sent email to {to_email}")
    except Exception as e:
        print(f"Error sending email: {e}")


async def reply_to_email(inbox_id, message_id, to_email, body):
    """Reply to an email"""
    try:
        await postnuvia.inboxes.messages.reply(
            inbox_id=inbox_id,
            message_id=message_id,
            to=[to_email],  # Required parameter for replies
            text=body
        )
        print(f"✓ Sent reply to {to_email}")
    except Exception as e:
        print(f"Error replying: {e}")


async def handle_manager_email(inbox_id, message_id, from_email, subject, body):
    """Handle email from sales manager - extract customer and send sales pitch"""
    global manager_email
    manager_email = from_email  # Remember manager for future notifications

    print(f"\n📧 Email from MANAGER: {from_email}")

    # Extract customer email
    customer_email = extract_customer_info(body)
    print(f"→ Extracted customer email: {customer_email}")

    if not customer_email:
        await reply_to_email(
            inbox_id,
            message_id,
            from_email,  # Reply back to the manager
            "I couldn't find a customer email address. Please include it in your message."
        )
        return

    # Generate sales pitch using AI
    system_prompt = """You are a helpful sales agent. Generate a brief, professional sales email
    based on the manager's request. Keep it under 150 words. Be friendly and professional."""

    messages = [{"role": "user", "content": f"Create a sales email based on this: {body}"}]
    sales_pitch = await get_ai_response(messages, system_prompt)

    # Send email to customer
    await send_email(
        inbox_id,
        customer_email,
        f"Introduction: {subject}" if subject else "Quick Introduction",
        sales_pitch
    )

    # Confirm to manager
    await reply_to_email(
        inbox_id,
        message_id,
        from_email,  # Reply back to the manager
        f"✓ I've sent an introduction email to {customer_email}.\n\nHere's what I sent:\n\n{sales_pitch}"
    )


async def handle_customer_email(inbox_id, message_id, thread_id, from_email, subject, body):
    """Handle email from customer - track conversation, detect intent, and notify manager"""
    print(f"\n📧 Email from CUSTOMER: {from_email}")

    # Track conversation history
    if thread_id not in conversations:
        conversations[thread_id] = []
    conversations[thread_id].append({"role": "user", "content": body})

    # Detect customer intent
    intent_keywords = {
        'interested': ['interested', 'demo', 'meeting', 'tell me more', 'sounds good'],
        'not_interested': ['not interested', 'no thank', 'not right now', 'maybe later'],
        'question': ['?', 'how', 'what', 'when', 'why', 'can you']
    }

    body_lower = body.lower()
    intent = 'question'  # default
    for key, keywords in intent_keywords.items():
        if any(keyword in body_lower for keyword in keywords):
            intent = key
            break

    # Generate AI response
    system_prompt = """You are a helpful sales agent. Answer customer questions professionally
    and helpfully. Keep responses brief (under 100 words). Be friendly but professional."""

    response = await get_ai_response(conversations[thread_id], system_prompt)

    # Reply to customer
    await reply_to_email(inbox_id, message_id, from_email, response)

    # Notify manager if strong intent signal
    if manager_email and intent in ['interested', 'not_interested']:
        status = "showing interest" if intent == 'interested' else "not interested at this time"
        await send_email(
            inbox_id,
            manager_email,
            f"Update: {from_email}",
            f"Customer {from_email} is {status}.\n\nTheir message:\n{body}\n\nMy response:\n{response}"
        )
        print(f"→ Notified manager about customer's {intent}")

    # Update conversation history
    conversations[thread_id].append({"role": "assistant", "content": response})


async def handle_new_email(message):
    """Process incoming email from WebSocket"""
    try:
        # Extract message data using object attributes
        inbox_id = message.inbox_id
        message_id = message.message_id
        thread_id = message.thread_id
        from_field = message.from_ or ""  # SDK uses from_
        from_email = extract_email(from_field)
        subject = message.subject or ""
        body = message.text or ""  # SDK uses text for the body

        print(f"\n{'='*60}")
        print(f"New email from: {from_email}")
        print(f"Subject: {subject}")
        print(f"{'='*60}")

        # Determine if from manager or customer
        if is_from_manager(body):
            await handle_manager_email(inbox_id, message_id, from_email, subject, body)
        else:
            await handle_customer_email(inbox_id, message_id, thread_id, from_email, subject, body)

    except Exception as e:
        print(f"Error handling email: {e}")


async def main():
    """Main WebSocket loop"""
    inbox_username = os.getenv("INBOX_USERNAME", "sales-agent")
    inbox_id = f"{inbox_username}@postnuvia.com"

    print(f"\nSales Agent starting...")
    print(f"Inbox: {inbox_id}")
    print(f"✓ Connecting to PostNuvia WebSocket...")

    # Connect to WebSocket
    try:
        async with postnuvia.websockets.connect() as socket:
            print(f"✓ Connected! Listening for emails...\n")

            # Subscribe to inbox
            await socket.send_subscribe(Subscribe(inbox_ids=[inbox_id]))

            # Listen for events
            async for event in socket:
                if isinstance(event, Subscribed):
                    print(f"✓ Subscribed to: {event.inbox_ids}\n")

                elif isinstance(event, MessageReceivedEvent):
                    print(f"📨 New email received!")
                    await handle_new_email(event.message)

    except (KeyboardInterrupt, asyncio.CancelledError):
        print("\n\nShutting down gracefully...")
    except Exception as e:
        print(f"\nError: {e}")


def run():
    """Run the main function"""
    try:
        asyncio.run(main())
    except KeyboardInterrupt:
        print("\n✓ Shutdown complete")


if __name__ == "__main__":
    run()
```

### Step 3: Create Requirements File

Create a file named `requirements.txt`:

```txt
postnuvia>=0.0.19
openai>=1.0.0
python-dotenv>=1.0.0
```

### Step 4: Install Dependencies

Install the required Python packages:

```bash
pip install -r requirements.txt
# or with pyproject.toml
pip install .
```

### Step 5: Configure Environment Variables

Create a `.env` file with your credentials:

```env
# PostNuvia Configuration
POSTNUVIA_API_KEY=your_postnuvia_api_key_here

# OpenAI Configuration
OPENAI_API_KEY=your_openai_api_key_here

# Inbox Settings
INBOX_USERNAME=sales-agent
```

> **Info**
>
> **Note:** Unlike webhook-based agents, you don't need ngrok or a public URL. The WebSocket connection is outbound only, so it works behind firewalls without any port forwarding.

## Code Walkthrough

Let's understand how the agent works by breaking down the key components.

### Architecture Overview

```
┌─────────────────────────────────────────────────────────────┐
│                      Your Python Script                      │
│                                                              │
│  ┌──────────────┐     ┌──────────────┐     ┌─────────────┐  │
│  │ AsyncAgent   │────▶│ WebSocket    │────▶│ Event       │  │
│  │ Mail Client  │     │ Connection   │     │ Handler     │  │
│  └──────────────┘     └──────────────┘     └─────────────┘  │
│         │                    ▲                    │          │
│         │                    │                    ▼          │
│         │              Real-time            ┌─────────────┐  │
│         │              Events               │ OpenAI      │  │
│         │                                   │ Integration │  │
│         ▼                                   └─────────────┘  │
│  ┌──────────────┐                                           │
│  │ Send/Reply   │◀──────────────────────────────────────────│
│  │ Emails       │                                           │
│  └──────────────┘                                           │
└─────────────────────────────────────────────────────────────┘
                              │
                              ▼
                    ┌──────────────────┐
                    │   PostNuvia      │
                    │   Cloud          │
                    └──────────────────┘
```

### 1. WebSocket Connection Setup

The core of this agent is the WebSocket connection:

```python
from postnuvia import AsyncPostNuvia, Subscribe, Subscribed, MessageReceivedEvent

# Initialize the async client
postnuvia = AsyncPostNuvia(api_key=os.getenv("POSTNUVIA_API_KEY"))

# Connect and subscribe
async with postnuvia.websockets.connect() as socket:
    await socket.send_subscribe(Subscribe(inbox_ids=[inbox_id]))
```

**Key points:**

* `AsyncPostNuvia` is the async version of the client
* `postnuvia.websockets.connect()` creates the WebSocket connection
* `Subscribe` specifies which inboxes to monitor

### 2. Event Handling Loop

The agent uses an async iterator pattern to process events:

```python
async for event in socket:
    if isinstance(event, Subscribed):
        print(f"✓ Subscribed to: {event.inbox_ids}")

    elif isinstance(event, MessageReceivedEvent):
        await handle_new_email(event.message)
```

**Event types:**

* `Subscribed` - Confirmation that subscription was successful
* `MessageReceivedEvent` - A new email arrived in the inbox

### 3. Email Processing Flow

The agent routes emails based on content:

```python
async def handle_new_email(message):
    # Extract fields from the message object
    inbox_id = message.inbox_id
    message_id = message.message_id
    thread_id = message.thread_id
    from_field = message.from_ or ""  # Note: SDK uses from_
    from_email = extract_email(from_field)
    subject = message.subject or ""
    body = message.text or ""

    # Route based on email content
    if is_from_manager(body):
        await handle_manager_email(inbox_id, message_id, from_email, subject, body)
    else:
        await handle_customer_email(inbox_id, message_id, thread_id, from_email, subject, body)
```

**Email extraction helper:**

```python
def extract_email(from_field):
    """Extract email from 'Name <email@example.com>' format"""
    match = re.search(r'<(.+?)>', from_field)
    return match.group(1) if match else from_field
```

### 4. Manager Email Handler

When the manager sends an email with customer info:

```python
async def handle_manager_email(inbox_id, message_id, from_email, subject, body):
    global manager_email
    manager_email = from_email  # Remember for notifications

    # Extract customer email using regex
    customer_email = extract_customer_info(body)

    if not customer_email:
        await reply_to_email(inbox_id, message_id, from_email,
            "I couldn't find a customer email. Please include it.")
        return

    # Generate AI sales pitch
    sales_pitch = await get_ai_response(
        [{"role": "user", "content": f"Create a sales email based on: {body}"}],
        "You are a helpful sales agent. Generate a brief, professional email..."
    )

    # Send to customer
    await send_email(inbox_id, customer_email, f"Introduction: {subject}", sales_pitch)

    # Confirm to manager
    await reply_to_email(inbox_id, message_id, from_email,
        f"✓ Sent email to {customer_email}.\n\nContent:\n{sales_pitch}")
```

### 5. Customer Email Handler with Intent Detection

The agent tracks conversations and detects customer intent:

```python
async def handle_customer_email(inbox_id, message_id, thread_id, from_email, subject, body):
    # Track conversation history per thread
    if thread_id not in conversations:
        conversations[thread_id] = []
    conversations[thread_id].append({"role": "user", "content": body})

    # Detect intent with keyword matching
    intent_keywords = {
        'interested': ['interested', 'demo', 'meeting', 'tell me more'],
        'not_interested': ['not interested', 'no thank', 'maybe later'],
        'question': ['?', 'how', 'what', 'when', 'why']
    }

    intent = 'question'  # default
    for key, keywords in intent_keywords.items():
        if any(kw in body.lower() for kw in keywords):
            intent = key
            break

    # Generate contextual AI response using conversation history
    response = await get_ai_response(conversations[thread_id], system_prompt)

    # Reply to customer
    await reply_to_email(inbox_id, message_id, from_email, response)

    # Notify manager of strong signals
    if manager_email and intent in ['interested', 'not_interested']:
        await send_email(inbox_id, manager_email, f"Update: {from_email}",
            f"Customer is {intent}.\n\nTheir message:\n{body}")
```

### 6. AI Response Generation

The agent uses OpenAI for generating responses:

```python
async def get_ai_response(messages, system_prompt):
    try:
        response = await openai.chat.completions.create(
            model="gpt-4o-mini",
            messages=[
                {"role": "system", "content": system_prompt},
                *messages  # Include conversation history
            ],
            temperature=0.7,
        )
        return response.choices[0].message.content
    except Exception as e:
        print(f"Error: {e}")
        return "I apologize, but I encountered an error."
```

**Key points:**

* Includes full conversation history for context
* Graceful error handling with fallback message

## Running the Agent

Start the agent:

```bash
python main.py
```

You should see output like this:

```
Sales Agent starting...
Inbox: sales-agent@postnuvia.com
✓ Connecting to PostNuvia WebSocket...
✓ Connected! Listening for emails...

✓ Subscribed to: ['sales-agent@postnuvia.com']
```

> **Success**
>
> **Success!** Your agent is now running and listening for emails in real-time.
>
> Leave this terminal window open - closing it will stop the agent.

## Testing Your Agent

Let's verify everything works with some test scenarios.

### Test Scenario: Manager Outreach Request

**Send this email from your personal email:**

```
To: sales-agent@postnuvia.com
Subject: New lead - AI startup
Body: Please reach out to this customer: customer-email@gmail.com
      They're interested in our API platform.
```

**Expected console output:**

```
============================================================
New email from: your-email@gmail.com
Subject: New lead - AI startup
============================================================

📧 Email from MANAGER: your-email@gmail.com
→ Extracted customer email: customer-email@gmail.com
✓ Sent email to customer-email@gmail.com
✓ Sent reply to your-email@gmail.com
```

**You'll receive:** A confirmation email with the sales pitch that was sent.

> **Success**
>
> **It works!** The agent processed emails in real-time, generated AI responses, and notified you about the interested customer.

## Customization

### Modifying Intent Detection

Update the `intent_keywords` dictionary in `handle_customer_email()`:

```python
intent_keywords = {
    'interested': ['interested', 'demo', 'meeting', 'pricing', 'sign up'],
    'not_interested': ['not interested', 'unsubscribe', 'remove me'],
    'question': ['?', 'how', 'what', 'when', 'can you'],
    'urgent': ['urgent', 'asap', 'immediately']  # Add new intent
}
```

### Adding More Inbox Subscriptions

Subscribe to multiple inboxes:

```python
await socket.send_subscribe(Subscribe(inbox_ids=[
    "sales-agent@postnuvia.com",
    "support-agent@postnuvia.com",
    "info@yourdomain.com"
]))
```

## Troubleshooting

### Common Issues

#### WebSocket connection refused or timeout

**Problem:** Cannot connect to PostNuvia WebSocket.

**Solutions:**

1. Verify your API key is correct:

```python
client = AsyncPostNuvia(api_key="your-key")
print(await client.inboxes.list())  # Should succeed
```

2. Check your internet connection and firewall settings
3. Ensure you're using `postnuvia>=0.0.19` which includes WebSocket support:

```bash
pip show postnuvia
```

#### Emails not being received

**Problem:** Agent is running but not receiving emails.

**Checklist:**

1. Verify the inbox exists:

```python
client = AsyncPostNuvia()
print(await client.inboxes.get("sales-agent@postnuvia.com"))
```

2. Check subscription confirmation in console output
3. Send test email to the correct inbox address
4. Verify the email isn't being filtered as spam

#### Agent crashes with asyncio errors

**Problem:** Async-related exceptions.

**Solutions:**

1. Ensure Python 3.11+ is installed:

```bash
python --version
```

2. Don't mix sync and async code improperly
3. Use `asyncio.run(main())` as the entry point

---

If you build something cool with PostNuvia, we'd love to hear about it. Share in our [Discord community](https://discord.gg/hTYatWYWBc)!