agentic_server/handler/
common.rs

1use axum::body::Body;
2use axum::http::HeaderMap;
3use axum::response::Response;
4use bytes::Bytes;
5use futures::StreamExt;
6use http::StatusCode;
7use tracing::warn;
8
9use agentic_core::executor::{BoxStream, ExecutorError};
10use agentic_core::proxy::{ProxyBody, ProxyResponse, error_response};
11use agentic_core::types::request_response::RequestPayload;
12
13pub(super) const MAX_BODY_SIZE: usize = 10 * 1024 * 1024;
14
15/// # Panics
16/// Panics if the response builder produces an invalid response (unreachable in practice).
17pub fn convert_response(resp: ProxyResponse) -> Response {
18    let mut builder = Response::builder().status(resp.status);
19    for (name, value) in &resp.headers {
20        builder = builder.header(name, value);
21    }
22    match resp.body {
23        ProxyBody::Full(bytes) => builder.body(Body::from(bytes)).expect("valid response"),
24        ProxyBody::Stream(stream) => builder.body(Body::from_stream(stream)).expect("valid response"),
25    }
26}
27
28/// # Panics
29/// Panics if the response builder produces an invalid response (unreachable in practice).
30pub fn executor_error_response(err: ExecutorError) -> Response {
31    let status = err.http_status();
32    if !matches!(err, ExecutorError::LLMRequest { .. }) {
33        warn!("executor error ({status}): {err}");
34    }
35    Response::builder()
36        .status(status)
37        .header("Content-Type", "application/json")
38        .body(Body::from(err.into_response_body()))
39        .expect("valid error response")
40}
41
42pub(super) async fn read_bytes(body: Body) -> Result<Bytes, Response> {
43    axum::body::to_bytes(body, MAX_BODY_SIZE).await.map_err(|_| {
44        convert_response(error_response(
45            StatusCode::PAYLOAD_TOO_LARGE,
46            "body_too_large",
47            "request body too large",
48        ))
49    })
50}
51
52pub(super) async fn read_and_parse(body: Body) -> Result<(Bytes, RequestPayload), Response> {
53    let bytes = read_bytes(body).await?;
54    let payload = serde_json::from_slice::<RequestPayload>(&bytes)
55        .map_err(|e| executor_error_response(ExecutorError::from(e)))?;
56    Ok((bytes, payload))
57}
58
59pub(super) fn extract_store(bytes: &[u8]) -> bool {
60    serde_json::from_slice::<serde_json::Value>(bytes)
61        .ok()
62        .and_then(|j| j.get("store").and_then(serde_json::Value::as_bool))
63        .unwrap_or(true)
64}
65
66pub(super) fn extract_bearer(headers: &HeaderMap, config_key: Option<&str>) -> Option<String> {
67    headers
68        .get("authorization")
69        .and_then(|v| v.to_str().ok())
70        .and_then(|v| v.strip_prefix("Bearer "))
71        .filter(|s| !s.is_empty())
72        .map(str::to_string)
73        .or_else(|| config_key.filter(|s| !s.is_empty()).map(str::to_string))
74}
75
76pub(super) fn sse_response(stream: BoxStream) -> Response {
77    let byte_stream = stream.map(|line| Ok::<Bytes, std::convert::Infallible>(Bytes::from(line)));
78    Response::builder()
79        .status(StatusCode::OK)
80        .header("Content-Type", "text/event-stream; charset=utf-8")
81        .header("Cache-Control", "no-cache")
82        .header("X-Accel-Buffering", "no")
83        .body(Body::from_stream(byte_stream))
84        .expect("valid SSE response")
85}
agentic_server/handler/common.rs

agentic_server/handler/
common.rs
