systemprompt_agent/services/a2a_server/streaming/
webhook_client.rs

1//! Webhook delivery for AGUI and A2A streaming events.
2//!
3//! The free [`broadcast_agui_event`] / [`broadcast_a2a_event`] entry points
4//! dispatch through a globally installed [`WebhookBroadcaster`]. Production
5//! installs nothing and gets the default [`HttpWebhookBroadcaster`]; tests
6//! call [`install_for_test`] with a recording fake. The indirection lets
7//! the deep callers in `event_loop`, `complete_handler`, `message_handler`,
8//! and `skills` stay as free-function calls while the harness still swaps
9//! the network for a deterministic spy.
10//!
11//! Copyright (c) systemprompt.io — Business Source License 1.1.
12//! See <https://systemprompt.io> for licensing details.
13
14use std::sync::{Arc, OnceLock};
15
16use async_trait::async_trait;
17use reqwest::Client;
18use serde::Serialize;
19use systemprompt_identifiers::UserId;
20use systemprompt_models::{A2AEvent, AgUiEvent, Config};
21
22#[derive(Debug, thiserror::Error)]
23pub enum WebhookError {
24    #[error("HTTP request failed: {0}")]
25    Request(#[from] reqwest::Error),
26    #[error("Webhook returned error status {status}: {message}")]
27    StatusError { status: u16, message: String },
28}
29
30/// `#[async_trait]` is required: the active broadcaster is stored and swapped
31/// as an `Arc<dyn WebhookBroadcaster>` (see `GLOBAL_BROADCASTER` below), so the
32/// trait must be `dyn`-compatible.
33#[async_trait]
34pub trait WebhookBroadcaster: Send + Sync + std::fmt::Debug {
35    async fn broadcast_agui(
36        &self,
37        user_id: &UserId,
38        event: AgUiEvent,
39        auth_token: &str,
40    ) -> Result<usize, WebhookError>;
41
42    async fn broadcast_a2a(
43        &self,
44        user_id: &UserId,
45        event: A2AEvent,
46        auth_token: &str,
47    ) -> Result<usize, WebhookError>;
48}
49
50#[derive(Serialize)]
51struct AgUiWebhookPayload {
52    #[serde(flatten)]
53    event: AgUiEvent,
54    user_id: UserId,
55}
56
57#[derive(Serialize)]
58struct A2AWebhookPayload {
59    #[serde(flatten)]
60    event: A2AEvent,
61    user_id: UserId,
62}
63
64fn get_api_url() -> String {
65    Config::get().map_or_else(
66        |_| "http://localhost:3000".to_owned(),
67        |c| c.api_internal_url.clone(),
68    )
69}
70
71#[derive(Debug, Default, Clone, Copy)]
72pub struct HttpWebhookBroadcaster;
73
74#[async_trait]
75impl WebhookBroadcaster for HttpWebhookBroadcaster {
76    async fn broadcast_agui(
77        &self,
78        user_id: &UserId,
79        event: AgUiEvent,
80        auth_token: &str,
81    ) -> Result<usize, WebhookError> {
82        let url = format!("{}/api/v1/webhook/agui", get_api_url());
83        let event_type = event.event_type();
84        if auth_token.is_empty() {
85            tracing::warn!(
86                event_type = ?event_type,
87                user_id = %user_id,
88                "AGUI broadcast with empty auth_token"
89            );
90        }
91        let payload = AgUiWebhookPayload {
92            event,
93            user_id: user_id.clone(),
94        };
95        post_and_decode(&url, auth_token, &payload, "AGUI").await
96    }
97
98    async fn broadcast_a2a(
99        &self,
100        user_id: &UserId,
101        event: A2AEvent,
102        auth_token: &str,
103    ) -> Result<usize, WebhookError> {
104        let url = format!("{}/api/v1/webhook/a2a", get_api_url());
105        let payload = A2AWebhookPayload {
106            event,
107            user_id: user_id.clone(),
108        };
109        post_and_decode(&url, auth_token, &payload, "A2A").await
110    }
111}
112
113#[derive(serde::Deserialize)]
114struct WebhookResponse {
115    connection_count: usize,
116}
117
118async fn post_and_decode<T: Serialize + Sync + ?Sized>(
119    url: &str,
120    auth_token: &str,
121    payload: &T,
122    kind: &str,
123) -> Result<usize, WebhookError> {
124    let client = Client::new();
125    let response = client
126        .post(url)
127        .header("Authorization", format!("Bearer {auth_token}"))
128        .header("Content-Type", "application/json")
129        .json(payload)
130        .send()
131        .await;
132    match response {
133        Ok(resp) if resp.status().is_success() => match resp.json::<WebhookResponse>().await {
134            Ok(r) => {
135                tracing::debug!(
136                    kind = kind,
137                    connection_count = r.connection_count,
138                    "broadcasted"
139                );
140                Ok(r.connection_count)
141            },
142            Err(e) => {
143                tracing::error!(kind = kind, error = %e, "response parse error");
144                Err(WebhookError::Request(e))
145            },
146        },
147        Ok(resp) => {
148            let status = resp.status().as_u16();
149            let message = resp
150                .text()
151                .await
152                .unwrap_or_else(|e| format!("<error reading response: {e}>"));
153            tracing::error!(kind = kind, status, message = %message, "event failed");
154            Err(WebhookError::StatusError { status, message })
155        },
156        Err(e) => {
157            tracing::error!(kind = kind, error = %e, "request error");
158            Err(WebhookError::Request(e))
159        },
160    }
161}
162
163static GLOBAL_BROADCASTER: OnceLock<Arc<dyn WebhookBroadcaster>> = OnceLock::new();
164
165fn default_broadcaster() -> Arc<dyn WebhookBroadcaster> {
166    Arc::new(HttpWebhookBroadcaster)
167}
168
169fn active_broadcaster() -> Arc<dyn WebhookBroadcaster> {
170    Arc::clone(GLOBAL_BROADCASTER.get_or_init(default_broadcaster))
171}
172
173pub fn install_for_test(broadcaster: Arc<dyn WebhookBroadcaster>) {
174    drop(GLOBAL_BROADCASTER.set(broadcaster));
175}
176
177pub async fn broadcast_agui_event(
178    user_id: &UserId,
179    event: AgUiEvent,
180    auth_token: &str,
181) -> Result<usize, WebhookError> {
182    active_broadcaster()
183        .broadcast_agui(user_id, event, auth_token)
184        .await
185}
186
187pub async fn broadcast_a2a_event(
188    user_id: &UserId,
189    event: A2AEvent,
190    auth_token: &str,
191) -> Result<usize, WebhookError> {
192    active_broadcaster()
193        .broadcast_a2a(user_id, event, auth_token)
194        .await
195}
196
197#[derive(Clone, Debug)]
198pub struct WebhookContext {
199    user_id: UserId,
200    auth_token: String,
201}
202
203impl WebhookContext {
204    pub fn new(user_id: UserId, auth_token: impl Into<String>) -> Self {
205        Self {
206            user_id,
207            auth_token: auth_token.into(),
208        }
209    }
210
211    pub const fn user_id(&self) -> &UserId {
212        &self.user_id
213    }
214
215    pub async fn broadcast_agui(&self, event: AgUiEvent) -> Result<usize, WebhookError> {
216        broadcast_agui_event(&self.user_id, event, &self.auth_token).await
217    }
218
219    pub async fn broadcast_a2a(&self, event: A2AEvent) -> Result<usize, WebhookError> {
220        broadcast_a2a_event(&self.user_id, event, &self.auth_token).await
221    }
222}
systemprompt_agent/services/a2a_server/streaming/webhook_client.rs

systemprompt_agent/services/a2a_server/streaming/
webhook_client.rs
