> For clean Markdown of any page, append .md to the page URL.
> For a complete documentation index, see https://agentinbox.space/llms.txt.
> For AI client integration (Claude Code, Cursor, etc.), connect to the MCP server at https://agentinbox.space/_mcp/server.

# Sales-агент на WebSocket

> AI sales-агент с WebSocket: обработка почты в реальном времени без опроса и без публичного URL вебхука.

## Обзор

Соберите sales-агента, который обрабатывает письма через WebSocket сразу по мере поступления. В отличие от вебхуков, не нужны ngrok и публичный URL — соединение к Agent Inbox исходящее, настройка минимальная.

Сценарий: менеджер передаёт агенту данные лида; агент ведёт переписку с клиентом и сигнализирует менеджеру о важных намерениях.

## Что вы получите

1. **WebSocket** — поток событий в реальном времени
2. **Письма менеджера** — извлечение email клиента и персонализированный аутрич
3. **Ответы клиента** — ответы ИИ с учётом контекста
4. **Уведомления менеджеру** — при сильных сигналах интереса / отказа

Схема:

```
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 и вебхук

| Критерий    | Вебхук                | WebSocket               |
| ----------- | --------------------- | ----------------------- |
| Настройка   | ngrok + публичный URL | Без лишних инструментов |
| Архитектура | Flask + HTTP          | Async Python            |
| Задержка    | HTTP round-trip       | Поток                   |
| Файрвол     | Открытый порт         | Только исходящие        |

Подробнее: [справочник WebSocket](/api-reference/websockets) и [документация SDK](https://github.com/PATTERN-AUTOMATION-LAB/agent-inbox-sdk?tab=readme-ov-file#websockets).

## Требования

> **Info**
>
> **Нужно:**
>
> * Python 3.11+
> * [Аккаунт Agent Inbox ](https://agentinbox-ru.vercel.app/)и API-ключ
> * [Аккаунт OpenAI](https://platform.openai.com) и API-ключ

## Проект

### Шаг 1: каталог

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

### Шаг 2: `main.py`

#### Показать полный код main.py

```python
"""
Sales Agent using Agent Inbox WebSocket

This is a simple example showing how to:
- Connect to Agent Inbox 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 agentinbox import AsyncAgentinbox, Subscribe, Subscribed, MessageReceivedEvent
from openai import AsyncOpenAI

# Load environment variables
load_dotenv()

# Initialize clients
agentinbox = AsyncAgentinbox(api_key=os.getenv("AGENTINBOX_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 agentinbox.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 agentinbox.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}@agentinbox.space"

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

    # Connect to WebSocket
    try:
        async with agentinbox.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()
```

### Шаг 3: `requirements.txt`

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

### Шаг 4: установка

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

### Шаг 5: `.env`

```env
# Agent Inbox Configuration
AGENTINBOX_API_KEY=your_agentinbox_api_key_here

# OpenAI Configuration
OPENAI_API_KEY=your_openai_api_key_here

# Inbox Settings
INBOX_USERNAME=sales-agent
```

> **Info**
>
> **Замечание:** в отличие от вебхуков, ngrok и публичный URL не нужны. WebSocket — исходящее соединение, работает за файрволом без проброса портов.

## Разбор кода

### Схема

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

### 1. Подключение WebSocket

Основа агента — постоянное соединение:

```python
from agentinbox import AsyncAgentinbox, Subscribe, Subscribed, MessageReceivedEvent

# Initialize the async client
agentinbox = AsyncAgentinbox(api_key=os.getenv("AGENTINBOX_API_KEY"))

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

**Кратко:**

* `AsyncAgentinbox` — асинхронный клиент
* `agentinbox.websockets.connect()` — WebSocket
* `Subscribe` — какие inbox слушать

### 2. Цикл событий

Асинхронный итератор по событиям:

```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)
```

**Типы событий:**

* `Subscribed` — подписка подтверждена
* `MessageReceivedEvent` — новое письмо

### 3. Маршрутизация писем

По содержимому письма:

```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)
```

**Разбор From:**

```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. Письмо от менеджера

Если в теле есть данные лида:

```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. Клиент и намерение

История треда и грубая классификация намерения по ключевым словам:

```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. Ответы OpenAI

Генерация текста:

```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."
```

**Важно:** в промпт попадает история переписки; при ошибке API — запасной текст.

## Запуск

```bash
python main.py
```

Ожидаемый вывод:

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

✓ Subscribed to: ['sales-agent@agentinbox.space']
```

> **Success**
>
> **Готово.** Агент слушает почту в реальном времени. Не закрывайте терминал.

## Проверка

### Сценарий: запрос менеджера

**Отправьте с личной почты:**

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

**Консоль:**

```
============================================================
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
```

**В почте:** подтверждение с текстом, отправленным клиенту.

> **Success**
>
> **Работает:** события в реальном времени, ответы ИИ, уведомление менеджера при интересе.

## Настройка

### Ключевые слова намерения

В `handle_customer_email()` правьте `intent_keywords`:

```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
}
```

### Несколько inbox

Подписка на несколько ящиков:

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

## Устранение неполадок

### Частые случаи

#### WebSocket: отказ или таймаут

**Проблема:** нет соединения с WebSocket Agent Inbox.

**Что проверить:**

1. Проверьте, что API-ключ верный:

```python
client = AsyncAgentinbox(api_key="your-key")
print(await client.inboxes.list())  # Должно выполниться без ошибки
```

2. Проверьте интернет и настройки файрвола
3. Убедитесь, что установлен `agentinbox>=0.0.19` с поддержкой WebSocket:

```bash
pip show agentinbox
```

#### Письма не приходят в агент

**Проблема:** процесс запущен, событий нет.

**Чеклист:**

1. Убедитесь, что inbox существует:

```python
client = AsyncAgentinbox()
print(await client.inboxes.get("sales-agent@agentinbox.space"))
```

2. Проверьте в консоли подтверждение подписки
3. Отправьте тестовое письмо на правильный адрес inbox
4. Убедитесь, что письмо не попадает в спам

#### Ошибки asyncio

**Проблема:** исключения в async-коде.

**Что сделать:**

1. Убедитесь, что установлен Python 3.11+:

```bash
python --version
```

2. Не смешивайте синхронный и асинхронный код некорректно
3. Используйте `asyncio.run(main())` как точку входа

---

Идеи и проекты на Agent Inbox — [info@patternautomation.com](mailto:info@patternautomation.com).