headless_lms_mcp/http/
mod.rs

1//! The actix adapter for `/api/v0/mcp`: maps HTTP requests to [`McpServer::handle`] and its
2//! responses back, streaming progress as server-sent events when the client accepts them.
3
4pub mod auth;
5
6use actix_web::{
7    HttpRequest, HttpResponse,
8    http::{StatusCode, header},
9    web::{self, Bytes, ServiceConfig},
10};
11use futures::stream::{self, StreamExt};
12use headless_lms_models::library::oauth::mcp::McpPrincipal;
13use rmcp::model::RequestStateCodec;
14use secrecy::ExposeSecret;
15use serde_json::Value;
16use tokio::sync::{mpsc, oneshot};
17
18use crate::protocol::dispatch::{self, McpServer, ServerConfig};
19use crate::protocol::headers::RequestHeaders;
20use crate::state::McpState;
21
22/// Room for `upload_file`'s largest file (25 MB) base64-encoded.
23const MAX_REQUEST_BYTES: usize = 36 * 1024 * 1024;
24
25/// The MCP server with the state its tools run against, shared by every worker.
26pub struct McpEndpoint {
27    pub server: McpServer<McpState>,
28    pub state: McpState,
29}
30
31impl McpEndpoint {
32    /// The endpoint with every catalog tool.
33    pub fn new(state: McpState) -> Self {
34        let app_conf = &state.app_conf;
35        let server_config = ServerConfig {
36            server_version: env!("CARGO_PKG_VERSION").to_string(),
37            resource_metadata_url: auth::protected_resource_metadata_url(app_conf),
38            request_state_codec: RequestStateCodec::try_new(
39                request_state_key(app_conf.jwt_password.expose_secret()).to_vec(),
40            )
41            .expect("a 32-byte key is long enough"),
42        };
43        Self {
44            server: McpServer::new(crate::handlers::registry(), server_config),
45            state,
46        }
47    }
48}
49
50/// A key for sealing `requestState`, derived from the JWT secret so that it can't be used to forge
51/// the secret's other signatures.
52fn request_state_key(secret: &str) -> [u8; 32] {
53    blake3::derive_key("headless-lms mcp request state v1", secret.as_bytes())
54}
55
56/// Adds the endpoint's route to a scope that is wrapped in [`auth::authenticate`] and has an
57/// [`McpEndpoint`] in its app data.
58pub fn configure(cfg: &mut ServiceConfig) {
59    cfg.app_data(web::PayloadConfig::new(MAX_REQUEST_BYTES))
60        .route("", web::route().to(handle))
61        .route("/", web::route().to(handle));
62}
63
64fn into_actix(response: dispatch::HttpResponse) -> HttpResponse {
65    let mut builder = HttpResponse::build(
66        StatusCode::from_u16(response.status).unwrap_or(StatusCode::INTERNAL_SERVER_ERROR),
67    );
68    for (name, value) in &response.headers {
69        builder.insert_header((name.as_str(), value.as_str()));
70    }
71    match response.body {
72        Some(body) => builder.body(body.to_string()),
73        None => builder.finish(),
74    }
75}
76
77fn accepts_event_stream(req: &HttpRequest) -> bool {
78    req.headers()
79        .get_all(header::ACCEPT)
80        .filter_map(|v| v.to_str().ok())
81        .any(|v| v.contains("text/event-stream"))
82}
83
84fn has_progress_token(message: &serde_json::Result<Value>) -> bool {
85    message
86        .as_ref()
87        .ok()
88        .and_then(|message| message.pointer("/params/_meta/progressToken"))
89        .is_some_and(|token| token.is_string() || token.is_number())
90}
91
92fn sse_event(message: &Value) -> Bytes {
93    Bytes::from(format!("event: message\ndata: {message}\n\n"))
94}
95
96/// The plain JSON answer of a call that finished before streaming started.
97fn finished(response: Result<dispatch::HttpResponse, oneshot::error::RecvError>) -> HttpResponse {
98    match response {
99        Ok(response) => into_actix(response),
100        Err(_) => HttpResponse::InternalServerError().finish(),
101    }
102}
103
104async fn handle(
105    req: HttpRequest,
106    body: Bytes,
107    endpoint: web::Data<McpEndpoint>,
108    principal: web::ReqData<McpPrincipal>,
109) -> HttpResponse {
110    let principal = principal.into_inner();
111    let headers = RequestHeaders::new(
112        req.headers()
113            .iter()
114            .filter_map(|(name, value)| Some((name.as_str(), value.to_str().ok()?.to_string()))),
115    );
116    let request = dispatch::HttpRequest {
117        method: req.method().to_string(),
118        headers,
119        body: serde_json::from_slice::<Value>(&body),
120    };
121    if !(accepts_event_stream(&req) && has_progress_token(&request.body)) {
122        let response = endpoint
123            .server
124            .handle(&endpoint.state, &principal, request, None)
125            .await;
126        return into_actix(response);
127    }
128
129    let (progress_sender, mut progress) = mpsc::unbounded_channel();
130    let (done_sender, mut done) = oneshot::channel();
131    let task_endpoint = endpoint.clone();
132    actix_web::rt::spawn(async move {
133        let response = task_endpoint
134            .server
135            .handle(
136                &task_endpoint.state,
137                &principal,
138                request,
139                Some(progress_sender),
140            )
141            .await;
142        let _ = done_sender.send(response);
143    });
144    // Plain JSON unless progress comes first, so errors keep their HTTP status.
145    let first = tokio::select! {
146        biased;
147        response = &mut done => return finished(response),
148        first = progress.recv() => first,
149    };
150    let Some(first) = first else {
151        return finished(done.await);
152    };
153    let notifications = stream::once(async move { first })
154        .chain(stream::unfold(progress, |mut progress| async move {
155            progress.recv().await.map(|n| (n, progress))
156        }))
157        .map(|n| sse_event(&n));
158    let last = stream::once(async move {
159        match done.await {
160            Ok(response) => sse_event(&response.body.unwrap_or(Value::Null)),
161            Err(_) => Bytes::new(),
162        }
163    });
164    HttpResponse::Ok()
165        .insert_header((header::CONTENT_TYPE, "text/event-stream"))
166        .insert_header((header::CACHE_CONTROL, "no-cache"))
167        .streaming(notifications.chain(last).map(Ok::<_, actix_web::Error>))
168}
169
170#[cfg(test)]
171mod tests {
172    use super::*;
173
174    #[test]
175    fn progress_token_detection() {
176        let parse = |body: &[u8]| serde_json::from_slice::<Value>(body);
177        assert!(has_progress_token(&parse(
178            br#"{"params":{"_meta":{"progressToken":"t1"}}}"#
179        )));
180        assert!(has_progress_token(&parse(
181            br#"{"params":{"_meta":{"progressToken":3}}}"#
182        )));
183        assert!(!has_progress_token(&parse(br#"{"params":{"_meta":{}}}"#)));
184        assert!(!has_progress_token(&parse(b"not json")));
185    }
186
187    #[test]
188    fn request_state_key_depends_on_the_secret() {
189        assert_ne!(request_state_key("a"), request_state_key("b"));
190        assert_eq!(request_state_key("a"), request_state_key("a"));
191    }
192}
headless_lms_mcp/http/mod.rs

headless_lms_mcp/http/
mod.rs
