merge: feat/mcp-server

2026-03-25 20:12:24 +03:00
parent a776ae4a73 752288ca70
commit 6296b04105
11 changed files with 1031 additions and 156 deletions
@@ -5,6 +5,7 @@ MCPAAS_DATABASE_URL=postgres://rmcp:change-me@postgres:5432/rmcp
 MCPAAS_STORAGE_ROOT=/var/lib/rmcp/storage
 MCPAAS_ADMIN_BIND=0.0.0.0:3001
 MCPAAS_MCP_BIND=0.0.0.0:3002
+MCPAAS_MCP_REFRESH_MS=5000
 MCPAAS_LOG_LEVEL=info
 MCPAAS_SECRET_PROVIDER=env
 MCPAAS_PUBLIC_BASE_URL=https://rmcp.example.com
@@ -850,11 +850,13 @@ dependencies = [
 "mcpaas-runtime",
 "mcpaas-schema",
 "reqwest",
+ "serde",
 "serde_json",
 "sqlx",
 "tokio",
 "tracing",
 "tracing-subscriber",
+ "uuid",
 ]

 [[package]]
@@ -2,21 +2,21 @@

 ## Current

-### `feat/ui-v1`
+### `feat/mcp-server`

 Status: completed

 DoD:

- UI covers operation list, create, test, sample upload, draft generation and publish
- UI works against `admin-api` without mock backend
- critical form helpers are covered by frontend tests
- production UI build works in Docker container
- CI checks UI build and tests together with Rust workspace
+- `mcp-server` exposes a single MCP HTTP endpoint
+- initialize and initialized lifecycle work with session handling
+- tools are listed and called through MCP JSON-RPC methods
+- published tool catalog refreshes without service restart
+- MCP integration tests cover initialize, list, call and refresh flow

 ## Next

- `feat/mcp-server`
+- `feat/graphql-support`

 ## Backlog

@@ -9,10 +9,13 @@ version.workspace = true
 axum.workspace = true
 mcpaas-registry = { path = "../../crates/mcpaas-registry" }
 mcpaas-runtime = { path = "../../crates/mcpaas-runtime" }
+mcpaas-schema = { path = "../../crates/mcpaas-schema" }
+serde.workspace = true
 serde_json.workspace = true
-tokio.workspace = true
+tokio = { workspace = true, features = ["sync"] }
 tracing.workspace = true
 tracing-subscriber.workspace = true
+uuid.workspace = true

 [dev-dependencies]
 mcpaas-core = { path = "../../crates/mcpaas-core" }
