soma_mcp/
rmcp_server.rs

1//! `SomaRmcpServer` — the `ServerHandler` implementation.
2//!
3//! This is the adapter between the rmcp crate and your application. It:
4//!   - Advertises tools, resources, and prompts to MCP clients
5//!   - Enforces auth scopes on every call
6//!   - Delegates business logic to `tools.rs` → `app.rs` → `soma-client`
7//!
8//! **Customize**: rename `SomaRmcpServer`. Update action metadata in
9//! `src/actions.rs` to keep schemas, scope rules, and dispatch in sync.
10
11use std::time::Instant;
12
13use rmcp::{
14    model::{
15        CallToolRequestParams, CallToolResult, GetPromptRequestParams, GetPromptResult,
16        Implementation, ListPromptsResult, ListResourceTemplatesResult, ListResourcesResult,
17        ListToolsResult, PaginatedRequestParams, ReadResourceRequestParams, ReadResourceResult,
18        Resource, ResourceContents, ResourceTemplate, ServerCapabilities, ServerInfo, Tool,
19    },
20    service::{Peer, RequestContext},
21    ErrorData, RoleServer, ServerHandler,
22};
23use rmcp_traces::TraceTrust;
24use serde_json::{Map, Value};
25
26use soma_application::{ApplicationError, ExecutionContext, ReadResourceRequest, ResourceContent};
27use soma_domain::{token_limit::MAX_RESPONSE_BYTES, TraceContext};
28use soma_mcp_server::{
29    conformance,
30    response_paging::{
31        response_page_request, strip_response_page_params, tool_result_from_cached_page,
32        tool_result_from_json, ResponsePagingOptions,
33    },
34};
35use soma_provider_core::ProviderResource;
36
37use super::{
38    gateway_proxy, prompts,
39    protocol_errors::{application_error_payload, tool_error_result, unknown_tool_error},
40    rmcp_auth::{
41        principal, protected_route_scope, protected_scope_allows_service, require_auth_context,
42        AuthContext,
43    },
44    schemas::tool_definitions_for_catalogs as tool_definitions,
45    state::McpState,
46    tools::execute_tool,
47    trace_resolution, ACTION_DISCRIMINATOR_FIELD,
48};
49
50macro_rules! trace_summary_event {
51    ($level:ident, $trace_resolution:expr, $trace_context_conflict:expr, $message:literal, $($field:tt)*) => {
52        tracing::$level!(
53            $($field)*
54            trace_id_prefix = ?$trace_resolution.summary.trace_id_prefix(),
55            span_id_prefix = ?$trace_resolution.summary.span_id_prefix(),
56            trace_sampled = ?$trace_resolution.summary.sampled(),
57            trace_trust = ?$trace_resolution.summary.trust(),
58            has_tracestate = $trace_resolution.summary.has_tracestate(),
59            baggage_member_count = $trace_resolution.summary.baggage_member_count(),
60            sensitive_baggage_member_count = $trace_resolution.summary.sensitive_baggage_member_count(),
61            trace_invalid_count = $trace_resolution.summary.invalid_count(),
62            trace_invalid_reasons = ?$trace_resolution.summary.invalid_reasons(),
63            http_trace_headers_present = $trace_resolution.http_trace_headers_present,
64            trace_context_conflict = $trace_context_conflict,
65            $message
66        );
67    };
68}
69
70// ── server ────────────────────────────────────────────────────────────────────
71
72#[derive(Clone)]
73pub struct SomaRmcpServer {
74    state: McpState,
75}
76
77pub fn rmcp_server(state: McpState) -> SomaRmcpServer {
78    SomaRmcpServer { state }
79}
80
81impl ServerHandler for SomaRmcpServer {
82    // ── tools ─────────────────────────────────────────────────────────────────
83
84    async fn list_tools(
85        &self,
86        _request: Option<PaginatedRequestParams>,
87        context: RequestContext<RoleServer>,
88    ) -> Result<ListToolsResult, ErrorData> {
89        let auth = require_auth_context(&self.state, &context)?;
90        let route_scope = protected_route_scope(&context);
91        let execution_context = execution_context(&self.state, &context, auth);
92        let soma_allowed = protected_scope_allows_service(route_scope, "soma");
93        refresh_file_providers(&self.state)?;
94        let mut tools = if soma_allowed {
95            rmcp_tool_definitions(&self.state)?
96        } else {
97            Vec::new()
98        };
99        tools.extend(
100            gateway_proxy::list_tools_for_subject_and_scope(
101                self.state.application(),
102                route_scope,
103                &execution_context,
104            )
105            .await?,
106        );
107        if soma_allowed && self.state.config().conformance_fixtures {
108            tools.extend(conformance::tool_definitions());
109        }
110        tracing::debug!(tool_count = tools.len(), "MCP tools listed");
111        Ok(ListToolsResult {
112            tools,
113            ..Default::default()
114        })
115    }
116
117    async fn call_tool(
118        &self,
119        request: CallToolRequestParams,
120        context: RequestContext<RoleServer>,
121    ) -> Result<CallToolResult, ErrorData> {
122        let tool_name = request.name.to_string();
123
124        // Extract action before scope check so a missing action returns the
125        // more useful "action is required" validation error, not DENY_SCOPE.
126        let action_opt: Option<String> = request
127            .arguments
128            .as_ref()
129            .and_then(|m| m.get("action"))
130            .and_then(Value::as_str)
131            .map(ToOwned::to_owned);
132
133        let response_page = match response_page_request(request.arguments.as_ref()) {
134            Ok(response_page) => response_page,
135            Err(error) => {
136                tracing::warn!("MCP tool rejected response paging params");
137                return Err(error);
138            }
139        };
140        let auth = match require_auth_context(&self.state, &context) {
141            Ok(auth) => auth,
142            Err(error) => {
143                tracing::warn!("MCP tool rejected auth context");
144                return Err(error);
145            }
146        };
147        let trace_resolution = trace_resolution_for_call(&self.state, &context);
148        let trace_context_conflict = trace_resolution.http_trace_headers_present
149            && trace_resolution::meta_has_any_trace_key(&context.meta);
150        let route_scope = protected_route_scope(&context);
151        let execution_context =
152            execution_context_with_trace(&self.state, auth, trace_resolution.trace_context.clone());
153        let soma_allowed = protected_scope_allows_service(route_scope, "soma");
154        if soma_allowed && self.state.config().conformance_fixtures {
155            if let Some(result) = conformance::call_tool(&tool_name) {
156                return Ok(result);
157            }
158        }
159        if tool_name != "soma" {
160            if let Some(result) = gateway_proxy::call_tool_for_subject_and_scope(
161                self.state.application(),
162                &tool_name,
163                request.arguments.clone(),
164                route_scope,
165                &execution_context,
166            )
167            .await
168            {
169                if result.is_error == Some(true) {
170                    trace_summary_event!(
171                        warn,
172                        trace_resolution,
173                        trace_context_conflict,
174                        "MCP gateway tool execution failed",
175                        tool = %tool_name,
176                        action = action_opt.as_deref().unwrap_or_default(),
177                    );
178                } else {
179                    trace_summary_event!(
180                        info,
181                        trace_resolution,
182                        trace_context_conflict,
183                        "MCP gateway tool execution completed",
184                        tool = %tool_name,
185                        action = action_opt.as_deref().unwrap_or_default(),
186                    );
187                }
188                return Ok(result);
189            }
190            trace_summary_event!(
191                warn,
192                trace_resolution,
193                trace_context_conflict,
194                "MCP tool rejected unknown tool",
195                tool = %tool_name,
196                action = action_opt.as_deref().unwrap_or_default(),
197            );
198            return Err(unknown_tool_error(&tool_name));
199        }
200        if !soma_allowed {
201            return Err(unknown_tool_error(&tool_name));
202        }
203        let action: String = action_opt.unwrap_or_default();
204        if let Some(cursor) = response_page.cursor().map(str::to_owned) {
205            trace_summary_event!(
206                info,
207                trace_resolution,
208                trace_context_conflict,
209                "MCP tool returned cached response page",
210                tool = %tool_name,
211                action = empty_action_as_none(&action).unwrap_or_default(),
212            );
213            return tool_result_from_cached_page(
214                self.state.response_pages(),
215                &cursor,
216                response_page,
217                response_paging_options(),
218                &tool_name,
219                empty_action_as_none(&action),
220            );
221        }
222
223        let mut arguments = request
224            .arguments
225            .map(Value::Object)
226            .unwrap_or_else(|| Value::Object(Map::new()));
227        strip_response_page_params(&mut arguments);
228        let continuation_args = arguments.as_object().cloned();
229
230        // Clone the peer so we can pass it to the tool dispatcher.
231        // The peer is needed for elicitation (asking the client for user input).
232        let peer: Peer<RoleServer> = context.peer.clone();
233        let started = Instant::now();
234        trace_summary_event!(
235            info,
236            trace_resolution,
237            trace_context_conflict,
238            "MCP tool execution started",
239            tool = %tool_name,
240            action = %action,
241        );
242
243        match execute_tool(&self.state, &tool_name, arguments, &peer, execution_context).await {
244            Ok(result) => {
245                trace_summary_event!(
246                    info,
247                    trace_resolution,
248                    trace_context_conflict,
249                    "MCP tool execution completed",
250                    tool = %tool_name,
251                    action = %action,
252                    elapsed_ms = started.elapsed().as_millis(),
253                );
254                tool_result_from_json(
255                    result,
256                    self.state.response_pages(),
257                    response_page,
258                    response_paging_options(),
259                    &tool_name,
260                    empty_action_as_none(&action),
261                    continuation_args.as_ref(),
262                )
263            }
264            Err(error) => {
265                let application_error = error.downcast_ref::<ApplicationError>();
266                if application_error.is_some_and(ApplicationError::is_validation) {
267                    trace_summary_event!(
268                        warn,
269                        trace_resolution,
270                        trace_context_conflict,
271                        "MCP tool rejected invalid params",
272                        tool = %tool_name,
273                        action = %action,
274                        elapsed_ms = started.elapsed().as_millis(),
275                    );
276                } else {
277                    trace_summary_event!(
278                        error,
279                        trace_resolution,
280                        trace_context_conflict,
281                        "MCP tool execution failed",
282                        tool = %tool_name,
283                        action = %action,
284                        elapsed_ms = started.elapsed().as_millis(),
285                        service_error_kind = application_error
286                            .and_then(ApplicationError::service_error_kind),
287                        private_diagnostics = application_error
288                            .and_then(ApplicationError::private_diagnostics),
289                        error = %error,
290                    );
291                }
292                tool_error_result(application_error_payload(
293                    &error,
294                    &tool_name,
295                    empty_action_as_none(&action),
296                ))
297            }
298        }
299    }
300
301    // ── resources ─────────────────────────────────────────────────────────────
302
303    async fn list_resources(
304        &self,
305        _request: Option<PaginatedRequestParams>,
306        context: RequestContext<RoleServer>,
307    ) -> Result<ListResourcesResult, ErrorData> {
308        let auth = require_auth_context(&self.state, &context)?;
309        let route_scope = protected_route_scope(&context);
310        let execution_context = execution_context(&self.state, &context, auth);
311        refresh_file_providers(&self.state)?;
312        let soma_allowed = protected_scope_allows_service(route_scope, "soma");
313        let mut resources = if soma_allowed {
314            let mut resources = vec![schema_resource()];
315            resources.extend(
316                self.state
317                    .application()
318                    .list_resources()
319                    .iter()
320                    .map(rmcp_resource_from_catalog_resource),
321            );
322            resources
323        } else {
324            Vec::new()
325        };
326        resources.extend(
327            gateway_proxy::list_resources_for_subject_and_scope(
328                self.state.application(),
329                route_scope,
330                &execution_context,
331            )
332            .await?,
333        );
334        if soma_allowed && self.state.config().conformance_fixtures {
335            resources.extend(conformance::resources());
336        }
337        Ok(ListResourcesResult {
338            resources,
339            ..Default::default()
340        })
341    }
342
343    async fn list_resource_templates(
344        &self,
345        _request: Option<PaginatedRequestParams>,
346        context: RequestContext<RoleServer>,
347    ) -> Result<ListResourceTemplatesResult, ErrorData> {
348        require_auth_context(&self.state, &context)?;
349        refresh_file_providers(&self.state)?;
350        let mut resource_templates: Vec<ResourceTemplate> = self
351            .state
352            .application()
353            .list_resource_templates()
354            .into_iter()
355            .map(|template| {
356                let mut built = ResourceTemplate::new(template.uri_template, template.name)
357                    .with_description(template.description.clone());
358                if let Some(mime_type) = template.mime_type {
359                    built = built.with_mime_type(mime_type);
360                }
361                built
362            })
363            .collect();
364        if self.state.config().conformance_fixtures {
365            resource_templates.extend(conformance::resource_templates());
366        }
367        Ok(ListResourceTemplatesResult {
368            resource_templates,
369            ..Default::default()
370        })
371    }
372
373    async fn read_resource(
374        &self,
375        request: ReadResourceRequestParams,
376        context: RequestContext<RoleServer>,
377    ) -> Result<ReadResourceResult, ErrorData> {
378        let auth = require_auth_context(&self.state, &context)?;
379        let route_scope = protected_route_scope(&context);
380        let execution_context = execution_context(&self.state, &context, auth);
381        let soma_allowed = protected_scope_allows_service(route_scope, "soma");
382        if soma_allowed && self.state.config().conformance_fixtures {
383            if let Some(result) = conformance::read_resource(&request.uri) {
384                return Ok(result);
385            }
386        }
387        if let Some(result) = gateway_proxy::read_resource_for_subject_and_scope(
388            self.state.application(),
389            &request.uri,
390            route_scope,
391            &execution_context,
392        )
393        .await?
394        {
395            return Ok(result);
396        }
397        if !soma_allowed {
398            return Err(ErrorData::invalid_params(
399                format!("unknown resource: {}", request.uri),
400                None,
401            ));
402        }
403        refresh_file_providers(&self.state)?;
404        if request.uri == SCHEMA_RESOURCE_URI {
405            let schema = tool_definitions_for_state(&self.state);
406            let text = serde_json::to_string_pretty(&schema).map_err(|e| {
407                ErrorData::internal_error(format!("serialization error: {e}"), None)
408            })?;
409            return Ok(ReadResourceResult::new(vec![ResourceContents::text(
410                text,
411                SCHEMA_RESOURCE_URI,
412            )
413            .with_mime_type("application/json")]));
414        }
415
416        let output = self
417            .state
418            .application()
419            .read_resource(
420                ReadResourceRequest {
421                    uri: request.uri.clone(),
422                },
423                execution_context,
424            )
425            .await
426            .map_err(|error| resource_read_error(&request.uri, &error))?;
427        Ok(ReadResourceResult::new(vec![
428            resource_contents_from_output(&request.uri, output),
429        ]))
430    }
431
432    // ── prompts ───────────────────────────────────────────────────────────────
433
434    async fn list_prompts(
435        &self,
436        _request: Option<PaginatedRequestParams>,
437        context: RequestContext<RoleServer>,
438    ) -> Result<ListPromptsResult, ErrorData> {
439        let auth = require_auth_context(&self.state, &context)?;
440        let route_scope = protected_route_scope(&context);
441        let execution_context = execution_context(&self.state, &context, auth);
442        let soma_allowed = protected_scope_allows_service(route_scope, "soma");
443        let mut result = if soma_allowed {
444            prompts::list_prompts()
445        } else {
446            ListPromptsResult::default()
447        };
448        refresh_file_providers(&self.state)?;
449        if soma_allowed {
450            result.prompts.extend(prompts::provider_prompts(
451                &self.state.application().list_prompts(),
452            ));
453        }
454        result.prompts.extend(
455            gateway_proxy::list_prompts_for_subject_and_scope(
456                self.state.application(),
457                route_scope,
458                &execution_context,
459            )
460            .await?,
461        );
462        if soma_allowed && self.state.config().conformance_fixtures {
463            result.prompts.extend(conformance::prompts());
464        }
465        Ok(result)
466    }
467
468    async fn get_prompt(
469        &self,
470        request: GetPromptRequestParams,
471        context: RequestContext<RoleServer>,
472    ) -> Result<GetPromptResult, ErrorData> {
473        let auth = require_auth_context(&self.state, &context)?;
474        let route_scope = protected_route_scope(&context);
475        let execution_context = execution_context(&self.state, &context, auth);
476        let soma_allowed = protected_scope_allows_service(route_scope, "soma");
477        if soma_allowed && self.state.config().conformance_fixtures {
478            if let Some(result) = conformance::get_prompt(request.clone()) {
479                return Ok(result);
480            }
481        }
482        if let Some(result) = gateway_proxy::get_prompt_for_subject_and_scope(
483            self.state.application(),
484            &request.name,
485            request.arguments.clone(),
486            route_scope,
487            &execution_context,
488        )
489        .await?
490        {
491            return Ok(result);
492        }
493        if !soma_allowed {
494            return Err(ErrorData::invalid_params(
495                format!("unknown prompt: {}", request.name),
496                None,
497            ));
498        }
499        refresh_file_providers(&self.state)?;
500        match self
501            .state
502            .application()
503            .get_prompt(&request.name, &execution_context)
504        {
505            Ok(prompt) => return Ok(prompts::provider_prompt_result(prompt)),
506            Err(error) if error.code == "insufficient_scope" => {
507                return Err(ErrorData::invalid_request(
508                    format!("forbidden: {}", error.message),
509                    None,
510                ));
511            }
512            Err(error) if error.code == "prompt_not_found" => {}
513            Err(error) => {
514                return Err(ErrorData::internal_error(error.message, None));
515            }
516        }
517        prompts::get_prompt(request).map_err(|e| ErrorData::invalid_params(e.to_string(), None))
518    }
519
520    // ── server info ───────────────────────────────────────────────────────────
521
522    fn get_info(&self) -> ServerInfo {
523        ServerInfo::new(
524            ServerCapabilities::builder()
525                .enable_tools()
526                .enable_resources()
527                .enable_prompts()
528                .build(),
529        )
530        .with_server_info(Implementation::new(
531            self.state.config().server_name.clone(),
532            env!("CARGO_PKG_VERSION"),
533        ))
534        .with_instructions(SERVER_INSTRUCTIONS)
535    }
536}
537
538fn response_paging_options() -> ResponsePagingOptions {
539    ResponsePagingOptions {
540        max_response_bytes: MAX_RESPONSE_BYTES,
541        action_discriminator_field: ACTION_DISCRIMINATOR_FIELD,
542    }
543}
544
545const SERVER_INSTRUCTIONS: &str = "\
546Soma is a batteries-included RMCP runtime for shipping provider-backed MCP servers. \
547It exposes one action-dispatched `soma` tool plus first-class MCP prompt and resource surfaces. \
548Homepage: https://soma.dinglebear.ai. Repository: https://github.com/dinglebear-ai/soma. \
549Node package: soma-rmcp. Binary: soma. \
550Config home: ~/.soma or SOMA_HOME. License: MIT. Author: dinglebear.ai. \
551Use drop-in providers to add tools, prompts, and resources without rewriting transport, auth, \
552schema, paging, config, Docker, plugin, or release plumbing. A new server comes online by adding \
553provider files under providers/tools, providers/prompts, providers/resources, or another configured \
554provider source. Clients should discover `soma://schema/mcp-tool` before invoking actions, call \
555`status` or `help` to inspect available providers, and send JSON action arguments matching the \
556advertised schema. Responses are structured JSON; large payloads may be paged through Soma's \
557resource paging flow.";
558
559// ── resource definitions ──────────────────────────────────────────────────────
560
561/// URI for the schema resource. **Customize**: change `soma` to your service name.
562const SCHEMA_RESOURCE_URI: &str = "soma://schema/mcp-tool";
563
564fn schema_resource() -> Resource {
565    Resource::new(SCHEMA_RESOURCE_URI, "soma tool schema")
566        .with_description("JSON schema for the Soma MCP tool and its action-based parameters")
567        .with_mime_type("application/json")
568}
569
570fn rmcp_resource_from_catalog_resource(resource: &ProviderResource) -> Resource {
571    let mut built = Resource::new(resource.uri_template.clone(), resource.name.clone())
572        .with_description(resource.description.clone());
573    if let Some(mime_type) = &resource.mime_type {
574        built = built.with_mime_type(mime_type.clone());
575    }
576    built
577}
578
579fn resource_contents_from_output(uri: &str, output: ResourceContent) -> ResourceContents {
580    match output {
581        ResourceContent::Text { text, mime_type } => {
582            let mut contents = ResourceContents::text(text, uri);
583            if let Some(mime_type) = mime_type {
584                contents = contents.with_mime_type(mime_type);
585            }
586            contents
587        }
588        ResourceContent::Blob {
589            blob_base64,
590            mime_type,
591        } => {
592            let mut contents = ResourceContents::blob(blob_base64, uri);
593            if let Some(mime_type) = mime_type {
594                contents = contents.with_mime_type(mime_type);
595            }
596            contents
597        }
598    }
599}
600
601/// Maps an application resource failure to the protocol-level
602/// `ErrorData` MCP `resources/read` expects — there is no structured
603/// tool-result-style "isError" channel for resource reads the way
604/// `call_tool` has, so every failure kind maps to `ErrorData`.
605fn resource_read_error(uri: &str, error: &ApplicationError) -> ErrorData {
606    match error.code.as_str() {
607        "unknown_resource" => ErrorData::invalid_params(format!("unknown resource: {uri}"), None),
608        "insufficient_scope" => {
609            ErrorData::invalid_request(format!("forbidden: {}", error.message), None)
610        }
611        _ => ErrorData::internal_error(error.message.to_string(), None),
612    }
613}
614
615// ── tool definition conversion ────────────────────────────────────────────────
616
617fn rmcp_tool_definitions(state: &McpState) -> Result<Vec<Tool>, ErrorData> {
618    tool_definitions_for_state(state)
619        .into_iter()
620        .map(rmcp_tool_from_json)
621        .collect()
622}
623
624fn refresh_file_providers(state: &McpState) -> Result<(), ErrorData> {
625    state
626        .application()
627        .refresh_providers_in_place()
628        .map_err(|error| ErrorData::internal_error(error.to_string(), None))
629}
630
631fn tool_definitions_for_state(state: &McpState) -> Vec<Value> {
632    let snapshot = state.application().catalog_snapshot();
633    tool_definitions(&snapshot.catalogs)
634}
635
636fn rmcp_tool_from_json(value: Value) -> Result<Tool, ErrorData> {
637    soma_mcp_server::protocol::tool_from_json_definition(value)
638}
639
640fn empty_action_as_none(action: &str) -> Option<&str> {
641    if action.is_empty() {
642        None
643    } else {
644        Some(action)
645    }
646}
647
648fn execution_context(
649    state: &McpState,
650    request: &RequestContext<RoleServer>,
651    auth: Option<&AuthContext>,
652) -> ExecutionContext {
653    state.execution_context(
654        Some(principal(auth)),
655        trace_context_from_meta(&request.meta),
656    )
657}
658
659fn execution_context_with_trace(
660    state: &McpState,
661    auth: Option<&AuthContext>,
662    trace: Option<TraceContext>,
663) -> ExecutionContext {
664    state.execution_context(Some(principal(auth)), trace)
665}
666
667/// Resolve trace metadata for one authenticated `call_tool` invocation. `Off`
668/// mode returns without ever touching `RequestContext.extensions`.
669fn trace_resolution_for_call(
670    state: &McpState,
671    context: &RequestContext<RoleServer>,
672) -> trace_resolution::TraceResolution {
673    let mode = state.config().trace_headers;
674    if mode == soma_config::TraceHeaderMode::Off {
675        return trace_resolution::TraceResolution::from_meta_only(&context.meta);
676    }
677    let headers = context
678        .extensions
679        .get::<http::request::Parts>()
680        .map(|parts| &parts.headers);
681    trace_resolution::resolve_trace_resolution(mode, &context.meta, headers)
682}
683
684fn trace_context_from_meta(meta: &rmcp::model::Meta) -> Option<TraceContext> {
685    let fields = soma_mcp_server::trace::raw_trace_fields_from_meta(meta, TraceTrust::Untrusted)?;
686    Some(TraceContext {
687        traceparent: fields.traceparent,
688        tracestate: fields.tracestate,
689    })
690}
691
692#[cfg(test)]
693#[path = "rmcp_server_tests.rs"]
694mod tests;
soma_mcp/rmcp_server.rs

soma_mcp/
rmcp_server.rs
