//! MCP tool definitions and handlers for the STDIO server
//!
//! Each tool maps to existing NetGet functionality (server management, protocol listing, etc.)
//! Tools use rmcp's macro system for automatic schema generation.
use std::sync::Arc;
use tokio::sync::mpsc;
use rmcp::handler::server::{router::tool::ToolRouter, wrapper::Parameters};
use rmcp::model::{
CallToolResult, Content, Implementation, InitializeRequestParams, InitializeResult,
ServerCapabilities,
};
use rmcp::service::{RequestContext, RoleServer};
use rmcp::{tool, tool_handler, tool_router, ErrorData as McpError, ServerHandler};
use schemars::JsonSchema;
use serde::{Deserialize, Deserializer};
use tracing::info;
/// Deserialize Option<u16> from either a number or a string (MCP clients may send "6178" instead of 6178)
fn deserialize_option_u16_flexible<'de, D>(deserializer: D) -> Result<Option<u16>, D::Error>
where
D: Deserializer<'de>,
{
#[derive(Deserialize)]
#[serde(untagged)]
enum NumOrString {
Number(u16),
String(String),
}
match Option::<NumOrString>::deserialize(deserializer)? {
Some(NumOrString::Number(n)) => Ok(Some(n)),
Some(NumOrString::String(s)) => s.parse().map(Some).map_err(serde::de::Error::custom),
None => Ok(None),
}
}
/// Deserialize u32 from either a number or a string
fn deserialize_u32_flexible<'de, D>(deserializer: D) -> Result<u32, D::Error>
where
D: Deserializer<'de>,
{
#[derive(Deserialize)]
#[serde(untagged)]
enum NumOrString {
Number(u32),
String(String),
}
match NumOrString::deserialize(deserializer)? {
NumOrString::Number(n) => Ok(n),
NumOrString::String(s) => s.parse().map_err(serde::de::Error::custom),
}
}
/// Deserialize u64 from either a number or a string
fn deserialize_u64_flexible<'de, D>(deserializer: D) -> Result<u64, D::Error>
where
D: Deserializer<'de>,
{
#[derive(Deserialize)]
#[serde(untagged)]
enum NumOrString {
Number(u64),
String(String),
}
match NumOrString::deserialize(deserializer)? {
NumOrString::Number(n) => Ok(n),
NumOrString::String(s) => s.parse().map_err(serde::de::Error::custom),
}
}
/// Deserialize Option<u32> from either a number or a string
fn deserialize_option_u32_flexible<'de, D>(deserializer: D) -> Result<Option<u32>, D::Error>
where
D: Deserializer<'de>,
{
#[derive(Deserialize)]
#[serde(untagged)]
enum NumOrString {
Number(u32),
String(String),
}
match Option::<NumOrString>::deserialize(deserializer)? {
Some(NumOrString::Number(n)) => Ok(Some(n)),
Some(NumOrString::String(s)) => s.parse().map(Some).map_err(serde::de::Error::custom),
None => Ok(None),
}
}
use crate::cli::Args;
use crate::llm::agent_queue::LlmRequestQueue;
use crate::llm::OllamaClient;
use crate::settings::Settings;
use crate::state::app_state::AppState;
/// Shared state for the MCP server
///
/// One instance is shared across all MCP sessions (a single session for STDIO,
/// potentially many for the HTTP transport).
pub(crate) struct SharedState {
app_state: AppState,
llm_client: OllamaClient,
/// Present only in `--llm-agent` mode: the queue protocol servers submit LLM
/// requests to and the MCP tools answer from.
agent_queue: Option<Arc<LlmRequestQueue>>,
_status_tx: mpsc::UnboundedSender<String>,
background_tasks: Vec<tokio::task::JoinHandle<()>>,
task_owner: tokio_util::sync::CancellationToken,
// On drop the token requests a graceful finish of the current claim/spawn
// batch. Aborting that batch could leave claimed tasks without an executor.
_task_ticker: tokio::task::JoinHandle<()>,
}
impl Drop for SharedState {
fn drop(&mut self) {
self.task_owner.cancel();
for task in &self.background_tasks {
task.abort();
}
}
}
/// NetGet MCP STDIO service - exposes NetGet capabilities as MCP tools
#[derive(Clone)]
pub struct NetGetMcpService {
state: Arc<SharedState>,
tool_router: ToolRouter<Self>,
}
// === Tool parameter structs ===
#[derive(Debug, Deserialize, JsonSchema)]
pub struct ListProtocolsParams {
/// Filter by "server", "client", or "all"
#[serde(default = "default_all")]
pub r#type: Option<String>,
/// Include experimental/honeypot protocols
#[serde(default)]
pub include_disabled: Option<bool>,
}
fn default_all() -> Option<String> {
Some("all".to_string())
}
#[derive(Debug, Deserialize, JsonSchema)]
pub struct StartServerParams {
/// Protocol name. Use list_protocols tool to see all available protocols.
pub protocol: String,
/// Port to listen on. Omit it for the protocol's well-known port (6379 for redis, 53 for
/// dns); when that needs root this process lacks, or is already in use, the server starts
/// on an OS-assigned port and the result says why. Pass 0 for an OS-assigned port. An
/// explicit port, 0 included, is always used as given.
#[serde(default, deserialize_with = "deserialize_option_u16_flexible")]
pub port: Option<u16>,
/// Natural language instruction for the LLM that handles each request.
/// This is the LLM FALLBACK path: every request/event triggers a (slow,
/// billable) model call. Use it only when responses genuinely require
/// reasoning or vary unpredictably. For deterministic behavior (echo,
/// fixed/canned responses, simple routing, high throughput) prefer
/// `event_handlers` with a script or static handler instead — see below.
#[serde(default)]
pub instruction: Option<String>,
/// Host address to bind to (default: 127.0.0.1)
#[serde(default)]
pub host: Option<String>,
/// Event handlers that decide how incoming events are handled WITHOUT an LLM
/// call. Strongly preferred over `instruction` whenever the behavior is
/// deterministic — a script/static handler runs in-process (instant, free,
/// deterministic), while `instruction` pays a model round-trip per request.
///
/// Array of objects: { "event_pattern": "<event id>" | "*", "handler": {...} }
/// matched in order, first match wins. `handler` is one of:
/// - {"type":"script","language":"python"|"javascript","code":"..."} — runs
/// the script per event. The event is JSON on stdin: read the payload via
/// data['event'][<field>] and print {"actions":[ ... ]} to stdout. Field
/// names vary by protocol (e.g. Telnet event `telnet_message_received` has
/// `message`; HTTP `http_request` has `method`/`path`/`headers`/`body`).
/// - {"type":"static","actions":[ ... ]} — always emit these fixed actions.
/// - {"type":"llm","instruction":"..."} — fall back to the LLM (reasoning).
///
/// Example — a Telnet echo server with ZERO LLM calls (echoes each line back
/// verbatim):
/// [{"event_pattern":"telnet_message_received","handler":{"type":"script",
/// "language":"python",
/// "code":"import json,sys;d=json.load(sys.stdin);print(json.dumps({'actions':[{'type':'send_telnet_message','message':d['event']['message']}]}))"}}]
/// Always confirm a protocol's exact event ids, field names, and action names
/// with get_protocol_docs before writing a handler.
#[serde(default)]
pub event_handlers: Option<Vec<serde_json::Value>>,
/// Optional protocol-specific startup parameters (JSON object). For example,
/// HTTP accepts a `request_filter` (array of {methods, path regex, headers}
/// rules — only matching requests reach the LLM; the rest get `filtered_response`,
/// default 404) and a `filtered_response` ({status, body, headers}). Use
/// get_protocol_docs for a protocol's available startup parameters.
#[serde(default)]
pub startup_params: Option<serde_json::Value>,
/// Network interface to bind, for interface-bound (layer 2 / raw socket)
/// protocols such as `arp`, `datalink`, `icmp` and `isis` — e.g. "en0", "lo".
/// Ignored by port-based protocols. Check get_protocol_docs: if the protocol
/// is interface-bound its docs list `interface` instead of `port`.
#[serde(default)]
pub interface: Option<String>,
/// Source MAC address for layer 2 protocols (e.g. "02:00:00:00:00:01").
/// Only meaningful together with `interface`.
#[serde(default)]
pub mac_address: Option<String>,
/// Make the server speak first on connect instead of waiting for the client
/// — the FTP/SMTP-style greeting banner. Only protocols that declare a
/// `send_first` startup parameter honour it (tcp, redis, postgresql, mysql,
/// mongodb, jsonrpc, irc, ldap, …); for the rest it is ignored with a
/// warning. The banner content comes from the handler for the protocol's
/// "connection opened" event.
#[serde(default)]
pub send_first: Option<bool>,
/// Seed the server's LLM memory before the first request. Only affects the
/// `instruction` path; `event_handlers` never read memory.
#[serde(default)]
pub initial_memory: Option<String>,
/// Natural-language instructions for the automatic feedback loop that
/// adjusts this server's behaviour as it runs.
#[serde(default)]
pub feedback_instructions: Option<String>,
/// Server-scoped scheduled LLM tasks, removed when the server stops. Array of
/// objects: {"task_id":"...", "recurring":true, "interval_secs":60,
/// "instruction":"..."} for a repeating task, or {"task_id":"...",
/// "recurring":false, "delay_secs":10, "instruction":"..."} for a one-shot.
/// Optional: "max_executions" (number), "context" (any JSON).
#[serde(default)]
pub scheduled_tasks: Option<Vec<serde_json::Value>>,
}
#[derive(Debug, Deserialize, JsonSchema)]
pub struct StartClientParams {
/// Protocol name. Use list_protocols with type "client" to see what is
/// available in this build, and get_protocol_docs for a protocol's event ids,
/// actions and startup parameters.
pub protocol: String,
/// Remote server to connect to, as "host:port" (e.g. "127.0.0.1:6379").
pub remote_addr: String,
/// Natural language instruction for the LLM that handles each event on this
/// connection. This is the LLM FALLBACK path: every event triggers a (slow,
/// billable) model call. Prefer `event_handlers` whenever the behavior is
/// deterministic.
#[serde(default)]
pub instruction: Option<String>,
/// Event handlers that decide how events are handled WITHOUT an LLM call.
/// Same shape as start_server's: an array of
/// { "event_pattern": "<event id>" | "*", "handler": {...} }, matched in
/// order, first match wins; `handler` is one of {"type":"script",...},
/// {"type":"static","actions":[...]} or {"type":"llm","instruction":"..."}.
/// Use get_protocol_docs for the protocol's client event ids and action names.
#[serde(default)]
pub event_handlers: Option<Vec<serde_json::Value>>,
/// Optional protocol-specific startup parameters (JSON object). Use
/// get_protocol_docs for a protocol's available startup parameters.
#[serde(default)]
pub startup_params: Option<serde_json::Value>,
/// Seed the client's LLM memory before the first event.
#[serde(default)]
pub initial_memory: Option<String>,
/// Natural-language instructions for the automatic feedback loop that adjusts
/// this client's behaviour as it runs.
#[serde(default)]
pub feedback_instructions: Option<String>,
/// Client-scoped scheduled LLM tasks. Same shape as start_server's:
/// {"task_id":"...", "recurring":true, "interval_secs":60, "instruction":"..."}
/// or {"task_id":"...", "recurring":false, "delay_secs":10, "instruction":"..."}.
#[serde(default)]
pub scheduled_tasks: Option<Vec<serde_json::Value>>,
}
#[derive(Debug, Deserialize, JsonSchema)]
pub struct StopClientParams {
/// Client ID to stop
#[serde(deserialize_with = "deserialize_u32_flexible")]
pub client_id: u32,
}
#[derive(Debug, Deserialize, JsonSchema)]
pub struct ClientStatusParams {
/// Client ID to query
#[serde(deserialize_with = "deserialize_u32_flexible")]
pub client_id: u32,
}
#[derive(Debug, Deserialize, JsonSchema)]
pub struct StopServerParams {
/// Server ID to stop
#[serde(deserialize_with = "deserialize_u32_flexible")]
pub server_id: u32,
}
#[derive(Debug, Deserialize, JsonSchema)]
pub struct ServerStatusParams {
/// Server ID to query
#[serde(deserialize_with = "deserialize_u32_flexible")]
pub server_id: u32,
}
#[derive(Debug, Deserialize, JsonSchema)]
pub struct SetModelParams {
/// Model name (e.g., "qwen3-coder:30b", "llama3.2:latest")
pub model: String,
}
#[derive(Debug, Deserialize, JsonSchema)]
pub struct GetProtocolDocsParams {
/// Protocol name (e.g., "http", "dns", "tcp")
pub protocol: String,
}
#[derive(Debug, Deserialize, JsonSchema)]
pub struct ListAccessLogsParams {
/// Maximum number of recent entries to return (default 20)
#[serde(default, deserialize_with = "deserialize_option_u32_flexible")]
pub limit: Option<u32>,
}
#[derive(Debug, Deserialize, JsonSchema)]
pub struct GetAccessLogParams {
/// Access-log entry id (from list_access_logs)
#[serde(deserialize_with = "deserialize_u64_flexible")]
pub id: u64,
}
#[derive(Debug, Deserialize, JsonSchema)]
pub struct UpdateServerInstructionParams {
/// Server ID
#[serde(deserialize_with = "deserialize_u32_flexible")]
pub server_id: u32,
/// New instruction text
pub instruction: String,
}
#[derive(Debug, Deserialize, JsonSchema)]
pub struct UpdateServerParams {
/// Server ID to update (from list_servers)
#[serde(deserialize_with = "deserialize_u32_flexible")]
pub server_id: u32,
/// New natural-language LLM instruction (hot-applied, connections preserved).
#[serde(default)]
pub instruction: Option<String>,
/// Replacement event handlers (hot-applied). Same shape as start_server's.
#[serde(default)]
pub event_handlers: Option<Vec<serde_json::Value>>,
/// Seed/replace the server's LLM memory (hot-applied).
#[serde(default)]
pub initial_memory: Option<String>,
/// Startup parameters to merge in. Validated against the protocol schema;
/// applying them triggers a clean stop+start (connections dropped, new id).
#[serde(default)]
pub startup_params: Option<serde_json::Value>,
/// New port to bind. Triggers a restart.
#[serde(default, deserialize_with = "deserialize_option_u16_flexible")]
pub port: Option<u16>,
/// New host/IP to bind. Triggers a restart.
#[serde(default)]
pub host: Option<String>,
/// New feedback-loop instructions (hot-applied).
#[serde(default)]
pub feedback_instructions: Option<String>,
/// Scheduled tasks to add (hot-applied). Same shape as start_server's.
#[serde(default)]
pub scheduled_tasks: Option<Vec<serde_json::Value>>,
}
#[derive(Debug, Deserialize, JsonSchema)]
pub struct UpdateClientParams {
/// Client ID to update
#[serde(deserialize_with = "deserialize_u32_flexible")]
pub client_id: u32,
/// New natural-language LLM instruction (hot-applied).
#[serde(default)]
pub instruction: Option<String>,
/// Replacement event handlers (hot-applied).
#[serde(default)]
pub event_handlers: Option<Vec<serde_json::Value>>,
/// Seed/replace the client's LLM memory (hot-applied).
#[serde(default)]
pub initial_memory: Option<String>,
/// Startup parameters to merge in. Validated; triggers a reconnect (new id).
#[serde(default)]
pub startup_params: Option<serde_json::Value>,
/// New remote address "host:port". Triggers a reconnect.
#[serde(default)]
pub remote_addr: Option<String>,
/// New feedback-loop instructions (hot-applied).
#[serde(default)]
pub feedback_instructions: Option<String>,
/// Scheduled tasks to add (hot-applied).
#[serde(default)]
pub scheduled_tasks: Option<Vec<serde_json::Value>>,
}
#[derive(Debug, Deserialize, JsonSchema)]
pub struct GetNextLlmRequestParams {
/// Seconds to long-poll for a request when none is immediately pending
/// (0 = return right away). Capped at 120 to stay under the client's tool-call timeout.
#[serde(default, deserialize_with = "deserialize_option_u32_flexible")]
pub wait_seconds: Option<u32>,
}
#[derive(Debug, Deserialize, JsonSchema)]
pub struct AnswerLlmRequestParams {
/// The request id returned by get_next_llm_request
#[serde(deserialize_with = "deserialize_u64_flexible")]
pub request_id: u64,
/// NetGet actions to execute, as a JSON array. Each object has a `type` field
/// naming the action plus its parameters (same shape shown by get_access_log).
pub actions: Vec<serde_json::Value>,
}
#[derive(Debug, Deserialize, JsonSchema)]
pub struct SendToClientParams {
/// Client ID (from list_clients)
#[serde(deserialize_with = "deserialize_u32_flexible")]
pub client_id: u32,
/// One of the client protocol's actions, as a JSON object with a `type` field naming it
/// plus its parameters, e.g. {"type":"send_tcp_data","data":"hello\n"}. get_protocol_docs
/// lists the actions a client accepts, with parameters and examples.
pub action: serde_json::Value,
/// Seconds to wait for the client's loop to execute the action and report (default 30,
/// max 120).
#[serde(default, deserialize_with = "deserialize_option_u32_flexible")]
pub timeout_secs: Option<u32>,
}
#[derive(Debug, Deserialize, JsonSchema)]
pub struct SendToPeerParams {
/// Server ID (from list_servers)
#[serde(deserialize_with = "deserialize_u32_flexible")]
pub server_id: u32,
/// Connection ID of the live peer (server_status lists them, and marks which accept
/// send_to_peer)
#[serde(deserialize_with = "deserialize_u32_flexible")]
pub connection_id: u32,
/// One of the server protocol's actions, as a JSON object with a `type` field, e.g.
/// {"type":"send_tcp_data","data":"hello\n"}.
pub action: serde_json::Value,
/// Seconds to wait for the connection to execute the action and report (default 30,
/// max 120).
#[serde(default, deserialize_with = "deserialize_option_u32_flexible")]
pub timeout_secs: Option<u32>,
}
#[derive(Debug, Deserialize, JsonSchema)]
pub struct DisconnectPeerParams {
/// Server ID (from list_servers)
#[serde(deserialize_with = "deserialize_u32_flexible")]
pub server_id: u32,
/// Connection ID of the live peer to hang up on (from server_status)
#[serde(deserialize_with = "deserialize_u32_flexible")]
pub connection_id: u32,
}
#[derive(Debug, Deserialize, JsonSchema)]
pub struct AnswerInterceptParams {
/// The parked request's id (from list_intercepts)
#[serde(deserialize_with = "deserialize_u64_flexible")]
pub intercept_id: u64,
/// The answer: a JSON array of action objects, each {"type":"<name>", ...parameters},
/// using only the actions list_intercepts offers for this request. An EMPTY array is a
/// real answer — "acknowledge, say nothing": nothing is written and the connection
/// carries on. `{{event.<field>}}` references are filled from the request's event data,
/// exactly as in a static handler.
pub actions: Vec<serde_json::Value>,
}
#[derive(Debug, Deserialize, JsonSchema)]
pub struct FailInterceptParams {
/// The parked request's id (from list_intercepts)
#[serde(deserialize_with = "deserialize_u64_flexible")]
pub intercept_id: u64,
}
/// Default wait for `send_to_client` / `send_to_peer`, the dashboard composer's own figure.
const INJECT_TIMEOUT_DEFAULT_SECS: u32 = 30;
/// Ceiling on a caller-supplied wait, under typical MCP client tool-call timeouts.
const INJECT_TIMEOUT_MAX_SECS: u32 = 120;
/// Ensure a named pipe (FIFO) exists at `path`, creating it if absent.
#[cfg(unix)]
fn ensure_fifo(path: &std::path::Path) -> anyhow::Result<()> {
use std::os::unix::ffi::OsStrExt;
use std::os::unix::fs::FileTypeExt;
match std::fs::symlink_metadata(path) {
Ok(metadata) if metadata.file_type().is_fifo() => return Ok(()),
Ok(_) => anyhow::bail!(
"--llm-agent-pipe path {} must be a FIFO, not a regular file, directory, or symlink",
path.display()
),
Err(error) if error.kind() == std::io::ErrorKind::NotFound => {}
Err(error) => return Err(error.into()),
}
let c_path = std::ffi::CString::new(path.as_os_str().as_bytes())
.map_err(|_| anyhow::anyhow!("invalid FIFO path: {}", path.display()))?;
// 0o600: owner read/write only.
let rc = unsafe { libc::mkfifo(c_path.as_ptr(), 0o600) };
if rc != 0 {
return Err(std::io::Error::last_os_error())
.map_err(|e| anyhow::anyhow!("failed to create FIFO {}: {}", path.display(), e));
}
Ok(())
}
#[cfg(not(unix))]
fn ensure_fifo(_path: &std::path::Path) -> anyhow::Result<()> {
anyhow::bail!("--llm-agent-pipe (named pipes) is only supported on Unix platforms")
}
/// How often the MCP-mode reaper runs. Matches the TUI's cleanup tick.
const REAPER_INTERVAL_SECS: u64 = 5;
/// How long a `Stopped` server stays visible before being reclaimed.
const SERVER_CLEANUP_TIMEOUT_SECS: u64 = 30;
/// How long a closed connection stays visible before being reclaimed.
const CONNECTION_CLEANUP_TIMEOUT_SECS: u64 = 10;
/// How long an idle connectionless (UDP-style) connection stays visible.
const CONNECTIONLESS_CLEANUP_TIMEOUT_SECS: u64 = 10;
/// Start the background reaper that reclaims stopped servers, closed connections
/// and finished conversations in MCP mode.
///
/// The TUI drives this from its own event loop; MCP mode has no such loop, so
/// without this task nothing ever called `cleanup_old_servers`. `close_server`
/// (whether issued by the LLM or by the `stop_server` tool's TUI-equivalent path)
/// only marks a server `Stopped`, so those entries — and, before the cleanup moved
/// into `remove_server`, their scheduled tasks — accumulated for the life of the
/// process.
///
/// It also drains due `provide_feedback` batches. `start_server` accepts
/// `feedback_instructions`, which makes `provide_feedback` an advertised tool, so
/// without this an MCP caller could switch the loop on and have every entry
/// accumulate unread — the TUI and non-interactive modes drive the drain from their
/// task timer. Scheduled tasks are driven separately by `spawn_task_ticker`.
///
/// SharedState owns this handle and aborts it when the final service goes away.
fn spawn_state_reaper(
app_state: AppState,
llm_client: crate::llm::OllamaClient,
status_tx: mpsc::UnboundedSender<String>,
) -> tokio::task::JoinHandle<()> {
tokio::spawn(async move {
let mut ticker =
tokio::time::interval(std::time::Duration::from_secs(REAPER_INTERVAL_SECS));
loop {
ticker.tick().await;
crate::llm::feedback::execute_due_feedback(&app_state, &llm_client, &status_tx).await;
app_state
.cleanup_old_servers(SERVER_CLEANUP_TIMEOUT_SECS)
.await;
app_state
.cleanup_closed_connections(CONNECTION_CLEANUP_TIMEOUT_SECS)
.await;
app_state
.cleanup_old_connections(CONNECTIONLESS_CLEANUP_TIMEOUT_SECS)
.await;
app_state.cleanup_old_conversations().await;
}
})
}
/// How often the MCP-mode scheduled-task ticker fires. Matches the 1s cadence the
/// TUI event loop and the non-interactive `run_server` runner use for their task
/// timers, so a task's declared `interval_secs`/`delay_secs` are honoured to the
/// same resolution in every mode.
const TASK_TICK_INTERVAL_SECS: u64 = 1;
/// Start the background scheduled-task ticker for MCP mode.
///
/// The TUI event loop and the non-interactive runner both drive
/// `execute_due_tasks_public` from a 1s timer; MCP mode had no such loop, so tasks
/// created by `start_server`'s `scheduled_tasks` array or the `schedule_task` action
/// sat at `Scheduled` forever and never fired. This restores that timer for both the
/// STDIO and HTTP transports (they share this one `SharedState`).
///
/// Scoping is not re-implemented here: `execute_due_tasks_public` reads the live task
/// set from `AppState` on every tick, so Global / Server / Connection scoping — and
/// the removal of server- and connection-scoped tasks by `remove_server` on
/// `stop_server`/`stop_all` — are honoured exactly as the TUI honours them. A stopped
/// server's tasks are gone from state before the next tick, so they cannot fire; only
/// Global tasks (which the TUI also keeps) persist.
///
/// SharedState owns this handle, so dropping the final service stops ticking.
fn spawn_task_ticker(
app_state: AppState,
llm_client: crate::llm::OllamaClient,
status_tx: mpsc::UnboundedSender<String>,
owner: tokio_util::sync::CancellationToken,
) -> tokio::task::JoinHandle<()> {
tokio::spawn(async move {
let mut ticker =
tokio::time::interval(std::time::Duration::from_secs(TASK_TICK_INTERVAL_SECS));
// A tick missed while a previous batch was executing should not stampede a
// burst of catch-up ticks, exactly as the TUI's task timer skips them.
ticker.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);
loop {
tokio::select! {
biased;
_ = owner.cancelled() => break,
_ = ticker.tick() => {}
}
crate::cli::execute_due_tasks_owned_public(
&app_state,
&llm_client,
&status_tx,
owner.clone(),
)
.await;
}
})
}
/// The port a server is actually listening on.
///
/// `ServerInstance::port` is the port that was *requested*, which is `0` for an OS-assigned
/// one; the address it really bound is in `local_addr`. Reporting the requested value meant an
/// MCP caller who started a server with `"port": 0` — the documented way to get an ephemeral
/// port — was told it was listening on port 0 and had no way at all to discover the real one.
/// `list_servers`, `get_server_details`, `get_status` and the `stop_server` confirmation all
/// had it.
///
/// Same defect as the dashboard's edit form had (`ServerPrefill::from_server`), from the same
/// cause: `port: 0` now passes through to the protocol's own bind, so every reader of `.port`
/// has to resolve it.
fn bound_port(server: &crate::state::server::ServerInstance) -> u16 {
server
.local_addr
.map(|a| a.port())
.filter(|p| *p != 0)
.unwrap_or(server.port)
}
// === Tool implementations ===
#[tool_router]
impl NetGetMcpService {
pub async fn new(args: &Args, settings: Settings) -> anyhow::Result<Self> {
let state = Self::create_shared_state(args, settings).await?;
Ok(Self::new_with_shared_state(state))
}
/// Create the shared state used by all MCP sessions.
///
/// For STDIO this is called once for the single session; for the HTTP
/// transport it is created once and shared across all sessions so they
/// see the same servers/clients.
pub(crate) async fn create_shared_state(
args: &Args,
settings: Settings,
) -> anyhow::Result<Arc<SharedState>> {
// Create app state.
//
// This is the same startup configuration the TUI (`cli::run`), the
// one-shot runner (`cli::non_interactive`) and `--client` all perform.
// MCP used to skip every bit of it — it built a bare `AppState::new()`,
// which hard-codes `RateLimiterConfig::default()`, `ScriptingMode::On`
// and `include_disabled_protocols: false`. So `--llm-max-concurrent`,
// `--llm-token-limit`, `--llm-token-window`, `--env`, `--no-scripts`,
// `--handler` and `--include-disabled-protocols` were all accepted and
// silently ignored in the mode this project documents as its primary
// programmatic surface. Anything added to the TUI's startup below
// belongs here too.
let base_url = args
.openai_url
.clone()
.or_else(|| args.ollama_url.clone())
.unwrap_or_else(|| "http://localhost:11434".to_string());
let app_state = AppState::new_with_options(args.include_disabled_protocols, base_url);
app_state
.set_min_stability(args.parse_min_stability()?)
.await;
app_state
.configure_rate_limiter(args.build_rate_limiter_config())
.await?;
// Scripting mode: CLI arg > saved setting. Unlike the TUI we do not
// validate availability by bailing, because there is no operator to read
// the message: an unavailable runtime is reported and the mode is left
// alone rather than killing the transport at handshake time.
if let Some(mode) = args
.parse_scripting_mode()?
.or_else(|| settings.parse_scripting_mode())
{
let scripting_env = app_state.get_scripting_env().await;
let available = match mode {
crate::state::app_state::ScriptingMode::On
| crate::state::app_state::ScriptingMode::Off => true,
crate::state::app_state::ScriptingMode::Python => scripting_env.python.is_some(),
crate::state::app_state::ScriptingMode::JavaScript => {
scripting_env.javascript.is_some()
}
crate::state::app_state::ScriptingMode::Go => scripting_env.go.is_some(),
crate::state::app_state::ScriptingMode::Perl => scripting_env.perl.is_some(),
};
if available {
app_state.set_selected_scripting_mode(mode).await;
} else {
tracing::warn!(
"Requested scripting environment {} is not available on this system; leaving scripting mode unchanged",
mode
);
}
}
if let Some(handler_mode) = args.parse_event_handler_mode()? {
app_state.set_event_handler_mode(handler_mode).await;
}
// Web search: ASK needs a TUI to prompt with, and MCP has none, so it
// degrades to OFF exactly as the non-interactive runner does.
let mut web_search_mode = settings.get_web_search_mode();
if web_search_mode == crate::state::app_state::WebSearchMode::Ask {
web_search_mode = crate::state::app_state::WebSearchMode::Off;
}
app_state.set_web_search_mode(web_search_mode).await;
// Build the LLM client. Three mutually-exclusive sources:
// --llm-agent → queue requests for the calling MCP agent (no model)
// --openai-url → OpenAI-compatible endpoint
// (default) → local Ollama
let (llm_client, agent_queue) = if args.llm_agent {
let pipe = args.llm_agent_pipe.clone();
if let Some(ref path) = pipe {
ensure_fifo(path)?;
}
let queue = Arc::new(LlmRequestQueue::new(pipe));
let timeout = std::time::Duration::from_secs(args.llm_agent_timeout);
let client = OllamaClient::new_queue(queue.clone(), timeout);
// Pre-seed a placeholder model so `ensure_model_selected` never reaches
// out to Ollama's /api/tags (no real model is involved in this mode).
if args.model.is_none() {
app_state.set_ollama_model(Some("agent".to_string())).await;
}
(client, Some(queue))
} else {
// `with_app_state` is what records token usage against AppState (the
// `/usage` accounting); `with_mock_config_file` is what lets a test
// drive this transport without a model. Both were missing here while
// every other startup path set them.
let client = crate::cli::create_llm_client(args)?
.with_mock_config_file(args.mock_config_file.clone())
.with_app_state(app_state.clone());
(client, None)
};
// Status channel (messages go to stderr in MCP mode)
let (status_tx, mut status_rx) = mpsc::unbounded_channel();
// Set up model if specified: CLI arg > saved setting, matching every
// other startup path (MCP used to read only the CLI arg, so a model
// chosen in the TUI and persisted to settings was ignored here).
// (`--llm-agent` is excluded from the fallback: it pre-seeded the
// placeholder "agent" model above and contacts no model at all.)
let configured_model = args.model.clone().or_else(|| {
if args.llm_agent {
None
} else {
settings.model.clone()
}
});
if let Some(model) = configured_model {
app_state.set_ollama_model(Some(model)).await;
}
// Reap stopped servers, closed connections and finished conversations.
// Without this, MCP mode never reclaims them: `cleanup_old_servers` was only
// ticked by the TUI event loop, so an LLM-initiated `close_server` (which
// marks the server Stopped rather than removing it) left the entry — and its
// scheduled tasks — in AppState for the life of the process.
let reaper = spawn_state_reaper(app_state.clone(), llm_client.clone(), status_tx.clone());
// Fire scheduled tasks. Without this, `start_server`'s `scheduled_tasks` and
// the `schedule_task` action create tasks that never execute in MCP mode —
// only the TUI and non-interactive runners ticked `execute_due_tasks`.
let task_owner = tokio_util::sync::CancellationToken::new();
let ticker = spawn_task_ticker(
app_state.clone(),
llm_client.clone(),
status_tx.clone(),
task_owner.clone(),
);
// The receiver belongs to the same service lifetime as its producers.
let status_drain = tokio::spawn(async move {
while let Some(msg) = status_rx.recv().await {
eprintln!("[NETGET] {}", msg);
}
});
let state = Arc::new(SharedState {
app_state,
llm_client,
agent_queue,
_status_tx: status_tx,
background_tasks: vec![reaper, status_drain],
task_owner,
_task_ticker: ticker,
});
Ok(state)
}
/// Create a service instance backed by pre-built shared state
/// (used by the HTTP transport, which creates one service per session)
pub(crate) fn new_with_shared_state(state: Arc<SharedState>) -> Self {
Self {
state,
tool_router: Self::tool_router(),
}
}
#[tool(
description = "List all available network protocols that can be started as servers or clients"
)]
async fn list_protocols(
&self,
Parameters(params): Parameters<ListProtocolsParams>,
) -> Result<CallToolResult, McpError> {
let filter_type = params.r#type.as_deref().unwrap_or("all");
let include_disabled = params.include_disabled.unwrap_or(false);
let mut result = String::new();
if filter_type == "all" || filter_type == "server" {
let registry = crate::protocol::server_registry::registry();
let protocols = registry.available_protocols();
result.push_str("## Server Protocols\n\n");
for name in &protocols {
if let Some(proto) = registry.get(name) {
let metadata = proto.metadata();
if !include_disabled
&& matches!(
metadata.state,
crate::protocol::metadata::DevelopmentState::Incomplete
)
{
continue;
}
result.push_str(&format!(
"- **{}** ({}) - {}\n",
name,
metadata.state.as_str(),
metadata.implementation
));
}
}
result.push('\n');
}
if filter_type == "all" || filter_type == "client" {
let registry = &crate::protocol::CLIENT_REGISTRY;
let mut protocols = registry.list_protocols();
protocols.sort();
result.push_str("## Client Protocols\n\n");
result.push_str("Start one with `start_client` (needs a `remote_addr`).\n\n");
for name in &protocols {
if let Some(proto) = registry.get(name) {
let metadata = proto.metadata();
if !include_disabled
&& matches!(
metadata.state,
crate::protocol::metadata::DevelopmentState::Incomplete
)
{
continue;
}
result.push_str(&format!(
"- **{}** ({}) - {}\n",
name,
metadata.state.as_str(),
proto.description()
));
} else {
result.push_str(&format!("- **{}**\n", name));
}
}
result.push('\n');
}
Ok(CallToolResult::success(vec![Content::text(result)]))
}
#[tool(
description = "Start a network protocol server. Choose HOW it responds: for DETERMINISTIC behavior (echo, fixed/canned responses, simple routing, high throughput) pass `event_handlers` with a script or static handler — these run in-process with NO LLM call (instant, free, reproducible). Only use the natural-language `instruction` when responses genuinely need reasoning or vary unpredictably; it invokes NetGet's LLM on every request. Prefer event_handlers whenever the logic is fixed."
)]
async fn start_server(
&self,
Parameters(params): Parameters<StartServerParams>,
) -> Result<CallToolResult, McpError> {
let instruction = params.instruction.unwrap_or_else(|| {
format!(
"You are a {} server. Handle requests appropriately.",
params.protocol
)
});
// Drive the new server with NetGet's configured LLM client (Ollama/OpenAI from CLI args)
self.state
.app_state
.set_llm_client(self.state.llm_client.clone())
.await;
info!(
"MCP: Starting {} server on {}:{}",
params.protocol,
params.host.as_deref().unwrap_or("127.0.0.1"),
params.port.unwrap_or(0)
);
let scheduled_tasks = match parse_scheduled_tasks(params.scheduled_tasks) {
Ok(tasks) => tasks,
Err(e) => {
return Ok(CallToolResult::error(vec![Content::text(format!(
"Invalid scheduled_tasks: {}",
e
))]));
}
};
let (status_tx, mut status_rx) = mpsc::unbounded_channel();
let server_id = match crate::cli::server_startup::start_server_from_action(
&self.state.app_state,
params.mac_address.clone(), // mac_address
params.interface.clone(), // interface
params.host.clone(), // host
params.port, // port
¶ms.protocol, // protocol
params.send_first.unwrap_or(false), // send_first
params.initial_memory, // initial_memory
instruction, // instruction
params.startup_params, // startup_params
params.event_handlers, // event_handlers
scheduled_tasks, // scheduled_tasks
params.feedback_instructions, // feedback_instructions
status_tx, // status_tx
)
.await
{
Ok(id) => id,
Err(e) => {
return Ok(CallToolResult::error(vec![Content::text(format!(
"Failed to start server: {}",
e
))]));
}
};
// Collect startup status messages
let mut messages = Vec::new();
while let Ok(msg) = status_rx.try_recv() {
messages.push(msg);
}
let result = format!(
"Server #{} ({}) started.\n{}",
server_id.as_u32(),
params.protocol,
messages.join("\n")
);
Ok(CallToolResult::success(vec![Content::text(result)]))
}
#[tool(description = "Stop a running server by its ID")]
async fn stop_server(
&self,
Parameters(params): Parameters<StopServerParams>,
) -> Result<CallToolResult, McpError> {
let server_id = crate::state::ServerId::new(params.server_id);
match self.state.app_state.get_server(server_id).await {
Some(server) => {
let protocol = server.protocol_name.clone();
let port = bound_port(&server);
self.state.app_state.remove_server(server_id).await;
Ok(CallToolResult::success(vec![Content::text(format!(
"Server #{} ({} on port {}) stopped",
params.server_id, protocol, port
))]))
}
None => Ok(CallToolResult::error(vec![Content::text(format!(
"Server #{} not found",
params.server_id
))])),
}
}
#[tool(
description = "List all running servers with their status, protocol, port, and connection count"
)]
async fn list_servers(&self) -> Result<CallToolResult, McpError> {
let servers = self.state.app_state.get_all_servers().await;
if servers.is_empty() {
return Ok(CallToolResult::success(vec![Content::text(
"No servers currently running.",
)]));
}
let mut result = String::from("## Running Servers\n\n");
for server in &servers {
result.push_str(&format!(
"- **Server #{}**: {} on port {} ({})\n Instruction: {}\n Memory: {}\n\n",
server.id.as_u32(),
server.protocol_name,
bound_port(server),
server.status,
if server.instruction.is_empty() {
"(none)"
} else {
&server.instruction
},
if server.memory.is_empty() {
"(empty)"
} else {
&server.memory
},
));
}
Ok(CallToolResult::success(vec![Content::text(result)]))
}
#[tool(description = "Get detailed status of a specific server")]
async fn server_status(
&self,
Parameters(params): Parameters<ServerStatusParams>,
) -> Result<CallToolResult, McpError> {
let server_id = crate::state::ServerId::new(params.server_id);
match self.state.app_state.get_server(server_id).await {
Some(server) => {
let result = format!(
"## Server #{}\n\n\
- **Protocol**: {}\n\
- **Port**: {}\n\
- **Status**: {}\n\
- **Instruction**: {}\n\
- **Memory**: {}\n",
server.id.as_u32(),
server.protocol_name,
bound_port(&server),
server.status,
if server.instruction.is_empty() {
"(none)"
} else {
&server.instruction
},
if server.memory.is_empty() {
"(empty)"
} else {
&server.memory
},
);
let mut result = result;
// Connections, with the id send_to_peer / disconnect_peer take and whether
// each accepts them (only protocols that register a peer handle do).
let mut connections: Vec<_> = server.connections.values().collect();
connections.sort_by_key(|c| c.id.as_u32());
if connections.is_empty() {
result.push_str("- **Connections**: none\n");
} else {
result.push_str("\n### Connections\n\n");
if let Some(reason) = crate::protocol::server_registry::registry()
.request_only_reason(&server.protocol_name)
{
result.push_str(&format!(
"No connection accepts send_to_peer: {}.\n\n",
reason
));
}
for conn in connections {
let injectable = self
.state
.app_state
.has_peer_handle(server_id, conn.id.as_u32())
.await;
result.push_str(&format!(
"- connection_id **{}** — {} ({:?}), {} byte(s) in / {} out{}\n",
conn.id.as_u32(),
conn.remote_addr,
conn.status,
conn.bytes_received,
conn.bytes_sent,
if injectable {
" — accepts send_to_peer / disconnect_peer"
} else {
""
}
));
}
}
Ok(CallToolResult::success(vec![Content::text(result)]))
}
None => Ok(CallToolResult::error(vec![Content::text(format!(
"Server #{} not found",
params.server_id
))])),
}
}
#[tool(
description = "Get overall NetGet status including the configured model and running servers"
)]
async fn get_status(&self) -> Result<CallToolResult, McpError> {
let servers = self.state.app_state.get_all_servers().await;
let model = self
.state
.app_state
.get_ollama_model()
.await
.unwrap_or_else(|| "(auto-select)".to_string());
let mut result = format!(
"## NetGet Status\n\n\
- **Model**: {}\n\
- **LLM backend**: {}\n\
- **Running servers**: {}\n",
model,
self.state.llm_client.backend_type(),
servers.len(),
);
if !servers.is_empty() {
result.push_str("\n### Servers\n");
for server in &servers {
result.push_str(&format!(
"- #{}: {} on port {} ({})\n",
server.id.as_u32(),
server.protocol_name,
bound_port(server),
server.status
));
}
}
Ok(CallToolResult::success(vec![Content::text(result)]))
}
#[tool(description = "Change the default LLM model used for new servers/clients")]
async fn set_model(
&self,
Parameters(params): Parameters<SetModelParams>,
) -> Result<CallToolResult, McpError> {
self.state
.app_state
.set_ollama_model(Some(params.model.clone()))
.await;
Ok(CallToolResult::success(vec![Content::text(format!(
"Model changed to: {}",
params.model
))]))
}
#[tool(
description = "Get MCP-shaped documentation for a protocol: the exact start_server/start_client arguments that apply, the protocol's event ids with their field names and types (needed to write an event_handlers script), its action names with parameter schemas and examples, its startup parameters, its privilege requirement, and its maturity."
)]
async fn get_protocol_docs(
&self,
Parameters(params): Parameters<GetProtocolDocsParams>,
) -> Result<CallToolResult, McpError> {
match crate::mcp_stdio::docs::render_protocol_docs(¶ms.protocol, &self.state.app_state)
.await
{
Some(text) => Ok(CallToolResult::success(vec![Content::text(text)])),
None => {
// Protocol not compiled in - list what's available
let server_registry = crate::protocol::server_registry::registry();
let available = server_registry.available_protocols();
Ok(CallToolResult::error(vec![Content::text(format!(
"Protocol '{}' not found. It may not be compiled into this build.\n\n\
Available server protocols ({}):\n{}\n\n\
Build with the protocol feature enabled: cargo build --features mcp-stdio,{}",
params.protocol,
available.len(),
available
.iter()
.map(|p| format!(" - {}", p))
.collect::<Vec<_>>()
.join("\n"),
params.protocol.to_lowercase(),
))]))
}
}
}
#[tool(description = "Update the LLM instruction for a running server")]
async fn update_server_instruction(
&self,
Parameters(params): Parameters<UpdateServerInstructionParams>,
) -> Result<CallToolResult, McpError> {
let server_id = crate::state::ServerId::new(params.server_id);
let updated = self
.state
.app_state
.with_server_mut(server_id, |server| {
server.instruction = params.instruction.clone();
})
.await;
match updated {
Some(()) => Ok(CallToolResult::success(vec![Content::text(format!(
"Server #{} instruction updated",
params.server_id
))])),
None => Ok(CallToolResult::error(vec![Content::text(format!(
"Server #{} not found",
params.server_id
))])),
}
}
#[tool(
description = "Update a RUNNING server in place, by its id, instead of starting a second one. Supply only the fields to change. instruction / event_handlers / initial_memory / feedback_instructions / scheduled_tasks are hot-applied WITHOUT dropping connections; port / host / startup_params require a clean stop+start (connections dropped, server gets a NEW id). Unknown id and invalid startup_params error cleanly and leave the running server untouched. Prefer this over start_server when a server already exists."
)]
async fn update_server(
&self,
Parameters(params): Parameters<UpdateServerParams>,
) -> Result<CallToolResult, McpError> {
// Keep AppState's LLM client in sync in case the update restarts the server.
self.state
.app_state
.set_llm_client(self.state.llm_client.clone())
.await;
let scheduled_tasks = match parse_scheduled_tasks(params.scheduled_tasks) {
Ok(t) => t,
Err(e) => {
return Ok(CallToolResult::error(vec![Content::text(format!(
"Invalid scheduled_tasks: {}",
e
))]))
}
};
let form = crate::cli::management::ServerForm {
protocol: String::new(),
port: params.port,
host: params.host,
interface: None,
mac_address: None,
send_first: false,
instruction: params.instruction,
initial_memory: params.initial_memory,
startup_params: params.startup_params,
event_handlers: params.event_handlers,
scheduled_tasks,
feedback_instructions: params.feedback_instructions,
};
let (status_tx, _status_rx) = mpsc::unbounded_channel();
match crate::cli::management::update_server(
&self.state.app_state,
crate::state::ServerId::new(params.server_id),
form,
status_tx,
)
.await
{
Ok(outcome) => Ok(CallToolResult::success(vec![Content::text(
outcome.summary,
)])),
Err(e) => Ok(CallToolResult::error(vec![Content::text(e.to_string())])),
}
}
#[tool(
description = "Update a RUNNING client in place, by its id, instead of connecting a second one. Supply only the fields to change. instruction / event_handlers / initial_memory / feedback_instructions / scheduled_tasks are hot-applied; remote_addr / startup_params trigger a reconnect (new id). Unknown id and invalid startup_params error cleanly and leave the running client untouched."
)]
async fn update_client(
&self,
Parameters(params): Parameters<UpdateClientParams>,
) -> Result<CallToolResult, McpError> {
self.state
.app_state
.set_llm_client(self.state.llm_client.clone())
.await;
let scheduled_tasks = match parse_scheduled_tasks(params.scheduled_tasks) {
Ok(t) => t,
Err(e) => {
return Ok(CallToolResult::error(vec![Content::text(format!(
"Invalid scheduled_tasks: {}",
e
))]))
}
};
let form = crate::cli::management::ClientForm {
protocol: String::new(),
remote_addr: params.remote_addr,
instruction: params.instruction,
initial_memory: params.initial_memory,
startup_params: params.startup_params,
event_handlers: params.event_handlers,
scheduled_tasks,
feedback_instructions: params.feedback_instructions,
};
let (status_tx, _status_rx) = mpsc::unbounded_channel();
match crate::cli::management::update_client(
&self.state.app_state,
crate::state::ClientId::new(params.client_id),
form,
self.state.llm_client.clone(),
status_tx,
)
.await
{
Ok(outcome) => Ok(CallToolResult::success(vec![Content::text(
outcome.summary,
)])),
Err(e) => Ok(CallToolResult::error(vec![Content::text(e.to_string())])),
}
}
#[tool(
description = "List recent request/response access-log entries across all servers (newest first). Use get_access_log with an entry id to see the full request and response."
)]
async fn list_access_logs(
&self,
Parameters(params): Parameters<ListAccessLogsParams>,
) -> Result<CallToolResult, McpError> {
let limit = params.limit.unwrap_or(20) as usize;
let entries = self.state.app_state.list_access_logs(Some(limit)).await;
if entries.is_empty() {
return Ok(CallToolResult::success(vec![Content::text(
"No requests logged yet. Access logs are recorded as servers handle requests.",
)]));
}
let now_ms = crate::utils::clock::SystemTime::now()
.duration_since(crate::utils::clock::UNIX_EPOCH)
.map(|d| d.as_millis() as u64)
.unwrap_or(0);
let mut out = format!("## Recent requests ({} shown)\n\n", entries.len());
for e in &entries {
let age = now_ms.saturating_sub(e.unix_ms) / 1000;
let owner = match (e.server_id, e.client_id) {
(Some(id), _) => format!("server #{id}"),
(None, Some(id)) => format!("client #{id}"),
(None, None) => "unknown".to_string(),
};
out.push_str(&format!(
"- **#{}** [{}s ago] {} ({}) — {} — {} → {}\n",
e.id,
age,
owner,
e.protocol,
e.event_type,
summarize_request(&e.request),
summarize_response(&e.response),
));
}
out.push_str("\nUse `get_access_log` with an id for the full request and response.");
Ok(CallToolResult::success(vec![Content::text(out)]))
}
#[tool(
description = "Get the full request and response for a single access-log entry by id (from list_access_logs)"
)]
async fn get_access_log(
&self,
Parameters(params): Parameters<GetAccessLogParams>,
) -> Result<CallToolResult, McpError> {
match self.state.app_state.get_access_log(params.id).await {
Some(e) => {
let now_ms = crate::utils::clock::SystemTime::now()
.duration_since(crate::utils::clock::UNIX_EPOCH)
.map(|d| d.as_millis() as u64)
.unwrap_or(0);
let age = now_ms.saturating_sub(e.unix_ms) / 1000;
let request = serde_json::to_string_pretty(&e.request)
.unwrap_or_else(|_| e.request.to_string());
let response = serde_json::to_string_pretty(&e.response)
.unwrap_or_else(|_| "[]".to_string());
let owner = match (e.server_id, e.client_id) {
(Some(id), _) => format!("Server #{id}"),
(None, Some(id)) => format!("Client #{id}"),
(None, None) => "Unknown".to_string(),
};
let result = format!(
"## Access log #{}\n\n\
- **Owner**: {} ({})\n\
- **Connection**: {}\n\
- **Event**: {}\n\
- **When**: {}s ago (unix_ms {})\n\n\
### Request\n```json\n{}\n```\n\n\
### Response (actions)\n```json\n{}\n```\n",
e.id,
owner,
e.protocol,
e.connection_id
.map(|c| c.to_string())
.unwrap_or_else(|| "(connectionless)".to_string()),
e.event_type,
age,
e.unix_ms,
request,
response,
);
Ok(CallToolResult::success(vec![Content::text(result)]))
}
None => Ok(CallToolResult::error(vec![Content::text(format!(
"No access-log entry #{} (it may have aged out of the buffer). Use list_access_logs to see current entries.",
params.id
))])),
}
}
#[tool(description = "Stop all running servers and connections")]
async fn stop_all(&self) -> Result<CallToolResult, McpError> {
let server_ids = self.state.app_state.get_all_server_ids().await;
let count = server_ids.len();
for id in server_ids {
self.state.app_state.remove_server(id).await;
}
Ok(CallToolResult::success(vec![Content::text(format!(
"Stopped {} server(s)",
count
))]))
}
#[tool(
description = "Connect a network protocol client to a remote server. The mirror of start_server: choose HOW it responds the same way — for DETERMINISTIC behavior pass `event_handlers` with a script or static handler (in-process, NO LLM call, instant and reproducible); only use the natural-language `instruction` when the decision genuinely needs reasoning, since it invokes NetGet's LLM on every event. Use get_protocol_docs for the protocol's client event ids, action names and startup parameters."
)]
async fn start_client(
&self,
Parameters(params): Parameters<StartClientParams>,
) -> Result<CallToolResult, McpError> {
let instruction = params.instruction.unwrap_or_else(|| {
format!(
"You are a {} client connected to {}. Handle responses appropriately.",
params.protocol, params.remote_addr
)
});
let scheduled_tasks = match parse_scheduled_tasks(params.scheduled_tasks) {
Ok(tasks) => tasks,
Err(e) => {
return Ok(CallToolResult::error(vec![Content::text(format!(
"Invalid scheduled_tasks: {}",
e
))]));
}
};
// Drive the new client with NetGet's configured LLM client, exactly as
// start_server does (the client half reads it from the argument rather
// than from AppState, but keep AppState in sync for anything that does).
self.state
.app_state
.set_llm_client(self.state.llm_client.clone())
.await;
info!(
"MCP: Connecting {} client to {}",
params.protocol, params.remote_addr
);
let client_id = match crate::cli::client_startup::start_client_from_action(
&self.state.app_state,
¶ms.protocol,
¶ms.remote_addr,
instruction,
params.startup_params,
params.initial_memory,
params.event_handlers,
scheduled_tasks,
params.feedback_instructions,
self.state.llm_client.clone(),
None, // status_tx: MCP has no status stream, drain to tracing
)
.await
{
Ok(id) => id,
Err(e) => {
return Ok(CallToolResult::error(vec![Content::text(format!(
"Failed to start client: {}",
e
))]));
}
};
Ok(CallToolResult::success(vec![Content::text(format!(
"Client #{} ({}) connected to {}.",
client_id.as_u32(),
params.protocol,
params.remote_addr
))]))
}
#[tool(
description = "Stop a client by its ID. Aborts every background task the client owns (read loop, transport driver, in-flight requests), which releases its socket and stops any further LLM calls, then drops it from NetGet's state so it disappears from list_clients and client_status. Note that a handful of client protocols do not yet register their tasks; for those this still only drops the bookkeeping — see src/mcp_stdio/CLAUDE.md for the current list."
)]
async fn stop_client(
&self,
Parameters(params): Parameters<StopClientParams>,
) -> Result<CallToolResult, McpError> {
let client_id = crate::state::ClientId::new(params.client_id);
match self.state.app_state.remove_client(client_id).await {
Some(client) => Ok(CallToolResult::success(vec![Content::text(format!(
"Client #{} ({} -> {}) stopped: its registered background tasks were \
aborted and it was removed from NetGet's state.",
params.client_id, client.protocol_name, client.remote_addr
))])),
None => Ok(CallToolResult::error(vec![Content::text(format!(
"Client #{} not found",
params.client_id
))])),
}
}
#[tool(
description = "List all clients with their protocol, remote address, status and instruction"
)]
async fn list_clients(&self) -> Result<CallToolResult, McpError> {
let clients = self.state.app_state.get_all_clients().await;
if clients.is_empty() {
return Ok(CallToolResult::success(vec![Content::text(
"No clients currently connected.",
)]));
}
let mut clients = clients;
clients.sort_by_key(|c| c.id.as_u32());
let mut result = String::from("## Clients\n\n");
for client in &clients {
result.push_str(&format!(
"- **Client #{}**: {} -> {} ({})\n Instruction: {}\n Memory: {}\n\n",
client.id.as_u32(),
client.protocol_name,
client.remote_addr,
client.status,
if client.instruction.is_empty() {
"(none)"
} else {
&client.instruction
},
if client.memory.is_empty() {
"(empty)"
} else {
&client.memory
},
));
}
Ok(CallToolResult::success(vec![Content::text(result)]))
}
#[tool(description = "Get detailed status of a specific client")]
async fn client_status(
&self,
Parameters(params): Parameters<ClientStatusParams>,
) -> Result<CallToolResult, McpError> {
let client_id = crate::state::ClientId::new(params.client_id);
match self.state.app_state.get_client(client_id).await {
Some(client) => {
let mut result = format!(
"## Client #{}\n\n\
- **Protocol**: {}\n\
- **Remote address**: {}\n\
- **Status**: {}\n\
- **Instruction**: {}\n\
- **Memory**: {}\n",
client.id.as_u32(),
client.protocol_name,
client.remote_addr,
client.status,
if client.instruction.is_empty() {
"(none)"
} else {
&client.instruction
},
if client.memory.is_empty() {
"(empty)"
} else {
&client.memory
},
);
if let Some(conn) = &client.connection {
result.push_str(&format!(
"- **Local address**: {}\n\
- **Connected address**: {}\n\
- **Bytes sent/received**: {}/{}\n\
- **Packets sent/received**: {}/{}\n",
conn.local_addr
.map(|a| a.to_string())
.unwrap_or_else(|| "(unknown)".to_string()),
conn.connected_addr
.map(|a| a.to_string())
.unwrap_or_else(|| "(unknown)".to_string()),
conn.bytes_sent,
conn.bytes_received,
conn.packets_sent,
conn.packets_received,
));
}
Ok(CallToolResult::success(vec![Content::text(result)]))
}
None => Ok(CallToolResult::error(vec![Content::text(format!(
"Client #{} not found",
params.client_id
))])),
}
}
#[tool(
description = "Put one action on the wire through a RUNNING client, now, without waiting for an event — the dashboard's [ send ]. `action` is one of the client protocol's actions as a JSON object ({\"type\":\"send_tcp_data\",\"data\":\"hello\\n\"}); its `type` is checked against the client's action set first (get_protocol_docs lists them with parameters and examples). The client's own connection loop executes it and the result says what happened: bytes sent, executed without writing, rejected by the protocol, or disconnected. Clients whose protocol has not adopted the command channel return an error immediately rather than hanging."
)]
async fn send_to_client(
&self,
Parameters(params): Parameters<SendToClientParams>,
) -> Result<CallToolResult, McpError> {
let client_id = crate::state::ClientId::new(params.client_id);
let Some(client) = self.state.app_state.get_client(client_id).await else {
return Ok(CallToolResult::error(vec![Content::text(format!(
"Client #{} not found",
params.client_id
))]));
};
let vocabulary = match crate::mcp_stdio::control::client_vocabulary(
&self.state.app_state,
&client.protocol_name,
) {
Ok(v) => v,
Err(e) => return Ok(CallToolResult::error(vec![Content::text(e)])),
};
if let Err(e) = crate::mcp_stdio::control::check_action_types(
std::slice::from_ref(¶ms.action),
&vocabulary,
&[],
) {
return Ok(CallToolResult::error(vec![Content::text(format!(
"Client #{} ({}): {}",
params.client_id, client.protocol_name, e
))]));
}
// Fail fast rather than waiting out the timeout on a client that has no channel.
if !self.state.app_state.has_client_handle(client_id).await {
return Ok(CallToolResult::error(vec![Content::text(format!(
"Client #{} ({}, {}) does not accept injected actions: it is not connected, or \
its protocol has not adopted the command channel. Its actions can only be \
produced by its instruction or event_handlers.",
params.client_id, client.protocol_name, client.status
))]));
}
let timeout = std::time::Duration::from_secs(
params
.timeout_secs
.unwrap_or(INJECT_TIMEOUT_DEFAULT_SECS)
.clamp(1, INJECT_TIMEOUT_MAX_SECS) as u64,
);
match self
.state
.app_state
.send_to_client(client_id, params.action, timeout)
.await
{
Ok(crate::state::client_handles::ClientSendOutcome::Rejected { error }) => {
Ok(CallToolResult::error(vec![Content::text(format!(
"Client #{}: the protocol rejected the action: {}",
params.client_id, error
))]))
}
Ok(outcome) => Ok(CallToolResult::success(vec![Content::text(format!(
"Client #{}: {}",
params.client_id,
crate::mcp_stdio::control::describe_outcome(&outcome)
))])),
Err(e) => {
// A client parks its events for a manual handler inside its own loop, so
// while one of its requests waits in list_intercepts nothing drains the
// command channel. Say so, or the failure reads as a dead client.
let waiting = self
.state
.app_state
.list_intercepts()
.await
.into_iter()
.filter(|v| {
v.owner == crate::state::intercepts::InterceptOwner::Client(client_id)
})
.map(|v| format!("#{}", v.id))
.collect::<Vec<_>>();
let hint = if waiting.is_empty() {
String::new()
} else {
format!(
" — its loop is parked on request {} waiting for an answer \
(list_intercepts / answer_intercept); it takes commands again once \
that is answered",
waiting.join(", ")
)
};
Ok(CallToolResult::error(vec![Content::text(format!(
"{e}{hint}"
))]))
}
}
}
#[tool(
description = "Put one action on the wire to ONE live peer of a running server, now, without waiting for the peer to speak — the dashboard's [ message ] on a peer row. `connection_id` comes from server_status, which also marks the connections that accept this (only protocols that register a peer handle do; tcp and telnet among them). `action` is one of the server protocol's actions ({\"type\":\"send_tcp_data\",\"data\":\"hi\\n\"}), checked against its action set first. Errors clearly when the protocol or connection has no peer handle, and with the protocol's own reason when it only answers requests (HTTP and the protocols on it)."
)]
async fn send_to_peer(
&self,
Parameters(params): Parameters<SendToPeerParams>,
) -> Result<CallToolResult, McpError> {
let server_id = crate::state::ServerId::new(params.server_id);
let Some(server) = self.state.app_state.get_server(server_id).await else {
return Ok(CallToolResult::error(vec![Content::text(format!(
"Server #{} not found",
params.server_id
))]));
};
if let Some(reason) =
crate::protocol::server_registry::registry().request_only_reason(&server.protocol_name)
{
return Ok(CallToolResult::error(vec![Content::text(format!(
"Server #{} ({}) cannot message a peer: {}. It speaks only in answer to a \
request — answer those through its event handlers, the model, or a `manual` \
rule and answer_intercept.",
params.server_id, server.protocol_name, reason
))]));
}
let vocabulary = match crate::mcp_stdio::control::peer_vocabulary(
&self.state.app_state,
&server.protocol_name,
) {
Ok(v) => v,
Err(e) => return Ok(CallToolResult::error(vec![Content::text(e)])),
};
if let Err(e) = crate::mcp_stdio::control::check_action_types(
std::slice::from_ref(¶ms.action),
&vocabulary,
&[],
) {
return Ok(CallToolResult::error(vec![Content::text(format!(
"Server #{} ({}): {}",
params.server_id, server.protocol_name, e
))]));
}
self.inject_into_peer(
server_id,
&server.protocol_name,
params.connection_id,
params.action,
params.timeout_secs,
)
.await
}
#[tool(
description = "Hang up on ONE live peer of a running server from the server's side — the dashboard's [ disconnect ]. Runs the protocol's own close_connection through the connection's peer handle (half-close, and the connection is marked closed at once). Only for protocols that register a peer handle and declare close_connection; errors clearly otherwise."
)]
async fn disconnect_peer(
&self,
Parameters(params): Parameters<DisconnectPeerParams>,
) -> Result<CallToolResult, McpError> {
let server_id = crate::state::ServerId::new(params.server_id);
let Some(server) = self.state.app_state.get_server(server_id).await else {
return Ok(CallToolResult::error(vec![Content::text(format!(
"Server #{} not found",
params.server_id
))]));
};
let declares_close = crate::mcp_stdio::control::peer_vocabulary(
&self.state.app_state,
&server.protocol_name,
)
.map(|v| v.iter().any(|a| a.name == "close_connection"))
.unwrap_or(false);
if !declares_close {
return Ok(CallToolResult::error(vec![Content::text(format!(
"Server #{} ({}) declares no close_connection action, so there is no \
server-side hang-up to run for one peer. stop_server closes every connection.",
params.server_id, server.protocol_name
))]));
}
self.inject_into_peer(
server_id,
&server.protocol_name,
params.connection_id,
serde_json::json!({"type": "close_connection"}),
None,
)
.await
}
#[tool(
description = "List every request parked for a human answer by a `manual` event handler ({\"event_pattern\":\"*\",\"handler\":{\"type\":\"manual\",\"timeout_secs\":300}}) — the dashboard's \"waiting for YOUR answer\" rows. For each: its id, the server or client and connection it came from, the event type and its data, the actions it may be answered with (name(params), * = required), and how many seconds remain before it fails closed on its own. Answer with answer_intercept, refuse with fail_intercept."
)]
async fn list_intercepts(&self) -> Result<CallToolResult, McpError> {
let views = self.state.app_state.list_intercepts().await;
if views.is_empty() {
return Ok(CallToolResult::success(vec![Content::text(
"No requests are waiting for an answer.",
)]));
}
let now = crate::state::intercepts::now_unix_ms();
let mut out = format!("## Requests waiting for an answer ({})\n\n", views.len());
for view in &views {
let (protocol, vocabulary) =
match crate::mcp_stdio::control::intercept_vocabulary(&self.state.app_state, view)
.await
{
Ok(pair) => pair,
Err(e) => (format!("unknown — {e}"), Vec::new()),
};
let owner = match view.owner {
crate::state::intercepts::InterceptOwner::Server(id) => {
format!("server #{} ({})", id.as_u32(), protocol)
}
crate::state::intercepts::InterceptOwner::Client(id) => {
format!("client #{} ({})", id.as_u32(), protocol)
}
};
let data = view
.event_data
.as_ref()
.map(|d| serde_json::to_string(d).unwrap_or_else(|_| d.to_string()))
.unwrap_or_else(|| "(none)".to_string());
let answers: Vec<String> = vocabulary
.iter()
.map(crate::mcp_stdio::control::action_signature)
.collect();
out.push_str(&format!(
"### intercept_id {}\n\
- **From**: {}{}\n\
- **Event**: `{}` — {}\n\
- **Event data**: `{}`\n\
- **Answer with**: {}\n\
- **Fails closed in**: {}s (of {}s)\n\n",
view.id,
owner,
view.connection_id
.map(|c| format!(", connection #{c}"))
.unwrap_or_default(),
view.event_type,
view.description,
data,
if answers.is_empty() {
"(no protocol actions — answer with [] or fail it)".to_string()
} else {
answers.join(", ")
},
view.seconds_until_fail_closed(now),
view.timeout_secs,
));
}
out.push_str(
"answer_intercept {intercept_id, actions:[…]} answers (an empty array = \
acknowledge and say nothing); fail_intercept {intercept_id} refuses now, and the \
peer gets the protocol's fail-closed reply. get_protocol_docs has each action's \
parameters and an example.",
);
Ok(CallToolResult::success(vec![Content::text(out)]))
}
#[tool(
description = "Answer a parked request (from list_intercepts) — the dashboard's [ Compose answer… ] / [ Answer with nothing ]. `actions` is a JSON array of action objects using only the actions list_intercepts offers for it; they run exactly as a static handler's would (same executor, same {{event.field}} interpolation). An EMPTY array is a real answer: acknowledge and say nothing — nothing is written, the connection carries on — which is distinct from letting it time out (that fails closed). Errors if the request is no longer waiting (answered, refused, timed out, or its connection closed)."
)]
async fn answer_intercept(
&self,
Parameters(params): Parameters<AnswerInterceptParams>,
) -> Result<CallToolResult, McpError> {
let Some(view) = self
.state
.app_state
.list_intercepts()
.await
.into_iter()
.find(|v| v.id == params.intercept_id)
else {
return Ok(CallToolResult::error(vec![Content::text(format!(
"Request #{} is not waiting (already answered, refused, timed out, or its \
connection closed). list_intercepts shows what is.",
params.intercept_id
))]));
};
let (protocol, vocabulary) =
match crate::mcp_stdio::control::intercept_vocabulary(&self.state.app_state, &view)
.await
{
Ok(pair) => pair,
Err(e) => return Ok(CallToolResult::error(vec![Content::text(e)])),
};
if let Err(e) = crate::mcp_stdio::control::check_action_types(
¶ms.actions,
&vocabulary,
crate::events::handler::common_action_names(),
) {
return Ok(CallToolResult::error(vec![Content::text(format!(
"Request #{} ({} `{}`) was NOT answered and is still waiting: {}",
params.intercept_id, protocol, view.event_type, e
))]));
}
let names: Vec<String> = params
.actions
.iter()
.filter_map(|a| a.get("type").and_then(|t| t.as_str()).map(str::to_string))
.collect();
match self
.state
.app_state
.resolve_intercept(params.intercept_id, params.actions)
.await
{
Ok(()) if names.is_empty() => {
Ok(CallToolResult::success(vec![Content::text(format!(
"Answered request #{} with nothing: acknowledged, no reply written, the \
connection carries on.",
params.intercept_id
))]))
}
Ok(()) => Ok(CallToolResult::success(vec![Content::text(format!(
"Answered request #{} with {}. The waiting connection executes them now; \
list_access_logs shows the outcome.",
params.intercept_id,
names.join(", ")
))])),
Err(e) => Ok(CallToolResult::error(vec![Content::text(e)])),
}
}
#[tool(
description = "Refuse a parked request (from list_intercepts) now instead of letting it wait out its timeout — the dashboard's [ Fail closed ]. The peer gets exactly what it would get if the model had failed: the protocol's own fail-closed reply (for tcp, a half-close; for http, an error status), never an invented success. There is no separate 'dismiss': refusing a parked request IS failing it closed."
)]
async fn fail_intercept(
&self,
Parameters(params): Parameters<FailInterceptParams>,
) -> Result<CallToolResult, McpError> {
if self
.state
.app_state
.dismiss_intercept(params.intercept_id)
.await
{
Ok(CallToolResult::success(vec![Content::text(format!(
"Refused request #{}: the peer gets the protocol's fail-closed reply.",
params.intercept_id
))]))
} else {
Ok(CallToolResult::error(vec![Content::text(format!(
"Request #{} is not waiting (already answered, refused or timed out).",
params.intercept_id
))]))
}
}
#[tool(
description = "Agent-LLM mode only (netget --llm-agent): fetch the next queued LLM request for a protocol server so YOU (the calling agent) can answer it in place of a model. Optionally long-poll with wait_seconds. Returns the request id, the prompt (server instruction + the triggering event), and the actions you may use. Reply by calling answer_llm_request with that id and a JSON array of actions. Returns '(no pending requests)' if none arrive within the wait."
)]
async fn get_next_llm_request(
&self,
Parameters(params): Parameters<GetNextLlmRequestParams>,
) -> Result<CallToolResult, McpError> {
let Some(queue) = self.state.agent_queue.as_ref() else {
return Ok(CallToolResult::error(vec![Content::text(
"Agent-LLM mode is not enabled. Start netget with --llm-agent to queue LLM requests for the agent.",
)]));
};
// Cap the long-poll to stay comfortably under typical MCP client tool-call timeouts.
let wait = params.wait_seconds.unwrap_or(0).min(120);
let req = queue
.wait_and_claim(std::time::Duration::from_secs(wait as u64))
.await;
let Some(req) = req else {
return Ok(CallToolResult::success(vec![Content::text(
"(no pending requests)",
)]));
};
// Render the prompt (system + user messages) and the available actions.
let mut prompt = String::new();
for m in &req.messages {
prompt.push_str(&format!("### {}\n{}\n\n", m.role, m.content));
}
let actions_json =
serde_json::to_string_pretty(&req.tools).unwrap_or_else(|_| "[]".to_string());
let now_ms = crate::utils::clock::SystemTime::now()
.duration_since(crate::utils::clock::UNIX_EPOCH)
.map(|d| d.as_millis() as u64)
.unwrap_or(0);
let age = now_ms.saturating_sub(req.created_unix_ms) / 1000;
let result = format!(
"## LLM request #{}\n\n\
- **Queued**: {}s ago\n\n\
### Prompt\n{}\n\
### Available actions (tool schemas)\n```json\n{}\n```\n\n\
To answer, call `answer_llm_request` with:\n\
```json\n{{ \"request_id\": {}, \"actions\": [ /* one or more action objects */ ] }}\n```\n\
Each action object needs a `type` field naming the action, plus its parameters. \
See `get_access_log` for worked examples of the action JSON a protocol expects.",
req.id, age, prompt, actions_json, req.id,
);
Ok(CallToolResult::success(vec![Content::text(result)]))
}
#[tool(
description = "Agent-LLM mode only: answer a queued LLM request (from get_next_llm_request). `actions` is a JSON array of NetGet action objects — each with a `type` field and its parameters — that the protocol server will execute (e.g. [{\"type\":\"send_tcp_data\",\"data\":\"...\"}]). This unblocks the waiting connection."
)]
async fn answer_llm_request(
&self,
Parameters(params): Parameters<AnswerLlmRequestParams>,
) -> Result<CallToolResult, McpError> {
let Some(queue) = self.state.agent_queue.as_ref() else {
return Ok(CallToolResult::error(vec![Content::text(
"Agent-LLM mode is not enabled. Start netget with --llm-agent.",
)]));
};
match queue.answer(params.request_id, params.actions) {
Ok(()) => Ok(CallToolResult::success(vec![Content::text(format!(
"Answered LLM request #{}.",
params.request_id
))])),
Err(e) => Ok(CallToolResult::error(vec![Content::text(e.to_string())])),
}
}
#[tool(
description = "Agent-LLM mode only: list outstanding queued LLM requests (pending and claimed-but-unanswered) for monitoring."
)]
async fn list_llm_requests(&self) -> Result<CallToolResult, McpError> {
let Some(queue) = self.state.agent_queue.as_ref() else {
return Ok(CallToolResult::error(vec![Content::text(
"Agent-LLM mode is not enabled. Start netget with --llm-agent.",
)]));
};
let pending = queue.list();
if pending.is_empty() {
return Ok(CallToolResult::success(vec![Content::text(
"No outstanding LLM requests.",
)]));
}
let now_ms = crate::utils::clock::SystemTime::now()
.duration_since(crate::utils::clock::UNIX_EPOCH)
.map(|d| d.as_millis() as u64)
.unwrap_or(0);
let mut out = String::from("## Outstanding LLM requests\n\n");
for r in &pending {
let age = now_ms.saturating_sub(r.created_unix_ms) / 1000;
out.push_str(&format!(
"- **#{}** — {} — queued {}s ago\n",
r.id,
if r.claimed { "claimed" } else { "pending" },
age
));
}
Ok(CallToolResult::success(vec![Content::text(out)]))
}
}
impl NetGetMcpService {
/// Shared body of `send_to_peer` and `disconnect_peer`: check the connection has a peer
/// handle (a clear error when the protocol or connection has none), then execute the
/// action inside that connection's own task and report the outcome.
async fn inject_into_peer(
&self,
server_id: crate::state::ServerId,
protocol_name: &str,
connection_id: u32,
action: serde_json::Value,
timeout_secs: Option<u32>,
) -> Result<CallToolResult, McpError> {
if !self
.state
.app_state
.has_peer_handle(server_id, connection_id)
.await
{
return Ok(CallToolResult::error(vec![Content::text(format!(
"Connection #{} on server #{} ({}) does not accept injected actions: either \
it is not a live connection of that server (server_status lists them), or \
the {} protocol does not register a peer handle for its connections.",
connection_id,
server_id.as_u32(),
protocol_name,
protocol_name
))]));
}
let timeout = std::time::Duration::from_secs(
timeout_secs
.unwrap_or(INJECT_TIMEOUT_DEFAULT_SECS)
.clamp(1, INJECT_TIMEOUT_MAX_SECS) as u64,
);
match self
.state
.app_state
.send_to_peer(server_id, connection_id, action, timeout)
.await
{
Ok(crate::state::client_handles::ClientSendOutcome::Rejected { error }) => {
Ok(CallToolResult::error(vec![Content::text(format!(
"Connection #{} on server #{}: the protocol rejected the action: {}",
connection_id,
server_id.as_u32(),
error
))]))
}
Ok(outcome) => Ok(CallToolResult::success(vec![Content::text(format!(
"Connection #{} on server #{}: {}",
connection_id,
server_id.as_u32(),
crate::mcp_stdio::control::describe_outcome(&outcome)
))])),
Err(e) => Ok(CallToolResult::error(vec![Content::text(e.to_string())])),
}
}
/// The `AppState` this service drives.
///
/// `SharedState` is crate-private and the `#[tool]` methods are private to the
/// router, so an integration test can only observe a tool's *text* result. That
/// is not enough to assert teardown: `stop_server` reports "stopped" whether or
/// not the server's scheduled tasks went with it. This accessor is what lets
/// `tests/mcp_stop_cleanup_test.rs` check the state behind the tool surface.
///
/// Handing out a clone is safe — `AppState` is an `Arc` over the shared lock, so
/// this is the same state the tools mutate, not a snapshot.
pub fn app_state(&self) -> AppState {
self.state.app_state.clone()
}
}
#[tool_handler]
impl ServerHandler for NetGetMcpService {
fn get_info(&self) -> <RoleServer as rmcp::service::ServiceRole>::Info {
InitializeResult::new(ServerCapabilities::builder().enable_tools().build())
.with_server_info(Implementation::new("netget", env!("CARGO_PKG_VERSION")))
.with_instructions(
"NetGet - LLM-controlled network protocol server. \
Use list_protocols to discover available protocols, then start_server to \
create one. \
IMPORTANT — choose the handling mode deliberately: for deterministic \
behavior (echo servers, fixed/canned responses, simple routing, high \
throughput) pass `event_handlers` with a script or static handler, which \
run in-process with NO LLM call (instant, free, reproducible). Reserve the \
natural-language `instruction` (the LLM fallback, one model call per \
request) for responses that genuinely need reasoning or vary \
unpredictably. Use get_protocol_docs to see a protocol's event ids and \
actions before writing a handler.\n\n\
CLIENTS: start_client / list_clients / client_status / stop_client are \
the mirror of the server tools, for connecting OUT to a remote server \
instead of listening. They take the same instruction and event_handlers \
arguments plus a remote_addr. stop_client aborts the client's background \
tasks, releasing its socket and stopping further LLM calls.\n\n\
DRIVING BY HAND: send_to_client puts one of a client's actions on the wire \
now; send_to_peer / disconnect_peer do the same for one live connection of a \
server (server_status lists connection ids). A `manual` event handler parks \
each matching event for YOU: list_intercepts shows them, answer_intercept \
answers (an empty array acknowledges without replying), fail_intercept \
refuses with the protocol's fail-closed reply.\n\n\
AGENT-LLM MODE (when netget was started with --llm-agent): there is no \
model — YOU answer the LLM calls. When a server needs a reasoned \
response it queues a request; fetch it with get_next_llm_request \
(optionally long-poll via wait_seconds), then reply with \
answer_llm_request(request_id, actions). If --llm-agent-pipe was set, a \
line with the new request id is written to that FIFO on each enqueue so \
you can block-read it instead of polling. Unanswered requests error out \
after the configured timeout. list_llm_requests shows what is outstanding.",
)
}
fn initialize(
&self,
request: InitializeRequestParams,
context: RequestContext<RoleServer>,
) -> impl std::future::Future<Output = Result<InitializeResult, McpError>> + Send + '_ {
async move {
info!(
"MCP client '{}' v{} initialized",
request.client_info.name, request.client_info.version,
);
// Retain peer info for the connection so rmcp can answer later requests
if context.peer.peer_info().is_none() {
context.peer.set_peer_info(request);
}
Ok(self.get_info())
}
}
}
/// Parse the MCP `scheduled_tasks` argument into typed task definitions.
///
/// Kept as raw JSON on the wire because `ServerTaskDefinition` does not derive
/// `JsonSchema`; a bad entry is reported to the caller rather than silently
/// dropped.
fn parse_scheduled_tasks(
tasks: Option<Vec<serde_json::Value>>,
) -> Result<Option<Vec<crate::llm::actions::common::ServerTaskDefinition>>, String> {
let Some(tasks) = tasks else {
return Ok(None);
};
if tasks.is_empty() {
return Ok(None);
}
let mut parsed = Vec::with_capacity(tasks.len());
for (i, task) in tasks.into_iter().enumerate() {
match serde_json::from_value::<crate::llm::actions::common::ServerTaskDefinition>(task) {
Ok(def) => parsed.push(def),
Err(e) => {
return Err(format!(
"entry {}: {}. Each entry needs \"task_id\" (string), \"recurring\" (boolean) \
and \"instruction\" (string); recurring tasks also take \"interval_secs\", \
one-shot tasks \"delay_secs\".",
i, e
));
}
}
}
Ok(Some(parsed))
}
/// Build a short one-line summary of a request (event data) for the log list.
fn summarize_request(request: &serde_json::Value) -> String {
// HTTP-style requests: method + path
let method = request.get("method").and_then(|v| v.as_str());
let path = request.get("path").and_then(|v| v.as_str());
if let (Some(m), Some(p)) = (method, path) {
return format!("{} {}", m, p);
}
if let Some(p) = path {
return p.to_string();
}
// Fall back to a compact, truncated form of the request JSON
let compact = request.to_string();
crate::utils::truncate_with_suffix(&compact, 79, "…")
}
/// Build a short one-line summary of the response action array for the log list.
fn summarize_response(response: &[serde_json::Value]) -> String {
if response.is_empty() {
return "(no response)".to_string();
}
let parts: Vec<String> = response
.iter()
.map(|a| {
let ty = a.get("type").and_then(|v| v.as_str()).unwrap_or("action");
// Surface an HTTP status code if present
match a
.get("status_code")
.or_else(|| a.get("status"))
.and_then(|v| v.as_u64())
{
Some(code) => format!("{}({})", ty, code),
None => ty.to_string(),
}
})
.collect();
parts.join(", ")
}