@@ -0,0 +1,558 @@
+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<String>,
+}
+
+#[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<String>,
+) -> 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<Value> {
+    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<Arc<AppState>>, 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<Arc<AppState>>,
+    headers: HeaderMap,
+    Json(message): Json<Value>,
+) -> 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::<Vec<_>>();
+
+                    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<AppState>, 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<AppState>, 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<AppState>,
+    headers: &HeaderMap,
+    message: &Value,
+) -> Result<SessionState, Response> {
+    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<String, StatusCode> {
+    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<Option<String>, 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::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::<Vec<_>>()
+        }),
+    }
+}
+
+impl AllowedOrigins {
+    fn new(public_base_url: Option<String>) -> 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<String> {
+    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
+}
@@ -0,0 +1,76 @@
+use std::{
+    collections::HashMap,
+    sync::Arc,
+    time::{Duration, Instant},
+};
+
+use mcpaas_registry::{PostgresRegistry, RegistryError, RegistryOperation};
+use tokio::sync::RwLock;
+
+#[derive(Clone)]
+pub struct PublishedToolCatalog {
+    registry: PostgresRegistry,
+    refresh_interval: Duration,
+    cached: Arc<RwLock<CachedCatalog>>,
+}
+
+#[derive(Default)]
+struct CachedCatalog {
+    loaded_at: Option<Instant>,
+    tools: Vec<RegistryOperation>,
+    tools_by_name: HashMap<String, RegistryOperation>,
+}
+
+impl PublishedToolCatalog {
+    pub fn new(registry: PostgresRegistry, refresh_interval: Duration) -> Self {
+        Self {
+            registry,
+            refresh_interval,
+            cached: Arc::new(RwLock::new(CachedCatalog::default())),
+        }
+    }
+
+    pub async fn list_tools(&self) -> Result<Vec<RegistryOperation>, RegistryError> {
+        self.refresh_if_stale().await?;
+        let guard = self.cached.read().await;
+        Ok(guard.tools.clone())
+    }
+
+    pub async fn get_tool(
+        &self,
+        tool_name: &str,
+    ) -> Result<Option<RegistryOperation>, RegistryError> {
+        self.refresh_if_stale().await?;
+        let guard = self.cached.read().await;
+        Ok(guard.tools_by_name.get(tool_name).cloned())
+    }
+
+    async fn refresh_if_stale(&self) -> Result<(), RegistryError> {
+        let should_refresh = {
+            let guard = self.cached.read().await;
+
+            match guard.loaded_at {
+                Some(loaded_at) => loaded_at.elapsed() >= self.refresh_interval,
+                None => true,
+            }
+        };
+
+        if !should_refresh {
+            return Ok(());
+        }
+
+        let tools = self.registry.list_published_operations().await?;
+        let tools_by_name = tools
+            .iter()
+            .cloned()
+            .map(|operation| (operation.name.clone(), operation))
+            .collect::<HashMap<_, _>>();
+        let mut guard = self.cached.write().await;
+
+        guard.loaded_at = Some(Instant::now());
+        guard.tools = tools;
+        guard.tools_by_name = tools_by_name;
+
+        Ok(())
+    }
+}
@@ -0,0 +1,63 @@
+use serde_json::{Value, json};
+
+pub const CURRENT_PROTOCOL_VERSION: &str = "2025-11-25";
+pub const DEFAULT_PROTOCOL_VERSION: &str = "2025-03-26";
+pub const SUPPORTED_PROTOCOL_VERSIONS: &[&str] = &[
+    DEFAULT_PROTOCOL_VERSION,
+    "2025-06-18",
+    CURRENT_PROTOCOL_VERSION,
+];
+
+pub fn request_id(message: &Value) -> Value {
+    message.get("id").cloned().unwrap_or(Value::Null)
+}
+
+pub fn method_name(message: &Value) -> Option<&str> {
+    message.get("method").and_then(Value::as_str)
+}
+
+pub fn params(message: &Value) -> Value {
+    message
+        .get("params")
+        .cloned()
+        .unwrap_or_else(|| Value::Object(Default::default()))
+}
+
+pub fn is_notification(message: &Value) -> bool {
+    method_name(message).is_some() && message.get("id").is_none()
+}
+
+pub fn is_request(message: &Value) -> bool {
+    method_name(message).is_some() && message.get("id").is_some()
+}
+
+pub fn is_response(message: &Value) -> bool {
+    method_name(message).is_none()
+        && (message.get("result").is_some() || message.get("error").is_some())
+}
+
+pub fn jsonrpc_result(id: Value, result: Value) -> Value {
+    json!({
+        "jsonrpc": "2.0",
+        "id": id,
+        "result": result
+    })
+}
+
+pub fn jsonrpc_error(id: Value, code: i64, message: impl Into<String>) -> Value {
+    json!({
+        "jsonrpc": "2.0",
+        "id": id,
+        "error": {
+            "code": code,
+            "message": message.into()
+        }
+    })
+}
+
+pub fn negotiated_protocol_version(requested: &str) -> Option<&'static str> {
+    SUPPORTED_PROTOCOL_VERSIONS
+        .iter()
+        .copied()
+        .find(|version| *version == requested)
+}
@@ -1,22 +1,15 @@
-use std::{env, net::SocketAddr, sync::Arc};
+mod app;
+mod catalog;
+mod jsonrpc;
+mod session;
+
+use std::{env, net::SocketAddr, time::Duration};

-use axum::{
-    Json, Router,
-    extract::{Path, State},
-    http::StatusCode,
-    routing::{get, post},
-};
 use mcpaas_registry::PostgresRegistry;
-use mcpaas_runtime::{RuntimeExecutor, RuntimeOperation};
-use serde_json::{Value, json};
 use tokio::net::TcpListener;
 use tracing::info;

