systemprompt_api/services/gateway/protocol/outbound/
mod.rs

1//! Outbound protocol adapters: canonical model to upstream provider.
2//!
3//! The [`OutboundAdapter`] trait sends a [`CanonicalRequest`] to an upstream
4//! provider and yields an [`OutboundOutcome`] — a buffered response or a stream
5//! of canonical events. Adapters register themselves via
6//! [`OutboundAdapterRegistration`] (collected by `inventory`) so the upstream
7//! registry can resolve one by provider tag. Implementations cover Anthropic,
8//! `OpenAI` Chat Completions, and `OpenAI` Responses.
9//!
10//! Copyright (c) systemprompt.io — Business Source License 1.1.
11//! See <https://systemprompt.io> for licensing details.
12
13pub mod anthropic;
14pub mod gemini;
15pub mod openai_chat;
16pub mod openai_responses;
17
18use std::sync::Arc;
19
20use anyhow::Result;
21use async_trait::async_trait;
22use futures_util::stream::BoxStream;
23use systemprompt_models::profile::GatewayRoute;
24use systemprompt_models::services::ai::ModelLimits;
25use thiserror::Error;
26
27use super::canonical::CanonicalRequest;
28use super::canonical_response::{CanonicalEvent, CanonicalResponse};
29
30/// Upstream provider failure, carried inside the `anyhow::Error` an adapter
31/// returns so the route layer can recover the real HTTP status by downcast
32/// instead of flattening every failure to 502.
33#[derive(Debug, Error)]
34pub enum UpstreamError {
35    #[error("{provider} returned {status}: {message}")]
36    Status {
37        provider: String,
38        status: u16,
39        message: String,
40        body: bytes::Bytes,
41        retry_after: Option<String>,
42        request_id: Option<String>,
43    },
44    #[error("{provider} request failed: {source}")]
45    Transport {
46        provider: String,
47        #[source]
48        source: reqwest::Error,
49    },
50}
51
52impl UpstreamError {
53    pub async fn from_response(provider: &str, response: reqwest::Response) -> Self {
54        let status = response.status().as_u16();
55        let header = |name: &str| {
56            response
57                .headers()
58                .get(name)
59                .and_then(|v| v.to_str().ok())
60                .map(ToOwned::to_owned)
61        };
62        let retry_after = header("retry-after");
63        let request_id = header("request-id").or_else(|| header("x-request-id"));
64        let body = response.bytes().await.unwrap_or_default();
65        Self::Status {
66            provider: provider.to_owned(),
67            status,
68            message: extract_upstream_message(&String::from_utf8_lossy(&body)),
69            body,
70            retry_after,
71            request_id,
72        }
73    }
74}
75
76// Why: one process-wide client — a client per request would open a fresh
77// connection pool and TLS handshake on every gateway call.
78pub(in crate::services::gateway) fn http_client() -> &'static reqwest::Client {
79    static CLIENT: std::sync::OnceLock<reqwest::Client> = std::sync::OnceLock::new();
80    CLIENT.get_or_init(reqwest::Client::new)
81}
82
83pub(in crate::services::gateway) async fn send_checked(
84    provider: &str,
85    req: reqwest::RequestBuilder,
86) -> Result<reqwest::Response> {
87    let response = req.send().await.map_err(|e| {
88        anyhow::Error::new(UpstreamError::Transport {
89            provider: provider.to_owned(),
90            source: e,
91        })
92    })?;
93    if !response.status().is_success() {
94        return Err(anyhow::Error::new(
95            UpstreamError::from_response(provider, response).await,
96        ));
97    }
98    Ok(response)
99}
100
101pub fn extract_upstream_message(body: &str) -> String {
102    serde_json::from_str::<serde_json::Value>(body)
103        .ok()
104        .and_then(|v| v["error"]["message"].as_str().map(ToOwned::to_owned))
105        .unwrap_or_else(|| body.chars().take(500).collect())
106}
107
108#[derive(Debug)]
109pub struct OutboundCtx<'a> {
110    pub route: &'a GatewayRoute,
111    pub endpoint: &'a str,
112    pub api_key: &'a str,
113    pub request: &'a CanonicalRequest,
114    pub upstream_model: &'a str,
115    pub model_limits: Option<ModelLimits>,
116    pub forward_headers: &'a [(String, String)],
117    pub raw_body: Option<&'a bytes::Bytes>,
118}
119
120#[expect(
121    missing_debug_implementations,
122    reason = "variants hold streaming bodies that intentionally do not implement Debug"
123)]
124pub enum OutboundOutcome {
125    Buffered(Box<CanonicalResponse>),
126    Streaming(BoxStream<'static, Result<CanonicalEvent, String>>),
127    RawBuffered {
128        body: bytes::Bytes,
129        content_type: Option<String>,
130        canonical: Box<CanonicalResponse>,
131    },
132    RawStreaming {
133        content_type: Option<String>,
134        stream: BoxStream<'static, Result<bytes::Bytes, String>>,
135    },
136}
137
138/// The exact bytes an adapter will put on the wire.
139///
140/// `raw_lane` records that the bytes started as the caller's own; they are
141/// still normalised in place, so they are not byte-identical to what arrived.
142#[derive(Debug, Clone)]
143pub struct PreparedBody {
144    pub bytes: bytes::Bytes,
145    pub raw_lane: bool,
146}
147
148#[async_trait]
149pub trait OutboundAdapter: Send + Sync {
150    fn build_body(&self, ctx: &OutboundCtx<'_>) -> Result<PreparedBody>;
151
152    async fn send(&self, ctx: OutboundCtx<'_>, body: &PreparedBody) -> Result<OutboundOutcome>;
153}
154
155#[derive(Debug, Clone, Copy)]
156pub struct OutboundAdapterRegistration {
157    pub tag: &'static str,
158    pub factory: fn() -> Arc<dyn OutboundAdapter>,
159}
160
161inventory::collect!(OutboundAdapterRegistration);
systemprompt_api/services/gateway/protocol/outbound/mod.rs

systemprompt_api/services/gateway/protocol/outbound/
mod.rs
