use std::net::SocketAddr;
use std::sync::Arc;
use axum::Router;
use axum::extract::{ConnectInfo, Request, State};
use axum::http::{HeaderMap, StatusCode};
use axum::middleware::{self, Next};
use axum::response::{IntoResponse, Json, Response};
use axum::routing::get;
use rmcp::transport::streamable_http_server::session::local::LocalSessionManager;
use rmcp::transport::streamable_http_server::{StreamableHttpServerConfig, StreamableHttpService};
use serde_json::json;
use tokio::sync::Mutex as TokioMutex;
use super::auth_extractor::extract_request_credentials;
use super::localhost_detector::is_localhost;
use super::metrics;
use super::types::HttpServerConfig;
use crate::core::component_registry::{ComponentRegistry, ComponentStatus};
use crate::core::mcp_server::McpifyServer;
#[derive(Clone)]
struct AppState {
registry: Arc<TokioMutex<ComponentRegistry>>,
header_location: &'static str,
header_name: String,
}
async fn healthz(State(state): State<AppState>) -> impl IntoResponse {
let registry = state.registry.lock().await;
let status = registry.overall_status();
let code = if status == ComponentStatus::Unhealthy {
StatusCode::SERVICE_UNAVAILABLE
} else {
StatusCode::OK
};
(
code,
Json(json!({
"status": format!("{status:?}"),
"components": registry.snapshot().len(),
})),
)
}
async fn metrics_handler() -> impl IntoResponse {
metrics::increment("http_requests_total");
(StatusCode::OK, metrics::render_metrics())
}
async fn auth_gate(
State(state): State<AppState>,
ConnectInfo(addr): ConnectInfo<SocketAddr>,
headers: HeaderMap,
request: Request,
next: Next,
) -> Response {
metrics::increment("http_requests_total");
let relaxed = is_localhost(addr.ip());
if !relaxed
&& extract_request_credentials(&headers, state.header_location, &state.header_name).is_err()
{
return (
StatusCode::UNAUTHORIZED,
Json(json!({ "error": "missing or invalid credentials for this API's auth method" })),
)
.into_response();
}
next.run(request).await
}
async fn set_cors_header(cors_allow: Option<String>, request: Request, next: Next) -> Response {
let mut response = next.run(request).await;
if let Some(origin) = cors_allow
&& let Ok(value) = origin.parse()
{
response
.headers_mut()
.insert("Access-Control-Allow-Origin", value);
}
response
}
pub async fn start_http_server(
mcp_service_factory: impl Fn() -> Result<McpifyServer, std::io::Error> + Send + Sync + 'static,
config: &HttpServerConfig,
registry: Arc<TokioMutex<ComponentRegistry>>,
header_location: &'static str,
header_name: String,
) -> anyhow::Result<()> {
let mut allowed_hosts = vec![
"localhost".to_string(),
"127.0.0.1".to_string(),
"::1".to_string(),
];
allowed_hosts.push(config.host.clone());
allowed_hosts.push(format!("{}:{}", config.host, config.port));
let mcp_service = StreamableHttpService::new(
mcp_service_factory,
LocalSessionManager::default().into(),
StreamableHttpServerConfig::default().with_allowed_hosts(allowed_hosts),
);
let state = AppState {
registry,
header_location,
header_name,
};
let cors_allow = config.cors_allow.clone();
let mcp_router = Router::new()
.nest_service("/mcp", mcp_service)
.layer(middleware::from_fn_with_state(state.clone(), auth_gate));
let app = Router::new()
.route("/healthz", get(healthz))
.route("/metrics", get(metrics_handler))
.merge(mcp_router)
.layer(middleware::from_fn(move |req, next| {
set_cors_header(cors_allow.clone(), req, next)
}))
.with_state(state);
let addr: SocketAddr = format!("{}:{}", config.host, config.port).parse()?;
let listener = tokio::net::TcpListener::bind(addr).await?;
axum::serve(
listener,
app.into_make_service_with_connect_info::<SocketAddr>(),
)
.await?;
Ok(())
}