headless_lms_mcp/protocol/
dispatch.rs

1//! JSON-RPC dispatch for one HTTP request: version negotiation, header checks, `server/discover`,
2//! `tools/list` and `tools/call`, including confirmation round trips through `requestState`.
3
4use std::time::Duration;
5
6use headless_lms_utils::stable_digest::stable_digest;
7use rmcp::model::{
8    CacheScope, CallToolResult, ContentBlock, DiscoverResult, ElicitRequest, ElicitRequestParams,
9    ErrorCode, ErrorData, Implementation, InputRequest, InputRequests, InputRequiredResult,
10    JsonRpcError, JsonRpcResponse, JsonRpcVersion2_0, ListToolsResult, MetaObject, RequestId,
11    RequestStateCodec, ResultType, SealOptions, ServerCapabilities, ToolsCapability,
12};
13use serde::Serialize;
14use serde_json::{Map, Value, json};
15use tokio::sync::mpsc::UnboundedSender;
16
17use super::context::ClientContext;
18use super::error_codes;
19use super::headers::{self, PROTOCOL_VERSION_HEADER, RequestHeaders};
20use super::progress::ProgressReporter;
21use super::{PROTOCOL_VERSION, SETUP_DOCS_URL, www_authenticate_challenge};
22use crate::hints;
23use crate::instructions::SERVER_INSTRUCTIONS;
24use crate::tools::{
25    CONFIRMATION_KEY, ConfirmationRequest, McpPrincipal, McpScope, RequestContext, RetryState,
26    Tool, ToolError, ToolOutcome, ToolRegistry, ToolResult,
27};
28
29const SERVER_INFO_META_KEY: &str = "io.modelcontextprotocol/serverInfo";
30const CONFIRMATION_TTL: Duration = Duration::from_secs(15 * 60);
31const SERVER_NAME: &str = "courses-moocfi";
32const SERVER_TITLE: &str = "courses.mooc.fi course material";
33/// Short, so scope and role changes show up promptly.
34const TOOLS_LIST_TTL_MS: u64 = 60_000;
35const DISCOVER_TTL_MS: u64 = 300_000;
36
37/// Deployment settings of an [`McpServer`].
38pub(crate) struct ServerConfig {
39    pub server_version: String,
40    /// The RFC 9728 metadata URL advertised in `WWW-Authenticate`.
41    pub resource_metadata_url: String,
42    /// Seals `requestState`.
43    pub request_state_codec: RequestStateCodec,
44}
45
46/// One HTTP request to the MCP endpoint, framework-independent.
47pub struct HttpRequest {
48    pub method: String,
49    pub headers: RequestHeaders,
50    pub body: serde_json::Result<Value>,
51}
52
53/// The answer to one HTTP request, framework-independent.
54#[derive(Debug, Clone, PartialEq)]
55pub struct HttpResponse {
56    pub status: u16,
57    pub headers: Vec<(String, String)>,
58    /// None for 202/405 responses without a body.
59    pub body: Option<Value>,
60}
61
62impl HttpResponse {
63    fn json(status: u16, body: Value) -> Self {
64        Self {
65            status,
66            headers: vec![("Content-Type".into(), "application/json".into())],
67            body: Some(body),
68        }
69    }
70
71    fn empty(status: u16) -> Self {
72        Self {
73            status,
74            headers: Vec::new(),
75            body: None,
76        }
77    }
78}
79
80struct RpcError {
81    status: u16,
82    code: ErrorCode,
83    message: String,
84    data: Option<Value>,
85    headers: Vec<(String, String)>,
86}
87
88impl RpcError {
89    fn new(status: u16, code: ErrorCode, message: impl Into<String>) -> Self {
90        Self {
91            status,
92            code,
93            message: message.into(),
94            data: None,
95            headers: Vec::new(),
96        }
97    }
98
99    fn parse_error(message: impl Into<String>) -> Self {
100        Self::new(400, ErrorCode::PARSE_ERROR, message)
101    }
102
103    fn invalid_request(message: impl Into<String>) -> Self {
104        Self::new(400, ErrorCode::INVALID_REQUEST, message)
105    }
106
107    /// Params the protocol layer rejects before any tool is chosen.
108    fn invalid_params(message: impl Into<String>) -> Self {
109        Self::new(400, ErrorCode::INVALID_PARAMS, message)
110    }
111
112    fn header_mismatch(message: impl Into<String>) -> Self {
113        Self::new(400, ErrorCode::HEADER_MISMATCH, message)
114    }
115
116    fn unsupported_protocol_version(message: impl Into<String>) -> Self {
117        Self::new(400, ErrorCode::UNSUPPORTED_PROTOCOL_VERSION, message)
118    }
119
120    fn method_not_found(message: impl Into<String>) -> Self {
121        Self::new(404, ErrorCode::METHOD_NOT_FOUND, message)
122    }
123
124    fn unauthorized(message: impl Into<String>) -> Self {
125        Self::new(403, error_codes::UNAUTHORIZED, message)
126    }
127
128    /// A `tools/call` that passed the protocol checks but can't run; answered with HTTP 200,
129    /// unlike [`RpcError::invalid_params`].
130    fn tool_call(code: ErrorCode, message: impl Into<String>) -> Self {
131        Self::new(200, code, message)
132    }
133
134    /// A `tools/call` that failed on the server's side; the details go to the log only.
135    fn internal() -> Self {
136        Self::tool_call(
137            ErrorCode::INTERNAL_ERROR,
138            "Internal error; the call changed nothing",
139        )
140    }
141
142    fn with_data(mut self, data: Value) -> Self {
143        self.data = Some(data);
144        self
145    }
146
147    /// `id` is `None` when the request's id couldn't be read.
148    fn into_response(self, id: Option<RequestId>) -> HttpResponse {
149        let error = JsonRpcError::new(id, ErrorData::new(self.code, self.message, self.data));
150        let mut response = HttpResponse::json(self.status, to_json(&error));
151        response.headers.extend(self.headers);
152        response
153    }
154}
155
156/// The MCP server: answers each request with the tools of its registry, run against the state
157/// `S`.
158pub struct McpServer<S> {
159    registry: ToolRegistry<S>,
160    config: ServerConfig,
161}
162
163impl<S: Send + Sync + 'static> McpServer<S> {
164    /// A server answering with the tools of `registry`.
165    pub(crate) fn new(registry: ToolRegistry<S>, config: ServerConfig) -> Self {
166        Self { registry, config }
167    }
168
169    /// Handles one request. `progress_sink`, when given, receives progress notifications the
170    /// transport may stream before the response.
171    pub async fn handle(
172        &self,
173        state: &S,
174        principal: &McpPrincipal,
175        request: HttpRequest,
176        progress_sink: Option<UnboundedSender<Value>>,
177    ) -> HttpResponse {
178        if request.method != "POST" {
179            let mut response = HttpResponse::empty(405);
180            response.headers.push(("Allow".into(), "POST".into()));
181            return response;
182        }
183        let mut object = match request.body {
184            Ok(Value::Object(object)) => object,
185            Ok(_) => {
186                return RpcError::invalid_request(
187                    "Invalid Request: send one JSON-RPC object per request; batches aren't supported",
188                )
189                .into_response(None);
190            }
191            Err(e) => {
192                return RpcError::parse_error(format!("Parse error: {e}")).into_response(None);
193            }
194        };
195        let is_response = object.contains_key("result") || object.contains_key("error");
196        let invalid = || {
197            RpcError::invalid_request(
198                "Invalid Request: a request needs a string method and a string or number id",
199            )
200            .into_response(None)
201        };
202        let (id, method) = match (object.remove("id"), object.remove("method")) {
203            (Some(id), Some(Value::String(method))) => match serde_json::from_value(id) {
204                Ok(id) => (id, method),
205                Err(_) => return invalid(),
206            },
207            (None, Some(Value::String(_))) => return HttpResponse::empty(202),
208            (Some(_), None) if is_response => return HttpResponse::empty(202),
209            _ => return invalid(),
210        };
211        match self
212            .dispatch(
213                state,
214                principal,
215                &request.headers,
216                &method,
217                object.remove("params"),
218                progress_sink,
219            )
220            .await
221        {
222            Ok(result) => HttpResponse::json(
223                200,
224                to_json(&JsonRpcResponse {
225                    jsonrpc: JsonRpcVersion2_0,
226                    id,
227                    result,
228                }),
229            ),
230            Err(error) => error.into_response(Some(id)),
231        }
232    }
233
234    async fn dispatch(
235        &self,
236        state: &S,
237        principal: &McpPrincipal,
238        headers: &RequestHeaders,
239        method: &str,
240        params: Option<Value>,
241        progress_sink: Option<UnboundedSender<Value>>,
242    ) -> Result<Value, RpcError> {
243        if method == "initialize" {
244            let requested = params
245                .as_ref()
246                .and_then(|p| p.get("protocolVersion"))
247                .cloned()
248                .unwrap_or(Value::Null);
249            return Err(self.unsupported_version(requested));
250        }
251        let mut params = match params {
252            Some(Value::Object(params)) => params,
253            _ => Map::new(),
254        };
255        let client = ClientContext::from_meta(match params.remove("_meta") {
256            Some(Value::Object(meta)) => meta,
257            _ => Map::new(),
258        });
259        let tool_name = params
260            .get("name")
261            .and_then(Value::as_str)
262            .unwrap_or_default();
263        let tool = match method {
264            "tools/call" => self.registry.get(tool_name),
265            _ => None,
266        };
267        if method != "server/discover" {
268            self.check_version(&client, headers)?;
269            let header_params = tool.map_or(&[][..], |tool| &tool.definition.header_params);
270            headers::validate(headers, method, &params, header_params)
271                .map_err(RpcError::header_mismatch)?;
272        }
273        match method {
274            "server/discover" => Ok(self.discover()),
275            "tools/list" => Ok(self.list_tools(principal)),
276            "tools/call" => {
277                let Some(tool) = tool else {
278                    return Err(RpcError::tool_call(
279                        ErrorCode::INVALID_PARAMS,
280                        format!("Unknown tool: {tool_name}"),
281                    ));
282                };
283                let progress = ProgressReporter::new(client.progress_token.clone(), progress_sink);
284                self.call_tool(state, principal, client, tool, params, progress)
285                    .await
286            }
287            other => Err(RpcError::method_not_found(format!(
288                "Method not found: {other}"
289            ))),
290        }
291    }
292
293    fn unsupported_version(&self, requested: Value) -> RpcError {
294        RpcError::unsupported_protocol_version(format!(
295            "This server supports MCP protocol revision {PROTOCOL_VERSION} only. Update the agent, or enable its {PROTOCOL_VERSION} support (Codex needs an opt-in); see {SETUP_DOCS_URL}"
296        ))
297        .with_data(json!({"requested": requested, "supported": [PROTOCOL_VERSION], "setup": SETUP_DOCS_URL}))
298    }
299
300    fn check_version(
301        &self,
302        client: &ClientContext,
303        headers: &RequestHeaders,
304    ) -> Result<(), RpcError> {
305        let Some(version) = &client.protocol_version else {
306            return Err(RpcError::invalid_params(
307                "Invalid params: request _meta must include io.modelcontextprotocol/protocolVersion and io.modelcontextprotocol/clientCapabilities",
308            ));
309        };
310        if *version != PROTOCOL_VERSION {
311            return Err(self.unsupported_version(json!(version)));
312        }
313        match headers.single(PROTOCOL_VERSION_HEADER) {
314            Ok(Some(header)) if header == version.as_str() => Ok(()),
315            Ok(Some(header)) => Err(RpcError::header_mismatch(format!(
316                "MCP-Protocol-Version header {header} doesn't match the request _meta version {}",
317                version.as_str()
318            ))),
319            Ok(None) => Err(RpcError::header_mismatch(
320                "missing MCP-Protocol-Version header",
321            )),
322            Err(reason) => Err(RpcError::header_mismatch(reason)),
323        }
324    }
325
326    fn server_info_meta(&self) -> MetaObject {
327        let server_info =
328            Implementation::new(SERVER_NAME, &self.config.server_version).with_title(SERVER_TITLE);
329        MetaObject::from(Map::from_iter([(
330            SERVER_INFO_META_KEY.to_string(),
331            to_json(&server_info),
332        )]))
333    }
334
335    fn discover(&self) -> Value {
336        let mut tools = ToolsCapability::default();
337        tools.list_changed = Some(false);
338        let mut capabilities = ServerCapabilities::default();
339        capabilities.tools = Some(tools);
340        let mut result = DiscoverResult::new(vec![PROTOCOL_VERSION], capabilities);
341        result.instructions = Some(SERVER_INSTRUCTIONS.to_string());
342        result.ttl_ms = DISCOVER_TTL_MS;
343        result.meta = Some(self.server_info_meta());
344        to_json(&result)
345    }
346
347    fn list_tools(&self, principal: &McpPrincipal) -> Value {
348        to_json(&ListToolsResult {
349            result_type: Some(ResultType::COMPLETE),
350            meta: Some(self.server_info_meta()),
351            next_cursor: None,
352            ttl_ms: Some(TOOLS_LIST_TTL_MS),
353            cache_scope: Some(CacheScope::Private),
354            tools: self.registry.listings_for(principal),
355        })
356    }
357
358    async fn call_tool(
359        &self,
360        state: &S,
361        principal: &McpPrincipal,
362        client: ClientContext,
363        tool: &Tool<S>,
364        mut params: Map<String, Value>,
365        progress: ProgressReporter,
366    ) -> Result<Value, RpcError> {
367        let name = tool.definition.name;
368        if !principal.has_scope(tool.definition.scope) {
369            return Err(self.insufficient_scope(principal, tool.definition.scope));
370        }
371        let mut arguments = match params.remove("arguments") {
372            None | Some(Value::Null) => Map::new(),
373            Some(Value::Object(arguments)) => arguments,
374            Some(_) => {
375                return Err(RpcError::tool_call(
376                    ErrorCode::INVALID_PARAMS,
377                    "arguments must be an object",
378                ));
379            }
380        };
381        let skill_version = arguments.remove("skill_version");
382        let skill_version = skill_version.as_ref().and_then(Value::as_str);
383        if skill_version.is_none() && tool.definition.requires_skill {
384            return Ok(self.tool_result(ToolError::from(hints::skill_missing()).into_result()));
385        }
386        let skill_hint = match hints::skill_version(skill_version) {
387            Ok(hint) => hint,
388            Err(outdated) => {
389                return Ok(self.tool_result(ToolError::from(outdated).into_result()));
390            }
391        };
392        let binding = confirmation_binding(principal, name, &arguments)?;
393        let retry = match params.remove("requestState") {
394            Some(Value::String(sealed)) => Some(self.open_request_state(
395                &sealed,
396                binding.as_bytes(),
397                params.remove("inputResponses"),
398            )?),
399            _ => None,
400        };
401        let sealing_binding = tool
402            .handler
403            .asks_for_confirmation()
404            .then_some(binding.as_str());
405        let ctx = RequestContext {
406            principal: principal.clone(),
407            client,
408            tool_name: name,
409            retry,
410            progress,
411        };
412        match tool.handler.call_with_json(state, &ctx, arguments).await {
413            Ok(ToolOutcome::Complete(result)) => {
414                Ok(self.tool_result(result.with_hints(skill_hint)))
415            }
416            Ok(ToolOutcome::NeedsConfirmation(request)) => {
417                Ok(self.confirmation(&ctx, request, sealing_binding, skill_hint))
418            }
419            Err(ToolError::InsufficientScope { required }) => {
420                Err(self.insufficient_scope(principal, required))
421            }
422            Err(ToolError::Internal(message)) => {
423                tracing::error!(tool = %ctx.tool_name, %message, "MCP tool failed");
424                Err(RpcError::internal())
425            }
426            Err(error) => Ok(self.tool_result(error.into_result().with_hints(skill_hint))),
427        }
428    }
429
430    /// The 403 for a missing scope. Its challenge asks for the scopes the token has as well, since
431    /// a client stepping up requests exactly the challenged scopes and would lose the others.
432    fn insufficient_scope(&self, principal: &McpPrincipal, required: McpScope) -> RpcError {
433        let scope = required.as_str();
434        let mut error = RpcError::unauthorized(format!(
435            "This needs the {scope} scope, which the agent's sign-in doesn't include"
436        ))
437        .with_data(json!({"required_scope": scope}));
438        let challenged: Vec<&str> = McpScope::ALL
439            .into_iter()
440            .filter(|s| *s == required || principal.has_scope(*s))
441            .map(McpScope::as_str)
442            .collect();
443        error.headers.push((
444            "WWW-Authenticate".into(),
445            www_authenticate_challenge(
446                &self.config.resource_metadata_url,
447                &challenged.join(" "),
448                Some((
449                    "insufficient_scope",
450                    "the token lacks a scope this tool needs",
451                )),
452            ),
453        ));
454        error
455    }
456
457    fn tool_result(&self, result: ToolResult) -> Value {
458        let structured = result.structured_content();
459        let content = vec![ContentBlock::text(structured.to_string())];
460        let mut call_result = if result.is_error {
461            CallToolResult::error(content)
462        } else {
463            CallToolResult::success(content)
464        };
465        call_result.structured_content = Some(structured);
466        call_result.meta = Some(self.server_info_meta());
467        to_json(&call_result)
468    }
469
470    /// `binding` is `None` when the handler doesn't declare that it asks for confirmation; the
471    /// text fallback is then the only safe answer.
472    fn confirmation(
473        &self,
474        ctx: &RequestContext,
475        request: ConfirmationRequest,
476        binding: Option<&str>,
477        skill_hint: Option<hints::Hint>,
478    ) -> Value {
479        let is_usable =
480            ctx.client.supports_url_elicitation() && ctx.client.can_use_input_required();
481        let (true, Some(binding)) = (is_usable, binding) else {
482            return self.tool_result(request.fallback.with_hints(skill_hint));
483        };
484        let options = SealOptions::new()
485            .associated_data(binding.as_bytes())
486            .ttl(CONFIRMATION_TTL);
487        // The confirmation is bound through the associated data; the state itself carries nothing.
488        let sealed = self.config.request_state_codec.seal_with(&[], &options);
489        let elicitation = ElicitRequest::new(ElicitRequestParams::UrlElicitationParams {
490            meta: None,
491            message: request.message,
492            url: request.url,
493            elicitation_id: request.elicitation_id,
494        });
495        let input_requests = InputRequests::from([(
496            CONFIRMATION_KEY.to_string(),
497            InputRequest::Elicitation(elicitation),
498        )]);
499        to_json(
500            &InputRequiredResult::new(Some(input_requests), Some(sealed))
501                .with_meta(self.server_info_meta()),
502        )
503    }
504
505    fn open_request_state(
506        &self,
507        sealed: &str,
508        associated_data: &[u8],
509        responses: Option<Value>,
510    ) -> Result<RetryState, RpcError> {
511        let invalid = || {
512            RpcError::tool_call(
513                error_codes::INVALID_REQUEST_STATE,
514                "requestState is invalid or expired; repeat the call without it",
515            )
516        };
517        self.config
518            .request_state_codec
519            .open_with(sealed, associated_data)
520            .map_err(|_| invalid())?;
521        let input_responses = match responses {
522            Some(Value::Object(responses)) => responses,
523            _ => Map::new(),
524        };
525        Ok(RetryState { input_responses })
526    }
527}
528
529/// Binds a confirmation to the user, the tool and its exact arguments, so a sealed state can't
530/// be replayed by someone else or for another change.
531fn confirmation_binding(
532    principal: &McpPrincipal,
533    tool: &str,
534    arguments: &Map<String, Value>,
535) -> Result<String, RpcError> {
536    // Canonical because serde_json's `Map` keeps keys sorted (the `preserve_order` feature is off).
537    let arguments = serde_json::to_vec(arguments).map_err(|error| {
538        tracing::error!(%tool, %error, "MCP tool arguments didn't serialize");
539        RpcError::internal()
540    })?;
541    Ok(stable_digest(&[
542        principal.actor.user_id.as_bytes(),
543        tool.as_bytes(),
544        &arguments,
545    ]))
546}
547
548/// A protocol value as JSON; one that doesn't serialize is logged and becomes `null`.
549fn to_json(value: &impl Serialize) -> Value {
550    serde_json::to_value(value).unwrap_or_else(|error| {
551        tracing::error!(%error, "an MCP protocol value didn't serialize");
552        Value::Null
553    })
554}
555
556#[cfg(test)]
557#[path = "dispatch_tests.rs"]
558mod tests;
headless_lms_mcp/protocol/dispatch.rs

headless_lms_mcp/protocol/
dispatch.rs
