use std::{sync::Arc, time::Duration}; use axum::{ Json, Router, extract::State, http::{ HeaderMap, HeaderValue, StatusCode, header::{self, ACCEPT}, }, response::{IntoResponse, Response}, routing::get, }; use mcpaas_registry::{PostgresRegistry, RegistryOperation}; use mcpaas_runtime::{RuntimeError, RuntimeExecutor, RuntimeOperation}; use serde::{Deserialize, Serialize}; use serde_json::{Value, json}; use crate::{ catalog::PublishedToolCatalog, jsonrpc::{ CURRENT_PROTOCOL_VERSION, DEFAULT_PROTOCOL_VERSION, is_notification, is_request, is_response, jsonrpc_error, jsonrpc_result, method_name, negotiated_protocol_version, params, request_id, }, session::{SessionState, SessionStore}, }; const HEADER_MCP_SESSION_ID: &str = "MCP-Session-Id"; const HEADER_MCP_PROTOCOL_VERSION: &str = "MCP-Protocol-Version"; #[derive(Clone)] pub struct AppState { catalog: PublishedToolCatalog, runtime: RuntimeExecutor, sessions: SessionStore, allowed_origins: AllowedOrigins, } #[derive(Clone)] pub struct AllowedOrigins { public_origin: Option, } #[derive(Debug, Serialize, Deserialize)] struct InitializeParams { #[serde(rename = "protocolVersion")] protocol_version: String, } #[derive(Debug, Serialize, Deserialize)] struct ToolCallParams { name: String, #[serde(default)] arguments: Value, } pub fn build_app( registry: PostgresRegistry, refresh_interval: Duration, public_base_url: Option, ) -> Router { let state = Arc::new(AppState { catalog: PublishedToolCatalog::new(registry, refresh_interval), runtime: RuntimeExecutor::new(), sessions: SessionStore::new(), allowed_origins: AllowedOrigins::new(public_base_url), }); Router::new() .route("/health", get(health)) .route("/", get(mcp_get).post(mcp_post).delete(mcp_delete)) .route("/mcp", get(mcp_get).post(mcp_post).delete(mcp_delete)) .with_state(state) } async fn health() -> Json { Json(json!({ "service": "mcp-server", "status": "ok" })) } async fn mcp_get() -> Response { StatusCode::METHOD_NOT_ALLOWED.into_response() } async fn mcp_delete(State(state): State>, headers: HeaderMap) -> Response { match session_id_from_headers(&headers) { Ok(Some(session_id)) => { if state.sessions.delete(&session_id).await { StatusCode::NO_CONTENT.into_response() } else { StatusCode::NOT_FOUND.into_response() } } Ok(None) => StatusCode::BAD_REQUEST.into_response(), Err(status) => status.into_response(), } } async fn mcp_post( State(state): State>, headers: HeaderMap, Json(message): Json, ) -> Response { if let Err(status) = validate_origin(&state.allowed_origins, &headers) { return status.into_response(); } if let Err(status) = validate_accept_header(&headers) { return status.into_response(); } if is_response(&message) || is_notification(&message) && method_name(&message).is_none() { return StatusCode::ACCEPTED.into_response(); } let protocol_version = match protocol_version_from_headers(&headers) { Ok(value) => value, Err(status) => return status.into_response(), }; match method_name(&message) { Some("initialize") if is_request(&message) => handle_initialize(state, &message).await, Some("notifications/initialized") if is_notification(&message) => { handle_initialized_notification(state, &headers).await } Some("ping") if is_request(&message) => { let session = match require_initialized_session(&state, &headers, &message).await { Ok(session) => session, Err(response) => return response, }; json_response( StatusCode::OK, jsonrpc_result( request_id(&message), json!({ "protocolVersion": session.protocol_version }), ), None, Some(&protocol_version), ) } Some("tools/list") if is_request(&message) => { let session = match require_initialized_session(&state, &headers, &message).await { Ok(session) => session, Err(response) => return response, }; match state.catalog.list_tools().await { Ok(tools) => { let definitions = tools.iter().map(tool_definition).collect::>(); json_response( StatusCode::OK, jsonrpc_result(request_id(&message), json!({ "tools": definitions })), None, Some(&session.protocol_version), ) } Err(error) => internal_jsonrpc_error(&message, error), } } Some("tools/call") if is_request(&message) => { let session = match require_initialized_session(&state, &headers, &message).await { Ok(session) => session, Err(response) => return response, }; let tool_call_params: ToolCallParams = match serde_json::from_value(params(&message)) { Ok(value) => value, Err(error) => { return json_response( StatusCode::OK, jsonrpc_error(request_id(&message), -32602, error.to_string()), None, Some(&session.protocol_version), ); } }; let arguments = if tool_call_params.arguments.is_null() { json!({}) } else { tool_call_params.arguments }; match state.catalog.get_tool(&tool_call_params.name).await { Ok(Some(operation)) => { match state .runtime .execute(&RuntimeOperation::from(operation), &arguments) .await { Ok(output) => json_response( StatusCode::OK, jsonrpc_result( request_id(&message), json!({ "content": [ { "type": "text", "text": serde_json::to_string_pretty(&output).unwrap_or_else(|_| "{}".to_owned()) } ], "structuredContent": output, "isError": false }), ), None, Some(&session.protocol_version), ), Err(error) => json_response( StatusCode::OK, jsonrpc_result( request_id(&message), json!({ "content": [ { "type": "text", "text": error.to_string() } ], "structuredContent": { "error": { "code": runtime_error_code(&error), "message": error.to_string() } }, "isError": true }), ), None, Some(&session.protocol_version), ), } } Ok(None) => json_response( StatusCode::OK, jsonrpc_error( request_id(&message), -32602, format!("tool {} was not found", tool_call_params.name), ), None, Some(&session.protocol_version), ), Err(error) => internal_jsonrpc_error(&message, error), } } Some(method) if is_notification(&message) => { let _ = method; StatusCode::ACCEPTED.into_response() } Some(method) => json_response( StatusCode::OK, jsonrpc_error( request_id(&message), -32601, format!("method {method} is not supported"), ), None, Some(&protocol_version), ), None => json_response( StatusCode::BAD_REQUEST, jsonrpc_error(Value::Null, -32600, "invalid JSON-RPC message"), None, Some(&protocol_version), ), } } async fn handle_initialize(state: Arc, message: &Value) -> Response { let initialize_params: InitializeParams = match serde_json::from_value(params(message)) { Ok(value) => value, Err(error) => { return json_response( StatusCode::OK, jsonrpc_error(request_id(message), -32602, error.to_string()), None, Some(DEFAULT_PROTOCOL_VERSION), ); } }; let Some(protocol_version) = negotiated_protocol_version(&initialize_params.protocol_version) else { return json_response( StatusCode::OK, jsonrpc_error( request_id(message), -32602, format!( "unsupported protocol version {}", initialize_params.protocol_version ), ), None, Some(DEFAULT_PROTOCOL_VERSION), ); }; let session_id = state.sessions.create(protocol_version).await; json_response( StatusCode::OK, jsonrpc_result( request_id(message), json!({ "protocolVersion": protocol_version, "capabilities": { "tools": { "listChanged": false } }, "serverInfo": { "name": "rmcp-mcp-server", "version": env!("CARGO_PKG_VERSION") } }), ), Some(&session_id), Some(protocol_version), ) } async fn handle_initialized_notification(state: Arc, headers: &HeaderMap) -> Response { match session_id_from_headers(headers) { Ok(Some(session_id)) => { if state.sessions.mark_initialized(&session_id).await { StatusCode::ACCEPTED.into_response() } else { StatusCode::NOT_FOUND.into_response() } } Ok(None) => StatusCode::BAD_REQUEST.into_response(), Err(status) => status.into_response(), } } async fn require_initialized_session( state: &Arc, headers: &HeaderMap, message: &Value, ) -> Result { let session_id = match session_id_from_headers(headers) { Ok(Some(session_id)) => session_id, Ok(None) => return Err(StatusCode::BAD_REQUEST.into_response()), Err(status) => return Err(status.into_response()), }; let Some(session) = state.sessions.get(&session_id).await else { return Err(StatusCode::NOT_FOUND.into_response()); }; if !session.initialized { return Err(json_response( StatusCode::OK, jsonrpc_error(request_id(message), -32002, "session is not initialized"), None, Some(&session.protocol_version), )); } Ok(session) } fn validate_origin( allowed_origins: &AllowedOrigins, headers: &HeaderMap, ) -> Result<(), StatusCode> { let Some(origin) = headers.get(header::ORIGIN) else { return Ok(()); }; let Ok(origin) = origin.to_str() else { return Err(StatusCode::FORBIDDEN); }; if allowed_origins.is_allowed(origin) { return Ok(()); } Err(StatusCode::FORBIDDEN) } fn validate_accept_header(headers: &HeaderMap) -> Result<(), StatusCode> { let Some(accept) = headers.get(ACCEPT) else { return Err(StatusCode::BAD_REQUEST); }; let Ok(accept) = accept.to_str() else { return Err(StatusCode::BAD_REQUEST); }; let normalized = accept.to_ascii_lowercase(); if normalized.contains("application/json") && normalized.contains("text/event-stream") { return Ok(()); } Err(StatusCode::NOT_ACCEPTABLE) } fn protocol_version_from_headers(headers: &HeaderMap) -> Result { let Some(version) = headers.get(HEADER_MCP_PROTOCOL_VERSION) else { return Ok(DEFAULT_PROTOCOL_VERSION.to_owned()); }; let Ok(version) = version.to_str() else { return Err(StatusCode::BAD_REQUEST); }; if negotiated_protocol_version(version).is_some() { return Ok(version.to_owned()); } Err(StatusCode::BAD_REQUEST) } fn session_id_from_headers(headers: &HeaderMap) -> Result, StatusCode> { let Some(session_id) = headers.get(HEADER_MCP_SESSION_ID) else { return Ok(None); }; let Ok(session_id) = session_id.to_str() else { return Err(StatusCode::BAD_REQUEST); }; Ok(Some(session_id.to_owned())) } fn internal_jsonrpc_error(message: &Value, error: impl std::fmt::Display) -> Response { json_response( StatusCode::INTERNAL_SERVER_ERROR, jsonrpc_error(request_id(message), -32603, error.to_string()), None, Some(DEFAULT_PROTOCOL_VERSION), ) } fn runtime_error_code(error: &RuntimeError) -> &'static str { match error { RuntimeError::Schema(_) => "schema_validation_error", RuntimeError::Mapping(_) => "mapping_error", RuntimeError::GraphqlAdapter(_) => "adapter_execution_error", RuntimeError::GrpcAdapter(_) => "adapter_execution_error", RuntimeError::RestAdapter(_) => "adapter_execution_error", RuntimeError::UnsupportedProtocol { .. } => "unsupported_protocol", RuntimeError::InvalidPreparedRequest { .. } => "runtime_error", } } fn json_response( status: StatusCode, payload: Value, session_id: Option<&str>, protocol_version: Option<&str>, ) -> Response { let mut response = (status, Json(payload)).into_response(); if let Some(session_id) = session_id { response.headers_mut().insert( HEADER_MCP_SESSION_ID, HeaderValue::from_str(session_id) .unwrap_or_else(|_| HeaderValue::from_static("invalid")), ); } if let Some(protocol_version) = protocol_version { response.headers_mut().insert( HEADER_MCP_PROTOCOL_VERSION, HeaderValue::from_str(protocol_version) .unwrap_or_else(|_| HeaderValue::from_static(CURRENT_PROTOCOL_VERSION)), ); } response } fn tool_definition(operation: &RegistryOperation) -> Value { json!({ "name": operation.name, "title": operation.tool_description.title, "description": operation.tool_description.description, "inputSchema": schema_to_json_schema(&operation.input_schema) }) } fn schema_to_json_schema(schema: &mcpaas_schema::Schema) -> Value { match schema.kind { mcpaas_schema::SchemaKind::Object => { let mut properties = serde_json::Map::new(); let mut required = Vec::new(); for (field_name, field_schema) in &schema.fields { properties.insert(field_name.clone(), schema_to_json_schema(field_schema)); if field_schema.required { required.push(Value::String(field_name.clone())); } } json!({ "type": "object", "properties": properties, "required": required }) } mcpaas_schema::SchemaKind::Array => json!({ "type": "array", "items": schema .items .as_deref() .map(schema_to_json_schema) .unwrap_or_else(|| json!({})) }), mcpaas_schema::SchemaKind::String => json!({ "type": "string" }), mcpaas_schema::SchemaKind::Integer => json!({ "type": "integer" }), mcpaas_schema::SchemaKind::Number => json!({ "type": "number" }), mcpaas_schema::SchemaKind::Boolean => json!({ "type": "boolean" }), mcpaas_schema::SchemaKind::Enum => json!({ "type": "string", "enum": schema.enum_values }), mcpaas_schema::SchemaKind::Null => json!({ "type": "null" }), mcpaas_schema::SchemaKind::Oneof => json!({ "anyOf": schema.variants.iter().map(schema_to_json_schema).collect::>() }), } } impl AllowedOrigins { fn new(public_base_url: Option) -> Self { Self { public_origin: public_base_url.and_then(|value| extract_origin(&value)), } } fn is_allowed(&self, origin: &str) -> bool { if origin.starts_with("http://localhost") || origin.starts_with("http://127.0.0.1") || origin.starts_with("https://localhost") || origin.starts_with("https://127.0.0.1") { return true; } match &self.public_origin { Some(public_origin) => public_origin == origin, None => false, } } } fn extract_origin(url: &str) -> Option { let mut parts = url.split('/'); let scheme = parts.next()?; let empty = parts.next()?; let authority = parts.next()?; if empty.is_empty() { return Some(format!("{scheme}//{authority}")); } None }