use std::sync::Arc;
use crate::application::mcp::initialize::server_capabilities_json;
use crate::application::mcp::input_required::{InputRound, capability_for, is_input_required};
use crate::application::server::builder::{DispatchError, Server};
use crate::application::server::context::{PeerSession, RequestContext};
use crate::domain::feature::{Feature, FeatureProfile};
use crate::domain::idempotency::{Claim, IdempotencyKey};
use crate::domain::ir::JsonObject;
use crate::domain::ir::LogLevel;
use crate::domain::registry::PrimitiveKind;
use crate::domain::{
ProtocolVersion, SupportedVersions, client_method_defined_in, discover_defined_in,
handshake_defined_in,
};
mod negotiate;
mod params;
mod shape;
#[cfg(test)]
mod tests;
use negotiate::negotiate;
use params::{parsed_param, split_call_params, string_param};
use shape::{stamp_result, wrap_listing};
#[derive(Clone)]
pub struct Dispatcher {
server: Arc<Server>,
}
impl core::fmt::Debug for Dispatcher {
fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
f.debug_struct("Dispatcher")
.field("name", &self.server.name())
.field("tool_count", &self.server.tool_count())
.finish_non_exhaustive()
}
}
#[derive(Clone, Copy)]
enum SubscribeOp {
Subscribe,
Unsubscribe,
}
impl Dispatcher {
#[must_use]
pub const fn new(server: Arc<Server>) -> Self {
Self { server }
}
#[must_use]
pub const fn server(&self) -> &Arc<Server> {
&self.server
}
#[must_use]
pub fn state(&self) -> &Arc<crate::application::server::context::StateMap> {
self.server.state()
}
pub fn handshake(
&self,
ctx: &RequestContext,
params: &JsonObject,
) -> Result<(JsonObject, PeerSession), DispatchError> {
let (version, profile) = negotiate(ctx, params)?;
let reply = self.initialize_result(version, profile)?;
Ok((reply, ctx.peer().clone().with_profile(profile)))
}
fn initialize_result(
&self,
version: ProtocolVersion,
profile: FeatureProfile,
) -> Result<JsonObject, DispatchError> {
if !handshake_defined_in(version) {
return Err(DispatchError::UnknownMethod);
}
let result = serde_json::json!({
"protocolVersion": version.as_wire(),
"capabilities": server_capabilities_json(profile),
"serverInfo": {
"name": self.server.name(),
"version": self.server.version(),
},
});
JsonObject::from_value(result).map_err(|_| DispatchError::MalformedInitialize)
}
pub async fn dispatch(
&self,
ctx: &RequestContext,
method: &str,
params: &JsonObject,
) -> Result<JsonObject, DispatchError> {
let version = ctx.peer().negotiated();
let answered = self.route(ctx, method, params).await?;
let stamped = stamp_result(answered, Cacheable::No, version);
Ok(self.identify(stamped, version))
}
fn identify(&self, result: JsonObject, version: ProtocolVersion) -> JsonObject {
if !result_envelope_defined_in(version) {
return result;
}
let Ok(serde_json::Value::Object(mut members)) =
serde_json::from_str::<serde_json::Value>(result.as_str())
else {
return result;
};
let meta = members
.entry("_meta")
.or_insert_with(|| serde_json::Value::Object(serde_json::Map::new()));
if let Some(meta) = meta.as_object_mut() {
meta.insert(
"io.modelcontextprotocol/serverInfo".to_owned(),
serde_json::json!({
"name": self.server.name(),
"version": self.server.version(),
}),
);
}
JsonObject::from_value(serde_json::Value::Object(members)).unwrap_or(result)
}
fn input_round(&self, params: &JsonObject) -> Result<InputRound, DispatchError> {
let responses = match parsed_param(params, "inputResponses")? {
None | Some(serde_json::Value::Null) => None,
Some(serde_json::Value::Object(map)) => Some(map),
Some(_) => return Err(DispatchError::MalformedParams),
};
let state = match parsed_param(params, "requestState")? {
None | Some(serde_json::Value::Null) => None,
Some(serde_json::Value::String(sealed)) => {
let key = self
.server
.request_state_key()
.ok_or(DispatchError::TamperedRequestState)?;
Some(
key.open(&sealed)
.ok_or(DispatchError::TamperedRequestState)?
.to_owned(),
)
}
Some(_) => return Err(DispatchError::MalformedParams),
};
Ok(InputRound {
responses,
state,
key: self.server.request_state_key().cloned(),
})
}
fn check_interim(
ctx: &RequestContext,
method: &str,
result: JsonObject,
) -> Result<JsonObject, DispatchError> {
let Ok(serde_json::Value::Object(members)) =
serde_json::from_str::<serde_json::Value>(result.as_str())
else {
return Ok(result);
};
if !is_input_required(&members) {
return Ok(result);
}
if !accepts_input_round(method) || !result_envelope_defined_in(ctx.peer().negotiated()) {
return Err(DispatchError::HandlerFailed(
"an InputRequiredResult is not defined for this request".to_owned(),
));
}
let mut required: Vec<&'static str> = Vec::new();
if let Some(requests) = members
.get("inputRequests")
.and_then(serde_json::Value::as_object)
{
for request in requests.values() {
let Some(capability) = request
.get("method")
.and_then(serde_json::Value::as_str)
.and_then(capability_for)
else {
continue;
};
let feature = match capability {
"sampling" => Feature::Sampling,
"elicitation" => Feature::Elicitation,
_ => Feature::Roots,
};
if !ctx.peer().allows(feature) && !required.contains(&capability) {
required.push(capability);
}
}
}
if !required.is_empty() {
return Err(DispatchError::MissingClientCapability { required });
}
Ok(result)
}
async fn route(
&self,
ctx: &RequestContext,
method: &str,
params: &JsonObject,
) -> Result<JsonObject, DispatchError> {
let version = ctx.peer().negotiated();
if !client_method_defined_in(version, method) {
return Err(DispatchError::UnknownMethod);
}
let round;
let ctx = if accepts_input_round(method) && result_envelope_defined_in(version) {
round = ctx.clone().with_input_round(self.input_round(params)?);
&round
} else {
ctx
};
let answered = self.route_defined(ctx, method, params).await?;
Self::check_interim(ctx, method, answered)
}
async fn route_defined(
&self,
ctx: &RequestContext,
method: &str,
params: &JsonObject,
) -> Result<JsonObject, DispatchError> {
match method {
"initialize" => self.initialize(ctx, params),
"server/discover" => Self::discover(ctx),
"ping" => Ok(JsonObject::default()),
"tools/list" => self.list(ctx, PrimitiveKind::Tool, "tools"),
"resources/list" => self.list(ctx, PrimitiveKind::Resource, "resources"),
"resources/templates/list" => {
self.list(ctx, PrimitiveKind::ResourceTemplate, "resourceTemplates")
}
"prompts/list" => self.list(ctx, PrimitiveKind::Prompt, "prompts"),
"tools/call" => self.call_tool(ctx, params).await,
"resources/read" => self.read_resource(ctx, params).await,
"prompts/get" => self.get_prompt(ctx, params).await,
"logging/setLevel" => Self::set_log_level(ctx, params),
"completion/complete" => Self::complete(ctx, params),
"resources/subscribe" => self.subscribe(ctx, params, SubscribeOp::Subscribe),
"resources/unsubscribe" => self.subscribe(ctx, params, SubscribeOp::Unsubscribe),
_ => Err(DispatchError::UnknownMethod),
}
}
fn discover(ctx: &RequestContext) -> Result<JsonObject, DispatchError> {
if !discover_defined_in(ctx.peer().negotiated()) {
return Err(DispatchError::UnknownMethod);
}
let result = serde_json::json!({
"resultType": "complete",
"supportedVersions": SupportedVersions::all()
.map_err(|_| DispatchError::UnnegotiableVersion)?
.into_iter()
.rev()
.map(|version| version.as_wire().into_owned())
.collect::<Vec<_>>(),
"ttlMs": 0,
"cacheScope": "private",
"capabilities": server_capabilities_json(ctx.peer().profile()),
});
JsonObject::from_value(result).map_err(|_| DispatchError::MalformedParams)
}
fn subscribe(
&self,
ctx: &RequestContext,
params: &JsonObject,
op: SubscribeOp,
) -> Result<JsonObject, DispatchError> {
Self::require(ctx, Feature::ResourcesSubscribe)?;
let uri = string_param(params, "uri")?;
if self.server.reader(&uri).is_none() {
return Err(DispatchError::UnknownResource);
}
let handle = self.server.handle();
match op {
SubscribeOp::Subscribe => handle.watch_resource(ctx.peer().sink(), &uri, None),
SubscribeOp::Unsubscribe => handle.unwatch_resource(ctx.peer().sink(), &uri),
}
Ok(JsonObject::default())
}
async fn read_resource(
&self,
ctx: &RequestContext,
params: &JsonObject,
) -> Result<JsonObject, DispatchError> {
Self::require(ctx, Feature::Resources)?;
let uri = string_param(params, "uri")?;
let reader = self
.server
.reader(&uri)
.ok_or(DispatchError::UnknownResource)?;
Ok(stamp_result(
reader(ctx.clone(), uri).await?,
Cacheable::Yes,
ctx.peer().negotiated(),
))
}
async fn get_prompt(
&self,
ctx: &RequestContext,
params: &JsonObject,
) -> Result<JsonObject, DispatchError> {
Self::require(ctx, Feature::Prompts)?;
let name = string_param(params, "name")?;
let arguments = match parsed_param(params, "arguments")? {
None | Some(serde_json::Value::Null) => JsonObject::default(),
Some(arguments) => {
JsonObject::from_value(arguments).map_err(|_| DispatchError::MalformedParams)?
}
};
let getter = self
.server
.getter(&name)
.ok_or(DispatchError::UnknownPrompt)?;
getter(ctx.clone(), arguments).await
}
fn set_log_level(
ctx: &RequestContext,
params: &JsonObject,
) -> Result<JsonObject, DispatchError> {
Self::require(ctx, Feature::Logging)?;
let level = string_param(params, "level")?;
let level = level
.parse::<LogLevel>()
.map_err(|_| DispatchError::MalformedParams)?;
ctx.peer().log_gate().set(level);
Ok(JsonObject::default())
}
fn complete(ctx: &RequestContext, _params: &JsonObject) -> Result<JsonObject, DispatchError> {
Self::require(ctx, Feature::Completions)?;
JsonObject::new(r#"{"completion":{"values":[]}}"#)
.map_err(|_| DispatchError::MalformedParams)
}
fn initialize(
&self,
ctx: &RequestContext,
params: &JsonObject,
) -> Result<JsonObject, DispatchError> {
let (version, profile) = negotiate(ctx, params)?;
self.initialize_result(version, profile)
}
fn list(
&self,
ctx: &RequestContext,
kind: PrimitiveKind,
key: &str,
) -> Result<JsonObject, DispatchError> {
Self::require(ctx, kind.feature())?;
let listing =
self.server
.registry()
.list_visible(ctx.peer().auth(), &ctx.peer().profile(), kind);
Ok(stamp_result(
wrap_listing(key, &listing),
Cacheable::Yes,
ctx.peer().negotiated(),
))
}
async fn call_tool(
&self,
ctx: &RequestContext,
params: &JsonObject,
) -> Result<JsonObject, DispatchError> {
Self::require(ctx, Feature::Tools)?;
let (name, arguments) = split_call_params(params)?;
let key = IdempotencyKey::for_method_call(
&ctx.peer().session(),
ctx.request_id(),
"tools/call",
&name,
);
let token = match self.server.idempotency().claim(&key)? {
Claim::Completed(recorded) => return Ok(recorded),
Claim::InFlight => {
return Err(DispatchError::DuplicateInFlight(
ctx.request_id().to_string(),
));
}
Claim::Claimed { token } => token,
};
let outcome = self.server.call(ctx, &name, arguments).await;
match outcome {
Ok(result) => {
self.server.idempotency().complete(&token, result.clone())?;
Ok(result)
}
Err(error) => {
self.server.idempotency().abandon(&token)?;
Err(error)
}
}
}
fn require(ctx: &RequestContext, feature: Feature) -> Result<(), DispatchError> {
if ctx.peer().allows(feature) {
return Ok(());
}
Err(DispatchError::NotCallable)
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum Cacheable {
Yes,
No,
}
fn accepts_input_round(method: &str) -> bool {
matches!(method, "tools/call" | "prompts/get" | "resources/read")
}
#[must_use]
pub const fn result_envelope_defined_in(version: ProtocolVersion) -> bool {
matches!(
version,
ProtocolVersion::V2026_07_28 | ProtocolVersion::Unknown(_)
)
}