mcpway 0.1.1

Run MCP stdio servers over SSE, WebSocket, and Streamable HTTP transports.
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

use std::net::SocketAddr;
use std::sync::Arc;

use crate::runtime::store::{RuntimeArgsStore, RuntimeArgsUpdate};
use crate::runtime::{RuntimeApplyResult, RuntimeScope, RuntimeUpdate};
use axum::extract::{ConnectInfo, Path, State};
use axum::http::StatusCode;
use axum::middleware::{self, Next};
use axum::response::{IntoResponse, Response};
use axum::routing::{get, post};
use axum::{Json, Router};
use futures::future::BoxFuture;

#[derive(Clone)]
pub struct AdminState {
    runtime: RuntimeArgsStore,
    handler: Arc<dyn Fn(RuntimeUpdate) -> BoxFuture<'static, RuntimeApplyResult> + Send + Sync>,
}

pub async fn spawn_admin_server(
    addr: SocketAddr,
    runtime: RuntimeArgsStore,
    handler: Arc<dyn Fn(RuntimeUpdate) -> BoxFuture<'static, RuntimeApplyResult> + Send + Sync>,
) {
    let state = AdminState { runtime, handler };

    let router = Router::new()
        .route("/runtime/defaults", post(update_defaults))
        .route("/runtime/session/:id", post(update_session))
        .route("/runtime/sessions", get(list_sessions))
        .with_state(state)
        .layer(middleware::from_fn(only_loopback));

    tracing::info!("Runtime admin endpoint listening on http://{addr}");

    let listener = match tokio::net::TcpListener::bind(addr).await {
        Ok(listener) => listener,
        Err(err) => {
            tracing::error!("Runtime admin bind error: {err}");
            return;
        }
    };
    let server = axum::serve(
        listener,
        router.into_make_service_with_connect_info::<SocketAddr>(),
    );
    if let Err(err) = server.await {
        tracing::error!("Runtime admin server error: {err}");
    }
}

async fn only_loopback(
    ConnectInfo(addr): ConnectInfo<SocketAddr>,
    req: axum::http::Request<axum::body::Body>,
    next: Next,
) -> Response {
    if !addr.ip().is_loopback() {
        return StatusCode::FORBIDDEN.into_response();
    }
    next.run(req).await
}

async fn update_defaults(
    State(state): State<AdminState>,
    Json(update): Json<RuntimeArgsUpdate>,
) -> impl IntoResponse {
    let update_msg = RuntimeUpdate {
        scope: RuntimeScope::Global,
        update,
    };
    Json((state.handler)(update_msg).await)
}

async fn update_session(
    State(state): State<AdminState>,
    Path(id): Path<String>,
    Json(update): Json<RuntimeArgsUpdate>,
) -> impl IntoResponse {
    let update_msg = RuntimeUpdate {
        scope: RuntimeScope::Session(id),
        update,
    };
    Json((state.handler)(update_msg).await)
}

async fn list_sessions(State(state): State<AdminState>) -> impl IntoResponse {
    let sessions = state.runtime.list_sessions().await;
    Json(sessions)
}