rmcp-actix-web 0.4.3

actix-web transport implementations for RMCP (Rust Model Context Protocol)
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124

//! SSE Service Composition Example
//!
//! This example demonstrates how to use framework-level composition to mount
//! SSE MCP services at custom paths within an existing actix-web application
//! using the unified builder pattern.
//!
//! ## Key Features Demonstrated
//!
//! - Using `SseService::builder()` to configure the service
//! - Mounting MCP services at custom paths using `.scope()`
//! - Integration with existing actix-web middleware and routes
//! - Unified builder pattern consistent with StreamableHttpService
//!
//! ## Usage
//!
//! ```bash
//! cargo run --example composition_sse_example
//! ```
//!
//! Then test with curl:
//! ```bash
//! # Connect to SSE endpoint at custom path
//! curl -N -H "Mcp-Session-Id: test-session" \
//!      http://127.0.0.1:8080/api/v1/mcp/sse
//!
//! # Send a message (in another terminal)
//! curl -X POST http://127.0.0.1:8080/api/v1/mcp/message \
//!      -H "Content-Type: application/json" \
//!      -H "Mcp-Session-Id: test-session" \
//!      -d '{"jsonrpc":"2.0","id":1,"method":"tools/list","params":{}}'
//! ```

use actix_web::{App, HttpResponse, HttpServer, Result, middleware, web};
use rmcp_actix_web::SseService;
use std::{sync::Arc, time::Duration};
use tracing_subscriber::{layer::SubscriberExt, util::SubscriberInitExt};

mod common;
use common::calculator::Calculator;

/// A simple health check endpoint to demonstrate integration with existing routes
async fn health_check() -> Result<HttpResponse> {
    Ok(HttpResponse::Ok().json(serde_json::json!({
        "status": "healthy",
        "service": "mcp-calculator",
        "version": "1.0.0"
    })))
}

/// Root endpoint that shows available services
async fn root() -> Result<HttpResponse> {
    Ok(HttpResponse::Ok().json(serde_json::json!({
        "message": "MCP Calculator Service",
        "endpoints": {
            "health": "/health",
            "mcp_sse": "/api/v1/mcp/sse",
            "mcp_post": "/api/v1/mcp/message"
        },
        "documentation": "https://modelcontextprotocol.io/"
    })))
}

#[actix_web::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    // Initialize tracing for better debugging
    tracing_subscriber::registry()
        .with(
            tracing_subscriber::EnvFilter::try_from_default_env()
                .unwrap_or_else(|_| "info".to_string().into()),
        )
        .with(tracing_subscriber::fmt::layer())
        .init();

    let bind_addr = "127.0.0.1:8080";
    tracing::info!("Starting SSE composition example server on {}", bind_addr);

    // Create SSE service using builder pattern with custom paths
    let sse_service = SseService::builder()
        .service_factory(Arc::new(|| Ok(Calculator::new()))) // Create new Calculator for each session
        .sse_path("/sse".to_string()) // Custom SSE endpoint path
        .post_path("/message".to_string()) // Custom message endpoint path
        .sse_keep_alive(Duration::from_secs(30)) // Keep-alive pings every 30 seconds
        .build();

    // Start the HTTP server with framework-level composition
    let server = HttpServer::new(move || {
        App::new()
            // Add logging middleware
            .wrap(middleware::Logger::default())
            .wrap(middleware::NormalizePath::trim())
            // Add custom application routes
            .route("/", web::get().to(root))
            .route("/health", web::get().to(health_check))
            // Mount the MCP service at a custom API path using scope()
            .service(web::scope("/api").service(
                web::scope("/v1").service(web::scope("/mcp").service(sse_service.clone().scope())),
            ))
    })
    .bind(bind_addr)?
    .run();

    tracing::info!("🚀 Server started successfully!");
    tracing::info!("📊 Health check: http://{}/health", bind_addr);
    tracing::info!("🔌 MCP SSE endpoint: http://{}/api/v1/mcp/sse", bind_addr);
    tracing::info!(
        "📨 MCP POST endpoint: http://{}/api/v1/mcp/message",
        bind_addr
    );
    tracing::info!("Press Ctrl+C to stop the server");

    // Run server until cancelled
    tokio::select! {
        result = server => {
            if let Err(e) = result {
                tracing::error!("HTTP server error: {}", e);
            }
        }
        _ = tokio::signal::ctrl_c() => {
            tracing::info!("Received Ctrl+C, shutting down gracefully");
        }
    }

    Ok(())
}