WebSockets | Agent Inbox | Документация

WebSocket даёт постоянное двунаправленное соединение с Agent Inbox для событий почты в реальном времени. В отличие от вебхуков, не нужен публичный URL и инструменты вроде ngrok.

Зачем WebSocket?

Возможность	Webhook	WebSocket
Настройка	Нужен публичный URL + ngrok	Дополнительные инструменты не нужны
Соединение	HTTP на каждое событие	Постоянное соединение
Направление	Agent Inbox → ваш сервер	Двунаправленно
Файрвол	Нужно открыть порт	Только исходящие
Задержка	HTTP round-trip	Поток «сразу»

Python SDK

В Python SDK есть синхронный и асинхронный клиенты WebSocket.

Асинхронное использование

import asyncio
from agentinbox import AsyncAgentinbox, Subscribe, Subscribed, MessageReceivedEvent
client = AsyncAgentinbox(api_key="YOUR_API_KEY")
async def main():
    async with client.websockets.connect() as socket:
        # Подписка на inbox
        await socket.send_subscribe(Subscribe(inbox_ids=["agent@agentinbox.space"]))
        # Обработка событий по мере поступления
        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())

Синхронное использование

from agentinbox import Agentinbox, Subscribe, Subscribed, MessageReceivedEvent
client = Agentinbox(api_key="YOUR_API_KEY")
with client.websockets.connect() as socket:
    # Подписка на inbox
    socket.send_subscribe(Subscribe(inbox_ids=["agent@agentinbox.space"]))
    # Обработка событий по мере поступления
    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}")

Обработчики событий

Можно использовать колбэки вместо итерации:

import asyncio
from agentinbox import AsyncAgentinbox, Subscribe, EventType
client = AsyncAgentinbox(api_key="YOUR_API_KEY")
async def main():
    async 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))
        # Подписка и прослушивание
        await socket.send_subscribe(Subscribe(inbox_ids=["agent@agentinbox.space"]))
        await socket.start_listening()
asyncio.run(main())

Для синхронного режима с обработчиками запустите слушатель в фоновом потоке:

import threading
from agentinbox import Agentinbox, Subscribe, EventType
client = Agentinbox(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@agentinbox.space"]))
    # Слушатель в фоне
    listener = threading.Thread(target=socket.start_listening, daemon=True)
    listener.start()
    listener.join()

TypeScript SDK

В TypeScript SDK клиент WebSocket с автоматическим переподключением.

Базовое использование

import { AgentinboxClient, Agentinbox } from "agentinbox";
const client = new AgentinboxClient({
  apiKey: process.env.AGENTINBOX_API_KEY,
});
async function main() {
  const socket = await client.websockets.connect();
  // Обработка событий
  socket.on("open", () => {
    console.log("Connected");
    // Подписка после открытия соединения
    socket.sendSubscribe({
      type: "subscribe",
      inboxIds: ["agent@agentinbox.space"],
    });
  });
  socket.on("message", (event: Agent Inbox.Subscribed | Agent Inbox.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

import { useEffect, useState } from "react";
import { AgentinboxClient, Agentinbox } from "agentinbox";
function useAgentinboxWebSocket(apiKey: string, inboxIds: string[]) {
  const [lastMessage, setLastMessage] = useState<Agent Inbox.MessageReceivedEvent | null>(null);
  const [isConnected, setIsConnected] = useState(false);
  useEffect(() => {
    const client = new AgentinboxClient({ 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 };
}

Параметры подписки

Можно фильтровать по inbox, pod или типу события:

Python:

from agentinbox import Subscribe
# Конкретные inbox
Subscribe(inbox_ids=["inbox1@agentinbox.space", "inbox2@agentinbox.space"])
# Pod
Subscribe(pod_ids=["pod-id-1", "pod-id-2"])
# Типы событий
Subscribe(
    inbox_ids=["agent@agentinbox.space"],
    event_types=["message.received", "message.sent"]
)

TypeScript:

// Конкретные inbox
socket.sendSubscribe({
  type: "subscribe",
  inboxIds: ["inbox1@agentinbox.space", "inbox2@agentinbox.space"],
});
// Pod
socket.sendSubscribe({
  type: "subscribe",
  podIds: ["pod-id-1", "pod-id-2"],
});
// Типы событий
socket.sendSubscribe({
  type: "subscribe",
  inboxIds: ["agent@agentinbox.space"],
  eventTypes: ["message.received", "message.sent"],
});

Типы событий

События соединения

Событие	Python	TypeScript	Описание
`subscribed`	`Subscribed`	`Agent Inbox.Subscribed`	Подписка подтверждена

События сообщений

Событие	Python	TypeScript	Описание
`message_received`	`MessageReceivedEvent`	`Agent Inbox.MessageReceivedEvent`	Новое входящее письмо
`message_sent`	`MessageSentEvent`	`Agent Inbox.MessageSentEvent`	Письмо отправлено
`message_delivered`	`MessageDeliveredEvent`	`Agent Inbox.MessageDeliveredEvent`	Письмо доставлено
`message_bounced`	`MessageBouncedEvent`	`Agent Inbox.MessageBouncedEvent`	Отскок
`message_complained`	`MessageComplainedEvent`	`Agent Inbox.MessageComplainedEvent`	Пометка как спам
`message_rejected`	`MessageRejectedEvent`	`Agent Inbox.MessageRejectedEvent`	Письмо отклонено

События доменов

Событие	Python	TypeScript	Описание
`domain_verified`	`DomainVerifiedEvent`	`Agent Inbox.DomainVerifiedEvent`	Верификация домена завершена

Поля сообщения

Объект event.message содержит:

Python	TypeScript	Описание
`inbox_id`	`inboxId`	Inbox, куда пришло письмо
`message_id`	`messageId`	Уникальный ID сообщения
`thread_id`	`threadId`	ID треда
`from_`	`from_`	Адрес отправителя
`to`	`to`	Список получателей
`subject`	`subject`	Тема
`text`	`text`	Текстовое тело
`html`	`html`	HTML-тело (если есть)
`attachments`	`attachments`	Вложения

Обработка ошибок

Python:

from agentinbox import AsyncAgentinbox, Subscribe, MessageReceivedEvent
from agentinbox.core.api_error import ApiError
client = AsyncAgentinbox(api_key="YOUR_API_KEY")
async def main():
    try:
        async with client.websockets.connect() as socket:
            await socket.send_subscribe(Subscribe(inbox_ids=["agent@agentinbox.space"]))
            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 { AgentinboxClient, Agentinbox, AgentinboxError  } from "agentinbox";
const client = new AgentinboxClient({
  apiKey: process.env.AGENTINBOX_API_KEY,
});
async function main() {
  try {
    const socket = await client.websockets.connect();
    socket.on("open", () => {
      socket.sendSubscribe({
        type: "subscribe",
        inboxIds: ["agent@agentinbox.space"],
      });
    });
    socket.on("message", (event: Agent Inbox.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 AgentinboxError) {
      console.error(`API error: ${err.statusCode} - ${err.message}`);
    } else {
      console.error("Connection error:", err);
    }
  }
}
main();

Скопировать в Cursor / Claude

"""
Agent Inbox 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, DomainVerifiedEvent
"""
from agentinbox import Agentinbox, Subscribe, Subscribed, MessageReceivedEvent
client = Agentinbox(api_key="YOUR_API_KEY")
with client.websockets.connect() as socket:
  socket.send_subscribe(Subscribe(inbox_ids=["agent@agentinbox.space"]))
  for event in socket:
    if isinstance(event, Subscribed): print(event.inbox_ids)
    elif isinstance(event, MessageReceivedEvent): print(event.message.subject)