-#[derive(Clone)]
-struct AppState {
-    registry: PostgresRegistry,
-    runtime: RuntimeExecutor,
-}
+use crate::app::build_app;

 #[tokio::main]
 async fn main() -> Result<(), Box<dyn std::error::Error>> {
@@ -29,13 +22,15 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {

    let database_url = env::var("MCPAAS_DATABASE_URL")?;
    let bind_addr = env::var("MCPAAS_MCP_BIND").unwrap_or_else(|_| "0.0.0.0:3002".into());
+    let public_base_url = env::var("MCPAAS_PUBLIC_BASE_URL").ok();
+    let refresh_interval = env::var("MCPAAS_MCP_REFRESH_MS")
+        .ok()
+        .and_then(|value| value.parse::<u64>().ok())
+        .map(Duration::from_millis)
+        .unwrap_or_else(|| Duration::from_secs(5));
    let socket_addr: SocketAddr = bind_addr.parse()?;
    let registry = PostgresRegistry::connect(&database_url).await?;
-    let state = Arc::new(AppState {
-        registry,
-        runtime: RuntimeExecutor::new(),
-    });
-    let app = build_app(state);
+    let app = build_app(registry, refresh_interval, public_base_url);
    let listener = TcpListener::bind(socket_addr).await?;

    info!("mcp-server listening on {}", socket_addr);
@@ -45,133 +40,33 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
    Ok(())
 }

-fn build_app(state: Arc<AppState>) -> Router {
-    Router::new()
-        .route("/health", get(health))
-        .route("/tools", get(list_tools))
-        .route("/tools/{tool_name}", post(call_tool))
-        .with_state(state)
-}
-
-async fn health() -> Json<serde_json::Value> {
-    Json(json!({
-        "service": "mcp-server",
-        "status": "ok"
-    }))
-}
-
-async fn list_tools(
-    State(state): State<Arc<AppState>>,
-) -> Result<Json<Value>, (StatusCode, Json<Value>)> {
-    let operations = state
-        .registry
-        .list_published_operations()
-        .await
-        .map_err(internal_error)?;
-
-    let tools = operations
-        .into_iter()
-        .map(|operation| {
-            json!({
-                "name": operation.name,
-                "title": operation.tool_description.title,
-                "description": operation.tool_description.description
-            })
-        })
-        .collect::<Vec<_>>();
-
-    Ok(Json(json!({ "tools": tools })))
-}
-
-async fn call_tool(
-    Path(tool_name): Path<String>,
-    State(state): State<Arc<AppState>>,
-    Json(input): Json<Value>,
-) -> Result<Json<Value>, (StatusCode, Json<Value>)> {
-    let operation = state
-        .registry
-        .list_published_operations()
-        .await
-        .map_err(internal_error)?
-        .into_iter()
-        .find(|operation| operation.name == tool_name)
-        .ok_or_else(|| {
-            (
-                StatusCode::NOT_FOUND,
-                Json(json!({
-                    "error": {
-                        "code": "not_found",
-                        "message": format!("tool {tool_name} was not found")
-                    }
-                })),
-            )
-        })?;
-
-    let output = state
-        .runtime
-        .execute(&RuntimeOperation::from(operation), &input)
-        .await
-        .map_err(runtime_error)?;
-
-    Ok(Json(json!({
-        "tool_name": tool_name,
-        "result": output
-    })))
-}
-
-fn internal_error(error: impl std::fmt::Display) -> (StatusCode, Json<Value>) {
-    (
-        StatusCode::INTERNAL_SERVER_ERROR,
-        Json(json!({
-            "error": {
-                "code": "internal_error",
-                "message": error.to_string()
-            }
-        })),
-    )
-}
-
-fn runtime_error(error: impl std::fmt::Display) -> (StatusCode, Json<Value>) {
-    (
-        StatusCode::BAD_GATEWAY,
-        Json(json!({
-            "error": {
-                "code": "runtime_error",
-                "message": error.to_string()
-            }
-        })),
-    )
-}
-
 #[cfg(test)]
 mod tests {
    use std::{
        collections::BTreeMap,
        env,
-        sync::Arc,
-        time::{SystemTime, UNIX_EPOCH},
+        time::{Duration, SystemTime, UNIX_EPOCH},
    };

-    use axum::{Json, Router, routing::post};
+    use axum::{Json, Router, http::header, routing::post};
    use mcpaas_core::{
        ExecutionConfig, HttpMethod, Operation, OperationId, OperationStatus, Protocol, RestTarget,
        Target, ToolDescription,
    };
    use mcpaas_mapping::{MappingRule, MappingSet};
    use mcpaas_registry::{PostgresRegistry, PublishRequest};
-    use mcpaas_runtime::RuntimeExecutor;
    use mcpaas_schema::{Schema, SchemaKind};
    use serde_json::{Value, json};
    use sqlx::{Executor, postgres::PgPoolOptions};
    use tokio::net::TcpListener;

-    use crate::{AppState, build_app};
+    use crate::app::build_app;

    #[tokio::test]
-    async fn lists_and_calls_published_rest_tool() {
+    async fn initializes_lists_and_calls_published_tool_via_mcp() {
        let registry = test_registry().await;
        let upstream_base_url = spawn_upstream_server().await;
-        let operation = test_operation(&upstream_base_url);
+        let operation = test_operation(&upstream_base_url, "crm_create_lead");

        registry
            .create_operation(&operation, Some("alice"))
@@ -187,24 +82,95 @@ mod tests {
            .await
            .unwrap();

-        let state = Arc::new(AppState {
+        let base_url = spawn_mcp_server(build_app(
            registry,
-            runtime: RuntimeExecutor::new(),
-        });
-        let base_url = spawn_mcp_server(build_app(state)).await;
+            Duration::from_millis(0),
+            Some("https://rmcp.example.com".to_owned()),
+        ))
+        .await;
        let client = reqwest::Client::new();
+        let initialized_session = initialize_session(&client, &base_url).await;

-        let tools = client
-            .get(format!("{base_url}/tools"))
+        let tools = post_jsonrpc(
+            &client,
+            &base_url,
+            Some(&initialized_session),
+            json!({
+                "jsonrpc": "2.0",
+                "id": 2,
+                "method": "tools/list",
+                "params": {}
+            }),
+        )
+        .await;
+        let call_result = post_jsonrpc(
+            &client,
+            &base_url,
+            Some(&initialized_session),
+            json!({
+                "jsonrpc": "2.0",
+                "id": 3,
+                "method": "tools/call",
+                "params": {
+                    "name": "crm_create_lead",
+                    "arguments": {
+                        "email": "user@example.com"
+                    }
+                }
+            }),
+        )
+        .await;
+
+        assert_eq!(tools["result"]["tools"][0]["name"], "crm_create_lead");
+        assert_eq!(
+            call_result["result"]["structuredContent"],
+            json!({ "id": "lead_123" })
+        );
+        assert_eq!(call_result["result"]["isError"], false);
+    }
+
+    #[tokio::test]
+    async fn requires_initialized_notification_before_tool_methods() {
+        let registry = test_registry().await;
+        let base_url = spawn_mcp_server(build_app(
+            registry,
+            Duration::from_millis(0),
+            Some("https://rmcp.example.com".to_owned()),
+        ))
+        .await;
+        let client = reqwest::Client::new();
+        let initialize_response = client
+            .post(format!("{base_url}/mcp"))
+            .header(header::ACCEPT, "application/json, text/event-stream")
+            .json(&json!({
+                "jsonrpc": "2.0",
+                "id": 1,
+                "method": "initialize",
+                "params": {
+                    "protocolVersion": "2025-11-25"
+                }
+            }))
            .send()
            .await
-            .unwrap()
-            .json::<Value>()
-            .await
            .unwrap();
-        let call_result = client
-            .post(format!("{base_url}/tools/crm_create_lead"))
-            .json(&json!({ "email": "user@example.com" }))
+        let session_id = initialize_response
+            .headers()
+            .get("MCP-Session-Id")
+            .unwrap()
+            .to_str()
+            .unwrap()
+            .to_owned();
+        let tools_list = client
+            .post(format!("{base_url}/mcp"))
+            .header(header::ACCEPT, "application/json, text/event-stream")
+            .header("MCP-Session-Id", session_id)
+            .header("MCP-Protocol-Version", "2025-11-25")
+            .json(&json!({
+                "jsonrpc": "2.0",
+                "id": 2,
+                "method": "tools/list",
+                "params": {}
+            }))
            .send()
            .await
            .unwrap()
@@ -212,8 +178,135 @@ mod tests {
            .await
            .unwrap();

-        assert_eq!(tools["tools"][0]["name"], "crm_create_lead");
-        assert_eq!(call_result["result"], json!({ "id": "lead_123" }));
+        assert_eq!(tools_list["error"]["code"], -32002);
+    }
+
+    #[tokio::test]
+    async fn refreshes_published_tools_without_restart() {
+        let registry = test_registry().await;
+        let upstream_base_url = spawn_upstream_server().await;
+        let base_url = spawn_mcp_server(build_app(
+            registry.clone(),
+            Duration::from_millis(0),
+            Some("https://rmcp.example.com".to_owned()),
+        ))
+        .await;
+        let client = reqwest::Client::new();
+        let initialized_session = initialize_session(&client, &base_url).await;
+
+        let before_publish = post_jsonrpc(
+            &client,
+            &base_url,
+            Some(&initialized_session),
+            json!({
+                "jsonrpc": "2.0",
+                "id": 2,
+                "method": "tools/list",
+                "params": {}
+            }),
+        )
+        .await;
+
+        let operation = test_operation(&upstream_base_url, "crm_publish_later");
+        registry
+            .create_operation(&operation, Some("alice"))
+            .await
+            .unwrap();
+        registry
+            .publish_operation(PublishRequest {
+                operation_id: &operation.id,
+                version: 1,
+                published_at: "2026-03-26T10:00:00Z",
+                published_by: Some("alice"),
+            })
+            .await
+            .unwrap();
+
+        let after_publish = post_jsonrpc(
+            &client,
+            &base_url,
+            Some(&initialized_session),
+            json!({
+                "jsonrpc": "2.0",
+                "id": 3,
+                "method": "tools/list",
+                "params": {}
+            }),
+        )
+        .await;
+
+        assert_eq!(before_publish["result"]["tools"], json!([]));
+        assert_eq!(
+            after_publish["result"]["tools"][0]["name"],
+            "crm_publish_later"
+        );
+    }
+
+    async fn initialize_session(client: &reqwest::Client, base_url: &str) -> String {
+        let initialize_response = client
+            .post(format!("{base_url}/mcp"))
+            .header(header::ACCEPT, "application/json, text/event-stream")
+            .json(&json!({
+                "jsonrpc": "2.0",
+                "id": 1,
+                "method": "initialize",
+                "params": {
+                    "protocolVersion": "2025-11-25"
+                }
+            }))
+            .send()
+            .await
+            .unwrap();
+        let session_id = initialize_response
+            .headers()
+            .get("MCP-Session-Id")
+            .unwrap()
+            .to_str()
+            .unwrap()
+            .to_owned();
+
+        let initialized_response = client
+            .post(format!("{base_url}/mcp"))
+            .header(header::ACCEPT, "application/json, text/event-stream")
+            .header("MCP-Session-Id", &session_id)
+            .header("MCP-Protocol-Version", "2025-11-25")
+            .json(&json!({
+                "jsonrpc": "2.0",
+                "method": "notifications/initialized",
+                "params": {}
+            }))
+            .send()
+            .await
+            .unwrap();
+
+        assert_eq!(initialized_response.status(), reqwest::StatusCode::ACCEPTED);
+
+        session_id
+    }
+
+    async fn post_jsonrpc(
+        client: &reqwest::Client,
+        base_url: &str,
+        session_id: Option<&str>,
+        payload: Value,
+    ) -> Value {
+        let mut request = client
+            .post(format!("{base_url}/mcp"))
+            .header(header::ACCEPT, "application/json, text/event-stream")
+            .header("MCP-Protocol-Version", "2025-11-25");
+
+        if let Some(session_id) = session_id {
+            request = request.header("MCP-Session-Id", session_id);
+        }
+
+        request
+            .json(&payload)
+            .send()
+            .await
+            .unwrap()
+            .json::<Value>()
+            .await
+            .unwrap()
    }

    async fn spawn_upstream_server() -> String {
@@ -273,10 +366,10 @@ mod tests {
            .unwrap()
    }

-    fn test_operation(base_url: &str) -> Operation<Schema, MappingSet> {
+    fn test_operation(base_url: &str, name: &str) -> Operation<Schema, MappingSet> {
        Operation {
-            id: OperationId::new("op_mcp_rest"),
-            name: "crm_create_lead".to_owned(),
+            id: OperationId::new(format!("op_{name}")),
+            name: name.to_owned(),
            display_name: "Create Lead".to_owned(),
            protocol: Protocol::Rest,
            status: OperationStatus::Published,
@@ -320,7 +413,7 @@ mod tests {
            },
            tool_description: ToolDescription {
                title: "Create Lead".to_owned(),
-                description: "Creates CRM lead".to_owned(),
+                description: "Creates a CRM lead".to_owned(),
                tags: Vec::new(),
                examples: Vec::new(),
            },
@@ -0,0 +1,59 @@
+use std::{collections::HashMap, sync::Arc};
+
+use tokio::sync::RwLock;
+use uuid::Uuid;
+
+#[derive(Clone)]
+pub struct SessionStore {
+    inner: Arc<RwLock<HashMap<String, SessionState>>>,
+}
+
+#[derive(Clone)]
+pub struct SessionState {
+    pub protocol_version: String,
+    pub initialized: bool,
+}
+
+impl SessionStore {
+    pub fn new() -> Self {
+        Self {
+            inner: Arc::new(RwLock::new(HashMap::new())),
+        }
+    }
+
+    pub async fn create(&self, protocol_version: &str) -> String {
+        let session_id = Uuid::now_v7().to_string();
+        let mut guard = self.inner.write().await;
+
+        guard.insert(
+            session_id.clone(),
+            SessionState {
+                protocol_version: protocol_version.to_owned(),
+                initialized: false,
+            },
+        );
+
+        session_id
+    }
+
+    pub async fn get(&self, session_id: &str) -> Option<SessionState> {
+        let guard = self.inner.read().await;
+        guard.get(session_id).cloned()
+    }
+
+    pub async fn mark_initialized(&self, session_id: &str) -> bool {
+        let mut guard = self.inner.write().await;
+
+        if let Some(session) = guard.get_mut(session_id) {
+            session.initialized = true;
+            return true;
+        }
+
+        false
+    }
+
+    pub async fn delete(&self, session_id: &str) -> bool {
+        let mut guard = self.inner.write().await;
+        guard.remove(session_id).is_some()
+    }
+}
@@ -100,13 +100,27 @@

 ## 8. MCP lifecycle

+MVP-контракт `mcp-server` строится вокруг JSON-RPC методов MCP:
+
+- `initialize`
+- `notifications/initialized`
+- `ping`
+- `tools/list`
+- `tools/call`
+
+Сессия создается на `initialize` и идентифицируется через `MCP-Session-Id`.
+Согласованная версия протокола возвращается и читается через `MCP-Protocol-Version`.
+
+Пока сессии хранятся in-memory внутри `mcp-server`, чего достаточно для MVP и demo-сценариев.
+
 ### Tool listing

-При старте и после reload:
+После `initialize` и `notifications/initialized`:

-1. `mcp-server` читает список published operations.
-2. Строит in-memory registry tools.
-3. Отдает их через MCP list tools.
+1. клиент вызывает `tools/list`;
+2. `mcp-server` перечитывает published operations по refresh policy;
+3. строит или обновляет in-memory catalog tools;
+4. отдает список tools через MCP JSON-RPC result.

 ### Tool call

@@ -122,8 +136,9 @@

 1. `admin-api` фиксирует published version в registry.
 2. `registry` обновляет published_operations.
-3. `mcp-server` получает reload signal или выполняет controlled refresh.
-4. Новый tool contract становится доступен MCP clients.
+3. `mcp-server` не требует restart и не опирается на ручной reload signal.
+4. `mcp-server` выполняет controlled refresh опубликованного каталога по interval-based policy.
+5. Новый tool contract становится доступен MCP clients.

 ## 10. Именование tools

@@ -61,6 +61,7 @@ var/mcpaas/
 - `MCPAAS_STORAGE_ROOT`
 - `MCPAAS_ADMIN_BIND`
 - `MCPAAS_MCP_BIND`
+- `MCPAAS_MCP_REFRESH_MS`
 - `MCPAAS_LOG_LEVEL`
 - `MCPAAS_SECRET_PROVIDER`
 - `MCPAAS_PUBLIC_BASE_URL`
@@ -71,6 +72,10 @@ var/mcpaas/
 - `MCPAAS_ADMIN_TOKEN`
 - `MCPAAS_MASTER_KEY`

+Стартовое значение для refresh published tools:
+
+- `MCPAAS_MCP_REFRESH_MS=5000`
+
 ## 6. Логирование и трассировка

 Для MVP нужно использовать: