protocol_droid/application/server/dispatch/mod.rs
1//! Routing a method name to the handler that serves it.
2//!
3//! # Why this is not on `Server`
4//!
5//! `Server::call` takes a tool name, because a tool call is the only thing a
6//! handler can answer. A peer, though, sends a _method_ name, and several
7//! methods never reach a tool handler at all — `tools/list` and `ping` have no
8//! handler to dispatch to. The method table lives here rather than on
9//! `Server` so the two stay apart: the builder is how a server is described,
10//! this is how a request against it is routed.
11//!
12//! # What the result is, and is not
13//!
14//! The result is an **IR-level** [`JsonObject`], not a wire frame. There is no
15//! request-side codec yet — `Codec` lifts and lowers a `ToolResult` and nothing
16//! else — so a result built here is the canonical form that T13/T14's transport
17//! will then lower for whichever revision negotiated. The in-process test that
18//! exercises this says the same thing in its own doc comment rather than
19//! pretending to cover the wire.
20//!
21//! # Which fields a listing carries
22//!
23//! Only the ones **every** revision expresses. `title` and `icons` arrived in
24//! `2026-07-28`, so emitting them unconditionally would hand a `2024-11-05`
25//! peer a field its revision has no name for. Gating a field on a revision is
26//! the codec's job and belongs with it; this module stays the common
27//! denominator, and the omission is recorded rather than papered over.
28
29use std::sync::Arc;
30
31use crate::application::mcp::initialize::server_capabilities_json;
32use crate::application::mcp::input_required::{InputRound, capability_for, is_input_required};
33use crate::application::server::builder::{DispatchError, Server};
34use crate::application::server::context::{PeerSession, RequestContext};
35use crate::domain::feature::{Feature, FeatureProfile};
36use crate::domain::idempotency::{Claim, IdempotencyKey};
37use crate::domain::ir::JsonObject;
38use crate::domain::ir::LogLevel;
39use crate::domain::registry::PrimitiveKind;
40use crate::domain::{
41 ProtocolVersion, SupportedVersions, client_method_defined_in, discover_defined_in,
42 handshake_defined_in,
43};
44
45mod negotiate;
46mod params;
47mod shape;
48
49#[cfg(test)]
50mod tests;
51
52use negotiate::negotiate;
53use params::{parsed_param, split_call_params, string_param};
54use shape::{stamp_result, wrap_listing};
55
56/// Routes methods to a server's handlers.
57#[derive(Clone)]
58pub struct Dispatcher {
59 server: Arc<Server>,
60}
61
62impl core::fmt::Debug for Dispatcher {
63 /// Prints the server's name and tool count.
64 ///
65 /// For the same reason `PeerSession` does: the registry behind it is a
66 /// large snapshot, and a post-mortem wants to know which server it was.
67 fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
68 f.debug_struct("Dispatcher")
69 .field("name", &self.server.name())
70 .field("tool_count", &self.server.tool_count())
71 .finish_non_exhaustive()
72 }
73}
74
75/// Which of the two subscription methods a request was.
76///
77/// They share every step — the same gate, the same existence check, the same
78/// empty answer — so one handler takes this instead of a `bool`, which would
79/// make a reader guess at the argument's polarity, and two near-identical
80/// handlers, which would be two places for the check to drift.
81#[derive(Clone, Copy)]
82enum SubscribeOp {
83 /// `resources/subscribe`.
84 Subscribe,
85 /// `resources/unsubscribe`.
86 Unsubscribe,
87}
88
89impl Dispatcher {
90 /// Routes to `server`.
91 #[must_use]
92 pub const fn new(server: Arc<Server>) -> Self {
93 Self { server }
94 }
95
96 /// The server requests are routed to.
97 #[must_use]
98 pub const fn server(&self) -> &Arc<Server> {
99 &self.server
100 }
101
102 /// The server's state map, for a transport handler that needs to
103 /// hand it to a [`RequestContext`].
104 ///
105 /// Cloning the `Arc` is one bump; the contents of the map are not
106 /// touched on the read path.
107 #[must_use]
108 pub fn state(&self) -> &Arc<crate::application::server::context::StateMap> {
109 self.server.state()
110 }
111
112 /// Runs the handshake and returns the session narrowed to what was
113 /// negotiated.
114 ///
115 /// [`dispatch`](Self::dispatch) answers `initialize` and discards the
116 /// negotiated profile, which leaves the caller gating the rest of the
117 /// connection on the profile the session was *constructed* with — a
118 /// revision chosen before the peer existed. That is the right answer only
119 /// when the peer happens to want the newest revision this build serves.
120 ///
121 /// The returned session is what the transport must carry forward: it holds
122 /// the profile resolved for the revision the client actually asked for, so
123 /// every later gate reads that. Transport-specific policy still belongs to
124 /// the transport; the profile does not belong to the caller.
125 ///
126 /// # Errors
127 ///
128 /// As [`dispatch`](Self::dispatch) for `initialize`, and
129 /// [`DispatchError::MalformedInitialize`] when the body does not carry a
130 /// readable version and capability block.
131 pub fn handshake(
132 &self,
133 ctx: &RequestContext,
134 params: &JsonObject,
135 ) -> Result<(JsonObject, PeerSession), DispatchError> {
136 let (version, profile) = negotiate(ctx, params)?;
137 let reply = self.initialize_result(version, profile)?;
138 Ok((reply, ctx.peer().clone().with_profile(profile)))
139 }
140
141 /// The `InitializeResult` body for a settled negotiation.
142 fn initialize_result(
143 &self,
144 version: ProtocolVersion,
145 profile: FeatureProfile,
146 ) -> Result<JsonObject, DispatchError> {
147 // SEP-2575 removed the handshake in `2026-07-28`. There is no result
148 // shape to build, so this is the spec's own answer — the method is not
149 // defined on this revision — and not a `FeatureUnavailable` dressed up
150 // as one.
151 if !handshake_defined_in(version) {
152 return Err(DispatchError::UnknownMethod);
153 }
154 let result = serde_json::json!({
155 "protocolVersion": version.as_wire(),
156 "capabilities": server_capabilities_json(profile),
157 "serverInfo": {
158 "name": self.server.name(),
159 "version": self.server.version(),
160 },
161 });
162 JsonObject::from_value(result).map_err(|_| DispatchError::MalformedInitialize)
163 }
164
165 /// Routes one request.
166 ///
167 /// The gate runs before anything is read out of `params`, so a method the
168 /// session cannot use does not even parse its arguments — the same order
169 /// `RequestContext`'s own gated calls use, and for the same reason.
170 ///
171 /// # Errors
172 ///
173 /// [`DispatchError::UnknownMethod`] for a name no method answers, and
174 /// [`DispatchError::NotCallable`] for a method whose feature the session
175 /// does not have. Both are refusals that wrote nothing.
176 pub async fn dispatch(
177 &self,
178 ctx: &RequestContext,
179 method: &str,
180 params: &JsonObject,
181 ) -> Result<JsonObject, DispatchError> {
182 let version = ctx.peer().negotiated();
183 let answered = self.route(ctx, method, params).await?;
184 // `resultType` belongs on every result at 2026-07-28 and on none
185 // before it, so it is stamped here rather than in each method: there
186 // are ten of them and a per-method call is a call someone forgets.
187 let stamped = stamp_result(answered, Cacheable::No, version);
188 Ok(self.identify(stamped, version))
189 }
190
191 /// Adds `_meta["io.modelcontextprotocol/serverInfo"]` at 2026-07-28.
192 ///
193 /// SEP-2575: with no handshake there is no `InitializeResult` to carry
194 /// `serverInfo`, so a server "SHOULD identify itself in each result's
195 /// `_meta`". Older revisions carry it once, in the handshake, and their
196 /// result schemas have no such key.
197 fn identify(&self, result: JsonObject, version: ProtocolVersion) -> JsonObject {
198 if !result_envelope_defined_in(version) {
199 return result;
200 }
201 let Ok(serde_json::Value::Object(mut members)) =
202 serde_json::from_str::<serde_json::Value>(result.as_str())
203 else {
204 return result;
205 };
206 let meta = members
207 .entry("_meta")
208 .or_insert_with(|| serde_json::Value::Object(serde_json::Map::new()));
209 if let Some(meta) = meta.as_object_mut() {
210 meta.insert(
211 "io.modelcontextprotocol/serverInfo".to_owned(),
212 serde_json::json!({
213 "name": self.server.name(),
214 "version": self.server.version(),
215 }),
216 );
217 }
218 JsonObject::from_value(serde_json::Value::Object(members)).unwrap_or(result)
219 }
220
221 /// The SEP-2322 round a retry carries, verified.
222 ///
223 /// Both members are optional and both are read only here. A `requestState`
224 /// that does not open under this server's key is refused before any
225 /// handler sees it — the handler is only ever handed a state this server
226 /// issued.
227 fn input_round(&self, params: &JsonObject) -> Result<InputRound, DispatchError> {
228 let responses = match parsed_param(params, "inputResponses")? {
229 None | Some(serde_json::Value::Null) => None,
230 Some(serde_json::Value::Object(map)) => Some(map),
231 Some(_) => return Err(DispatchError::MalformedParams),
232 };
233 let state = match parsed_param(params, "requestState")? {
234 None | Some(serde_json::Value::Null) => None,
235 Some(serde_json::Value::String(sealed)) => {
236 let key = self
237 .server
238 .request_state_key()
239 .ok_or(DispatchError::TamperedRequestState)?;
240 Some(
241 key.open(&sealed)
242 .ok_or(DispatchError::TamperedRequestState)?
243 .to_owned(),
244 )
245 }
246 Some(_) => return Err(DispatchError::MalformedParams),
247 };
248 Ok(InputRound {
249 responses,
250 state,
251 key: self.server.request_state_key().cloned(),
252 })
253 }
254
255 /// Checks an interim result against what the request may receive.
256 ///
257 /// Two rules from SEP-2322 and one from SEP-2575. An `InputRequiredResult`
258 /// exists only at `2026-07-28` and later, and only on `tools/call`,
259 /// `prompts/get` and `resources/read` — "servers MUST NOT send
260 /// `InputRequiredResult` responses on any client requests other than the
261 /// supported ones". And every input request it carries needs the client
262 /// capability for that request, which the client must have declared on
263 /// *this* request: otherwise the answer is `-32021`.
264 fn check_interim(
265 ctx: &RequestContext,
266 method: &str,
267 result: JsonObject,
268 ) -> Result<JsonObject, DispatchError> {
269 let Ok(serde_json::Value::Object(members)) =
270 serde_json::from_str::<serde_json::Value>(result.as_str())
271 else {
272 return Ok(result);
273 };
274 if !is_input_required(&members) {
275 return Ok(result);
276 }
277 if !accepts_input_round(method) || !result_envelope_defined_in(ctx.peer().negotiated()) {
278 return Err(DispatchError::HandlerFailed(
279 "an InputRequiredResult is not defined for this request".to_owned(),
280 ));
281 }
282 let mut required: Vec<&'static str> = Vec::new();
283 if let Some(requests) = members
284 .get("inputRequests")
285 .and_then(serde_json::Value::as_object)
286 {
287 for request in requests.values() {
288 let Some(capability) = request
289 .get("method")
290 .and_then(serde_json::Value::as_str)
291 .and_then(capability_for)
292 else {
293 continue;
294 };
295 let feature = match capability {
296 "sampling" => Feature::Sampling,
297 "elicitation" => Feature::Elicitation,
298 _ => Feature::Roots,
299 };
300 if !ctx.peer().allows(feature) && !required.contains(&capability) {
301 required.push(capability);
302 }
303 }
304 }
305 if !required.is_empty() {
306 return Err(DispatchError::MissingClientCapability { required });
307 }
308 Ok(result)
309 }
310
311 /// Routes one method to its handler.
312 ///
313 /// Split from [`dispatch`](Self::dispatch) so the result envelope is added
314 /// once, on the way out, instead of at each of the ten places a result is
315 /// built.
316 async fn route(
317 &self,
318 ctx: &RequestContext,
319 method: &str,
320 params: &JsonObject,
321 ) -> Result<JsonObject, DispatchError> {
322 let version = ctx.peer().negotiated();
323 // A method the negotiated revision does not define is `-32601`
324 // whatever this server could otherwise do with it: answering `ping` on
325 // `2026-07-28` would tell a client the method exists there.
326 if !client_method_defined_in(version, method) {
327 return Err(DispatchError::UnknownMethod);
328 }
329 let round;
330 let ctx = if accepts_input_round(method) && result_envelope_defined_in(version) {
331 round = ctx.clone().with_input_round(self.input_round(params)?);
332 &round
333 } else {
334 ctx
335 };
336 let answered = self.route_defined(ctx, method, params).await?;
337 Self::check_interim(ctx, method, answered)
338 }
339
340 /// Routes a method the negotiated revision defines.
341 async fn route_defined(
342 &self,
343 ctx: &RequestContext,
344 method: &str,
345 params: &JsonObject,
346 ) -> Result<JsonObject, DispatchError> {
347 match method {
348 // `initialize` is the one method with no gate, and for the
349 // opposite reason to `ping`. `ping` may be answered on a session
350 // that already exists; `initialize` is the request that *creates*
351 // the session, so consulting the feature profile to decide whether
352 // to serve it would ask the session whether it exists and take the
353 // answer from something that has not been built yet. The
354 // negotiation that does decide it has already happened — the
355 // revision in the result is the one the session settled on — so
356 // the gate has nothing left to contribute here.
357 //
358 // The one thing it *does* still have to contribute is which
359 // revision that is: on `2026-07-28` there is no handshake to
360 // perform, and the handler refuses rather than inventing one.
361 "initialize" => self.initialize(ctx, params),
362 // SEP-2575's replacement for the handshake, and defined only where
363 // the handshake is not. The two are complementary, so a revision
364 // never has both.
365 "server/discover" => Self::discover(ctx),
366 // `ping` is defined in every revision that has a session and
367 // answers on an otherwise idle one, so gating it would refuse the
368 // request it exists for.
369 "ping" => Ok(JsonObject::default()),
370 "tools/list" => self.list(ctx, PrimitiveKind::Tool, "tools"),
371 "resources/list" => self.list(ctx, PrimitiveKind::Resource, "resources"),
372 "resources/templates/list" => {
373 self.list(ctx, PrimitiveKind::ResourceTemplate, "resourceTemplates")
374 }
375 "prompts/list" => self.list(ctx, PrimitiveKind::Prompt, "prompts"),
376 "tools/call" => self.call_tool(ctx, params).await,
377 "resources/read" => self.read_resource(ctx, params).await,
378 "prompts/get" => self.get_prompt(ctx, params).await,
379 "logging/setLevel" => Self::set_log_level(ctx, params),
380 "completion/complete" => Self::complete(ctx, params),
381 "resources/subscribe" => self.subscribe(ctx, params, SubscribeOp::Subscribe),
382 "resources/unsubscribe" => self.subscribe(ctx, params, SubscribeOp::Unsubscribe),
383 _ => Err(DispatchError::UnknownMethod),
384 }
385 }
386
387 /// Serves `server/discover`: what this server speaks, and for how long a
388 /// client may keep the answer.
389 ///
390 /// SEP-2575 removed `initialize` from `2026-07-28` and made every request
391 /// carry its own version and capabilities in `_meta`. That is enough to
392 /// *use* the server and not enough to *choose* one, which is the gap this
393 /// fills: a client with no session and no `initialize` has no other way to
394 /// learn the revisions on offer.
395 ///
396 /// The capabilities are the profile's own advertisement rather than a
397 /// separate list, because a second list is a second opinion: a client that
398 /// discovered `tools` and then found `tools/call` refused would have been
399 /// told something this crate does not do.
400 ///
401 /// `ttlMs` is `0`. This server has no cache and re-derives the answer on
402 /// every call, so advertising a lifetime it cannot honour would tell a
403 /// client to reuse a result whose backing registry may since have changed.
404 /// Zero is the honest value and the specification permits it: "should be
405 /// considered immediately stale".
406 ///
407 /// `cacheScope` is `private`. The advertisement describes what this
408 /// process can do, and there is nothing user-specific in it, but a
409 /// capability set is also a map of what an operator has turned on, and
410 /// `"public"` invites an intermediary to share it across authorization
411 /// contexts. `"private"` is the safe reading of the same answer.
412 fn discover(ctx: &RequestContext) -> Result<JsonObject, DispatchError> {
413 if !discover_defined_in(ctx.peer().negotiated()) {
414 return Err(DispatchError::UnknownMethod);
415 }
416 let result = serde_json::json!({
417 "resultType": "complete",
418 "supportedVersions": SupportedVersions::all()
419 .map_err(|_| DispatchError::UnnegotiableVersion)?
420 .into_iter()
421 .rev()
422 .map(|version| version.as_wire().into_owned())
423 .collect::<Vec<_>>(),
424 "ttlMs": 0,
425 "cacheScope": "private",
426 "capabilities": server_capabilities_json(ctx.peer().profile()),
427 });
428 JsonObject::from_value(result).map_err(|_| DispatchError::MalformedParams)
429 }
430
431 /// Serves `resources/subscribe` and `resources/unsubscribe`.
432 ///
433 /// Both answer the empty object. A subscription is recorded against the
434 /// session's own sink, so [`ServerHandle::notify_resource_updated`] reaches
435 /// exactly the sessions that asked, and it ends with the session.
436 ///
437 /// [`ServerHandle::notify_resource_updated`]: crate::application::server::ServerHandle::notify_resource_updated
438 fn subscribe(
439 &self,
440 ctx: &RequestContext,
441 params: &JsonObject,
442 op: SubscribeOp,
443 ) -> Result<JsonObject, DispatchError> {
444 Self::require(ctx, Feature::ResourcesSubscribe)?;
445 let uri = string_param(params, "uri")?;
446 // The URI must be one this server lists. Accepting a subscription for
447 // a resource that does not exist is how a client ends up waiting for
448 // an update that can never come, and the peer cannot tell that from a
449 // server that simply has nothing to report.
450 if self.server.reader(&uri).is_none() {
451 return Err(DispatchError::UnknownResource);
452 }
453 let handle = self.server.handle();
454 match op {
455 SubscribeOp::Subscribe => handle.watch_resource(ctx.peer().sink(), &uri, None),
456 SubscribeOp::Unsubscribe => handle.unwatch_resource(ctx.peer().sink(), &uri),
457 }
458 Ok(JsonObject::default())
459 }
460
461 /// Serves `resources/read`.
462 ///
463 /// Not idempotency-keyed, unlike `tools/call`. A read has no side effect
464 /// the peer can observe, so a retried read costs another read and nothing
465 /// else — while a key would make two legitimate concurrent reads of the
466 /// same URI collide into `DuplicateInFlight`, which is exactly wrong for a
467 /// request whose whole purpose is to be repeatable.
468 async fn read_resource(
469 &self,
470 ctx: &RequestContext,
471 params: &JsonObject,
472 ) -> Result<JsonObject, DispatchError> {
473 Self::require(ctx, Feature::Resources)?;
474 let uri = string_param(params, "uri")?;
475 let reader = self
476 .server
477 .reader(&uri)
478 .ok_or(DispatchError::UnknownResource)?;
479 // One of the five cacheable results at 2026-07-28, and the only one
480 // that is not a listing — which is exactly why it is marked here
481 // rather than inferred from the method name.
482 Ok(stamp_result(
483 reader(ctx.clone(), uri).await?,
484 Cacheable::Yes,
485 ctx.peer().negotiated(),
486 ))
487 }
488
489 /// Serves `prompts/get`.
490 ///
491 /// Also not idempotency-keyed, for the same reason as `read_resource`: a
492 /// prompt getter produces a message list, and running it twice produces the
493 /// same list twice at no cost to anybody.
494 async fn get_prompt(
495 &self,
496 ctx: &RequestContext,
497 params: &JsonObject,
498 ) -> Result<JsonObject, DispatchError> {
499 Self::require(ctx, Feature::Prompts)?;
500 let name = string_param(params, "name")?;
501 let arguments = match parsed_param(params, "arguments")? {
502 None | Some(serde_json::Value::Null) => JsonObject::default(),
503 Some(arguments) => {
504 JsonObject::from_value(arguments).map_err(|_| DispatchError::MalformedParams)?
505 }
506 };
507 let getter = self
508 .server
509 .getter(&name)
510 .ok_or(DispatchError::UnknownPrompt)?;
511 getter(ctx.clone(), arguments).await
512 }
513
514 /// Serves `logging/setLevel`.
515 ///
516 /// Answers the empty object the method's own result type is defined as,
517 /// after checking the requested `level` is one the spec names. The check
518 /// is here rather than ignored because a server that accepts any string
519 /// reports a level it cannot honour: the client believes a filter is in
520 /// place and stops sending the records it would otherwise have expected,
521 /// and the peer has no way to tell "set to debug" from "set to a typo".
522 ///
523 /// The level is stored on the session's [`LogGate`], which every later
524 /// request's context shares, so [`RequestContext::log`] filters against
525 /// it from the next record on.
526 ///
527 /// [`LogGate`]: crate::application::server::LogGate
528 fn set_log_level(
529 ctx: &RequestContext,
530 params: &JsonObject,
531 ) -> Result<JsonObject, DispatchError> {
532 Self::require(ctx, Feature::Logging)?;
533 let level = string_param(params, "level")?;
534 // A `level` outside the spec's own set is a client error, and answers
535 // `InvalidParams` — the same answer as a missing member, because a
536 // peer cannot act on either and the difference is not one it can use.
537 let level = level
538 .parse::<LogLevel>()
539 .map_err(|_| DispatchError::MalformedParams)?;
540 ctx.peer().log_gate().set(level);
541 Ok(JsonObject::default())
542 }
543
544 /// Serves `completion/complete`.
545 ///
546 /// Returns an empty `values` array: the crate registers no completable
547 /// `PromptArgument` values, so the reference is well-defined — there is
548 /// nothing to offer, and the suite documents an empty array as a
549 /// conforming answer.
550 ///
551 /// `completion` is an object holding `values`, not a bare array. An empty
552 /// *array* at the top level answers `{"completion":[]}`, which is the
553 /// shape a first reading of the method name suggests and the one that
554 /// scores `0 passed, 2 failed` — the scenario reads
555 /// `r.completion.values`, finds `[]` has no member `values`, and reports
556 /// `Missing values array in completion`. The empty case is the one where
557 /// the distinction is invisible: both a missing `values` and an empty one
558 /// mean "no completions", so the error names a missing array that was
559 /// never absent, only wrapped wrongly.
560 fn complete(ctx: &RequestContext, _params: &JsonObject) -> Result<JsonObject, DispatchError> {
561 Self::require(ctx, Feature::Completions)?;
562 JsonObject::new(r#"{"completion":{"values":[]}}"#)
563 .map_err(|_| DispatchError::MalformedParams)
564 }
565
566 /// Answers the handshake: this server, the revision it speaks, and what it
567 /// can do.
568 ///
569 /// The result is the IR form of `InitializeResult` — `protocolVersion`,
570 /// `capabilities`, `serverInfo` — which are the three fields every
571 /// revision that has a handshake marks required. `instructions` is omitted
572 /// rather than emitted empty: it is optional, it is a human-facing hint
573 /// with no source on a [`Server`], and an empty string is worse than
574 /// absent, because a client that reads it has been told the server has no
575 /// instructions when what it means is that this server supplies none.
576 ///
577 /// # What decides the revision
578 ///
579 /// The session's, not the request's. `params.protocolVersion` is what the
580 /// client *offered*; by the time a request reaches a dispatcher the layer
581 /// that owns negotiation has already settled the revision, and the session
582 /// is running on it. Answering with the offer instead would let a client
583 /// move a session onto a revision the transport never negotiated for it,
584 /// and the profile gating every other method would be one revision ahead of
585 /// the wire.
586 ///
587 /// The offer is still read and still checked, for the two reasons a
588 /// malformed handshake matters. The version the session speaks is settled
589 /// by [`negotiate_server`] against **the client's own offer**, which is
590 /// why [`check_initialize_params`] returns it rather than discarding it.
591 ///
592 /// # What decides `capabilities`
593 ///
594 /// [`FeatureProfile::advertised_by_server`] — the profile, read back. The
595 /// advertisement is a fold over the same table every feature gate is a fold
596 /// over, so it cannot claim a capability the session will not act on, and
597 /// it cannot decline one it will.
598 ///
599 /// # Errors
600 ///
601 /// - [`DispatchError::UnnegotiableVersion`] when the session is on a
602 /// revision this build has no wire types for.
603 /// - [`DispatchError::UnknownMethod`] when the session's revision has no
604 /// handshake at all — `2026-07-28` and `draft` removed `initialize` and
605 /// its result, so a request for it on one of those revisions is asking
606 /// for a method the revision does not define.
607 /// - [`DispatchError::MalformedInitialize`] when `params` is not an object
608 /// with a string `protocolVersion` that parses, a `capabilities` object,
609 /// and a `clientInfo` carrying a `name`.
610 fn initialize(
611 &self,
612 ctx: &RequestContext,
613 params: &JsonObject,
614 ) -> Result<JsonObject, DispatchError> {
615 let (version, profile) = negotiate(ctx, params)?;
616 self.initialize_result(version, profile)
617 }
618
619 /// One of the four listing methods.
620 ///
621 /// All four are the same operation against a different kind, so they are
622 /// one function with the feature and the key as parameters rather than
623 /// four near-identical bodies.
624 fn list(
625 &self,
626 ctx: &RequestContext,
627 kind: PrimitiveKind,
628 key: &str,
629 ) -> Result<JsonObject, DispatchError> {
630 Self::require(ctx, kind.feature())?;
631 let listing =
632 self.server
633 .registry()
634 .list_visible(ctx.peer().auth(), &ctx.peer().profile(), kind);
635 Ok(stamp_result(
636 wrap_listing(key, &listing),
637 Cacheable::Yes,
638 ctx.peer().negotiated(),
639 ))
640 }
641
642 /// `tools/call`, through the server's own dispatch.
643 /// Runs one `tools/call`, at most once per request id.
644 ///
645 /// The order here is not a style choice; it is the claim:
646 ///
647 /// 1. **Check.** Claim the key before anything runs. Claiming after would
648 /// leave a window in which two callers both see an absent key.
649 /// 2. **Execute.** Only the caller that was handed the token runs the
650 /// tool. A caller that finds a completed key answers from the recording
651 /// and never reaches the handler, and one that finds an in-flight key
652 /// is refused rather than queued — see [`DispatchError::DuplicateInFlight`].
653 /// 3. **Persist.** Record the result, or release the claim, before
654 /// returning. A result that is returned but not recorded is a result a
655 /// retry will not find, and a key that is neither recorded nor released
656 /// is a key nobody can ever claim again.
657 ///
658 /// # Why a failed call releases rather than records
659 ///
660 /// An `Err` here means the tool produced no result — the gate refused it,
661 /// or the handler failed. There is nothing to replay, so recording it would
662 /// hand every future retry the same refusal for a condition that may have
663 /// been fixed, and would make a transient failure permanent. Releasing
664 /// makes the next attempt a fresh one, which is what a caller sending the
665 /// same id again is asking for.
666 ///
667 /// The consequence is that a tool which *ran* and then failed is retried.
668 /// That is the honest answer: this layer cannot tell a tool that had no
669 /// effect from one that charged a card and then failed to answer, and
670 /// guessing "no effect" would silently drop work. A tool that must not
671 /// run twice needs a key of its own — which is the tool's own port, not
672 /// this one.
673 ///
674 /// # What a dropped connection costs
675 ///
676 /// A client whose POST never got an answer retries with the same request
677 /// id. It gets the recorded result, and the tool ran once. This is the
678 /// only reason the store is on `Server` rather than left to the transport:
679 /// the transport sees a dropped socket and knows nothing about what the
680 /// tool did, and only the layer that ran it can answer.
681 async fn call_tool(
682 &self,
683 ctx: &RequestContext,
684 params: &JsonObject,
685 ) -> Result<JsonObject, DispatchError> {
686 Self::require(ctx, Feature::Tools)?;
687 let (name, arguments) = split_call_params(params)?;
688
689 // A request id is only unique within a session — every client numbers
690 // its first request `1` — so the key carries both. Keying on the id
691 // alone would answer one session's call with another's result.
692 //
693 // It also has to carry *what* was asked. A request id is a label for
694 // one request, not a promise that the same id always names the same
695 // operation: JSON-RPC 2.0 lets a peer reuse ids across a sequence, and
696 // the conformance suite's client restarts its counter per scenario
697 // against a connection it never closes. Keyed on the id alone, the
698 // second `tools/call` a session receives is answered with the first
699 // one's recording — a valid `CallToolResult` carrying another tool's
700 // output, which the caller has no way to detect. Repeating the *same*
701 // tool still replays, which is the case a retry needs.
702 let key = IdempotencyKey::for_method_call(
703 &ctx.peer().session(),
704 ctx.request_id(),
705 "tools/call",
706 &name,
707 );
708 let token = match self.server.idempotency().claim(&key)? {
709 // Recorded already. The answer is the recording, and the tool does
710 // not run.
711 Claim::Completed(recorded) => return Ok(recorded),
712 // Someone else holds it. Refusing is what keeps one side effect
713 // from running twice; the holder is a real request.
714 Claim::InFlight => {
715 return Err(DispatchError::DuplicateInFlight(
716 ctx.request_id().to_string(),
717 ));
718 }
719 Claim::Claimed { token } => token,
720 };
721
722 let outcome = self.server.call(ctx, &name, arguments).await;
723 match outcome {
724 Ok(result) => {
725 // The result is both the answer and the recording, and
726 // `complete` takes it by value, so it is cloned out. `JsonObject`
727 // holds its bytes in an `Arc`, so this is a pointer bump and
728 // not a copy of the document.
729 self.server.idempotency().complete(&token, result.clone())?;
730 Ok(result)
731 }
732 Err(error) => {
733 // Release before returning, so the key is claimable again. A
734 // failed record or release is not swallowed: it is reported,
735 // because a key stuck in `Claimed` is invisible to every later
736 // caller and looks exactly like a server that stopped working.
737 self.server.idempotency().abandon(&token)?;
738 Err(error)
739 }
740 }
741 }
742
743 /// The gate every method shares, in one place.
744 ///
745 /// Separate from `RequestContext`'s own private `require`, which returns a
746 /// `GatedError` naming the caller. A method is not a tool, so the refusal
747 /// here is a `DispatchError` and the two types do not merge — a caller
748 /// matching on the distinction can tell "you may not call this method"
749 /// from "this tool needs something you lack".
750 ///
751 /// An associated function rather than a method because it reads nothing
752 /// from the dispatcher: it asks the session, and a gate that could also
753 /// consult the server is a gate with a second opinion.
754 fn require(ctx: &RequestContext, feature: Feature) -> Result<(), DispatchError> {
755 if ctx.peer().allows(feature) {
756 return Ok(());
757 }
758 Err(DispatchError::NotCallable)
759 }
760}
761
762/// Wraps a listing's visible entries under `key`.
763///
764/// The document is assembled from values this module built, so the only way it
765/// could be invalid is a bug here. It falls back to `{}` rather than
766/// propagating because a listing that cannot be encoded should read as empty
767/// to the caller rather than take the whole response down.
768/// Whether a result carries SEP-2549's caching hints.
769///
770/// A second parameter to [`stamp_result`] rather than a flag on every method:
771/// the schema draws the line itself, and this names which side a result is on.
772#[derive(Debug, Clone, Copy, PartialEq, Eq)]
773enum Cacheable {
774 /// `ttlMs` and `cacheScope` belong on this result.
775 Yes,
776 /// They do not — a `CallToolResult` is not something to cache.
777 No,
778}
779
780/// Whether `method` may answer with a SEP-2322 `InputRequiredResult`.
781///
782/// `tasks/result` is the fourth the SEP names; it belongs to the tasks
783/// extension, which this crate does not serve.
784fn accepts_input_round(method: &str) -> bool {
785 matches!(method, "tools/call" | "prompts/get" | "resources/read")
786}
787
788/// Whether SEP-2549's result envelope exists in this revision.
789#[must_use]
790pub const fn result_envelope_defined_in(version: ProtocolVersion) -> bool {
791 matches!(
792 version,
793 ProtocolVersion::V2026_07_28 | ProtocolVersion::Unknown(_)
794 )
795}