use axum::Router;
use rmcp::{
handler::server::{tool::ToolRouter, wrapper::Parameters},
model::*,
tool, tool_handler, tool_router,
transport::streamable_http_server::{
session::local::LocalSessionManager, StreamableHttpServerConfig, StreamableHttpService,
},
ErrorData as McpError, ServerHandler,
};
use schemars::JsonSchema;
use serde::Deserialize;
use std::sync::{Arc, LazyLock, OnceLock};
use std::time::Duration;
use tokio_util::sync::CancellationToken;
use crate::channels::{get_channel_logs, get_channels_json};
use crate::debug::dbg::{get_dbg_logs, get_debug_dbg_entries_json};
use crate::debug::gauge::{get_debug_gauge_entries_json, get_debug_gauge_logs};
use crate::debug::val::{get_debug_val_entries_json, get_val_logs};
use crate::functions::{
get_function_logs_alloc, get_function_logs_timing, get_functions_alloc_json,
get_functions_timing_json,
};
use crate::futures::{get_future_logs_list, get_futures_json};
use crate::json::{
JsonChannelLogsList, JsonFunctionAllocLogsList, JsonFunctionTimingLogsList, JsonFutureLogsList,
JsonHttpLogsList, JsonProfilerStatus, JsonSqlLogsList, JsonStreamLogsList,
};
use crate::output::format_duration;
use crate::streams::{get_stream_logs, get_streams_json};
use crate::threads::get_threads_json;
fn id_from_number_or_string<'de, D>(deserializer: D) -> Result<u32, D::Error>
where
D: serde::Deserializer<'de>,
{
struct IdVisitor;
impl serde::de::Visitor<'_> for IdVisitor {
type Value = u32;
fn expecting(&self, f: &mut std::fmt::Formatter) -> std::fmt::Result {
f.write_str("an integer id or its string form")
}
fn visit_u64<E: serde::de::Error>(self, v: u64) -> Result<u32, E> {
u32::try_from(v).map_err(E::custom)
}
fn visit_i64<E: serde::de::Error>(self, v: i64) -> Result<u32, E> {
u32::try_from(v).map_err(E::custom)
}
fn visit_str<E: serde::de::Error>(self, v: &str) -> Result<u32, E> {
v.trim().parse().map_err(E::custom)
}
}
deserializer.deserialize_any(IdVisitor)
}
#[derive(Debug, Deserialize, JsonSchema)]
struct FunctionIdParam {
#[schemars(description = "Function ID from the functions_timing or functions_alloc response")]
#[serde(deserialize_with = "id_from_number_or_string")]
function_id: u32,
}
#[derive(Debug, Deserialize, JsonSchema)]
struct ChannelIdParam {
#[schemars(description = "Channel id from the channels list")]
#[serde(deserialize_with = "id_from_number_or_string")]
channel_id: u32,
}
#[derive(Debug, Deserialize, JsonSchema)]
struct StreamIdParam {
#[schemars(description = "Stream id from the streams list")]
#[serde(deserialize_with = "id_from_number_or_string")]
stream_id: u32,
}
#[derive(Debug, Deserialize, JsonSchema)]
struct FutureIdParam {
#[schemars(description = "Future id from the futures list")]
#[serde(deserialize_with = "id_from_number_or_string")]
future_id: u32,
}
#[derive(Debug, Deserialize, JsonSchema)]
struct SqlIdParam {
#[schemars(description = "SQL query id from the sql list")]
#[serde(deserialize_with = "id_from_number_or_string")]
sql_id: u32,
}
#[derive(Debug, Deserialize, JsonSchema)]
struct HttpIdParam {
#[schemars(description = "HTTP endpoint id from the http list")]
#[serde(deserialize_with = "id_from_number_or_string")]
http_id: u32,
}
#[derive(Debug, Deserialize, JsonSchema)]
struct GaugeIdParam {
#[schemars(description = "Gauge id from the gauges list")]
#[serde(deserialize_with = "id_from_number_or_string")]
gauge_id: u32,
}
#[derive(Debug, Deserialize, JsonSchema)]
struct DebugIdParam {
#[schemars(description = "Debug entry ID from the debug tool response")]
#[serde(deserialize_with = "id_from_number_or_string")]
debug_id: u32,
}
static MCP_SERVER_PORT: LazyLock<u16> = LazyLock::new(|| {
std::env::var("HOTPATH_MCP_PORT")
.ok()
.and_then(|p| p.parse::<u16>().ok())
.unwrap_or(6771)
});
#[derive(Clone)]
pub(crate) struct HotPathMcpServer {
tool_router: ToolRouter<Self>,
}
#[cfg_attr(feature = "hotpath-meta", hotpath_meta::measure_all)]
#[tool_router]
impl HotPathMcpServer {
pub fn new() -> Self {
Self {
tool_router: Self::tool_router(),
}
}
#[tool(
description = r#"Get execution timing metrics for all profiled functions.
Returns JSON array of functions sorted by total time. Each entry contains:
- name: fully qualified function name (e.g. "my_app::db::query")
- call_count: number of invocations
- total_ns: cumulative execution time in nanoseconds
- mean_ns, p50_ns, p95_ns, p99_ns: latency percentiles
Use this first to identify performance hotspots. Look for high p95/p99 values indicating tail latency issues."#
)]
async fn functions_timing(&self) -> Result<CallToolResult, McpError> {
log_debug("Tool called: functions_timing");
let formatted = get_functions_timing_json();
Ok(CallToolResult::success(vec![Content::text(to_json(
&formatted,
)?)]))
}
#[tool(
description = r#"Get memory allocation metrics per function (requires hotpath-alloc feature).
Returns JSON array with:
- name: function name
- alloc_bytes: total bytes allocated
- alloc_count: number of allocations
Returns error if hotpath-alloc feature is not enabled. Cross-reference with functions_timing to find functions that are both slow and allocation-heavy."#
)]
async fn functions_alloc(&self) -> Result<CallToolResult, McpError> {
log_debug("Tool called: functions_alloc");
match get_functions_alloc_json() {
Some(formatted) => Ok(CallToolResult::success(vec![Content::text(to_json(
&formatted,
)?)])),
None => Ok(CallToolResult::error(vec![Content::text(
"Memory profiling not available - enable hotpath-alloc feature",
)])),
}
}
#[tool(
description = r#"Get CPU sampling attribution per instrumented function (requires hotpath-cpu feature).
Returns JSON envelope with:
- status: "idle", "capturing", "ready", or "error"
- functions: when ready, array of functions with sample counts and CPU time attribution
- session/profile metadata
Use functions_cpu_snapshot to trigger an on-demand capture, then poll this tool until status is "ready". Returns error if hotpath-cpu feature is not enabled."#
)]
async fn functions_cpu(&self) -> Result<CallToolResult, McpError> {
log_debug("Tool called: functions_cpu");
#[cfg(feature = "hotpath-cpu")]
{
let envelope = crate::functions::cpu::get_cpu_envelope();
Ok(CallToolResult::success(vec![Content::text(to_json(
&envelope,
)?)]))
}
#[cfg(not(feature = "hotpath-cpu"))]
Ok(CallToolResult::error(vec![Content::text(
"CPU profiling not available - enable hotpath-cpu feature",
)]))
}
#[tool(
description = r#"Trigger an on-demand CPU sampling snapshot (requires hotpath-cpu feature).
Starts a background capture of CPU samples collected since profiling began. Returns immediately with status "capturing". Poll the functions_cpu tool until its status is "ready" to read the results.
Returns error if a snapshot is already in progress or the hotpath-cpu feature is not enabled."#
)]
async fn functions_cpu_snapshot(&self) -> Result<CallToolResult, McpError> {
log_debug("Tool called: functions_cpu_snapshot");
#[cfg(feature = "hotpath-cpu")]
{
if crate::functions::cpu::try_spawn_snapshot() {
Ok(CallToolResult::success(vec![Content::text(
r#"{"status":"capturing"}"#,
)]))
} else {
Ok(CallToolResult::error(vec![Content::text(
"Snapshot already in progress",
)]))
}
}
#[cfg(not(feature = "hotpath-cpu"))]
Ok(CallToolResult::error(vec![Content::text(
"CPU profiling not available - enable hotpath-cpu feature",
)]))
}
#[tool(
description = r#"Get metrics for all monitored async channels (tokio, crossbeam, std, futures-channel).
Returns JSON array with:
- id: channel identifier
- label: optional custom label
- channel_type: "bounded", "unbounded", or "oneshot"
- sent/received: message counts
- sent_per_sec/received_per_sec: throughput over the channel's active window (omitted when not derivable)
- state: "active", "closed"
Use to track channel throughput and identify stalled or closed channels."#
)]
async fn channels(&self) -> Result<CallToolResult, McpError> {
log_debug("Tool called: channels");
let channels = get_channels_json();
Ok(CallToolResult::success(vec![Content::text(to_json(
&channels,
)?)]))
}
#[tool(description = r#"Get metrics for all monitored async streams.
Returns JSON array with:
- id: stream identifier
- label: optional custom label
- items_yielded: count of items produced
- state: "active" or "closed"
Use to track stream throughput and identify stalled streams."#)]
async fn streams(&self) -> Result<CallToolResult, McpError> {
log_debug("Tool called: streams");
let streams = get_streams_json();
Ok(CallToolResult::success(vec![Content::text(to_json(
&streams,
)?)]))
}
#[tool(description = r#"Get lifecycle metrics for all monitored futures.
Returns JSON array with:
- id: future identifier
- label: optional custom label
- call_count: number of future invocations observed
- total_polls: cumulative number of poll calls across invocations
- total_poll_duration_ns: cumulative poll time in nanoseconds
High poll counts can indicate futures that wake frequently without making progress."#)]
async fn futures(&self) -> Result<CallToolResult, McpError> {
log_debug("Tool called: futures");
let futures = get_futures_json();
Ok(CallToolResult::success(vec![Content::text(to_json(
&futures,
)?)]))
}
#[tool(
description = r#"Get wait and acquire-time metrics for all monitored RwLocks.
Returns JSON with per-lock entries split into read and write sections. Each section contains:
- count: number of lock acquisitions
- wait: time blocked before the lock was granted (avg and configured percentiles, in nanoseconds)
- acquire: time the lock was held, granted to released (avg and configured percentiles, in nanoseconds)
High wait times indicate lock contention; high acquire times indicate long critical sections. Locks are instrumented via hotpath::rw_lock!(expr)."#
)]
async fn rw_locks(&self) -> Result<CallToolResult, McpError> {
log_debug("Tool called: rw_locks");
let rw_locks = crate::rw_locks::get_rw_locks_json();
Ok(CallToolResult::success(vec![Content::text(to_json(
&rw_locks,
)?)]))
}
#[tool(
description = r#"Get wait and acquire-time metrics for all monitored Mutexes.
Returns JSON with per-mutex entries containing:
- count: number of lock acquisitions
- wait: time blocked before the lock was granted (avg and configured percentiles, in nanoseconds)
- acquire: time the lock was held, granted to released (avg and configured percentiles, in nanoseconds)
High wait times indicate lock contention; high acquire times indicate long critical sections. Mutexes are instrumented via hotpath::mutex!(expr)."#
)]
async fn mutexes(&self) -> Result<CallToolResult, McpError> {
log_debug("Tool called: mutexes");
let mutexes = crate::mutexes::get_mutexes_json();
Ok(CallToolResult::success(vec![Content::text(to_json(
&mutexes,
)?)]))
}
#[tool(
description = r#"Get byte-level I/O metrics for all instrumented Read/Write/AsyncRead/AsyncWrite values.
Returns JSON with per-wrapper entries, each containing read/write/flush/shutdown operation stats:
- count: number of completed operations
- bytes: total bytes processed
- errors: failed operations (retryable WouldBlock/Interrupted conditions are not counted)
- avg, configured percentiles, and total duration; async durations span first poll to Ready, so they include async waiting time
I/O values are instrumented via hotpath::io!(expr). Wrapping the underlying resource (file, socket) measures actual resource I/O; wrapping a BufReader/BufWriter measures application-facing buffered operations."#
)]
async fn io(&self) -> Result<CallToolResult, McpError> {
log_debug("Tool called: io");
let io = crate::io::get_io_json();
Ok(CallToolResult::success(vec![Content::text(to_json(&io)?)]))
}
#[tool(
description = r#"Get execution-time metrics for SQL queries captured via the sqlx tracing layer.
Returns JSON array with one entry per normalized query (parameter-varied executions merge into one bucket):
- id: query identifier
- query: normalized SQL statement text
- call_count: number of executions
- avg, configured percentiles, and total duration in nanoseconds
Requires hotpath::sqlx_tracing_layer() added to the profiled application's tracing subscriber. Use sql_logs with a query id to get recent individual executions."#
)]
async fn sql(&self) -> Result<CallToolResult, McpError> {
log_debug("Tool called: sql");
let sql = crate::sql::get_sql_json();
Ok(CallToolResult::success(vec![Content::text(to_json(&sql)?)]))
}
#[tool(description = r#"Get detailed execution logs for a specific SQL query.
Returns JSON array of recent executions with timestamps and durations. Use sql first to get query IDs, then use this tool to get detailed logs."#)]
async fn sql_logs(&self, params: Parameters<SqlIdParam>) -> Result<CallToolResult, McpError> {
let sql_id = params.0.sql_id;
log_debug(&format!("Tool called: sql_logs({})", sql_id));
match crate::sql::get_sql_logs(sql_id) {
Some(logs) => {
let current_elapsed_ns = get_current_elapsed_ns();
let formatted = JsonSqlLogsList::from_logs(&logs, current_elapsed_ns);
Ok(CallToolResult::success(vec![Content::text(to_json(
&formatted,
)?)]))
}
None => Ok(CallToolResult::error(vec![Content::text(
"SQL query not found",
)])),
}
}
#[tool(
description = r#"Get execution-time metrics for HTTP requests captured via hotpath's reqwest instrumentation.
Returns JSON array with one entry per normalized endpoint (parameter-varied requests to the same route merge into one bucket, e.g. GET api.example.com/users/{id}):
- id: endpoint identifier
- endpoint: normalized METHOD host/path key
- count: number of requests
- errors: transport errors plus responses with status >= 400
- avg, configured percentiles, and total duration
Requires wrapping the reqwest client with hotpath::http!(...) (or attaching ReqwestHttpMiddleware to a reqwest-middleware stack). Use http_logs with an endpoint id to get recent individual requests."#
)]
async fn http(&self) -> Result<CallToolResult, McpError> {
log_debug("Tool called: http");
let http = crate::http::get_http_json();
Ok(CallToolResult::success(vec![Content::text(to_json(
&http,
)?)]))
}
#[tool(
description = r#"Get detailed request logs for a specific HTTP endpoint.
Returns JSON array of recent requests with timestamps, durations, and status codes ("error" for transport failures). Use http first to get endpoint IDs, then use this tool to get detailed logs."#
)]
async fn http_logs(&self, params: Parameters<HttpIdParam>) -> Result<CallToolResult, McpError> {
let http_id = params.0.http_id;
log_debug(&format!("Tool called: http_logs({})", http_id));
match crate::http::get_http_logs(http_id) {
Some(logs) => {
let current_elapsed_ns = get_current_elapsed_ns();
let formatted = JsonHttpLogsList::from_logs(&logs, current_elapsed_ns);
Ok(CallToolResult::success(vec![Content::text(to_json(
&formatted,
)?)]))
}
None => Ok(CallToolResult::error(vec![Content::text(
"HTTP endpoint not found",
)])),
}
}
#[tool(description = r#"Get CPU usage metrics for all monitored threads.
Returns JSON array with:
- name: thread name (e.g. "tokio-runtime-worker")
- cpu_percent: CPU utilization (0-100 per core)
Sampled at configurable interval (HOTPATH_THREADS_INTERVAL_MS env var, default 250ms). Useful for identifying CPU-bound threads."#)]
async fn threads(&self) -> Result<CallToolResult, McpError> {
log_debug("Tool called: threads");
let threads = get_threads_json();
Ok(CallToolResult::success(vec![Content::text(to_json(
&threads,
)?)]))
}
#[tool(description = r#"Get detailed timing logs for a specific function.
Returns JSON array of recent execution logs with timestamps and duration. Use functions_timing first to get function IDs, then use this tool to get detailed logs."#)]
async fn function_timing_logs(
&self,
params: Parameters<FunctionIdParam>,
) -> Result<CallToolResult, McpError> {
let function_id = params.0.function_id;
log_debug(&format!(
"Tool called: function_timing_logs({})",
function_id
));
match get_function_logs_timing(function_id) {
Some(logs) => {
let current_elapsed_ns = get_current_elapsed_ns();
let formatted = JsonFunctionTimingLogsList::from_logs(&logs, current_elapsed_ns);
Ok(CallToolResult::success(vec![Content::text(to_json(
&formatted,
)?)]))
}
None => Ok(CallToolResult::error(vec![Content::text(format!(
"Function with id {} not found",
function_id
))])),
}
}
#[tool(
description = r#"Get detailed allocation logs for a specific function (requires hotpath-alloc feature).
Returns JSON array of recent allocation logs. Use functions_alloc first to get function IDs, then use this tool to get detailed logs."#
)]
async fn function_alloc_logs(
&self,
params: Parameters<FunctionIdParam>,
) -> Result<CallToolResult, McpError> {
let function_id = params.0.function_id;
log_debug(&format!(
"Tool called: function_alloc_logs({})",
function_id
));
match get_function_logs_alloc(function_id) {
Some(logs) => {
let current_elapsed_ns = get_current_elapsed_ns();
let formatted = JsonFunctionAllocLogsList::from_logs(&logs, current_elapsed_ns);
Ok(CallToolResult::success(vec![Content::text(to_json(
&formatted,
)?)]))
}
None => Ok(CallToolResult::error(vec![Content::text(
"Memory profiling not available - enable hotpath-alloc feature",
)])),
}
}
#[tool(description = r#"Get detailed message logs for a specific channel.
Returns JSON array of recent send/receive events with timestamps. Use channels first to get channel IDs, then use this tool to get detailed logs."#)]
async fn channel_logs(
&self,
params: Parameters<ChannelIdParam>,
) -> Result<CallToolResult, McpError> {
let channel_id = params.0.channel_id;
log_debug(&format!("Tool called: channel_logs({})", channel_id));
match get_channel_logs(channel_id) {
Some(logs) => {
let current_elapsed_ns = get_current_elapsed_ns();
let formatted = JsonChannelLogsList::from_logs(&logs, current_elapsed_ns);
Ok(CallToolResult::success(vec![Content::text(to_json(
&formatted,
)?)]))
}
None => Ok(CallToolResult::error(vec![Content::text(
"Channel not found",
)])),
}
}
#[tool(description = r#"Get detailed item logs for a specific stream.
Returns JSON array of recent yield events with timestamps. Use streams first to get stream IDs, then use this tool to get detailed logs."#)]
async fn stream_logs(
&self,
params: Parameters<StreamIdParam>,
) -> Result<CallToolResult, McpError> {
let stream_id = params.0.stream_id;
log_debug(&format!("Tool called: stream_logs({})", stream_id));
match get_stream_logs(stream_id) {
Some(logs) => {
let current_elapsed_ns = get_current_elapsed_ns();
let formatted = JsonStreamLogsList::from_logs(&logs, current_elapsed_ns);
Ok(CallToolResult::success(vec![Content::text(to_json(
&formatted,
)?)]))
}
None => Ok(CallToolResult::error(vec![Content::text(
"Stream not found",
)])),
}
}
#[tool(description = r#"Get detailed call/poll logs for a specific future.
Returns JSON array of poll events and completion status. Use futures first to get future IDs, then use this tool to get detailed logs."#)]
async fn future_logs(
&self,
params: Parameters<FutureIdParam>,
) -> Result<CallToolResult, McpError> {
let future_id = params.0.future_id;
log_debug(&format!("Tool called: future_logs({})", future_id));
match get_future_logs_list(future_id) {
Some(calls) => {
let formatted = JsonFutureLogsList::from(&calls);
Ok(CallToolResult::success(vec![Content::text(to_json(
&formatted,
)?)]))
}
None => Ok(CallToolResult::error(vec![Content::text(
"Future not found",
)])),
}
}
#[tool(description = r#"Get metrics for all tracked gauges.
Returns JSON array with:
- id: gauge identifier
- key: gauge name/label
- current_value: current numeric value
- min_value: minimum value seen
- max_value: maximum value seen
- update_count: number of set/inc/dec operations
Use gauges to track numeric values that change over time like queue sizes, connection counts, or custom metrics."#)]
async fn gauges(&self) -> Result<CallToolResult, McpError> {
log_debug("Tool called: gauges");
let gauges = get_debug_gauge_entries_json();
Ok(CallToolResult::success(vec![Content::text(to_json(
&gauges,
)?)]))
}
#[tool(description = r#"Get detailed logs for a specific gauge.
Returns JSON array of recent value updates with timestamps. Use gauges first to get gauge IDs, then use this tool to get detailed logs."#)]
async fn gauge_logs(
&self,
params: Parameters<GaugeIdParam>,
) -> Result<CallToolResult, McpError> {
let gauge_id = params.0.gauge_id;
log_debug(&format!("Tool called: gauge_logs({})", gauge_id));
match get_debug_gauge_logs(gauge_id) {
Some(logs) => Ok(CallToolResult::success(vec![Content::text(to_json(
&logs,
)?)])),
None => Ok(CallToolResult::error(vec![Content::text(
"Gauge not found",
)])),
}
}
#[tool(description = r#"Get all dbg! debug entries.
Returns JSON array of debug entries with IDs, source locations, expressions, and current values. Use the returned IDs with dbg_logs to get detailed history."#)]
async fn dbg_entries(&self) -> Result<CallToolResult, McpError> {
log_debug("Tool called: dbg_entries");
let entries = get_debug_dbg_entries_json();
Ok(CallToolResult::success(vec![Content::text(to_json(
&entries,
)?)]))
}
#[tool(description = r#"Get all val! value tracking entries.
Returns JSON array of value entries with IDs, keys, source locations, and current values. Use the returned IDs with val_logs to get detailed history."#)]
async fn val_entries(&self) -> Result<CallToolResult, McpError> {
log_debug("Tool called: val_entries");
let entries = get_debug_val_entries_json();
Ok(CallToolResult::success(vec![Content::text(to_json(
&entries,
)?)]))
}
#[tool(description = r#"Get detailed logs for a specific dbg! debug entry.
Returns JSON array of recent debug values with timestamps. Use dbg_entries first to get entry IDs, then use this tool to get detailed logs."#)]
async fn dbg_logs(&self, params: Parameters<DebugIdParam>) -> Result<CallToolResult, McpError> {
let debug_id = params.0.debug_id;
log_debug(&format!("Tool called: dbg_logs({})", debug_id));
match get_dbg_logs(debug_id) {
Some(logs) => Ok(CallToolResult::success(vec![Content::text(to_json(
&logs,
)?)])),
None => Ok(CallToolResult::error(vec![Content::text(
"Debug entry not found",
)])),
}
}
#[tool(description = r#"Get detailed logs for a specific val! debug entry.
Returns JSON array of recent value updates with timestamps. Use val_entries first to get entry IDs, then use this tool to get detailed logs."#)]
async fn val_logs(&self, params: Parameters<DebugIdParam>) -> Result<CallToolResult, McpError> {
let debug_id = params.0.debug_id;
log_debug(&format!("Tool called: val_logs({})", debug_id));
match get_val_logs(debug_id) {
Some(logs) => Ok(CallToolResult::success(vec![Content::text(to_json(
&logs,
)?)])),
None => Ok(CallToolResult::error(vec![Content::text(
"Value entry not found",
)])),
}
}
#[tool(
description = r#"Get Tokio runtime metrics snapshot (requires tokio feature).
Returns JSON with per-worker stats (park count, busy duration, poll count, steal count) and global stats (alive tasks, queue depths, blocking threads, IO driver metrics). Requires calling hotpath::tokio_runtime!() in the profiled application."#
)]
async fn tokio_runtime(&self) -> Result<CallToolResult, McpError> {
log_debug("Tool called: tokio_runtime");
#[cfg(feature = "tokio")]
match crate::tokio_runtime::get_runtime_json() {
Some(snapshot) => {
return Ok(CallToolResult::success(vec![Content::text(to_json(
&snapshot,
)?)]));
}
None => {
return Ok(CallToolResult::error(vec![Content::text(
"Tokio runtime metrics not available - use hotpath::tokio_runtime!() to start collection",
)]));
}
}
#[cfg(not(feature = "tokio"))]
Ok(CallToolResult::error(vec![Content::text(
"Tokio runtime metrics not available - enable tokio feature",
)]))
}
#[tool(description = r#"Get profiler status including uptime.
Returns JSON with:
- uptime: human-readable duration since profiler started (e.g. "1m 23s", "2h 5m 30s")
Use to check if the profiler is running and how long it has been active."#)]
async fn profiler_status(&self) -> Result<CallToolResult, McpError> {
log_debug("Tool called: profiler_status");
let status = JsonProfilerStatus {
uptime: format_duration(get_current_elapsed_ns()),
pid: std::process::id(),
};
Ok(CallToolResult::success(vec![Content::text(to_json(
&status,
)?)]))
}
}
#[cfg_attr(feature = "hotpath-meta", hotpath_meta::measure(log = true))]
fn get_current_elapsed_ns() -> u64 {
crate::lib_on::current_elapsed_ns()
}
#[tool_handler(router = self.tool_router)]
impl ServerHandler for HotPathMcpServer {
fn get_info(&self) -> ServerInfo {
let mut server_info = Implementation::default();
server_info.name = "hotpath".into();
server_info.version = env!("CARGO_PKG_VERSION").into();
let mut info = ServerInfo::default();
info.protocol_version = ProtocolVersion::V_2024_11_05;
info.capabilities = ServerCapabilities::builder().enable_tools().build();
info.server_info = server_info;
info.instructions = Some(
"hotpath profiler metrics MCP server. Provides tools to query profiling data.".into(),
);
info
}
}
#[cfg_attr(feature = "hotpath-meta", hotpath_meta::measure(log = true))]
fn to_json<T: serde::Serialize>(value: &T) -> Result<String, McpError> {
serde_json::to_string(value)
.map_err(|e| McpError::internal_error(format!("Failed to serialize metrics: {}", e), None))
}
fn constant_time_eq(a: &[u8], b: &[u8]) -> bool {
if a.len() != b.len() {
return false;
}
a.iter()
.zip(b.iter())
.fold(0u8, |acc, (x, y)| acc | (x ^ y))
== 0
}
fn check_auth(expected: Option<&str>, provided: Option<&str>) -> bool {
match expected {
None => true,
Some(expected) => provided
.map(|p| constant_time_eq(p.as_bytes(), expected.as_bytes()))
.unwrap_or(false),
}
}
async fn auth_middleware(
request: axum::extract::Request,
next: axum::middleware::Next,
) -> Result<axum::response::Response, axum::http::StatusCode> {
let expected = std::env::var("HOTPATH_MCP_AUTH_TOKEN")
.ok()
.filter(|s| !s.is_empty());
let provided = request
.headers()
.get(axum::http::header::AUTHORIZATION)
.and_then(|h| h.to_str().ok());
if check_auth(expected.as_deref(), provided) {
Ok(next.run(request).await)
} else {
Err(axum::http::StatusCode::UNAUTHORIZED)
}
}
static MCP_SERVER_STARTED: OnceLock<()> = OnceLock::new();
#[cfg_attr(feature = "hotpath-meta", hotpath_meta::measure(log = true))]
pub(crate) fn start_mcp_server_once() {
MCP_SERVER_STARTED.get_or_init(|| {
let port = *MCP_SERVER_PORT;
let auth_enabled = std::env::var("HOTPATH_MCP_AUTH_TOKEN")
.ok()
.filter(|s| !s.is_empty())
.is_some();
log_debug(&format!(
"Starting MCP server on port {} (auth: {})",
port,
if auth_enabled { "enabled" } else { "disabled" }
));
std::thread::Builder::new()
.name("hp-mcp".into())
.spawn(move || {
let rt = tokio::runtime::Builder::new_current_thread()
.enable_all()
.build()
.expect("Failed to create MCP runtime");
rt.block_on(async move {
let cancellation_token = CancellationToken::new();
let mut config = StreamableHttpServerConfig::default();
config.sse_keep_alive = Some(Duration::from_secs(15));
config.stateful_mode = true;
config.cancellation_token = cancellation_token.clone();
let service = StreamableHttpService::new(
|| Ok(HotPathMcpServer::new()),
Arc::new(LocalSessionManager::default()),
config,
);
let app = Router::new()
.nest_service("/mcp", service)
.layer(axum::middleware::from_fn(auth_middleware));
let addr = format!("127.0.0.1:{}", port);
let listener = match tokio::net::TcpListener::bind(&addr).await {
Ok(l) => l,
Err(e) => {
log_debug(&format!("Failed to bind to {}: {}", addr, e));
return;
}
};
log_debug(&format!("Listening on http://{}/mcp", addr));
let _ = axum::serve(listener, app)
.with_graceful_shutdown(async move {
cancellation_token.cancelled().await;
})
.await;
});
})
.expect("Failed to spawn MCP server thread");
});
}
#[cfg(feature = "dev")]
fn log_debug(msg: &str) {
use std::io::Write;
use std::time::{SystemTime, UNIX_EPOCH};
let _ = std::fs::create_dir_all("log");
if let Ok(mut file) = std::fs::OpenOptions::new()
.create(true)
.append(true)
.open("log/development.log")
{
let secs = SystemTime::now()
.duration_since(UNIX_EPOCH)
.map(|d| d.as_secs())
.unwrap_or(0);
let sod = (secs % 86_400) as u32;
let (hour, min, sec) = (sod / 3600, (sod % 3600) / 60, sod % 60);
let _ = writeln!(
file,
"{:02}:{:02}:{:02} DEBUG [hotpath-mcp] {}",
hour, min, sec, msg
);
}
}
#[cfg(not(feature = "dev"))]
fn log_debug(_msg: &str) {}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn auth_disabled_allows_all() {
assert!(check_auth(None, None));
assert!(check_auth(None, Some("anything")));
}
#[test]
fn auth_enabled_rejects_missing() {
assert!(!check_auth(Some("secret"), None));
}
#[test]
fn auth_enabled_rejects_wrong() {
assert!(!check_auth(Some("secret"), Some("wrong")));
assert!(!check_auth(Some("secret"), Some("Secret")));
assert!(!check_auth(Some("secret"), Some("")));
}
#[test]
fn auth_enabled_accepts_correct() {
assert!(check_auth(Some("secret"), Some("secret")));
assert!(check_auth(Some("Bearer token"), Some("Bearer token")));
}
#[test]
fn id_param_accepts_number_and_string() {
let param: SqlIdParam = serde_json::from_str(r#"{"sql_id": 3}"#).unwrap();
assert_eq!(param.sql_id, 3);
let param: SqlIdParam = serde_json::from_str(r#"{"sql_id": "3"}"#).unwrap();
assert_eq!(param.sql_id, 3);
assert!(serde_json::from_str::<SqlIdParam>(r#"{"sql_id": "abc"}"#).is_err());
assert!(serde_json::from_str::<SqlIdParam>(r#"{"sql_id": -1}"#).is_err());
}
}