github_mcp/http/
server.rs1use std::net::SocketAddr;
4use std::sync::Arc;
5
6use axum::Router;
7use axum::extract::{ConnectInfo, Request, State};
8use axum::http::{HeaderMap, StatusCode};
9use axum::middleware::{self, Next};
10use axum::response::{IntoResponse, Json, Response};
11use axum::routing::get;
12use rmcp::transport::streamable_http_server::session::local::LocalSessionManager;
13use rmcp::transport::streamable_http_server::{StreamableHttpServerConfig, StreamableHttpService};
14use serde_json::json;
15use tokio::sync::Mutex as TokioMutex;
16
17use super::auth_extractor::extract_request_credentials;
18use super::localhost_detector::is_localhost;
19use super::metrics;
20use super::types::HttpServerConfig;
21use crate::core::component_registry::{ComponentRegistry, ComponentStatus};
22use crate::core::mcp_server::McpifyServer;
23
24#[derive(Clone)]
25struct AppState {
26 registry: Arc<TokioMutex<ComponentRegistry>>,
27 header_location: &'static str,
31 header_name: String,
32}
33
34async fn healthz(State(state): State<AppState>) -> impl IntoResponse {
35 let registry = state.registry.lock().await;
36 let status = registry.overall_status();
37 let code = if status == ComponentStatus::Unhealthy {
38 StatusCode::SERVICE_UNAVAILABLE
39 } else {
40 StatusCode::OK
41 };
42 (
43 code,
44 Json(json!({
45 "status": format!("{status:?}"),
46 "components": registry.snapshot().len(),
47 })),
48 )
49}
50
51async fn metrics_handler() -> impl IntoResponse {
52 metrics::increment("http_requests_total");
53 (StatusCode::OK, metrics::render_metrics())
54}
55
56async fn auth_gate(
74 State(state): State<AppState>,
75 ConnectInfo(addr): ConnectInfo<SocketAddr>,
76 headers: HeaderMap,
77 request: Request,
78 next: Next,
79) -> Response {
80 metrics::increment("http_requests_total");
81 let relaxed = is_localhost(addr.ip());
82
83 if !relaxed
84 && extract_request_credentials(&headers, state.header_location, &state.header_name).is_err()
85 {
86 return (
87 StatusCode::UNAUTHORIZED,
88 Json(json!({ "error": "missing or invalid credentials for this API's auth method" })),
89 )
90 .into_response();
91 }
92
93 next.run(request).await
94}
95
96async fn set_cors_header(cors_allow: Option<String>, request: Request, next: Next) -> Response {
97 let mut response = next.run(request).await;
98 if let Some(origin) = cors_allow
99 && let Ok(value) = origin.parse()
100 {
101 response
102 .headers_mut()
103 .insert("Access-Control-Allow-Origin", value);
104 }
105 response
106}
107
108pub async fn start_http_server(
112 mcp_service_factory: impl Fn() -> Result<McpifyServer, std::io::Error> + Send + Sync + 'static,
113 config: &HttpServerConfig,
114 registry: Arc<TokioMutex<ComponentRegistry>>,
115 header_location: &'static str,
116 header_name: String,
117) -> anyhow::Result<()> {
118 let mut allowed_hosts = vec![
119 "localhost".to_string(),
120 "127.0.0.1".to_string(),
121 "::1".to_string(),
122 ];
123 allowed_hosts.push(config.host.clone());
124 allowed_hosts.push(format!("{}:{}", config.host, config.port));
125
126 let mcp_service = StreamableHttpService::new(
127 mcp_service_factory,
128 LocalSessionManager::default().into(),
129 StreamableHttpServerConfig::default().with_allowed_hosts(allowed_hosts),
130 );
131
132 let state = AppState {
133 registry,
134 header_location,
135 header_name,
136 };
137 let cors_allow = config.cors_allow.clone();
138
139 let mcp_router = Router::new()
140 .nest_service("/mcp", mcp_service)
141 .layer(middleware::from_fn_with_state(state.clone(), auth_gate));
142
143 let app = Router::new()
144 .route("/healthz", get(healthz))
145 .route("/metrics", get(metrics_handler))
146 .merge(mcp_router)
147 .layer(middleware::from_fn(move |req, next| {
148 set_cors_header(cors_allow.clone(), req, next)
149 }))
150 .with_state(state);
151
152 let addr: SocketAddr = format!("{}:{}", config.host, config.port).parse()?;
153 let listener = tokio::net::TcpListener::bind(addr).await?;
154 axum::serve(
155 listener,
156 app.into_make_service_with_connect_info::<SocketAddr>(),
157 )
158 .await?;
159
160 Ok(())
161}