use self::args_parser::parse_session_args;
use crate::assets::{
self, AddNodesOptions, AssetSelector, AssetsLayout, SetupOptions, StageProtocolOptions,
};
use crate::control::{ControlRequest, ControlResponse, ControlResult, send_request};
use crate::process::{self, StartPlan};
use crate::state::{
ProcessKind, ProcessStatus, STATE_FILE_NAME, State, spawn_pid_sync_tasks,
spawn_pid_sync_tasks_for_ids,
};
use anyhow::{Context, Result, anyhow};
use backoff::ExponentialBackoff;
use backoff::backoff::Backoff;
use casper_types::contracts::ContractHash;
use casper_types::{
AddressableEntityHash, AsymmetricType, Digest, EntityVersion, Key, PricingMode, PublicKey,
SecretKey, TimeDiff, Transaction, TransactionHash, TransactionRuntimeParams, TransactionV1Hash,
URef,
};
use clap::{Args, ValueEnum};
use futures::StreamExt;
use nix::errno::Errno;
use nix::sys::signal::kill;
use nix::unistd::Pid;
use rmcp::handler::server::{router::tool::ToolRouter, wrapper::Parameters};
use rmcp::model::{CallToolResult, ServerCapabilities, ServerInfo};
use rmcp::service::ServiceExt;
use rmcp::transport::{
StreamableHttpServerConfig,
streamable_http_server::{session::local::LocalSessionManager, tower::StreamableHttpService},
};
use rmcp::{
ErrorData, ServerHandler, tool, tool_handler, tool_router, transport::stdio as mcp_stdio,
};
use serde::{Deserialize, Serialize};
use serde_json::{Value, json};
use std::collections::{HashMap, VecDeque};
use std::path::{Path, PathBuf};
use std::str::FromStr;
use std::sync::Arc;
use std::sync::atomic::{AtomicBool, AtomicU32, AtomicU64, Ordering};
use std::time::{Duration, Instant};
use tokio::fs as tokio_fs;
use tokio::io::{AsyncReadExt, AsyncWriteExt};
use tokio::net::UnixListener;
use tokio::sync::{Mutex, Notify};
use veles_casper_rust_sdk::TransactionV1Builder;
use veles_casper_rust_sdk::jsonrpc::CasperClient;
use veles_casper_rust_sdk::sse::event::SseEvent;
use veles_casper_rust_sdk::sse::{self, config::ListenerConfig};
const DEFAULT_HTTP_BIND: &str = "127.0.0.1:32100";
const DEFAULT_HTTP_PATH: &str = "/mcp";
const DEFAULT_NETWORK_NAME: &str = "casper-dev";
const DEFAULT_NODE_COUNT: u32 = 4;
const DEFAULT_DELAY_SECS: u64 = 3;
const DEFAULT_LOG_LEVEL: &str = "info";
const DEFAULT_NODE_LOG_FORMAT: &str = "json";
const DEFAULT_SEED: &str = "default";
const DEFAULT_TIMEOUT_SECS: u64 = 60;
const DEFAULT_LOG_PAGE_SIZE: usize = 200;
const DEFAULT_SSE_PAGE_SIZE: usize = 200;
const DEFAULT_SSE_HISTORY_CAPACITY: usize = 20_000;
const DEFAULT_PAYMENT_AMOUNT: u64 = 100_000_000_000;
mod args_parser;
#[derive(Debug, Clone, Copy, ValueEnum)]
#[value(rename_all = "kebab-case")]
pub enum McpTransport {
Stdio,
Http,
Both,
}
#[derive(Debug, Args, Clone)]
pub struct McpArgs {
#[arg(long, value_enum, default_value_t = McpTransport::Both)]
transport: McpTransport,
#[arg(long, default_value = DEFAULT_HTTP_BIND)]
http_bind: String,
#[arg(long, default_value = DEFAULT_HTTP_PATH)]
http_path: String,
#[arg(long, value_name = "PATH")]
net_path: Option<PathBuf>,
}
#[derive(Debug, Clone)]
struct McpServer {
manager: Arc<NetworkManager>,
tool_router: ToolRouter<Self>,
}
impl McpServer {
fn new(manager: Arc<NetworkManager>) -> Self {
Self {
manager,
tool_router: Self::tool_router(),
}
}
fn server_info() -> ServerInfo {
ServerInfo {
instructions: Some(
"Control multiple local Casper devnets. Start with spawn_network, then wait_network_ready before RPC or transaction tools. Do not use external casper-client binaries or curl; use MCP tools directly (for example get_transaction, wait_transaction, make_transaction_package_call, make_transaction_contract_call, make_transaction_session_wasm, send_transaction_signed).".to_string(),
),
capabilities: ServerCapabilities::builder().enable_tools().build(),
..Default::default()
}
}
}
#[tool_router]
impl McpServer {
#[tool(
description = "Spawn or resume a managed network. Defaults to force_setup=true for fresh devnet startup."
)]
async fn spawn_network(
&self,
Parameters(request): Parameters<SpawnNetworkRequest>,
) -> std::result::Result<CallToolResult, ErrorData> {
let result = self
.manager
.spawn_network(request)
.await
.map_err(to_mcp_error)?;
Ok(ok_value(
serde_json::to_value(result).map_err(internal_serde_error)?,
))
}
#[tool(
description = "Wait for a managed network to be ready: processes running, /status healthy, reactor Validate, and at least one block observed."
)]
async fn wait_network_ready(
&self,
Parameters(request): Parameters<WaitNetworkReadyRequest>,
) -> std::result::Result<CallToolResult, ErrorData> {
let result = self
.manager
.wait_network_ready(&request.network_name, request.timeout_seconds)
.await
.map_err(to_mcp_error)?;
Ok(ok_value(
serde_json::to_value(result).map_err(internal_serde_error)?,
))
}
#[tool(
description = "Stage a protocol upgrade from a named custom asset. When the network is running, sidecars are restarted after staging."
)]
async fn stage_protocol(
&self,
Parameters(request): Parameters<StageProtocolRequest>,
) -> std::result::Result<CallToolResult, ErrorData> {
let result = self
.manager
.stage_protocol(
&request.network_name,
request.asset.as_deref(),
request.custom_asset.as_deref(),
request.asset_name.as_deref(),
&request.protocol_version,
request.activation_point,
)
.await
.map_err(to_mcp_error)?;
Ok(ok_value(
serde_json::to_value(result).map_err(internal_serde_error)?,
))
}
#[tool(
description = "Despawn a managed network. By default it stops processes and keeps files; set purge=true to remove files."
)]
async fn despawn_network(
&self,
Parameters(request): Parameters<DespawnNetworkRequest>,
) -> std::result::Result<CallToolResult, ErrorData> {
let result = self
.manager
.despawn_network(&request.network_name, request.purge.unwrap_or(false))
.await
.map_err(to_mcp_error)?;
Ok(ok_value(
serde_json::to_value(result).map_err(internal_serde_error)?,
))
}
#[tool(description = "List discovered network names and managed/running state.")]
async fn list_networks(
&self,
_: Parameters<ListNetworksRequest>,
) -> std::result::Result<CallToolResult, ErrorData> {
let result = self.manager.list_networks().await.map_err(to_mcp_error)?;
Ok(ok_value(
serde_json::to_value(result).map_err(internal_serde_error)?,
))
}
#[tool(
description = "List managed network processes, optionally filtered by process name. Defaults to running_only=true."
)]
async fn managed_processes(
&self,
Parameters(request): Parameters<ManagedProcessesRequest>,
) -> std::result::Result<CallToolResult, ErrorData> {
let result = self
.manager
.managed_processes(request)
.await
.map_err(to_mcp_error)?;
Ok(ok_value(
serde_json::to_value(result).map_err(internal_serde_error)?,
))
}
#[tool(description = "Call node REST /status for a managed network node.")]
async fn status(
&self,
Parameters(request): Parameters<NodeScopedRequest>,
) -> std::result::Result<CallToolResult, ErrorData> {
let network = self
.manager
.get_network(&request.network_name)
.await
.map_err(to_mcp_error)?;
ensure_running_network(&network).await?;
let status = fetch_rest_status(request.node_id)
.await
.map_err(to_mcp_error)?;
Ok(ok_value(status))
}
#[tool(
description = "Run a raw JSON-RPC call against node sidecar endpoint. Useful as a generic query helper."
)]
async fn rpc_query(
&self,
Parameters(request): Parameters<RpcQueryRequest>,
) -> std::result::Result<CallToolResult, ErrorData> {
let network = self
.manager
.get_network(&request.network_name)
.await
.map_err(to_mcp_error)?;
ensure_running_network(&network).await?;
let value = self
.manager
.raw_rpc_query(request.node_id, &request.method, request.params)
.await
.map_err(to_mcp_error)?;
Ok(ok_value(value))
}
#[tool(description = "Typed balance query by account/public key/uref/entity identifier.")]
async fn rpc_query_balance(
&self,
Parameters(request): Parameters<RpcQueryBalanceRequest>,
) -> std::result::Result<CallToolResult, ErrorData> {
let network = self
.manager
.get_network(&request.network_name)
.await
.map_err(to_mcp_error)?;
ensure_running_network(&network).await?;
let block_id = normalize_optional_identifier(request.block_id.as_deref());
let state_root_hash = normalize_optional_identifier(request.state_root_hash.as_deref());
let params =
build_query_balance_params(&request.purse_identifier, &block_id, &state_root_hash)
.map_err(to_mcp_error)?;
let response = self
.manager
.raw_rpc_query(request.node_id, "query_balance", Some(params))
.await
.map_err(to_mcp_error)?;
Ok(ok_value(
extract_rpc_result(response).map_err(to_mcp_error)?,
))
}
#[tool(
description = "Typed global state query by key + optional path + optional block/state-root identifier. If no identifier is provided, the latest block hash is used."
)]
async fn rpc_query_global_state(
&self,
Parameters(request): Parameters<RpcQueryGlobalStateRequest>,
) -> std::result::Result<CallToolResult, ErrorData> {
let network = self
.manager
.get_network(&request.network_name)
.await
.map_err(to_mcp_error)?;
ensure_running_network(&network).await?;
let (block_id, state_root_hash) = resolve_global_state_identifier(
&request.network_name,
request.node_id,
request.block_id.as_deref(),
request.state_root_hash.as_deref(),
)
.await
.map_err(to_mcp_error)?;
let params = build_query_global_state_params(
&request.key,
request.path.unwrap_or_default(),
&block_id,
&state_root_hash,
)
.map_err(to_mcp_error)?;
let response = self
.manager
.raw_rpc_query(request.node_id, "query_global_state", Some(params))
.await
.map_err(to_mcp_error)?;
Ok(ok_value(
extract_rpc_result(response).map_err(to_mcp_error)?,
))
}
#[tool(description = "Get current block height using typed chain_get_block RPC.")]
async fn current_block_height(
&self,
Parameters(request): Parameters<CurrentBlockRequest>,
) -> std::result::Result<CallToolResult, ErrorData> {
let network = self
.manager
.get_network(&request.network_name)
.await
.map_err(to_mcp_error)?;
ensure_running_network(&network).await?;
let value = fetch_block_result(
&self.manager,
&request.network_name,
request.node_id,
request.block_id.as_deref(),
)
.await;
let value = match value {
Ok(value) => value,
Err(err) => {
if request.block_id.is_none()
&& let Some((low, high)) =
parse_no_such_block_range_from_error(&err.to_string())
{
return Ok(ok_value(json!({
"network_name": request.network_name,
"node_id": request.node_id,
"height": high,
"block": Value::Null,
"pending": true,
"available_block_range": {
"low": low,
"high": high,
},
"message": "latest block is not yet queryable; retry shortly",
})));
}
return Err(to_mcp_error(err));
}
};
let height = extract_block_height(&value).ok_or_else(|| {
ErrorData::internal_error("missing block height in RPC response", None)
})?;
Ok(ok_value(json!({
"network_name": request.network_name,
"node_id": request.node_id,
"height": height,
"block": value,
})))
}
#[tool(description = "Get current block payload using typed chain_get_block RPC.")]
async fn current_block(
&self,
Parameters(request): Parameters<CurrentBlockRequest>,
) -> std::result::Result<CallToolResult, ErrorData> {
let network = self
.manager
.get_network(&request.network_name)
.await
.map_err(to_mcp_error)?;
ensure_running_network(&network).await?;
let value = fetch_block_result(
&self.manager,
&request.network_name,
request.node_id,
request.block_id.as_deref(),
)
.await
.map_err(to_mcp_error)?;
Ok(ok_value(value))
}
#[tool(description = "Get paginated node stdout/stderr logs from on-disk files.")]
async fn get_node_logs(
&self,
Parameters(request): Parameters<GetNodeLogsRequest>,
) -> std::result::Result<CallToolResult, ErrorData> {
let network = self
.manager
.get_network(&request.network_name)
.await
.map_err(to_mcp_error)?;
let limit = request.limit.unwrap_or(DEFAULT_LOG_PAGE_SIZE);
if limit == 0 {
return Err(ErrorData::invalid_params(
"limit must be greater than 0",
None,
));
}
let file_name = match request.stream {
NodeLogStream::Stdout => "stdout.log",
NodeLogStream::Stderr => "stderr.log",
};
let path = network
.layout
.node_logs_dir(request.node_id)
.join(file_name);
let page = read_log_page(&path, request.before_line, limit)
.await
.map_err(to_mcp_error)?;
Ok(ok_value(
serde_json::to_value(page).map_err(internal_serde_error)?,
))
}
#[tool(
description = "Wait for the next SSE event after a sequence cursor with optional event type filters."
)]
async fn wait_next_sse(
&self,
Parameters(request): Parameters<WaitNextSseRequest>,
) -> std::result::Result<CallToolResult, ErrorData> {
let network = self
.manager
.get_network(&request.network_name)
.await
.map_err(to_mcp_error)?;
ensure_running_network(&network).await?;
let timeout = Duration::from_secs(request.timeout_seconds.unwrap_or(DEFAULT_TIMEOUT_SECS));
let filter = SseFilter {
include_event_types: request.include_event_types.unwrap_or_default(),
exclude_event_types: request.exclude_event_types.unwrap_or_default(),
};
let after = match request.after_sequence {
Some(value) => value,
None => network.sse_store.latest_sequence().await,
};
let event = network
.sse_store
.wait_next(after, &filter, timeout)
.await
.map_err(to_mcp_error)?;
Ok(ok_value(
serde_json::to_value(event).map_err(internal_serde_error)?,
))
}
#[tool(
description = "Fetch paginated SSE history with sequence cursor and optional event type filters."
)]
async fn sse_history(
&self,
Parameters(request): Parameters<SseHistoryRequest>,
) -> std::result::Result<CallToolResult, ErrorData> {
let network = self
.manager
.get_network(&request.network_name)
.await
.map_err(to_mcp_error)?;
let limit = request.limit.unwrap_or(DEFAULT_SSE_PAGE_SIZE);
if limit == 0 {
return Err(ErrorData::invalid_params(
"limit must be greater than 0",
None,
));
}
let filter = SseFilter {
include_event_types: request.include_event_types.unwrap_or_default(),
exclude_event_types: request.exclude_event_types.unwrap_or_default(),
};
let history = network
.sse_store
.history(request.before_sequence, limit, &filter)
.await;
Ok(ok_value(
serde_json::to_value(history).map_err(internal_serde_error)?,
))
}
#[tool(description = "List derived accounts from derived-accounts.csv for a network.")]
async fn list_derived_accounts(
&self,
Parameters(request): Parameters<ListDerivedAccountsRequest>,
) -> std::result::Result<CallToolResult, ErrorData> {
let accounts = self
.manager
.list_derived_accounts(&request.network_name)
.await
.map_err(to_mcp_error)?;
Ok(ok_value(
serde_json::to_value(accounts).map_err(internal_serde_error)?,
))
}
#[tool(
description = "Submit signed or unsigned transaction JSON. Signer key is derived from signer_path for this managed network. transaction must be a typed Transaction JSON object."
)]
async fn send_transaction_signed(
&self,
Parameters(request): Parameters<SendTransactionSignedRequest>,
) -> std::result::Result<CallToolResult, ErrorData> {
let network = self
.manager
.get_network(&request.network_name)
.await
.map_err(to_mcp_error)?;
ensure_running_network(&network).await?;
let signer = self
.manager
.derived_account_for_path(&network, &request.signer_path)
.await
.map_err(to_mcp_error)?;
verify_path_hash_consistency(&network.layout, &request.signer_path, &signer.account_hash)
.await
.map_err(to_mcp_error)?;
let mut transaction = parse_transaction_json(request.transaction)?;
transaction.sign(&signer.secret_key);
let rpc = mcp_rpc_client(&request.network_name, request.node_id).map_err(to_mcp_error)?;
let response = rpc
.put_transaction(transaction)
.await
.map_err(to_mcp_error)?;
Ok(ok_value(json!({
"network_name": request.network_name,
"node_id": request.node_id,
"transaction_hash": response.transaction_hash.to_hex_string(),
})))
}
#[tool(
description = "Create a stored package-name call transaction (make-transaction style). Returns transaction JSON for follow-up submission with send_transaction_signed. session_args accepts either: (1) array of {name,type,value} objects, e.g. [{\"name\":\"value\",\"type\":\"I32\",\"value\":\"1\"}], or (2) full RuntimeArgs JSON object. Legacy field name session_args_json is accepted as an alias, but values must be typed JSON (not encoded JSON strings). Not supported: object shorthand like {\"value\":1} or casper-client string args like [\"value:i32=1\"]. For composite CLTypes (List/Map/Tuple/Result/ByteArray), value must be hex bytes (0x...)."
)]
async fn make_transaction_package_call(
&self,
Parameters(request): Parameters<MakeTransactionPackageCallRequest>,
) -> std::result::Result<CallToolResult, ErrorData> {
let network = self
.manager
.get_network(&request.network_name)
.await
.map_err(to_mcp_error)?;
ensure_running_network(&network).await?;
let runtime = match (
request.runtime_transferred_value,
request.runtime_seed_hex.as_deref(),
) {
(None, None) => TransactionRuntimeParams::VmCasperV1,
(transferred_value, maybe_seed_hex) => {
let seed = parse_optional_seed_hex(maybe_seed_hex).map_err(to_mcp_error)?;
TransactionRuntimeParams::VmCasperV2 {
transferred_value: transferred_value.unwrap_or(0),
seed,
}
}
};
let mut builder = TransactionV1Builder::new_targeting_package_via_alias(
request.transaction_package_name.clone(),
request.transaction_package_version,
request.session_entry_point.clone(),
runtime,
);
if let Some(args_json) = request.session_args.as_ref()
&& let Some(runtime_args) = parse_session_args(args_json).map_err(to_mcp_error)?
{
builder = builder.with_runtime_args(runtime_args);
}
if let Some(ttl_millis) = request.ttl_millis {
builder = builder.with_ttl(TimeDiff::from_millis(ttl_millis));
}
let pricing = build_pricing_mode(
request.gas_price_tolerance,
Some(request.payment_amount.unwrap_or(DEFAULT_PAYMENT_AMOUNT)),
);
builder = builder.with_pricing_mode(pricing);
let chain_name = match request.chain_name {
Some(value) => value,
None => fetch_chain_name(&request.network_name, request.node_id)
.await
.map_err(to_mcp_error)?,
};
builder = builder.with_chain_name(chain_name.clone());
let mut signed = false;
let mut signer_path = None;
let mut derived_signer = None;
if let Some(path) = request.signer_path {
let signer = self
.manager
.derived_account_for_path(&network, &path)
.await
.map_err(to_mcp_error)?;
verify_path_hash_consistency(&network.layout, &path, &signer.account_hash)
.await
.map_err(to_mcp_error)?;
derived_signer = Some(signer);
signed = true;
signer_path = Some(path);
} else if let Some(initiator_public_key) = request.initiator_public_key {
let public_key = PublicKey::from_hex(initiator_public_key.trim())
.with_context(|| "failed to parse initiator_public_key as hex public key")
.map_err(to_mcp_error)?;
builder = builder.with_initiator_addr(public_key);
} else {
return Err(ErrorData::invalid_params(
"provide signer_path (for signed tx) or initiator_public_key (for unsigned tx)",
None,
));
}
if let Some(signer) = derived_signer.as_ref() {
builder = builder.with_secret_key(&signer.secret_key);
}
let tx = builder.build().map_err(to_mcp_error)?;
let tx_json = serde_json::to_value(Transaction::V1(tx)).map_err(internal_serde_error)?;
Ok(ok_value(json!({
"network_name": request.network_name,
"node_id": request.node_id,
"chain_name": chain_name,
"signed": signed,
"signer_path": signer_path,
"transaction": tx_json,
})))
}
#[tool(
description = "Create a stored contract-hash call transaction (make-transaction style). Returns transaction JSON for follow-up submission with send_transaction_signed. session_args accepts either: (1) array of {name,type,value} objects, e.g. [{\"name\":\"value\",\"type\":\"I32\",\"value\":\"1\"}], or (2) full RuntimeArgs JSON object. Legacy field name session_args_json is accepted as an alias, but values must be typed JSON (not encoded JSON strings). Not supported: object shorthand like {\"value\":1} or casper-client string args like [\"value:i32=1\"]. For composite CLTypes (List/Map/Tuple/Result/ByteArray), value must be hex bytes (0x...)."
)]
async fn make_transaction_contract_call(
&self,
Parameters(request): Parameters<MakeTransactionContractCallRequest>,
) -> std::result::Result<CallToolResult, ErrorData> {
let network = self
.manager
.get_network(&request.network_name)
.await
.map_err(to_mcp_error)?;
ensure_running_network(&network).await?;
let runtime = match (
request.runtime_transferred_value,
request.runtime_seed_hex.as_deref(),
) {
(None, None) => TransactionRuntimeParams::VmCasperV1,
(transferred_value, maybe_seed_hex) => {
let seed = parse_optional_seed_hex(maybe_seed_hex).map_err(to_mcp_error)?;
TransactionRuntimeParams::VmCasperV2 {
transferred_value: transferred_value.unwrap_or(0),
seed,
}
}
};
let contract_hash = parse_contract_hash_for_invocation(&request.transaction_contract_hash)
.map_err(to_mcp_error)?;
let mut builder = TransactionV1Builder::new_targeting_invocable_entity(
contract_hash,
request.session_entry_point.clone(),
runtime,
);
if let Some(args_json) = request.session_args.as_ref()
&& let Some(runtime_args) = parse_session_args(args_json).map_err(to_mcp_error)?
{
builder = builder.with_runtime_args(runtime_args);
}
if let Some(ttl_millis) = request.ttl_millis {
builder = builder.with_ttl(TimeDiff::from_millis(ttl_millis));
}
let pricing = build_pricing_mode(
request.gas_price_tolerance,
Some(request.payment_amount.unwrap_or(DEFAULT_PAYMENT_AMOUNT)),
);
builder = builder.with_pricing_mode(pricing);
let chain_name = match request.chain_name {
Some(value) => value,
None => fetch_chain_name(&request.network_name, request.node_id)
.await
.map_err(to_mcp_error)?,
};
builder = builder.with_chain_name(chain_name.clone());
let mut signed = false;
let mut signer_path = None;
let mut derived_signer = None;
if let Some(path) = request.signer_path {
let signer = self
.manager
.derived_account_for_path(&network, &path)
.await
.map_err(to_mcp_error)?;
verify_path_hash_consistency(&network.layout, &path, &signer.account_hash)
.await
.map_err(to_mcp_error)?;
derived_signer = Some(signer);
signed = true;
signer_path = Some(path);
} else if let Some(initiator_public_key) = request.initiator_public_key {
let public_key = PublicKey::from_hex(initiator_public_key.trim())
.with_context(|| "failed to parse initiator_public_key as hex public key")
.map_err(to_mcp_error)?;
builder = builder.with_initiator_addr(public_key);
} else {
return Err(ErrorData::invalid_params(
"provide signer_path (for signed tx) or initiator_public_key (for unsigned tx)",
None,
));
}
if let Some(signer) = derived_signer.as_ref() {
builder = builder.with_secret_key(&signer.secret_key);
}
let tx = builder.build().map_err(to_mcp_error)?;
let tx_json = serde_json::to_value(Transaction::V1(tx)).map_err(internal_serde_error)?;
Ok(ok_value(json!({
"network_name": request.network_name,
"node_id": request.node_id,
"chain_name": chain_name,
"transaction_contract_hash": request.transaction_contract_hash,
"signed": signed,
"signer_path": signer_path,
"transaction": tx_json,
})))
}
#[tool(
description = "Create a session wasm transaction (make-transaction style). Returns transaction JSON for follow-up submission with send_transaction_signed. session_args accepts either: (1) array of {name,type,value} objects, e.g. [{\"name\":\"value\",\"type\":\"I32\",\"value\":\"1\"}], or (2) full RuntimeArgs JSON object. Legacy field name session_args_json is accepted as an alias, but values must be typed JSON (not encoded JSON strings). Not supported: object shorthand like {\"value\":1} or casper-client string args like [\"value:i32=1\"]. For composite CLTypes (List/Map/Tuple/Result/ByteArray), value must be hex bytes (0x...)."
)]
async fn make_transaction_session_wasm(
&self,
Parameters(request): Parameters<MakeTransactionSessionWasmRequest>,
) -> std::result::Result<CallToolResult, ErrorData> {
let network = self
.manager
.get_network(&request.network_name)
.await
.map_err(to_mcp_error)?;
ensure_running_network(&network).await?;
let wasm_path = PathBuf::from(&request.wasm_path);
let wasm_bytes = tokio_fs::read(&wasm_path)
.await
.with_context(|| format!("failed to read wasm at {}", wasm_path.display()))
.map_err(to_mcp_error)?;
let runtime = match (
request.runtime_transferred_value,
request.runtime_seed_hex.as_deref(),
) {
(None, None) => TransactionRuntimeParams::VmCasperV1,
(transferred_value, maybe_seed_hex) => {
let seed = parse_optional_seed_hex(maybe_seed_hex).map_err(to_mcp_error)?;
TransactionRuntimeParams::VmCasperV2 {
transferred_value: transferred_value.unwrap_or(0),
seed,
}
}
};
let mut builder = TransactionV1Builder::new_session(
request.is_install_upgrade.unwrap_or(true),
wasm_bytes.into(),
runtime,
);
if let Some(args_json) = request.session_args.as_ref()
&& let Some(runtime_args) = parse_session_args(args_json).map_err(to_mcp_error)?
{
builder = builder.with_runtime_args(runtime_args);
}
if let Some(ttl_millis) = request.ttl_millis {
builder = builder.with_ttl(TimeDiff::from_millis(ttl_millis));
}
let pricing = build_pricing_mode(
request.gas_price_tolerance,
Some(request.payment_amount.unwrap_or(DEFAULT_PAYMENT_AMOUNT)),
);
builder = builder.with_pricing_mode(pricing);
let chain_name = match request.chain_name {
Some(value) => value,
None => fetch_chain_name(&request.network_name, request.node_id)
.await
.map_err(to_mcp_error)?,
};
builder = builder.with_chain_name(chain_name.clone());
let mut signed = false;
let mut signer_path = None;
let mut derived_signer = None;
if let Some(path) = request.signer_path {
let signer = self
.manager
.derived_account_for_path(&network, &path)
.await
.map_err(to_mcp_error)?;
verify_path_hash_consistency(&network.layout, &path, &signer.account_hash)
.await
.map_err(to_mcp_error)?;
derived_signer = Some(signer);
signed = true;
signer_path = Some(path);
} else if let Some(initiator_public_key) = request.initiator_public_key {
let public_key = PublicKey::from_hex(initiator_public_key.trim())
.with_context(|| "failed to parse initiator_public_key as hex public key")
.map_err(to_mcp_error)?;
builder = builder.with_initiator_addr(public_key);
} else {
return Err(ErrorData::invalid_params(
"provide signer_path (for signed tx) or initiator_public_key (for unsigned tx)",
None,
));
}
if let Some(signer) = derived_signer.as_ref() {
builder = builder.with_secret_key(&signer.secret_key);
}
let tx = builder.build().map_err(to_mcp_error)?;
let tx_json = serde_json::to_value(Transaction::V1(tx)).map_err(internal_serde_error)?;
Ok(ok_value(json!({
"network_name": request.network_name,
"node_id": request.node_id,
"chain_name": chain_name,
"wasm_path": request.wasm_path,
"signed": signed,
"signer_path": signer_path,
"transaction": tx_json,
})))
}
#[tool(
description = "Build, sign and submit a session wasm transaction from a derived account path. session_args accepts either: (1) array of {name,type,value} objects, e.g. [{\"name\":\"value\",\"type\":\"I32\",\"value\":\"1\"}], or (2) full RuntimeArgs JSON object. Legacy field name session_args_json is accepted as an alias, but values must be typed JSON (not encoded JSON strings). Not supported: object shorthand like {\"value\":1} or casper-client string args like [\"value:i32=1\"]. For composite CLTypes (List/Map/Tuple/Result/ByteArray), value must be hex bytes (0x...)."
)]
async fn send_session_wasm(
&self,
Parameters(request): Parameters<SendSessionWasmRequest>,
) -> std::result::Result<CallToolResult, ErrorData> {
let network = self
.manager
.get_network(&request.network_name)
.await
.map_err(to_mcp_error)?;
ensure_running_network(&network).await?;
let signer = self
.manager
.derived_account_for_path(&network, &request.signer_path)
.await
.map_err(to_mcp_error)?;
verify_path_hash_consistency(&network.layout, &request.signer_path, &signer.account_hash)
.await
.map_err(to_mcp_error)?;
let wasm_path = PathBuf::from(&request.wasm_path);
let wasm_bytes = tokio_fs::read(&wasm_path)
.await
.with_context(|| format!("failed to read wasm at {}", wasm_path.display()))
.map_err(to_mcp_error)?;
let runtime = match (
request.runtime_transferred_value,
request.runtime_seed_hex.as_deref(),
) {
(None, None) => TransactionRuntimeParams::VmCasperV1,
(transferred_value, maybe_seed_hex) => {
let seed = parse_optional_seed_hex(maybe_seed_hex).map_err(to_mcp_error)?;
TransactionRuntimeParams::VmCasperV2 {
transferred_value: transferred_value.unwrap_or(0),
seed,
}
}
};
let mut builder = TransactionV1Builder::new_session(
request.is_install_upgrade.unwrap_or(true),
wasm_bytes.into(),
runtime,
);
if let Some(args_json) = request.session_args.as_ref()
&& let Some(runtime_args) = parse_session_args(args_json).map_err(to_mcp_error)?
{
builder = builder.with_runtime_args(runtime_args);
}
if let Some(ttl_millis) = request.ttl_millis {
builder = builder.with_ttl(TimeDiff::from_millis(ttl_millis));
}
let pricing = build_pricing_mode(
request.gas_price_tolerance,
Some(request.payment_amount.unwrap_or(DEFAULT_PAYMENT_AMOUNT)),
);
builder = builder.with_pricing_mode(pricing);
let chain_name = match request.chain_name {
Some(value) => value,
None => fetch_chain_name(&request.network_name, request.node_id)
.await
.map_err(to_mcp_error)?,
};
let tx = builder
.with_chain_name(chain_name)
.with_secret_key(&signer.secret_key)
.build()
.map_err(to_mcp_error)?;
let rpc = mcp_rpc_client(&request.network_name, request.node_id).map_err(to_mcp_error)?;
let response = rpc
.put_transaction(Transaction::V1(tx))
.await
.map_err(to_mcp_error)?;
Ok(ok_value(json!({
"network_name": request.network_name,
"node_id": request.node_id,
"transaction_hash": response.transaction_hash.to_hex_string(),
})))
}
#[tool(
description = "Transfer tokens between derived account paths. from_path signs, to_path resolves recipient."
)]
async fn transfer_tokens(
&self,
Parameters(request): Parameters<TransferTokensRequest>,
) -> std::result::Result<CallToolResult, ErrorData> {
let network = self
.manager
.get_network(&request.network_name)
.await
.map_err(to_mcp_error)?;
ensure_running_network(&network).await?;
let from = self
.manager
.derived_account_for_path(&network, &request.from_path)
.await
.map_err(to_mcp_error)?;
let to = self
.manager
.derived_account_for_path(&network, &request.to_path)
.await
.map_err(to_mcp_error)?;
verify_path_hash_consistency(&network.layout, &request.from_path, &from.account_hash)
.await
.map_err(to_mcp_error)?;
verify_path_hash_consistency(&network.layout, &request.to_path, &to.account_hash)
.await
.map_err(to_mcp_error)?;
let amount = casper_types::U512::from_dec_str(&request.amount)
.with_context(|| "amount must be a decimal U512 string")
.map_err(to_mcp_error)?;
let to_public_key = PublicKey::from_hex(&to.public_key_hex)
.with_context(|| "failed to parse derived recipient public key")
.map_err(to_mcp_error)?;
let mut builder =
TransactionV1Builder::new_transfer(amount, None, to_public_key, request.transfer_id)
.map_err(to_mcp_error)?;
if let Some(ttl_millis) = request.ttl_millis {
builder = builder.with_ttl(TimeDiff::from_millis(ttl_millis));
}
let pricing = build_pricing_mode(
request.gas_price_tolerance,
Some(request.payment_amount.unwrap_or(DEFAULT_PAYMENT_AMOUNT)),
);
builder = builder.with_pricing_mode(pricing);
let chain_name = match request.chain_name {
Some(value) => value,
None => fetch_chain_name(&request.network_name, request.node_id)
.await
.map_err(to_mcp_error)?,
};
let tx = builder
.with_chain_name(chain_name)
.with_secret_key(&from.secret_key)
.build()
.map_err(to_mcp_error)?;
let rpc = mcp_rpc_client(&request.network_name, request.node_id).map_err(to_mcp_error)?;
let response = rpc
.put_transaction(Transaction::V1(tx))
.await
.map_err(to_mcp_error)?;
Ok(ok_value(json!({
"network_name": request.network_name,
"node_id": request.node_id,
"transaction_hash": response.transaction_hash.to_hex_string(),
"from_path": request.from_path,
"to_path": request.to_path,
"amount": request.amount,
})))
}
#[tool(
description = "Wait for transaction execution result with timeout and return execution effects/result payload."
)]
async fn wait_transaction(
&self,
Parameters(request): Parameters<WaitTransactionRequest>,
) -> std::result::Result<CallToolResult, ErrorData> {
let network = self
.manager
.get_network(&request.network_name)
.await
.map_err(to_mcp_error)?;
ensure_running_network(&network).await?;
let timeout = Duration::from_secs(request.timeout_seconds.unwrap_or(DEFAULT_TIMEOUT_SECS));
let poll = Duration::from_millis(request.poll_interval_millis.unwrap_or(1_000));
let tx_hash =
parse_transaction_hash_input(&request.transaction_hash).map_err(to_mcp_error)?;
let rpc = mcp_rpc_client(&request.network_name, request.node_id).map_err(to_mcp_error)?;
let deadline = Instant::now() + timeout;
loop {
let response = rpc.get_transaction(tx_hash).await.map_err(to_mcp_error)?;
if let Some(exec_info) = response.execution_info.clone()
&& exec_info.execution_result.is_some()
{
return Ok(ok_value(
serde_json::to_value(response).map_err(internal_serde_error)?,
));
}
if Instant::now() >= deadline {
return Err(ErrorData::resource_not_found(
format!(
"transaction {} execution result not available before timeout",
request.transaction_hash
),
None,
));
}
tokio::time::sleep(poll).await;
}
}
#[tool(
description = "Get transaction details via info_get_transaction (non-waiting). Returns typed JSON response from node."
)]
async fn get_transaction(
&self,
Parameters(request): Parameters<GetTransactionRequest>,
) -> std::result::Result<CallToolResult, ErrorData> {
let network = self
.manager
.get_network(&request.network_name)
.await
.map_err(to_mcp_error)?;
ensure_running_network(&network).await?;
let tx_hash =
parse_transaction_hash_input(&request.transaction_hash).map_err(to_mcp_error)?;
let rpc = mcp_rpc_client(&request.network_name, request.node_id).map_err(to_mcp_error)?;
let response = rpc.get_transaction(tx_hash).await.map_err(to_mcp_error)?;
Ok(ok_value(
serde_json::to_value(response).map_err(internal_serde_error)?,
))
}
}
#[tool_handler]
impl ServerHandler for McpServer {
fn get_info(&self) -> ServerInfo {
Self::server_info()
}
}
#[derive(Debug)]
struct NetworkManager {
assets_root: PathBuf,
managed: Mutex<HashMap<String, Arc<ManagedNetwork>>>,
http: reqwest::Client,
}
#[derive(Debug)]
struct ManagedNetwork {
layout: AssetsLayout,
state: Arc<Mutex<State>>,
node_count: AtomicU32,
rust_log: String,
seed: Arc<str>,
sse_store: Arc<SseStore>,
last_block_hook_hash: Mutex<Option<String>>,
shutdown: Arc<AtomicBool>,
control_shutdown: Arc<AtomicBool>,
asset_mutation_lock: Mutex<()>,
sse_tasks: Mutex<Vec<tokio::task::JoinHandle<()>>>,
control_task: Mutex<Option<tokio::task::JoinHandle<()>>>,
}
impl ManagedNetwork {
async fn is_running(&self) -> bool {
let state = self.state.lock().await;
processes_running(&state)
}
async fn stop(&self) -> Result<()> {
self.shutdown.store(true, Ordering::SeqCst);
self.control_shutdown.store(true, Ordering::SeqCst);
if let Some(control_task) = self.control_task.lock().await.take() {
control_task.abort();
let _ = control_task.await;
}
let _ = tokio_fs::remove_file(self.layout.control_socket_path()).await;
let tasks = {
let mut guard = self.sse_tasks.lock().await;
guard.drain(..).collect::<Vec<_>>()
};
for task in tasks {
task.abort();
let _ = task.await;
}
let mut state = self.state.lock().await;
process::stop(&mut state).await
}
}
impl NetworkManager {
async fn new(assets_root: PathBuf) -> Result<Self> {
let http = reqwest::Client::builder()
.no_proxy()
.timeout(Duration::from_secs(5))
.build()?;
Ok(Self {
assets_root,
managed: Mutex::new(HashMap::new()),
http,
})
}
async fn spawn_network(&self, request: SpawnNetworkRequest) -> Result<SpawnNetworkResponse> {
let network_name = request
.network_name
.unwrap_or_else(|| DEFAULT_NETWORK_NAME.to_string());
if network_name.trim().is_empty() {
return Err(anyhow!("network_name must not be empty"));
}
let maybe_existing = {
let mut managed = self.managed.lock().await;
if let Some(existing) = managed.get(&network_name)
&& existing.is_running().await
&& !request.force_setup.unwrap_or(true)
{
return Ok(SpawnNetworkResponse {
network_name,
node_count: existing.node_count.load(Ordering::SeqCst),
managed: true,
already_running: true,
forced_setup: false,
});
}
managed.remove(&network_name)
};
if let Some(existing) = maybe_existing {
let _ = existing.stop().await;
}
let force_setup = request.force_setup.unwrap_or(true);
let requested_nodes = request.node_count.unwrap_or(DEFAULT_NODE_COUNT);
let users = request.users;
let delay = request.delay.unwrap_or(DEFAULT_DELAY_SECS);
let log_level = request
.log_level
.unwrap_or_else(|| DEFAULT_LOG_LEVEL.to_string());
let node_log_format = request
.node_log_format
.unwrap_or_else(|| DEFAULT_NODE_LOG_FORMAT.to_string());
let seed: Arc<str> = Arc::from(request.seed.unwrap_or_else(|| DEFAULT_SEED.to_string()));
let layout = AssetsLayout::new(self.assets_root.clone(), network_name.clone());
let assets_exist = layout.exists().await;
let asset_selector = assets::optional_asset_selector(
request.asset.as_deref(),
request.custom_asset.as_deref(),
)?;
let setup_result = if force_setup {
assets::teardown(&layout).await?;
Some(
assets::setup_local(
&layout,
&SetupOptions {
nodes: requested_nodes,
users,
delay_seconds: delay,
network_name: network_name.clone(),
asset: asset_selector.clone(),
protocol_version: request.protocol_version.clone(),
chainspec_overrides: Vec::new(),
node_log_format,
seed: Arc::clone(&seed),
},
)
.await?,
)
} else if !assets_exist {
Some(
assets::setup_local(
&layout,
&SetupOptions {
nodes: requested_nodes,
users,
delay_seconds: delay,
network_name: network_name.clone(),
asset: asset_selector,
protocol_version: request.protocol_version.clone(),
chainspec_overrides: Vec::new(),
node_log_format,
seed: Arc::clone(&seed),
},
)
.await?,
)
} else {
None
};
if !layout.exists().await {
return Err(anyhow!(
"assets missing under {}; call spawn_network with force_setup=true",
layout.net_dir().display()
));
}
assets::ensure_network_hook_samples(&layout).await?;
if !force_setup && assets_exist {
let _ = assets::ensure_consensus_keys(&layout, Arc::clone(&seed)).await?;
}
let node_count = layout.count_nodes().await?;
if node_count == 0 {
return Err(anyhow!("network has no nodes to start"));
}
ensure_sidecar_available(&layout, node_count).await?;
let state_path = layout.net_dir().join(STATE_FILE_NAME);
if !tokio_fs::try_exists(&state_path).await.unwrap_or(false) {
let protocol_version = match &setup_result {
Some(result) => result.protocol_version.clone(),
None => latest_layout_protocol_version(&layout).await?,
};
assets::prepare_genesis_hooks(&layout, &protocol_version).await?;
}
let mut state = State::new(state_path).await?;
let rust_log = log_level.clone();
process::start(
&layout,
&StartPlan {
rust_log: rust_log.clone(),
},
&mut state,
)
.await?;
let managed = Arc::new(ManagedNetwork {
layout: layout.clone(),
state: Arc::new(Mutex::new(state)),
node_count: AtomicU32::new(node_count),
rust_log,
seed,
sse_store: Arc::new(SseStore::new(DEFAULT_SSE_HISTORY_CAPACITY)),
last_block_hook_hash: Mutex::new(None),
shutdown: Arc::new(AtomicBool::new(false)),
control_shutdown: Arc::new(AtomicBool::new(false)),
asset_mutation_lock: Mutex::new(()),
sse_tasks: Mutex::new(Vec::new()),
control_task: Mutex::new(None),
});
spawn_pid_sync_tasks(Arc::clone(&managed.state)).await;
self.spawn_sse_collectors(&managed).await;
if let Err(err) = self.spawn_control_server(&managed).await {
let _ = managed.stop().await;
return Err(err);
}
self.managed
.lock()
.await
.insert(network_name.clone(), Arc::clone(&managed));
Ok(SpawnNetworkResponse {
network_name,
node_count,
managed: true,
already_running: false,
forced_setup: force_setup,
})
}
async fn spawn_sse_collectors(&self, network: &Arc<ManagedNetwork>) {
let mut tasks = Vec::new();
for node_id in 1..=network.node_count.load(Ordering::SeqCst) {
let endpoint = assets::sse_endpoint(node_id);
let network = Arc::clone(network);
let task = tokio::spawn(async move {
run_sse_listener(network, node_id, endpoint).await;
});
tasks.push(task);
}
let mut guard = network.sse_tasks.lock().await;
guard.extend(tasks);
}
async fn spawn_control_server(&self, network: &Arc<ManagedNetwork>) -> Result<()> {
let socket_path = network.layout.control_socket_path();
if let Ok(metadata) = tokio_fs::symlink_metadata(&socket_path).await {
if metadata.is_dir() {
tokio_fs::remove_dir_all(&socket_path).await?;
} else {
tokio_fs::remove_file(&socket_path).await?;
}
}
if let Some(parent) = socket_path.parent() {
tokio_fs::create_dir_all(parent).await?;
}
let listener = UnixListener::bind(&socket_path)?;
let network_for_task = Arc::clone(network);
let shutdown = Arc::clone(&network_for_task.control_shutdown);
let task = tokio::spawn(async move {
loop {
if shutdown.load(Ordering::SeqCst) {
break;
}
let accepted =
tokio::time::timeout(Duration::from_millis(250), listener.accept()).await;
let (stream, _) = match accepted {
Ok(Ok(pair)) => pair,
Ok(Err(_)) => break,
Err(_) => continue,
};
let network = Arc::clone(&network_for_task);
tokio::spawn(async move {
handle_managed_control_stream(stream, network).await;
});
}
let _ = tokio_fs::remove_file(&socket_path).await;
});
*network.control_task.lock().await = Some(task);
Ok(())
}
async fn stage_protocol(
&self,
network_name: &str,
asset: Option<&str>,
custom_asset: Option<&str>,
legacy_asset_name: Option<&str>,
protocol_version: &str,
activation_point: u64,
) -> Result<StageProtocolResponse> {
let asset_selector = stage_asset_selector(asset, custom_asset, legacy_asset_name)?;
let managed = {
let guard = self.managed.lock().await;
guard.get(network_name).cloned()
};
if let Some(network) = managed
&& network.is_running().await
{
if let Ok(Some(current_era)) = read_current_era_from_status(1).await
&& activation_point <= current_era
{
return Err(anyhow!(
"activation point {} must be greater than current era {}",
activation_point,
current_era
));
}
let request = ControlRequest::StageProtocol {
asset: match &asset_selector {
AssetSelector::Versioned(asset) => Some(asset.clone()),
AssetSelector::LatestVersioned | AssetSelector::Custom(_) => None,
},
custom_asset: match &asset_selector {
AssetSelector::Custom(asset) => Some(asset.clone()),
AssetSelector::LatestVersioned | AssetSelector::Versioned(_) => None,
},
asset_name: None,
protocol_version: protocol_version.to_string(),
activation_point,
chainspec_overrides: Vec::new(),
restart_sidecars: true,
rust_log: Some(network.rust_log.clone()),
};
let socket_path = network.layout.control_socket_path();
return match send_request(&socket_path, &request).await {
Ok(ControlResponse::Ok { result }) => match result {
ControlResult::StageProtocol {
live_mode,
staged_nodes,
restarted_sidecars,
} => Ok(StageProtocolResponse {
network_name: network_name.to_string(),
live_mode,
staged_nodes,
restarted_sidecars,
}),
ControlResult::RuntimeStatus { .. } => Err(anyhow!(
"unexpected runtime_status response from {}",
socket_path.display()
)),
ControlResult::AddNodes { .. } => Err(anyhow!(
"unexpected add_nodes response from {}",
socket_path.display()
)),
},
Ok(ControlResponse::Error { error }) => Err(anyhow!(error)),
Err(err) => Err(anyhow!(
"failed to reach managed control socket {}: {}",
socket_path.display(),
err
)),
};
}
let layout = AssetsLayout::new(self.assets_root.clone(), network_name.to_string());
if !layout.exists().await {
return Err(anyhow!(
"assets for {} not found under {}",
network_name,
layout.net_dir().display()
));
}
let staged = assets::stage_protocol(
&layout,
&StageProtocolOptions {
asset: asset_selector,
protocol_version: protocol_version.to_string(),
activation_point,
chainspec_overrides: Vec::new(),
},
)
.await?;
Ok(StageProtocolResponse {
network_name: network_name.to_string(),
live_mode: false,
staged_nodes: staged.staged_nodes,
restarted_sidecars: Vec::new(),
})
}
async fn wait_network_ready(
&self,
network_name: &str,
timeout_seconds: Option<u64>,
) -> Result<WaitReadyResponse> {
let network = self.get_network(network_name).await?;
let timeout = Duration::from_secs(timeout_seconds.unwrap_or(DEFAULT_TIMEOUT_SECS));
let deadline = Instant::now() + timeout;
loop {
let running = ensure_running_network(&network).await.is_ok();
if running {
let status = check_rest_ready(&network).await;
if let Ok(rest_by_node) = status {
let state = network.state.lock().await;
let block_observed = state.last_block_height.is_some()
|| rest_by_node.values().any(rest_has_block);
if block_observed {
return Ok(WaitReadyResponse {
network_name: network_name.to_string(),
ready: true,
node_count: network.node_count.load(Ordering::SeqCst),
rest: rest_by_node,
last_block_height: state.last_block_height,
});
}
}
}
if Instant::now() >= deadline {
return Ok(WaitReadyResponse {
network_name: network_name.to_string(),
ready: false,
node_count: network.node_count.load(Ordering::SeqCst),
rest: HashMap::new(),
last_block_height: None,
});
}
tokio::time::sleep(Duration::from_millis(500)).await;
}
}
async fn despawn_network(&self, network_name: &str, purge: bool) -> Result<DespawnResponse> {
let managed = {
let mut guard = self.managed.lock().await;
guard.remove(network_name)
};
let Some(network) = managed else {
return Err(anyhow!("network '{}' is not managed", network_name));
};
network.stop().await?;
if purge {
assets::teardown(&network.layout).await?;
}
Ok(DespawnResponse {
network_name: network_name.to_string(),
purged: purge,
})
}
async fn list_networks(&self) -> Result<ListNetworksResponse> {
let discovered = discover_network_names(&self.assets_root).await?;
let managed_snapshot = {
let guard = self.managed.lock().await;
guard
.iter()
.map(|(name, network)| (name.clone(), Arc::clone(network)))
.collect::<Vec<_>>()
};
let mut rows = Vec::new();
for name in &discovered {
if let Some((_, network)) = managed_snapshot.iter().find(|(n, _)| n == name) {
rows.push(NetworkRow {
network_name: name.clone(),
discovered: true,
managed: true,
running: network.is_running().await,
node_count: Some(network.node_count.load(Ordering::SeqCst)),
});
} else {
let layout = AssetsLayout::new(self.assets_root.clone(), name.clone());
rows.push(NetworkRow {
network_name: name.clone(),
discovered: true,
managed: false,
running: false,
node_count: layout.count_nodes().await.ok(),
});
}
}
for (name, network) in managed_snapshot {
if !discovered.iter().any(|candidate| candidate == &name) {
rows.push(NetworkRow {
network_name: name,
discovered: false,
managed: true,
running: network.is_running().await,
node_count: Some(network.node_count.load(Ordering::SeqCst)),
});
}
}
rows.sort_by(|a, b| a.network_name.cmp(&b.network_name));
Ok(ListNetworksResponse { networks: rows })
}
async fn managed_processes(
&self,
request: ManagedProcessesRequest,
) -> Result<ManagedProcessesResponse> {
let network = self.get_network(&request.network_name).await?;
let running_only = request.running_only.unwrap_or(true);
let process_name = request
.process_name
.as_deref()
.map(str::trim)
.filter(|name| !name.is_empty())
.map(ToString::to_string);
let process_name_lc = process_name.as_ref().map(|name| name.to_ascii_lowercase());
let state = network.state.lock().await;
let mut processes = Vec::new();
for process in &state.processes {
if let Some(name_lc) = process_name_lc.as_deref()
&& !process.id.to_ascii_lowercase().contains(name_lc)
{
continue;
}
let pid = process.current_pid();
let running = matches!(process.last_status, ProcessStatus::Running)
&& pid.is_some_and(is_pid_running);
if running_only && !running {
continue;
}
processes.push(ManagedProcessRow {
id: process.id.clone(),
node_id: process.node_id,
kind: process_kind_name(&process.kind).to_string(),
pid,
running,
last_status: process_status_name(&process.last_status).to_string(),
command: process.command.clone(),
args: process.args.clone(),
cwd: process.cwd.clone(),
stdout_path: process.stdout_path.clone(),
stderr_path: process.stderr_path.clone(),
});
}
Ok(ManagedProcessesResponse {
network_name: request.network_name,
running_only,
process_name,
processes,
})
}
async fn get_network(&self, network_name: &str) -> Result<Arc<ManagedNetwork>> {
let guard = self.managed.lock().await;
guard.get(network_name).cloned().ok_or_else(|| {
anyhow!(
"network '{}' is not managed; call spawn_network first",
network_name
)
})
}
async fn stop_all_networks(&self) -> Result<()> {
let managed = {
let mut guard = self.managed.lock().await;
guard
.drain()
.map(|(_, network)| network)
.collect::<Vec<_>>()
};
let mut errors = Vec::new();
for network in managed {
if let Err(err) = network.stop().await {
errors.push(err.to_string());
}
}
if errors.is_empty() {
Ok(())
} else {
Err(anyhow!(errors.join("\n")))
}
}
async fn raw_rpc_query(
&self,
node_id: u32,
method: &str,
params: Option<Value>,
) -> Result<Value> {
let endpoint = assets::rpc_endpoint(node_id);
let payload = match params {
Some(params) => json!({
"id": 1,
"jsonrpc": "2.0",
"method": method,
"params": params,
}),
None => json!({
"id": 1,
"jsonrpc": "2.0",
"method": method,
}),
};
let response = self
.http
.post(endpoint)
.json(&payload)
.send()
.await?
.error_for_status()?;
Ok(response.json::<Value>().await?)
}
async fn list_derived_accounts(&self, network_name: &str) -> Result<Vec<DerivedAccountRow>> {
let layout = AssetsLayout::new(self.assets_root.clone(), network_name.to_string());
let csv = assets::derived_accounts_summary(&layout)
.await
.ok_or_else(|| {
anyhow!(
"missing derived-accounts.csv for network '{}'",
network_name
)
})?;
let seed = {
let managed = self.managed.lock().await;
managed
.get(network_name)
.map(|network| Arc::clone(&network.seed))
};
parse_derived_accounts_csv(&csv, seed).await
}
async fn derived_account_for_path(
&self,
network: &Arc<ManagedNetwork>,
path: &str,
) -> Result<DerivedSigner> {
let material =
assets::derive_account_from_seed_path(Arc::clone(&network.seed), path).await?;
let secret_key = SecretKey::from_pem(&material.secret_key_pem)?;
Ok(DerivedSigner {
public_key_hex: material.public_key_hex,
account_hash: material.account_hash,
secret_key,
})
}
}
async fn handle_managed_control_stream(
mut stream: tokio::net::UnixStream,
network: Arc<ManagedNetwork>,
) {
let mut request_bytes = Vec::new();
let response = match stream.read_to_end(&mut request_bytes).await {
Ok(_) => match serde_json::from_slice::<ControlRequest>(&request_bytes) {
Ok(request) => handle_managed_control_request(&network, request).await,
Err(err) => ControlResponse::Error {
error: format!("invalid control request: {}", err),
},
},
Err(err) => ControlResponse::Error {
error: format!("failed to read control request: {}", err),
},
};
let response_bytes = serde_json::to_vec(&response).unwrap_or_else(|err| {
format!(
"{{\"status\":\"error\",\"error\":\"failed to serialize control response: {}\"}}",
err
)
.into_bytes()
});
let _ = stream.write_all(&response_bytes).await;
let _ = stream.shutdown().await;
}
async fn spawn_added_sse_collectors(network: &Arc<ManagedNetwork>, node_ids: &[u32]) {
let mut tasks = Vec::new();
for node_id in node_ids {
let node_id = *node_id;
let endpoint = assets::sse_endpoint(node_id);
let network = Arc::clone(network);
let task = tokio::spawn(async move {
run_sse_listener(network, node_id, endpoint).await;
});
tasks.push(task);
}
let mut guard = network.sse_tasks.lock().await;
guard.extend(tasks);
}
async fn handle_managed_control_request(
network: &Arc<ManagedNetwork>,
request: ControlRequest,
) -> ControlResponse {
match request {
ControlRequest::RuntimeStatus => {
let state = network.state.lock().await;
ControlResponse::Ok {
result: ControlResult::RuntimeStatus {
running_node_ids: running_node_ids(&state),
last_block_height: state.last_block_height,
},
}
}
ControlRequest::StageProtocol {
asset,
custom_asset,
asset_name,
protocol_version,
activation_point,
chainspec_overrides,
restart_sidecars,
rust_log,
} => {
let asset_selector = match stage_asset_selector(
asset.as_deref(),
custom_asset.as_deref(),
asset_name.as_deref(),
) {
Ok(asset_selector) => asset_selector,
Err(err) => {
return ControlResponse::Error {
error: err.to_string(),
};
}
};
let _asset_mutation_guard = network.asset_mutation_lock.lock().await;
if let Err(err) =
assets::ensure_consensus_keys(&network.layout, Arc::clone(&network.seed)).await
{
return ControlResponse::Error {
error: format!("failed to recreate consensus keys: {}", err),
};
}
let staged = assets::stage_protocol(
&network.layout,
&StageProtocolOptions {
asset: asset_selector,
protocol_version,
activation_point,
chainspec_overrides,
},
)
.await;
let staged = match staged {
Ok(staged) => staged,
Err(err) => {
return ControlResponse::Error {
error: err.to_string(),
};
}
};
let mut restarted_sidecars = Vec::new();
if restart_sidecars {
let rust_log = rust_log.unwrap_or_else(|| network.rust_log.clone());
let mut state = network.state.lock().await;
match process::restart_sidecars(&network.layout, &mut state, &rust_log).await {
Ok(restarted) => {
for proc in restarted {
restarted_sidecars.push(proc.record.node_id);
}
}
Err(err) => {
return ControlResponse::Error {
error: err.to_string(),
};
}
}
}
ControlResponse::Ok {
result: ControlResult::StageProtocol {
live_mode: true,
staged_nodes: staged.staged_nodes,
restarted_sidecars,
},
}
}
ControlRequest::AddNodes { count } => {
let _asset_mutation_guard = network.asset_mutation_lock.lock().await;
let added = match assets::add_nodes(
&network.layout,
&AddNodesOptions {
count,
seed: Arc::clone(&network.seed),
},
)
.await
{
Ok(added) => added,
Err(err) => {
return ControlResponse::Error {
error: err.to_string(),
};
}
};
let started = {
let mut state = network.state.lock().await;
process::start_added_nodes(
&network.layout,
&mut state,
&added.added_node_ids,
added.total_nodes,
&network.rust_log,
)
.await
};
let started = match started {
Ok(started) => started,
Err(err) => {
let error =
rollback_added_nodes_after_start_error(&network.layout, &added, err).await;
return ControlResponse::Error { error };
}
};
let process_ids = started
.iter()
.map(|proc| proc.record.id.clone())
.collect::<Vec<_>>();
let started_processes = started.len() as u32;
network
.node_count
.store(added.total_nodes, Ordering::SeqCst);
spawn_added_sse_collectors(network, &added.added_node_ids).await;
spawn_pid_sync_tasks_for_ids(Arc::clone(&network.state), &process_ids).await;
ControlResponse::Ok {
result: ControlResult::AddNodes {
added_node_ids: added.added_node_ids,
total_nodes: added.total_nodes,
started_processes,
},
}
}
}
}
async fn rollback_added_nodes_after_start_error(
layout: &AssetsLayout,
added: &assets::AddNodesResult,
err: anyhow::Error,
) -> String {
match assets::rollback_added_nodes(layout, &added.added_node_ids).await {
Ok(()) => err.to_string(),
Err(rollback_err) => format!(
"{}; failed to roll back added node assets: {}",
err, rollback_err
),
}
}
async fn read_current_era_from_status(node_id: u32) -> Result<Option<u64>> {
let value = fetch_rest_status(node_id).await?;
Ok(value
.get("last_added_block_info")
.and_then(|info| info.get("era_id"))
.and_then(|era_id| {
era_id
.as_u64()
.or_else(|| era_id.as_str().and_then(|raw| raw.parse::<u64>().ok()))
}))
}
#[derive(Debug)]
struct SseStore {
sequence: AtomicU64,
events: Mutex<VecDeque<SseRecord>>,
notify: Notify,
capacity: usize,
}
impl SseStore {
fn new(capacity: usize) -> Self {
Self {
sequence: AtomicU64::new(0),
events: Mutex::new(VecDeque::new()),
notify: Notify::new(),
capacity,
}
}
async fn latest_sequence(&self) -> u64 {
self.sequence.load(Ordering::SeqCst)
}
async fn push(&self, node_id: u32, event: SseEvent) {
let sequence = self.sequence.fetch_add(1, Ordering::SeqCst) + 1;
let event_type = sse_event_type(&event).to_string();
let payload = sse_event_payload(&event);
let record = SseRecord {
sequence,
timestamp_rfc3339: timestamp_prefix(),
node_id,
event_type,
payload,
};
let mut guard = self.events.lock().await;
guard.push_back(record);
while guard.len() > self.capacity {
let _ = guard.pop_front();
}
drop(guard);
self.notify.notify_waiters();
}
async fn wait_next(
&self,
after_sequence: u64,
filter: &SseFilter,
timeout: Duration,
) -> Result<SseRecord> {
let deadline = Instant::now() + timeout;
loop {
if let Some(record) = self.find_first_after(after_sequence, filter).await {
return Ok(record);
}
if Instant::now() >= deadline {
return Err(anyhow!("timed out waiting for SSE event"));
}
let remaining = deadline.saturating_duration_since(Instant::now());
tokio::time::timeout(remaining, self.notify.notified()).await?;
}
}
async fn find_first_after(&self, after_sequence: u64, filter: &SseFilter) -> Option<SseRecord> {
let guard = self.events.lock().await;
guard
.iter()
.find(|event| event.sequence > after_sequence && filter.matches(event))
.cloned()
}
async fn history(
&self,
before_sequence: Option<u64>,
limit: usize,
filter: &SseFilter,
) -> SseHistoryPage {
let before = before_sequence.unwrap_or(u64::MAX);
let guard = self.events.lock().await;
let mut matched = guard
.iter()
.filter(|event| event.sequence < before)
.filter(|event| filter.matches(event))
.cloned()
.collect::<Vec<_>>();
let total = matched.len();
if matched.len() > limit {
matched = matched.split_off(matched.len() - limit);
}
let next_before_sequence = matched.first().map(|event| event.sequence);
SseHistoryPage {
total_matching: total,
returned: matched.len(),
next_before_sequence,
events: matched,
}
}
}
#[derive(Debug, Default)]
struct SseFilter {
include_event_types: Vec<String>,
exclude_event_types: Vec<String>,
}
impl SseFilter {
fn matches(&self, event: &SseRecord) -> bool {
let include = if self.include_event_types.is_empty() {
true
} else {
self.include_event_types
.iter()
.any(|name| name == &event.event_type)
};
if !include {
return false;
}
!self
.exclude_event_types
.iter()
.any(|name| name == &event.event_type)
}
}
#[derive(Debug, Clone, Serialize, Deserialize, rmcp::schemars::JsonSchema)]
struct SseRecord {
sequence: u64,
timestamp_rfc3339: String,
node_id: u32,
event_type: String,
payload: Value,
}
#[derive(Debug, Clone, Serialize, Deserialize, rmcp::schemars::JsonSchema)]
struct SseHistoryPage {
total_matching: usize,
returned: usize,
next_before_sequence: Option<u64>,
events: Vec<SseRecord>,
}
#[derive(Debug)]
struct DerivedSigner {
public_key_hex: String,
account_hash: String,
secret_key: SecretKey,
}
#[derive(Debug, Deserialize, rmcp::schemars::JsonSchema)]
struct SpawnNetworkRequest {
network_name: Option<String>,
asset: Option<String>,
custom_asset: Option<String>,
protocol_version: Option<String>,
node_count: Option<u32>,
users: Option<u32>,
delay: Option<u64>,
log_level: Option<String>,
node_log_format: Option<String>,
seed: Option<String>,
force_setup: Option<bool>,
}
#[derive(Debug, Serialize, rmcp::schemars::JsonSchema)]
struct SpawnNetworkResponse {
network_name: String,
node_count: u32,
managed: bool,
already_running: bool,
forced_setup: bool,
}
#[derive(Debug, Deserialize, rmcp::schemars::JsonSchema)]
struct WaitNetworkReadyRequest {
network_name: String,
timeout_seconds: Option<u64>,
}
#[derive(Debug, Serialize, rmcp::schemars::JsonSchema)]
struct WaitReadyResponse {
network_name: String,
ready: bool,
node_count: u32,
rest: HashMap<u32, Value>,
last_block_height: Option<u64>,
}
#[derive(Debug, Deserialize, rmcp::schemars::JsonSchema)]
struct StageProtocolRequest {
network_name: String,
asset: Option<String>,
custom_asset: Option<String>,
asset_name: Option<String>,
protocol_version: String,
activation_point: u64,
}
#[derive(Debug, Serialize, rmcp::schemars::JsonSchema)]
struct StageProtocolResponse {
network_name: String,
live_mode: bool,
staged_nodes: u32,
restarted_sidecars: Vec<u32>,
}
#[derive(Debug, Deserialize, rmcp::schemars::JsonSchema)]
struct DespawnNetworkRequest {
network_name: String,
purge: Option<bool>,
}
#[derive(Debug, Serialize, rmcp::schemars::JsonSchema)]
struct DespawnResponse {
network_name: String,
purged: bool,
}
#[derive(Debug, Deserialize, Default, rmcp::schemars::JsonSchema)]
struct ListNetworksRequest {}
#[derive(Debug, Serialize, rmcp::schemars::JsonSchema)]
struct ListNetworksResponse {
networks: Vec<NetworkRow>,
}
#[derive(Debug, Serialize, rmcp::schemars::JsonSchema)]
struct NetworkRow {
network_name: String,
discovered: bool,
managed: bool,
running: bool,
node_count: Option<u32>,
}
#[derive(Debug, Deserialize, rmcp::schemars::JsonSchema)]
struct ManagedProcessesRequest {
network_name: String,
process_name: Option<String>,
running_only: Option<bool>,
}
#[derive(Debug, Serialize, rmcp::schemars::JsonSchema)]
struct ManagedProcessesResponse {
network_name: String,
running_only: bool,
process_name: Option<String>,
processes: Vec<ManagedProcessRow>,
}
#[derive(Debug, Serialize, rmcp::schemars::JsonSchema)]
struct ManagedProcessRow {
id: String,
node_id: u32,
kind: String,
pid: Option<u32>,
running: bool,
last_status: String,
command: String,
args: Vec<String>,
cwd: String,
stdout_path: String,
stderr_path: String,
}
#[derive(Debug, Deserialize, rmcp::schemars::JsonSchema)]
struct NodeScopedRequest {
network_name: String,
node_id: u32,
}
#[derive(Debug, Deserialize, rmcp::schemars::JsonSchema)]
struct RpcQueryRequest {
network_name: String,
node_id: u32,
method: String,
params: Option<Value>,
}
#[derive(Debug, Deserialize, rmcp::schemars::JsonSchema)]
struct RpcQueryBalanceRequest {
network_name: String,
node_id: u32,
purse_identifier: String,
block_id: Option<String>,
state_root_hash: Option<String>,
}
#[derive(Debug, Deserialize, rmcp::schemars::JsonSchema)]
struct RpcQueryGlobalStateRequest {
network_name: String,
node_id: u32,
key: String,
path: Option<Vec<String>>,
block_id: Option<String>,
state_root_hash: Option<String>,
}
#[derive(Debug, Deserialize, rmcp::schemars::JsonSchema)]
struct CurrentBlockRequest {
network_name: String,
node_id: u32,
block_id: Option<String>,
}
#[derive(Debug, Clone, Copy, Deserialize, Serialize, rmcp::schemars::JsonSchema)]
#[serde(rename_all = "lowercase")]
enum NodeLogStream {
Stdout,
Stderr,
}
#[derive(Debug, Deserialize, rmcp::schemars::JsonSchema)]
struct GetNodeLogsRequest {
network_name: String,
node_id: u32,
stream: NodeLogStream,
limit: Option<usize>,
before_line: Option<usize>,
}
#[derive(Debug, Serialize, rmcp::schemars::JsonSchema)]
struct LogLine {
line_number: usize,
content: String,
}
#[derive(Debug, Serialize, rmcp::schemars::JsonSchema)]
struct LogPage {
path: String,
total_lines: usize,
returned: usize,
next_before_line: Option<usize>,
lines: Vec<LogLine>,
}
#[derive(Debug, Deserialize, rmcp::schemars::JsonSchema)]
struct WaitNextSseRequest {
network_name: String,
include_event_types: Option<Vec<String>>,
exclude_event_types: Option<Vec<String>>,
after_sequence: Option<u64>,
timeout_seconds: Option<u64>,
}
#[derive(Debug, Deserialize, rmcp::schemars::JsonSchema)]
struct SseHistoryRequest {
network_name: String,
include_event_types: Option<Vec<String>>,
exclude_event_types: Option<Vec<String>>,
before_sequence: Option<u64>,
limit: Option<usize>,
}
#[derive(Debug, Deserialize, rmcp::schemars::JsonSchema)]
struct ListDerivedAccountsRequest {
network_name: String,
}
#[derive(Debug, Serialize, Deserialize, rmcp::schemars::JsonSchema)]
struct DerivedAccountRow {
kind: String,
name: String,
key_type: String,
derivation: String,
path: String,
account_hash: String,
balance: String,
public_key: Option<String>,
}
#[derive(Debug, Deserialize, rmcp::schemars::JsonSchema)]
struct SendTransactionSignedRequest {
network_name: String,
node_id: u32,
signer_path: String,
transaction: Value,
}
#[derive(Debug, Deserialize, rmcp::schemars::JsonSchema)]
struct MakeTransactionPackageCallRequest {
network_name: String,
node_id: u32,
transaction_package_name: String,
transaction_package_version: Option<EntityVersion>,
session_entry_point: String,
#[serde(alias = "session_args_json")]
session_args: Option<Value>,
signer_path: Option<String>,
initiator_public_key: Option<String>,
chain_name: Option<String>,
ttl_millis: Option<u64>,
gas_price_tolerance: Option<u8>,
payment_amount: Option<u64>,
runtime_transferred_value: Option<u64>,
runtime_seed_hex: Option<String>,
}
#[derive(Debug, Deserialize, rmcp::schemars::JsonSchema)]
struct MakeTransactionContractCallRequest {
network_name: String,
node_id: u32,
transaction_contract_hash: String,
session_entry_point: String,
#[serde(alias = "session_args_json")]
session_args: Option<Value>,
signer_path: Option<String>,
initiator_public_key: Option<String>,
chain_name: Option<String>,
ttl_millis: Option<u64>,
gas_price_tolerance: Option<u8>,
payment_amount: Option<u64>,
runtime_transferred_value: Option<u64>,
runtime_seed_hex: Option<String>,
}
#[derive(Debug, Deserialize, rmcp::schemars::JsonSchema)]
struct MakeTransactionSessionWasmRequest {
network_name: String,
node_id: u32,
wasm_path: String,
#[serde(alias = "session_args_json")]
session_args: Option<Value>,
is_install_upgrade: Option<bool>,
signer_path: Option<String>,
initiator_public_key: Option<String>,
chain_name: Option<String>,
ttl_millis: Option<u64>,
gas_price_tolerance: Option<u8>,
payment_amount: Option<u64>,
runtime_transferred_value: Option<u64>,
runtime_seed_hex: Option<String>,
}
#[derive(Debug, Deserialize, rmcp::schemars::JsonSchema)]
struct SendSessionWasmRequest {
network_name: String,
node_id: u32,
signer_path: String,
wasm_path: String,
#[serde(alias = "session_args_json")]
session_args: Option<Value>,
chain_name: Option<String>,
is_install_upgrade: Option<bool>,
ttl_millis: Option<u64>,
gas_price_tolerance: Option<u8>,
payment_amount: Option<u64>,
runtime_transferred_value: Option<u64>,
runtime_seed_hex: Option<String>,
}
#[derive(Debug, Deserialize, rmcp::schemars::JsonSchema)]
struct TransferTokensRequest {
network_name: String,
node_id: u32,
from_path: String,
to_path: String,
amount: String,
transfer_id: Option<u64>,
chain_name: Option<String>,
ttl_millis: Option<u64>,
gas_price_tolerance: Option<u8>,
payment_amount: Option<u64>,
}
#[derive(Debug, Deserialize, rmcp::schemars::JsonSchema)]
struct WaitTransactionRequest {
network_name: String,
node_id: u32,
transaction_hash: String,
timeout_seconds: Option<u64>,
poll_interval_millis: Option<u64>,
}
#[derive(Debug, Deserialize, rmcp::schemars::JsonSchema)]
struct GetTransactionRequest {
network_name: String,
node_id: u32,
transaction_hash: String,
}
#[derive(Debug, Deserialize)]
#[serde(rename_all = "snake_case")]
struct RestStatusProbe {
reactor_state: Option<String>,
}
fn to_mcp_error(err: impl std::fmt::Display) -> ErrorData {
ErrorData::internal_error(err.to_string(), None)
}
fn internal_serde_error(err: serde_json::Error) -> ErrorData {
ErrorData::internal_error(format!("serde error: {}", err), None)
}
fn ok_value(value: Value) -> rmcp::model::CallToolResult {
rmcp::model::CallToolResult::structured(value)
}
fn parse_transaction_json(value: Value) -> std::result::Result<Transaction, ErrorData> {
if let Value::String(encoded) = &value {
let _ = encoded;
return Err(ErrorData::invalid_params(
"invalid transaction payload: encoded JSON strings are not supported. Provide transaction as a typed JSON object",
None,
));
}
if let Ok(transaction) = serde_json::from_value::<Transaction>(value.clone()) {
return Ok(transaction);
}
if let Some(inner) = value.get("transaction") {
return parse_transaction_json(inner.clone());
}
Err(ErrorData::invalid_params(
"failed to parse transaction JSON; expected Transaction object or { transaction: Transaction }, with typed JSON values only.",
None,
))
}
fn parse_transaction_hash_input(input: &str) -> Result<TransactionHash> {
let value = input.trim();
if value.is_empty() {
return Err(anyhow!("transaction_hash must not be empty"));
}
let unwrapped = value
.strip_prefix("transaction-v1-hash(")
.and_then(|inner| inner.strip_suffix(')'))
.or_else(|| {
value
.strip_prefix("deploy-hash(")
.and_then(|inner| inner.strip_suffix(')'))
})
.unwrap_or(value);
if unwrapped.contains("..") {
return Err(anyhow!(
"transaction_hash appears abbreviated; provide full hex digest"
));
}
let normalized = unwrapped.strip_prefix("0x").unwrap_or(unwrapped);
let digest = Digest::from_hex(normalized)
.map_err(|err| anyhow!("failed to parse transaction hash digest: {}", err))?;
Ok(TransactionHash::from(TransactionV1Hash::from(digest)))
}
fn mcp_rpc_client(network_name: &str, node_id: u32) -> Result<CasperClient> {
CasperClient::new(
network_name.to_string(),
vec![assets::rpc_endpoint(node_id)],
)
.map_err(|err| {
anyhow!(
"failed to initialize rpc client for node {}: {}",
node_id,
err
)
})
}
fn extract_rpc_result(response: Value) -> Result<Value> {
if let Some(error) = response.get("error") {
return Err(anyhow!("rpc request failed: {}", error));
}
response
.get("result")
.cloned()
.ok_or_else(|| anyhow!("rpc response missing result field"))
}
async fn fetch_block_result(
manager: &NetworkManager,
network_name: &str,
node_id: u32,
block_id: Option<&str>,
) -> Result<Value> {
let block_id = normalize_optional_identifier(block_id);
if block_id.is_empty() {
let rpc = mcp_rpc_client(network_name, node_id)?;
let result = rpc.get_block().await?;
return serde_json::to_value(result).map_err(Into::into);
}
let block_identifier = parse_block_identifier_value(&block_id)?;
let response = manager
.raw_rpc_query(
node_id,
"chain_get_block",
Some(json!({
"block_identifier": block_identifier
})),
)
.await?;
extract_rpc_result(response)
}
fn build_query_global_state_params(
key: &str,
path: Vec<String>,
block_id: &str,
state_root_hash: &str,
) -> Result<Value> {
let key = parse_query_key(key)?;
let state_identifier = parse_state_identifier(block_id, state_root_hash)?;
let mut params = serde_json::Map::new();
params.insert("key".to_string(), Value::String(key));
params.insert(
"path".to_string(),
Value::Array(path.into_iter().map(Value::String).collect()),
);
if let Some(state_identifier) = state_identifier {
params.insert("state_identifier".to_string(), state_identifier);
}
Ok(Value::Object(params))
}
fn build_query_balance_params(
purse_identifier: &str,
block_id: &str,
state_root_hash: &str,
) -> Result<Value> {
let purse_identifier = parse_purse_identifier(purse_identifier)?;
let state_identifier = parse_state_identifier(block_id, state_root_hash)?;
let mut params = serde_json::Map::new();
params.insert("purse_identifier".to_string(), purse_identifier);
if let Some(state_identifier) = state_identifier {
params.insert("state_identifier".to_string(), state_identifier);
}
Ok(Value::Object(params))
}
fn parse_state_identifier(block_id: &str, state_root_hash: &str) -> Result<Option<Value>> {
let block_id = block_id.trim();
if !block_id.is_empty() {
if block_id.len() == Digest::LENGTH * 2 {
Digest::from_hex(block_id)
.map_err(|err| anyhow!("invalid block hash digest in block_id: {}", err))?;
return Ok(Some(json!({
"BlockHash": block_id,
})));
}
let height = block_id
.parse::<u64>()
.map_err(|err| anyhow!("invalid block height in block_id: {}", err))?;
return Ok(Some(json!({
"BlockHeight": height,
})));
}
let state_root_hash = state_root_hash.trim();
if state_root_hash.is_empty() {
return Ok(None);
}
Digest::from_hex(state_root_hash)
.map_err(|err| anyhow!("invalid state_root_hash digest: {}", err))?;
Ok(Some(json!({
"StateRootHash": state_root_hash,
})))
}
fn parse_block_identifier_value(block_id: &str) -> Result<Value> {
let block_id = block_id.trim();
if block_id.is_empty() {
return Err(anyhow!("block identifier must not be empty"));
}
if block_id.len() == Digest::LENGTH * 2 {
Digest::from_hex(block_id)
.map_err(|err| anyhow!("invalid block hash digest in block_id: {}", err))?;
Ok(json!({
"Hash": block_id,
}))
} else {
let height = block_id
.parse::<u64>()
.map_err(|err| anyhow!("invalid block height in block_id: {}", err))?;
Ok(json!({
"Height": height,
}))
}
}
fn parse_query_key(key: &str) -> Result<String> {
let key = key.trim();
if key.is_empty() {
return Err(anyhow!("key must not be empty"));
}
if let Ok(contract_hash) = ContractHash::from_formatted_str(key) {
return Ok(Key::Hash(contract_hash.value()).to_formatted_string());
}
if let Ok(parsed) = Key::from_formatted_str(key) {
return Ok(parsed.to_formatted_string());
}
if let Ok(public_key) = PublicKey::from_hex(key) {
return Ok(Key::Account(public_key.to_account_hash()).to_formatted_string());
}
Err(anyhow!(
"failed to parse key for query; expected formatted Key or public key hex"
))
}
fn parse_contract_hash_for_invocation(input: &str) -> Result<AddressableEntityHash> {
let input = input.trim();
if input.is_empty() {
return Err(anyhow!("transaction_contract_hash must not be empty"));
}
if let Ok(contract_hash) = ContractHash::from_formatted_str(input) {
return Ok(contract_hash.into());
}
let digest_hex = input
.strip_prefix("hash-")
.or_else(|| input.strip_prefix("contract-hash-"))
.unwrap_or(input);
let bytes = hex_to_bytes(digest_hex)?;
if bytes.len() != Digest::LENGTH {
return Err(anyhow!(
"transaction_contract_hash must be {} bytes",
Digest::LENGTH
));
}
let mut hash = [0u8; Digest::LENGTH];
hash.copy_from_slice(&bytes);
Ok(ContractHash::new(hash).into())
}
fn parse_purse_identifier(input: &str) -> Result<Value> {
let input = input.trim();
if input.is_empty() {
return Err(anyhow!("purse_identifier must not be empty"));
}
if input.starts_with("account-hash-") {
casper_types::account::AccountHash::from_formatted_str(input)
.map_err(|err| anyhow!("invalid account hash purse identifier: {}", err))?;
return Ok(json!({
"main_purse_under_account_hash": input,
}));
}
if input.starts_with("entity-") {
return Ok(json!({
"main_purse_under_entity_addr": input,
}));
}
if input.starts_with("uref-") {
URef::from_formatted_str(input)
.map_err(|err| anyhow!("invalid uref purse identifier: {}", err))?;
return Ok(json!({
"purse_uref": input,
}));
}
let public_key = PublicKey::from_hex(input)
.map_err(|err| anyhow!("invalid public key purse identifier: {}", err))?;
Ok(json!({
"main_purse_under_public_key": public_key.to_hex_string(),
}))
}
fn build_pricing_mode(gas_price_tolerance: Option<u8>, payment_amount: Option<u64>) -> PricingMode {
if let Some(payment_amount) = payment_amount {
PricingMode::PaymentLimited {
payment_amount,
gas_price_tolerance: gas_price_tolerance.unwrap_or(5),
standard_payment: true,
}
} else {
PricingMode::Fixed {
gas_price_tolerance: gas_price_tolerance.unwrap_or(5),
additional_computation_factor: 0,
}
}
}
fn parse_optional_seed_hex(seed: Option<&str>) -> Result<Option<[u8; 32]>> {
let Some(seed) = seed else {
return Ok(None);
};
let seed = seed.trim();
if seed.is_empty() {
return Ok(None);
}
let cleaned = seed.strip_prefix("0x").unwrap_or(seed);
let bytes = hex_to_bytes(cleaned)?;
if bytes.len() != 32 {
return Err(anyhow!("runtime_seed_hex must be exactly 32 bytes"));
}
let mut out = [0u8; 32];
out.copy_from_slice(&bytes);
Ok(Some(out))
}
fn normalize_optional_identifier(value: Option<&str>) -> String {
value.map(str::trim).unwrap_or_default().to_string()
}
async fn resolve_global_state_identifier(
network_name: &str,
node_id: u32,
block_id: Option<&str>,
state_root_hash: Option<&str>,
) -> Result<(String, String)> {
let block_id = normalize_optional_identifier(block_id);
let state_root_hash = normalize_optional_identifier(state_root_hash);
if !block_id.is_empty() || !state_root_hash.is_empty() {
return Ok((block_id, state_root_hash));
}
let rpc = mcp_rpc_client(network_name, node_id)?;
let response = rpc.get_block().await?;
let latest_block = response
.block_with_signatures
.ok_or_else(|| anyhow!("latest block was not returned by chain_get_block"))?;
Ok((latest_block.block.hash().to_hex_string(), String::new()))
}
fn hex_to_bytes(input: &str) -> Result<Vec<u8>> {
if !input.len().is_multiple_of(2) {
return Err(anyhow!("hex string must have even length"));
}
let mut out = Vec::with_capacity(input.len() / 2);
let mut chars = input.chars();
while let (Some(hi), Some(lo)) = (chars.next(), chars.next()) {
let byte = ((hex_nibble(hi)? as u8) << 4) | (hex_nibble(lo)? as u8);
out.push(byte);
}
Ok(out)
}
fn hex_nibble(ch: char) -> Result<u32> {
ch.to_digit(16)
.ok_or_else(|| anyhow!("invalid hex character '{}'", ch))
}
async fn ensure_running_network(
network: &Arc<ManagedNetwork>,
) -> std::result::Result<(), ErrorData> {
if network.is_running().await {
Ok(())
} else {
Err(ErrorData::resource_not_found(
"network is not running; call spawn_network then wait_network_ready",
None,
))
}
}
async fn check_rest_ready(network: &Arc<ManagedNetwork>) -> Result<HashMap<u32, Value>> {
let node_ids = {
let state = network.state.lock().await;
state
.processes
.iter()
.filter_map(|process| {
if matches!(process.kind, ProcessKind::Node) {
Some(process.node_id)
} else {
None
}
})
.collect::<Vec<_>>()
};
if node_ids.is_empty() {
return Err(anyhow!("network has no node processes"));
}
let mut by_node = HashMap::new();
for node_id in node_ids {
let value = fetch_rest_status(node_id).await?;
let probe: RestStatusProbe = serde_json::from_value(value.clone())
.with_context(|| format!("invalid /status payload for node {}", node_id))?;
if probe.reactor_state.as_deref() != Some("Validate") {
return Err(anyhow!("node {} reactor is not Validate", node_id));
}
by_node.insert(node_id, value);
}
Ok(by_node)
}
fn rest_has_block(value: &Value) -> bool {
value
.get("last_added_block_info")
.and_then(|entry| entry.get("height"))
.and_then(Value::as_u64)
.is_some()
}
async fn fetch_rest_status(node_id: u32) -> Result<Value> {
let url = format!("{}/status", assets::rest_endpoint(node_id));
let client = reqwest::Client::builder()
.no_proxy()
.timeout(Duration::from_secs(4))
.build()?;
let response = client.get(url).send().await?.error_for_status()?;
Ok(response.json::<Value>().await?)
}
async fn claim_block_hook(network: &Arc<ManagedNetwork>, block_hash: &str) -> bool {
let mut last_block_hook_hash = network.last_block_hook_hash.lock().await;
if last_block_hook_hash.as_deref() == Some(block_hash) {
return false;
}
*last_block_hook_hash = Some(block_hash.to_string());
true
}
async fn run_sse_listener(network: Arc<ManagedNetwork>, node_id: u32, endpoint: String) {
let mut backoff = ExponentialBackoff::default();
let mut connection_version: Option<String> = None;
loop {
if network.shutdown.load(Ordering::SeqCst) {
return;
}
let config = match ListenerConfig::builder()
.with_endpoint(endpoint.clone())
.build()
{
Ok(config) => config,
Err(_) => {
if !sleep_backoff(&mut backoff).await {
return;
}
continue;
}
};
let stream = match sse::listener(config).await {
Ok(stream) => {
backoff.reset();
stream
}
Err(_) => {
if !sleep_backoff(&mut backoff).await {
return;
}
continue;
}
};
futures::pin_mut!(stream);
let mut failed = false;
while let Some(event) = stream.next().await {
if network.shutdown.load(Ordering::SeqCst) {
return;
}
match event {
Ok(event) => {
if let SseEvent::ApiVersion(version) = &event {
connection_version = Some(version.to_string());
}
if let SseEvent::BlockAdded { block_hash, block } = &event {
let _ = record_last_block_height(&network.state, block.height()).await;
assets::spawn_pending_post_genesis_hook(network.layout.clone());
if let Some(protocol_version) = connection_version.as_deref()
&& claim_block_hook(&network, &block_hash.to_string()).await
{
assets::spawn_block_added_hook(
network.layout.clone(),
protocol_version.to_string(),
json!({
"block_hash": block_hash.to_string(),
"height": block.height(),
"era_id": block.era_id().value(),
}),
);
}
}
network.sse_store.push(node_id, event).await;
}
Err(_) => {
failed = true;
break;
}
}
}
connection_version = None;
if failed && !sleep_backoff(&mut backoff).await {
return;
}
}
}
async fn record_last_block_height(state: &Arc<Mutex<State>>, height: u64) -> Result<()> {
let mut state = state.lock().await;
if state.last_block_height == Some(height) {
return Ok(());
}
state.last_block_height = Some(height);
state.touch().await
}
async fn sleep_backoff(backoff: &mut ExponentialBackoff) -> bool {
if let Some(delay) = backoff.next_backoff() {
tokio::time::sleep(delay).await;
true
} else {
false
}
}
fn sse_event_type(event: &SseEvent) -> &'static str {
match event {
SseEvent::ApiVersion(_) => "ApiVersion",
SseEvent::DeployAccepted(_) => "DeployAccepted",
SseEvent::BlockAdded { .. } => "BlockAdded",
SseEvent::DeployProcessed(_) => "DeployProcessed",
SseEvent::DeployExpired(_) => "DeployExpired",
SseEvent::TransactionAccepted(_) => "TransactionAccepted",
SseEvent::TransactionProcessed { .. } => "TransactionProcessed",
SseEvent::TransactionExpired { .. } => "TransactionExpired",
SseEvent::Fault { .. } => "Fault",
SseEvent::FinalitySignature(_) => "FinalitySignature",
SseEvent::Step { .. } => "Step",
SseEvent::Shutdown => "Shutdown",
}
}
fn process_kind_name(kind: &ProcessKind) -> &'static str {
match kind {
ProcessKind::Node => "node",
ProcessKind::Sidecar => "sidecar",
}
}
fn process_status_name(status: &ProcessStatus) -> &'static str {
match status {
ProcessStatus::Running => "running",
ProcessStatus::Stopped => "stopped",
ProcessStatus::Exited => "exited",
ProcessStatus::Unknown => "unknown",
ProcessStatus::Skipped => "skipped",
}
}
fn sse_event_payload(event: &SseEvent) -> Value {
match event {
SseEvent::ApiVersion(version) => json!({ "api_version": version.to_string() }),
SseEvent::DeployAccepted(payload) => payload.clone(),
SseEvent::BlockAdded { block_hash, block } => json!({
"block_hash": block_hash.to_string(),
"height": block.height(),
"era_id": block.era_id().value(),
}),
SseEvent::DeployProcessed(payload) => payload.clone(),
SseEvent::DeployExpired(payload) => payload.clone(),
SseEvent::TransactionAccepted(transaction) => {
json!({ "transaction_hash": transaction.hash().to_hex_string() })
}
SseEvent::TransactionProcessed {
transaction_hash,
execution_result,
messages,
..
} => json!({
"transaction_hash": transaction_hash.to_hex_string(),
"execution_result": execution_result,
"messages": messages,
}),
SseEvent::TransactionExpired { transaction_hash } => json!({
"transaction_hash": transaction_hash.to_hex_string(),
}),
SseEvent::Fault {
era_id,
public_key,
timestamp,
} => json!({
"era_id": era_id.value(),
"public_key": public_key.to_hex(),
"timestamp": timestamp,
}),
SseEvent::FinalitySignature(signature) => json!({
"block_hash": signature.block_hash().to_string(),
"era_id": signature.era_id().value(),
"signature": signature.signature().to_hex(),
}),
SseEvent::Step {
era_id,
execution_effects,
} => json!({
"era_id": era_id.value(),
"execution_effects": execution_effects.get(),
}),
SseEvent::Shutdown => json!({}),
}
}
fn timestamp_prefix() -> String {
time::OffsetDateTime::now_utc()
.format(&time::format_description::well_known::Rfc3339)
.unwrap_or_else(|_| "unknown-time".to_string())
}
fn processes_running(state: &State) -> bool {
if state.processes.is_empty() {
return false;
}
state.processes.iter().all(|process| {
matches!(process.last_status, ProcessStatus::Running)
&& process.current_pid().is_some_and(is_pid_running)
})
}
fn running_node_ids(state: &State) -> Vec<u32> {
let mut node_ids = std::collections::BTreeSet::new();
for process in &state.processes {
if !matches!(process.kind, ProcessKind::Node) {
continue;
}
if !matches!(process.last_status, ProcessStatus::Running) {
continue;
}
let Some(pid) = process.current_pid() else {
continue;
};
if !is_pid_running(pid) {
continue;
}
node_ids.insert(process.node_id);
}
node_ids.into_iter().collect()
}
fn is_pid_running(pid: u32) -> bool {
let pid = Pid::from_raw(pid as i32);
match kill(pid, None) {
Ok(()) => true,
Err(Errno::ESRCH) => false,
Err(_) => true,
}
}
async fn discover_network_names(assets_root: &Path) -> Result<Vec<String>> {
if !is_dir(assets_root).await {
return Ok(Vec::new());
}
let mut names = Vec::new();
let mut entries = tokio_fs::read_dir(assets_root).await?;
while let Some(entry) = entries.next_entry().await? {
if !entry.file_type().await?.is_dir() {
continue;
}
let name = entry.file_name().to_string_lossy().to_string();
if !name.is_empty() {
names.push(name);
}
}
names.sort();
Ok(names)
}
async fn ensure_sidecar_available(layout: &AssetsLayout, node_count: u32) -> Result<()> {
for node_id in 1..=node_count {
let version_dir = layout.latest_protocol_version_dir(node_id).await?;
let sidecar_bin = layout
.node_bin_dir(node_id)
.join(&version_dir)
.join("casper-sidecar");
let sidecar_cfg = layout
.node_config_root(node_id)
.join(&version_dir)
.join("sidecar.toml");
if !is_file(&sidecar_bin).await {
return Err(anyhow!(
"missing sidecar binary for node {}: {}",
node_id,
sidecar_bin.display()
));
}
if !is_file(&sidecar_cfg).await {
return Err(anyhow!(
"missing sidecar.toml for node {}: {}",
node_id,
sidecar_cfg.display()
));
}
}
Ok(())
}
async fn latest_layout_protocol_version(layout: &AssetsLayout) -> Result<String> {
Ok(layout
.latest_protocol_version_dir(1)
.await?
.replace('_', "."))
}
fn stage_asset_selector(
asset: Option<&str>,
custom_asset: Option<&str>,
legacy_asset_name: Option<&str>,
) -> Result<AssetSelector> {
let custom_asset = match (custom_asset, legacy_asset_name) {
(Some(_), Some(_)) => {
return Err(anyhow!(
"custom_asset and asset_name are mutually exclusive"
));
}
(Some(custom_asset), None) | (None, Some(custom_asset)) => Some(custom_asset),
(None, None) => None,
};
assets::required_asset_selector(asset, custom_asset)
}
async fn parse_derived_accounts_csv(
csv: &str,
seed: Option<Arc<str>>,
) -> Result<Vec<DerivedAccountRow>> {
let mut lines = csv.lines();
let header = lines
.next()
.ok_or_else(|| anyhow!("derived accounts csv is empty"))?;
if header.trim() != "kind,name,key_type,derivation,path,account_hash,balance" {
return Err(anyhow!("unexpected derived-accounts.csv header"));
}
let mut rows = Vec::new();
for line in lines {
let line = line.trim();
if line.is_empty() {
continue;
}
let parts = line.splitn(7, ',').collect::<Vec<_>>();
if parts.len() != 7 {
return Err(anyhow!("invalid derived account row: {}", line));
}
let path = parts[4].to_string();
let public_key = if let Some(seed) = &seed {
match assets::derive_account_from_seed_path(Arc::clone(seed), &path).await {
Ok(material) => Some(material.public_key_hex),
Err(_) => None,
}
} else {
None
};
rows.push(DerivedAccountRow {
kind: parts[0].to_string(),
name: parts[1].to_string(),
key_type: parts[2].to_string(),
derivation: parts[3].to_string(),
path,
account_hash: parts[5].to_string(),
balance: parts[6].to_string(),
public_key,
});
}
Ok(rows)
}
async fn verify_path_hash_consistency(
layout: &AssetsLayout,
path: &str,
expected_account_hash: &str,
) -> Result<()> {
let Some(csv) = assets::derived_accounts_summary(layout).await else {
return Ok(());
};
for line in csv.lines().skip(1) {
let line = line.trim();
if line.is_empty() {
continue;
}
let parts = line.splitn(7, ',').collect::<Vec<_>>();
if parts.len() != 7 {
continue;
}
if parts[4] == path {
if parts[5] != expected_account_hash {
return Err(anyhow!(
"derived account hash mismatch for path {}: csv={} derived={}",
path,
parts[5],
expected_account_hash
));
}
return Ok(());
}
}
Ok(())
}
async fn fetch_chain_name(network_name: &str, node_id: u32) -> Result<String> {
let rpc = mcp_rpc_client(network_name, node_id)?;
rpc.get_network_name().await.map_err(Into::into)
}
fn extract_block_height(value: &Value) -> Option<u64> {
value
.pointer("/block_with_signatures/block/header/height")
.and_then(Value::as_u64)
.or_else(|| {
value
.pointer("/block/block/header/height")
.and_then(Value::as_u64)
})
.or_else(|| {
value
.pointer("/block_with_signatures/Version2/block/Version2/header/height")
.and_then(Value::as_u64)
})
.or_else(|| {
value
.pointer("/block_with_signatures/Version1/block/Version1/header/height")
.and_then(Value::as_u64)
})
.or_else(|| find_first_height(value))
}
fn find_first_height(value: &Value) -> Option<u64> {
match value {
Value::Object(map) => {
if let Some(height) = map.get("height").and_then(Value::as_u64) {
return Some(height);
}
for nested in map.values() {
if let Some(height) = find_first_height(nested) {
return Some(height);
}
}
None
}
Value::Array(items) => {
for item in items {
if let Some(height) = find_first_height(item) {
return Some(height);
}
}
None
}
_ => None,
}
}
fn parse_no_such_block_range_from_error(error_text: &str) -> Option<(u64, u64)> {
let start = error_text.find('{')?;
let payload = &error_text[start..];
let value: Value = serde_json::from_str(payload).ok()?;
let code = value.get("code").and_then(Value::as_i64)?;
let message = value.get("message").and_then(Value::as_str)?;
if code != -32001 || !message.eq_ignore_ascii_case("No such block") {
return None;
}
let low = value
.pointer("/data/available_block_range/low")
.and_then(Value::as_u64)
.unwrap_or(0);
let high = value
.pointer("/data/available_block_range/high")
.and_then(Value::as_u64)
.unwrap_or(low);
Some((low, high))
}
async fn read_log_page(path: &Path, before_line: Option<usize>, limit: usize) -> Result<LogPage> {
let contents = tokio_fs::read_to_string(path)
.await
.with_context(|| format!("failed to read log file {}", path.display()))?;
let all_lines = contents
.lines()
.map(ToString::to_string)
.collect::<Vec<_>>();
let total_lines = all_lines.len();
let before = before_line.unwrap_or(total_lines + 1);
if before == 0 {
return Err(anyhow!("before_line must be >= 1"));
}
let end_exclusive = before.saturating_sub(1).min(total_lines);
let start = end_exclusive.saturating_sub(limit);
let mut lines = Vec::new();
for (idx, content) in all_lines[start..end_exclusive].iter().enumerate() {
lines.push(LogLine {
line_number: start + idx + 1,
content: content.clone(),
});
}
let next_before_line = if start == 0 { None } else { Some(start + 1) };
Ok(LogPage {
path: path.display().to_string(),
total_lines,
returned: lines.len(),
next_before_line,
lines,
})
}
async fn is_dir(path: &Path) -> bool {
tokio_fs::metadata(path)
.await
.map(|meta| meta.is_dir())
.unwrap_or(false)
}
async fn is_file(path: &Path) -> bool {
tokio_fs::metadata(path)
.await
.map(|meta| meta.is_file())
.unwrap_or(false)
}
pub async fn run(args: McpArgs) -> Result<()> {
let assets_root = match args.net_path {
Some(path) => path,
None => assets::default_assets_root()?,
};
let manager = Arc::new(NetworkManager::new(assets_root).await?);
let result = match args.transport {
McpTransport::Stdio => run_stdio(manager.clone()).await,
McpTransport::Http => run_http(manager.clone(), &args.http_bind, &args.http_path).await,
McpTransport::Both => run_both(manager.clone(), &args.http_bind, &args.http_path).await,
};
let stop_result = manager.stop_all_networks().await;
match (result, stop_result) {
(Ok(()), Ok(())) => Ok(()),
(Err(err), Ok(())) => Err(err),
(Ok(()), Err(stop_err)) => Err(stop_err),
(Err(run_err), Err(stop_err)) => Err(anyhow!(
"mcp server failed: {run_err}; additionally failed to stop networks: {stop_err}"
)),
}
}
async fn run_stdio(manager: Arc<NetworkManager>) -> Result<()> {
let service = McpServer::new(manager).serve(mcp_stdio()).await?;
service.waiting().await?;
Ok(())
}
async fn run_http(manager: Arc<NetworkManager>, bind: &str, path: &str) -> Result<()> {
let path = normalize_http_path(path);
let socket = std::net::SocketAddr::from_str(bind)
.with_context(|| format!("invalid http bind address '{}'", bind))?;
let service: StreamableHttpService<McpServer, LocalSessionManager> = StreamableHttpService::new(
{
let manager = manager.clone();
move || Ok(McpServer::new(manager.clone()))
},
Arc::new(LocalSessionManager::default()),
StreamableHttpServerConfig::default(),
);
let router = axum::Router::new().nest_service(&path, service);
let listener = tokio::net::TcpListener::bind(socket).await?;
axum::serve(listener, router)
.with_graceful_shutdown(async {
let _ = tokio::signal::ctrl_c().await;
})
.await?;
Ok(())
}
async fn run_both(manager: Arc<NetworkManager>, bind: &str, path: &str) -> Result<()> {
let path = normalize_http_path(path);
let socket = std::net::SocketAddr::from_str(bind)
.with_context(|| format!("invalid http bind address '{}'", bind))?;
let (shutdown_tx, mut shutdown_rx) = tokio::sync::watch::channel(false);
let mut http_task = {
let manager = manager.clone();
tokio::spawn(async move {
let service: StreamableHttpService<McpServer, LocalSessionManager> =
StreamableHttpService::new(
{
let manager = manager.clone();
move || Ok(McpServer::new(manager.clone()))
},
Arc::new(LocalSessionManager::default()),
StreamableHttpServerConfig::default(),
);
let router = axum::Router::new().nest_service(&path, service);
let listener = tokio::net::TcpListener::bind(socket).await?;
axum::serve(listener, router)
.with_graceful_shutdown(async move {
tokio::select! {
_ = async {
loop {
if *shutdown_rx.borrow() {
break;
}
if shutdown_rx.changed().await.is_err() {
break;
}
}
} => {}
_ = tokio::signal::ctrl_c() => {}
}
})
.await
.map_err(anyhow::Error::from)
})
};
let mut stdio_task = tokio::spawn(async move { run_stdio(manager).await });
let result = tokio::select! {
res = &mut stdio_task => match res {
Ok(inner) => inner,
Err(join_err) => Err(anyhow!("stdio task failed: {}", join_err)),
},
res = &mut http_task => match res {
Ok(inner) => inner,
Err(join_err) => Err(anyhow!("http task failed: {}", join_err)),
},
};
let _ = shutdown_tx.send(true);
let _ = tokio::time::timeout(Duration::from_secs(2), &mut http_task).await;
let _ = tokio::time::timeout(Duration::from_secs(2), &mut stdio_task).await;
result
}
fn normalize_http_path(path: &str) -> String {
if path.starts_with('/') {
path.to_string()
} else {
format!("/{}", path)
}
}
#[cfg(test)]
mod tests {
use super::*;
#[tokio::test]
async fn parse_derived_accounts_rows() {
let csv = "kind,name,key_type,derivation,path,account_hash,balance\nvalidator,node-1,secp256k1,bip32,m/44'/506'/0'/0/0,account-hash-a,100\nuser,user-1,secp256k1,bip32,m/44'/506'/0'/0/100,account-hash-b,200";
let rows = parse_derived_accounts_csv(csv, None).await.unwrap();
assert_eq!(rows.len(), 2);
assert_eq!(rows[0].path, "m/44'/506'/0'/0/0");
assert_eq!(rows[1].account_hash, "account-hash-b");
}
#[test]
fn sse_filter_include_exclude() {
let filter = SseFilter {
include_event_types: vec!["BlockAdded".to_string()],
exclude_event_types: vec!["TransactionAccepted".to_string()],
};
let block = SseRecord {
sequence: 1,
timestamp_rfc3339: "t".to_string(),
node_id: 1,
event_type: "BlockAdded".to_string(),
payload: json!({}),
};
let tx = SseRecord {
sequence: 2,
timestamp_rfc3339: "t".to_string(),
node_id: 1,
event_type: "TransactionAccepted".to_string(),
payload: json!({}),
};
assert!(filter.matches(&block));
assert!(!filter.matches(&tx));
}
#[tokio::test]
async fn sse_history_paginates_by_sequence() {
let store = SseStore::new(100);
for sequence in 1..=5 {
let mut guard = store.events.lock().await;
guard.push_back(SseRecord {
sequence,
timestamp_rfc3339: "t".to_string(),
node_id: 1,
event_type: "BlockAdded".to_string(),
payload: json!({ "sequence": sequence }),
});
drop(guard);
}
let filter = SseFilter::default();
let page = store.history(Some(6), 2, &filter).await;
assert_eq!(page.returned, 2);
assert_eq!(page.events[0].sequence, 4);
assert_eq!(page.events[1].sequence, 5);
assert_eq!(page.next_before_sequence, Some(4));
}
#[tokio::test]
async fn read_log_page_uses_before_cursor() {
let temp = tempfile::NamedTempFile::new().unwrap();
tokio_fs::write(temp.path(), "a\nb\nc\nd\n").await.unwrap();
let page = read_log_page(temp.path(), Some(5), 2).await.unwrap();
assert_eq!(page.lines.len(), 2);
assert_eq!(page.lines[0].line_number, 3);
assert_eq!(page.lines[1].line_number, 4);
assert_eq!(page.next_before_line, Some(3));
}
#[tokio::test]
async fn sidecar_preflight_fails_when_missing() {
let root = tempfile::tempdir().unwrap();
let layout = AssetsLayout::new(root.path().to_path_buf(), "test-net".to_string());
let node_bin = layout.node_bin_dir(1).join("2_0_0");
let node_cfg = layout.node_config_root(1).join("2_0_0");
tokio_fs::create_dir_all(&node_bin).await.unwrap();
tokio_fs::create_dir_all(&node_cfg).await.unwrap();
tokio_fs::write(node_bin.join("casper-node"), "bin")
.await
.unwrap();
tokio_fs::write(node_cfg.join("sidecar.toml"), "cfg")
.await
.unwrap();
let result = ensure_sidecar_available(&layout, 1).await;
assert!(result.is_err());
}
#[test]
fn normalize_optional_identifier_trims_and_defaults() {
assert_eq!(normalize_optional_identifier(None), "");
assert_eq!(normalize_optional_identifier(Some(" ")), "");
assert_eq!(normalize_optional_identifier(Some(" 123 ")), "123");
}
#[test]
fn parse_state_identifier_variants() {
assert_eq!(
parse_state_identifier("42", "").unwrap(),
Some(json!({ "BlockHeight": 42u64 }))
);
assert_eq!(
parse_state_identifier(
"2f6fbeebbe1bdf6f8ff05880edfa4e4f79849d2b4f0ecf65482177e4fabc1234",
""
)
.unwrap(),
Some(json!({
"BlockHash": "2f6fbeebbe1bdf6f8ff05880edfa4e4f79849d2b4f0ecf65482177e4fabc1234"
}))
);
assert_eq!(parse_state_identifier("", "").unwrap(), None,);
}
#[test]
fn parse_contract_hash_for_invocation_accepts_common_formats() {
let hash = "2f6fbeebbe1bdf6f8ff05880edfa4e4f79849d2b4f0ecf65482177e4fabc1234";
let contract = parse_contract_hash_for_invocation(&format!("contract-{}", hash)).unwrap();
let key_hash = parse_contract_hash_for_invocation(&format!("hash-{}", hash)).unwrap();
let raw = parse_contract_hash_for_invocation(hash).unwrap();
assert_eq!(contract.to_hex_string(), hash);
assert_eq!(key_hash.to_hex_string(), hash);
assert_eq!(raw.to_hex_string(), hash);
}
#[test]
fn parse_query_key_accepts_contract_hash_format() {
let hash = "2f6fbeebbe1bdf6f8ff05880edfa4e4f79849d2b4f0ecf65482177e4fabc1234";
let key = parse_query_key(&format!("contract-{}", hash)).unwrap();
assert_eq!(key, format!("hash-{}", hash));
}
#[test]
fn parse_transaction_json_rejects_escaped_string_payload() {
let tx_json = json!({
"Version1": {
"hash": "7eeb092361e31b4cc9885e3621f1470f29631338ecc703643c22da1d38fd81a9",
"payload": {
"initiator_addr": {
"PublicKey": "0202f9bae6a6c5a8345c2aa8339b54ff3fcf82d2f6a9cce1732e765c2cc403b3be9f"
},
"timestamp": "2026-02-27T18:03:18.541Z",
"ttl": "30m",
"chain_name": "casper-devnet",
"pricing_mode": {
"PaymentLimited": {
"payment_amount": 100000000000u64,
"gas_price_tolerance": 5,
"standard_payment": true
}
},
"fields": {
"args": {"Named": []},
"entry_point": {"Custom": "counter_inc"},
"scheduling": "Standard",
"target": {
"Stored": {
"id": {
"ByPackageName": {
"name": "counter_package_name",
"version": null
}
},
"runtime": "VmCasperV1"
}
}
}
},
"approvals": [{
"signer": "0202f9bae6a6c5a8345c2aa8339b54ff3fcf82d2f6a9cce1732e765c2cc403b3be9f",
"signature": "02c64336e5ed2832bdb84adb3f334d585548ee096066aa9d0797c11ab3f074ec9d7bd396994bc9b9c239342be801bc385a9c5083779bace4dfe0b400d4a13c07db"
}]
}
});
let direct = parse_transaction_json(tx_json.clone()).unwrap();
let wrapped = parse_transaction_json(json!({ "transaction": tx_json })).unwrap();
let encoded = serde_json::to_string(&direct).unwrap();
let from_string_err = parse_transaction_json(Value::String(encoded.clone()));
let wrapped_string_err = parse_transaction_json(json!({
"transaction": encoded
}));
assert_eq!(
direct.hash().to_hex_string(),
wrapped.hash().to_hex_string()
);
assert!(from_string_err.is_err());
assert!(wrapped_string_err.is_err());
}
#[test]
fn parse_no_such_block_range_from_error_extracts_bounds() {
let error = "casper client error: response for rpc-id 1 chain_get_block is json-rpc error: {\"code\":-32001,\"message\":\"No such block\",\"data\":{\"message\":\"no block found for the provided identifier\",\"available_block_range\":{\"low\":0,\"high\":0}}}";
let range = parse_no_such_block_range_from_error(error).unwrap();
assert_eq!(range, (0, 0));
}
#[test]
fn send_transaction_signed_request_requires_transaction_field() {
let payload = json!({
"network_name": "casper-devnet",
"node_id": 1,
"signer_path": "m/44'/506'/0'/0/100",
"transaction": { "Version1": { "hash": "abc" } }
});
let request: SendTransactionSignedRequest = serde_json::from_value(payload).unwrap();
assert!(request.transaction.get("Version1").is_some());
let legacy_payload = json!({
"network_name": "casper-devnet",
"node_id": 1,
"signer_path": "m/44'/506'/0'/0/100",
"transaction_json": { "Version1": { "hash": "def" } }
});
let legacy_err = serde_json::from_value::<SendTransactionSignedRequest>(legacy_payload)
.unwrap_err()
.to_string();
assert!(legacy_err.contains("missing field `transaction`"));
}
}