agentmail/
realtime.rs

1//! Realtime inbox events over WebSocket (`websockets` feature).
2//!
3//! The API pushes the same events webhooks deliver (message lifecycle,
4//! domain verification, calendar activity) over a WebSocket, so agents
5//! without a public URL can receive mail without exposing an endpoint.
6//!
7//! Protocol (mirrors the official SDKs): connect to `{websocket}/v0` with
8//! the API key as the `api_key` query parameter and a bearer `Authorization`
9//! header, send one `{"type":"subscribe", ...}` frame, and read back a
10//! `subscribed` acknowledgement followed by typed event frames until the
11//! server closes the connection. [`RealtimeStream::next_event`] yields
12//! [`RealtimeEvent`]s; unknown event types surface as
13//! [`RealtimeEvent::Other`] rather than failing, so additions don't break
14//! readers. There is no automatic reconnection: on
15//! [`RealtimeEvent::Closed`], call [`Client::connect_realtime`] again.
16//!
17//! The websocket host is not derivable from the API host by scheme alone
18//! (`https://api.agentmail.to` -> `wss://ws.agentmail.to`, EU:
19//! `wss://ws.agentmail.eu`), so [`Client::connect_realtime`] derives it by
20//! swapping the `api.` host label for `ws.` and overriding is available via
21//! `AGENTMAIL_WEBSOCKET_URL` (with [`Client::from_env`]) or
22//! [`Client::with_websocket_url`].
23
24use crate::types::{CalendarAttendee, CalendarEvent, Domain, Message, Recurrence, Thread};
25use crate::{Client, Error};
26use futures_util::{SinkExt, StreamExt};
27use serde::{Deserialize, Serialize};
28use tokio_tungstenite::tungstenite::Message as WsFrame;
29use tokio_tungstenite::tungstenite::client::IntoClientRequest;
30
31/// The production realtime host. See the module docs for how the URL is
32/// derived and overridden.
33pub const DEFAULT_WEBSOCKET_URL: &str = "wss://ws.agentmail.to";
34
35/// A subscribe request: which events, and for which inboxes or pods. Empty
36/// filters mean "everything the key can see".
37#[derive(Clone, Debug, Default, Serialize)]
38pub struct Subscribe {
39    /// Event types to receive (e.g. `message.received`); empty means all.
40    #[serde(skip_serializing_if = "Vec::is_empty")]
41    pub event_types: Vec<String>,
42    /// Limit events to these inboxes.
43    #[serde(skip_serializing_if = "Vec::is_empty")]
44    pub inbox_ids: Vec<String>,
45    /// Limit events to these pods.
46    #[serde(skip_serializing_if = "Vec::is_empty")]
47    pub pod_ids: Vec<String>,
48}
49
50#[derive(Serialize)]
51struct SubscribeFrame<'a> {
52    #[serde(rename = "type")]
53    kind: &'static str,
54    #[serde(flatten)]
55    sub: &'a Subscribe,
56}
57
58/// The server's acknowledgement of a [`Subscribe`] frame.
59#[derive(Clone, Debug, Deserialize)]
60pub struct Subscribed {
61    /// The active event filter (empty means all).
62    #[serde(default)]
63    pub event_types: Vec<String>,
64    /// The active inbox filter.
65    #[serde(default)]
66    pub inbox_ids: Vec<String>,
67    /// The active pod filter.
68    #[serde(default)]
69    pub pod_ids: Vec<String>,
70}
71
72/// A per-recipient delivery status, from bounce events.
73#[derive(Clone, Debug, Deserialize)]
74pub struct EventRecipient {
75    /// The recipient address.
76    #[serde(default)]
77    pub address: Option<String>,
78    /// Per-recipient status reported by the upstream provider.
79    #[serde(default)]
80    pub status: Option<String>,
81}
82
83/// A calendar event's state before an update, all fields optional.
84#[derive(Clone, Debug, Deserialize)]
85pub struct CalendarEventPrevious {
86    /// Title before the update.
87    #[serde(default)]
88    pub title: Option<String>,
89    /// Description before the update.
90    #[serde(default)]
91    pub description: Option<String>,
92    /// Location before the update.
93    #[serde(default)]
94    pub location: Option<String>,
95    /// Stored metadata before the update.
96    #[serde(default)]
97    pub metadata: Option<serde_json::Value>,
98    /// Confirmation state before the update.
99    #[serde(default)]
100    pub status: Option<String>,
101    /// Whether the event was all-day.
102    #[serde(default)]
103    pub all_day: Option<bool>,
104    /// Display start before the update.
105    #[serde(default)]
106    pub start: Option<String>,
107    /// Display end before the update.
108    #[serde(default)]
109    pub end: Option<String>,
110    /// IANA timezone before the update.
111    #[serde(default)]
112    pub timezone: Option<String>,
113    /// Duration interpretation before the update.
114    #[serde(default)]
115    pub duration_mode: Option<String>,
116    /// Recurrence rule before the update.
117    #[serde(default)]
118    pub recurrence: Option<Recurrence>,
119    /// Attendee list before the update.
120    #[serde(default)]
121    pub attendees: Vec<CalendarAttendee>,
122}
123
124/// One event off the wire, decoded.
125#[derive(Clone, Debug, Deserialize)]
126pub struct MessageReceivedEvent {
127    /// `message.received`, or `.spam` / `.blocked` / `.unauthenticated` for
128    /// mail that was filtered.
129    #[serde(default)]
130    pub event_type: Option<String>,
131    /// Server-issued event id.
132    #[serde(default)]
133    pub event_id: Option<String>,
134    /// The received message.
135    pub message: Message,
136    /// The thread the message opened.
137    pub thread: Thread,
138}
139
140macro_rules! envelope_event {
141    ($name:ident { $payload:ident : $payload_ty:ty $(, $extra:ident : $extra_ty:ty)* }) => {
142/// A decoded event frame; see the module docs for the protocol.
143#[derive(Clone, Debug, Deserialize)]
144pub struct $name {
145            /// The event type on the wire (e.g. `message.sent`).
146            #[serde(default)]
147            pub event_type: Option<String>,
148            /// Server-issued event id.
149            #[serde(default)]
150            pub event_id: Option<String>,
151            /// Inbox the event concerns, when scoped to one.
152            #[serde(default)]
153            pub inbox_id: Option<String>,
154            /// The event's payload.
155            pub $payload: $payload_ty,
156            $(/// Extra field carried by this event kind.
157            #[serde(default)]
158            pub $extra: $extra_ty,)*
159        }
160    };
161}
162
163/// Send/delivery details, from `message.sent` and `message.delivered`.
164#[derive(Clone, Debug, Deserialize)]
165pub struct DispatchPayload {
166    /// Inbox that sent (or owns) the message.
167    #[serde(default)]
168    pub inbox_id: Option<String>,
169    /// Thread the message is filed under.
170    #[serde(default)]
171    pub thread_id: Option<String>,
172    /// The message.
173    #[serde(default)]
174    pub message_id: Option<String>,
175    /// When the send/delivery happened (RFC 3339).
176    #[serde(default)]
177    pub timestamp: Option<String>,
178    /// Addresses the message reached.
179    #[serde(default)]
180    pub recipients: Vec<String>,
181}
182
183envelope_event!(MessageSentEvent {
184    send: DispatchPayload
185});
186envelope_event!(MessageDeliveredEvent {
187    delivery: DispatchPayload
188});
189envelope_event!(MessageOpenedEvent { open: OpenPayload });
190envelope_event!(MessageBouncedEvent {
191    bounce: BouncePayload
192});
193envelope_event!(MessageComplainedEvent {
194    complaint: ComplaintPayload
195});
196envelope_event!(MessageRejectedEvent {
197    reject: RejectPayload
198});
199envelope_event!(DomainVerifiedEvent { domain: Domain });
200envelope_event!(CalendarEventCreatedEvent {
201    calendar_event: CalendarEvent
202});
203envelope_event!(CalendarEventDeletedEvent {
204    calendar_event: CalendarEvent
205});
206envelope_event!(CalendarEventRespondedEvent {
207    calendar_event: CalendarEvent
208});
209envelope_event!(CalendarEventUpdatedEvent { calendar_event: CalendarEvent, previous: Option<CalendarEventPrevious> });
210envelope_event!(CalendarEventStartingEvent { calendar_event: CalendarEvent, scheduled_at: Option<String> });
211envelope_event!(CalendarEventEndingEvent { calendar_event: CalendarEvent, scheduled_at: Option<String> });
212
213/// Recipient-open details, from `message.opened`.
214#[derive(Clone, Debug, Deserialize)]
215pub struct OpenPayload {
216    /// Inbox the message was sent from.
217    #[serde(default)]
218    pub inbox_id: Option<String>,
219    /// Thread the message is filed under.
220    #[serde(default)]
221    pub thread_id: Option<String>,
222    /// The opened message.
223    #[serde(default)]
224    pub message_id: Option<String>,
225    /// When the open was recorded (RFC 3339).
226    #[serde(default)]
227    pub timestamp: Option<String>,
228}
229
230/// Bounce details, from `message.bounced`.
231#[derive(Clone, Debug, Deserialize)]
232pub struct BouncePayload {
233    /// Inbox that sent the message.
234    #[serde(default)]
235    pub inbox_id: Option<String>,
236    /// Thread the message is filed under.
237    #[serde(default)]
238    pub thread_id: Option<String>,
239    /// The bounced message.
240    #[serde(default)]
241    pub message_id: Option<String>,
242    /// When the bounce was recorded (RFC 3339).
243    #[serde(default)]
244    pub timestamp: Option<String>,
245    /// Bounce class reported by the provider.
246    #[serde(default)]
247    pub sub_type: Option<String>,
248    /// Per-recipient bounce statuses.
249    #[serde(default)]
250    pub recipients: Vec<EventRecipient>,
251}
252
253/// Complaint details, from `message.complained`.
254#[derive(Clone, Debug, Deserialize)]
255pub struct ComplaintPayload {
256    /// Inbox that sent the message.
257    #[serde(default)]
258    pub inbox_id: Option<String>,
259    /// Thread the message is filed under.
260    #[serde(default)]
261    pub thread_id: Option<String>,
262    /// The complained-about message.
263    #[serde(default)]
264    pub message_id: Option<String>,
265    /// When the complaint was recorded (RFC 3339).
266    #[serde(default)]
267    pub timestamp: Option<String>,
268    /// Complaint class reported by the provider.
269    #[serde(default)]
270    pub sub_type: Option<String>,
271    /// Addresses that complained.
272    #[serde(default)]
273    pub recipients: Vec<String>,
274}
275
276/// Rejection details, from `message.rejected`.
277#[derive(Clone, Debug, Deserialize)]
278pub struct RejectPayload {
279    /// Inbox that tried to send the message.
280    #[serde(default)]
281    pub inbox_id: Option<String>,
282    /// Thread the message would have been filed under.
283    #[serde(default)]
284    pub thread_id: Option<String>,
285    /// The rejected message.
286    #[serde(default)]
287    pub message_id: Option<String>,
288    /// When the rejection happened (RFC 3339).
289    #[serde(default)]
290    pub timestamp: Option<String>,
291    /// Why the message was rejected.
292    #[serde(default)]
293    pub reason: Option<String>,
294}
295
296/// One frame from the stream.
297#[derive(Clone, Debug)]
298pub enum RealtimeEvent {
299    /// The server accepted our [`Subscribe`] filter.
300    Subscribed(Subscribed),
301    /// Mail arrived (also fires for `.spam` / `.blocked` /
302    /// `.unauthenticated` deliveries; see
303    /// [`MessageReceivedEvent::event_type`]).
304    MessageReceived(MessageReceivedEvent),
305    /// The API accepted an outgoing message for delivery.
306    MessageSent(MessageSentEvent),
307    /// An upstream provider accepted the message.
308    MessageDelivered(MessageDeliveredEvent),
309    /// A recipient's client rendered the message's tracking pixel.
310    MessageOpened(MessageOpenedEvent),
311    /// A recipient's provider bounced the message.
312    MessageBounced(MessageBouncedEvent),
313    /// A recipient filed a spam complaint.
314    MessageComplained(MessageComplainedEvent),
315    /// The API refused to deliver an outgoing message.
316    MessageRejected(MessageRejectedEvent),
317    /// A domain finished verifying.
318    DomainVerified(DomainVerifiedEvent),
319    /// A calendar event was created.
320    CalendarEventCreated(CalendarEventCreatedEvent),
321    /// A calendar event was updated (`previous` holds the old state).
322    CalendarEventUpdated(CalendarEventUpdatedEvent),
323    /// A calendar event was deleted.
324    CalendarEventDeleted(CalendarEventDeletedEvent),
325    /// A calendar event is about to start.
326    CalendarEventStarting(CalendarEventStartingEvent),
327    /// A calendar event is about to end.
328    CalendarEventEnding(CalendarEventEndingEvent),
329    /// An attendee replied to an invite.
330    CalendarEventResponded(CalendarEventRespondedEvent),
331    /// A frame this client version doesn't know; the raw JSON is preserved.
332    Other(serde_json::Value),
333    /// A binary frame (passed through undecoded).
334    Binary(Vec<u8>),
335    /// The server closed the connection; reconnect with
336    /// [`Client::connect_realtime`].
337    Closed,
338}
339
340/// An open realtime stream. Send one [`Subscribe`] (done for you by
341/// [`Client::connect_realtime`]), then poll [`RealtimeStream::next_event`].
342pub struct RealtimeStream {
343    ws: tokio_tungstenite::WebSocketStream<
344        tokio_tungstenite::MaybeTlsStream<tokio::net::TcpStream>,
345    >,
346}
347
348fn realtime_error(e: tokio_tungstenite::tungstenite::Error) -> Error {
349    Error::Realtime(e.to_string())
350}
351
352impl RealtimeStream {
353    /// Send a new subscribe filter on the open connection.
354    pub async fn subscribe(&mut self, subscribe: &Subscribe) -> Result<(), Error> {
355        let frame = SubscribeFrame {
356            kind: "subscribe",
357            sub: subscribe,
358        };
359        let text = serde_json::to_string(&frame).map_err(|e| Error::Decode {
360            reason: e.to_string(),
361            body: String::new(),
362        })?;
363        self.ws
364            .send(WsFrame::Text(text.into()))
365            .await
366            .map_err(realtime_error)
367    }
368
369    /// Await the next frame. Control frames (ping/pong) are answered
370    /// transparently; a server close yields [`RealtimeEvent::Closed`]
371    /// (and then keeps yielding it).
372    pub async fn next_event(&mut self) -> Result<RealtimeEvent, Error> {
373        loop {
374            match self.ws.next().await {
375                Some(Ok(WsFrame::Text(text))) => {
376                    let value: serde_json::Value =
377                        serde_json::from_str(&text).map_err(|e| Error::Decode {
378                            reason: e.to_string(),
379                            body: text.to_string(),
380                        })?;
381                    return Ok(decode_event(value));
382                }
383                Some(Ok(WsFrame::Binary(bytes))) => {
384                    return Ok(RealtimeEvent::Binary(bytes.to_vec()));
385                }
386                // Pings are answered by tungstenite; ignore the rest.
387                Some(Ok(WsFrame::Ping(_) | WsFrame::Pong(_))) => continue,
388                Some(Ok(WsFrame::Close(_))) | None => return Ok(RealtimeEvent::Closed),
389                Some(Ok(WsFrame::Frame(_))) => continue,
390                Some(Err(e)) => return Err(realtime_error(e)),
391            }
392        }
393    }
394}
395
396fn decode_event(value: serde_json::Value) -> RealtimeEvent {
397    match value.get("type").and_then(|t| t.as_str()) {
398        Some("subscribed") => serde_json::from_value(value.clone())
399            .map(|s: Subscribed| RealtimeEvent::Subscribed(s))
400            .unwrap_or_else(|_| RealtimeEvent::Other(value)),
401        Some("event") => decode_domain_event(value),
402        _ => RealtimeEvent::Other(value),
403    }
404}
405
406fn decode_domain_event(value: serde_json::Value) -> RealtimeEvent {
407    use RealtimeEvent as E;
408    let event_type = value
409        .get("event_type")
410        .and_then(|t| t.as_str())
411        .unwrap_or("");
412    let decoded = match event_type {
413        "message.received"
414        | "message.received.spam"
415        | "message.received.blocked"
416        | "message.received.unauthenticated" => {
417            serde_json::from_value(value.clone()).map(E::MessageReceived)
418        }
419        "message.sent" => serde_json::from_value(value.clone()).map(E::MessageSent),
420        "message.delivered" => serde_json::from_value(value.clone()).map(E::MessageDelivered),
421        "message.opened" => serde_json::from_value(value.clone()).map(E::MessageOpened),
422        "message.bounced" => serde_json::from_value(value.clone()).map(E::MessageBounced),
423        "message.complained" => serde_json::from_value(value.clone()).map(E::MessageComplained),
424        "message.rejected" => serde_json::from_value(value.clone()).map(E::MessageRejected),
425        "domain.verified" => serde_json::from_value(value.clone()).map(E::DomainVerified),
426        "calendar.event.created" => {
427            serde_json::from_value(value.clone()).map(E::CalendarEventCreated)
428        }
429        "calendar.event.updated" => {
430            serde_json::from_value(value.clone()).map(E::CalendarEventUpdated)
431        }
432        "calendar.event.deleted" => {
433            serde_json::from_value(value.clone()).map(E::CalendarEventDeleted)
434        }
435        "calendar.event.starting" => {
436            serde_json::from_value(value.clone()).map(E::CalendarEventStarting)
437        }
438        "calendar.event.ending" => {
439            serde_json::from_value(value.clone()).map(E::CalendarEventEnding)
440        }
441        "calendar.event.responded" => {
442            serde_json::from_value(value.clone()).map(E::CalendarEventResponded)
443        }
444        _ => return RealtimeEvent::Other(value),
445    };
446    decoded.unwrap_or(RealtimeEvent::Other(value))
447}
448
449fn derive_websocket_url(base_url: &str) -> String {
450    let (scheme, rest) = if let Some(rest) = base_url.strip_prefix("https://") {
451        ("wss://", rest)
452    } else if let Some(rest) = base_url.strip_prefix("http://") {
453        ("ws://", rest)
454    } else {
455        return DEFAULT_WEBSOCKET_URL.to_string();
456    };
457    let host = rest.split('/').next().unwrap_or(rest);
458    // api.agentmail.to -> ws.agentmail.to, x402.api.agentmail.to ->
459    // x402.ws.agentmail.to (the `api` label swaps wherever it sits);
460    // hosts without one (mock servers) pass through.
461    let ws_host = host
462        .split('.')
463        .map(|label| if label == "api" { "ws" } else { label })
464        .collect::<Vec<_>>()
465        .join(".");
466    format!("{scheme}{ws_host}")
467}
468
469impl Client {
470    /// The websocket URL this client will dial, after derivation and any
471    /// [`Client::with_websocket_url`] override.
472    #[cfg(feature = "websockets")]
473    pub fn websocket_url(&self) -> String {
474        self.ws_url_override
475            .clone()
476            .unwrap_or_else(|| derive_websocket_url(&self.base_url))
477    }
478
479    /// Override the derived websocket host (e.g. for a proxy or a mock
480    /// server on a different port).
481    ///
482    /// Requires the `websockets` feature (off by default).
483    #[cfg(feature = "websockets")]
484    #[cfg_attr(docsrs, doc(cfg(feature = "websockets")))]
485    pub fn with_websocket_url(mut self, url: impl Into<String>) -> Self {
486        self.ws_url_override = Some(url.into().trim_end_matches('/').to_string());
487        self
488    }
489
490    /// Open the realtime event stream and send `subscribe`.
491    ///
492    /// The URL is [`Client::websocket_url`] + `/v0`, with the API key both in
493    /// the `api_key` query parameter and the `Authorization` header (the
494    /// official SDKs do the same). No automatic reconnection: on
495    /// [`RealtimeEvent::Closed`], call this again.
496    ///
497    /// Requires the `websockets` feature (off by default).
498    #[cfg(feature = "websockets")]
499    #[cfg_attr(docsrs, doc(cfg(feature = "websockets")))]
500    pub async fn connect_realtime(&self, subscribe: &Subscribe) -> Result<RealtimeStream, Error> {
501        let url = format!(
502            "{}/v0?api_key={}",
503            self.websocket_url(),
504            crate::util::urlish(&self.api_key)
505        );
506        let mut request = url.as_str().into_client_request().map_err(realtime_error)?;
507        request.headers_mut().insert(
508            "Authorization",
509            format!("Bearer {}", self.api_key)
510                .parse()
511                .map_err(|e| Error::Realtime(format!("invalid authorization header: {e}")))?,
512        );
513        let (ws, _response) = tokio_tungstenite::connect_async(request)
514            .await
515            .map_err(realtime_error)?;
516        let mut stream = RealtimeStream { ws };
517        stream.subscribe(subscribe).await?;
518        Ok(stream)
519    }
520}
521
522#[cfg(test)]
523mod tests {
524    use super::*;
525
526    #[test]
527    fn url_derivation_matches_upstream_environments() {
528        // From the official SDK's environment table.
529        assert_eq!(
530            derive_websocket_url("https://api.agentmail.to"),
531            "wss://ws.agentmail.to"
532        );
533        assert_eq!(
534            derive_websocket_url("https://api.agentmail.eu"),
535            "wss://ws.agentmail.eu"
536        );
537        assert_eq!(
538            derive_websocket_url("https://x402.api.agentmail.to"),
539            "wss://x402.ws.agentmail.to"
540        );
541        // Mock servers keep their host.
542        assert_eq!(
543            derive_websocket_url("http://127.0.0.1:8080"),
544            "ws://127.0.0.1:8080"
545        );
546    }
547
548    #[test]
549    fn subscribe_frame_has_const_type_and_skips_empty() {
550        let frame = serde_json::to_value(SubscribeFrame {
551            kind: "subscribe",
552            sub: &Subscribe {
553                inbox_ids: vec!["ib_1".into()],
554                ..Default::default()
555            },
556        })
557        .unwrap();
558        assert_eq!(
559            frame,
560            serde_json::json!({"type": "subscribe", "inbox_ids": ["ib_1"]})
561        );
562    }
563
564    #[test]
565    fn unknown_event_types_decode_as_other() {
566        let event = decode_event(serde_json::json!({
567            "type": "event", "event_type": "carrier.pigeon", "data": 1
568        }));
569        assert!(matches!(event, RealtimeEvent::Other(_)));
570    }
571}
agentmail/realtime.rs

agentmail/
realtime.rs
