feat: implement MCP streamable HTTP server

apps/mcp-server/src/main.rs

@@ -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(),
             },