mod alias;
mod bench;
mod config;
mod editor;
mod elicit;
mod find;
mod sampling;
mod session;
mod style;
mod subscribe;
mod vars;
mod wire;
use std::collections::HashMap;
use std::future::Future;
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::{Arc, RwLock};
use std::time::Duration;
use clap::{Parser, ValueEnum};
use nu_ansi_term::{Color, Style};
use tower_mcp::client::{
ChannelTransport, HttpClientConfig, HttpClientTransport, McpClient, McpClientBuilder,
NotificationHandler, StdioClientTransport,
};
use tower_mcp::protocol::{
Content, DiscoverResult, Implementation, InitializeResult, LogLevel, PromptDefinition,
ResourceDefinition, ResourceTemplateDefinition, ServerCapabilities, TaskObject, ToolDefinition,
};
use tower_mcp::{ProtocolSupport, ProtocolSupportError};
use alias::Aliases;
use elicit::ReplClientHandler;
use session::{Connector, Session, is_not_initialized, is_session_lost};
use style::{json_pretty, paint, tag, task_status_style};
use wire::{TracingTransport, wire};
#[derive(Clone, Copy, Debug, Default, PartialEq, Eq, ValueEnum)]
enum ProtocolMode {
#[default]
Stable,
#[value(name = "2026-07-28", alias = "final")]
Final,
}
impl ProtocolMode {
fn support(self) -> Result<ProtocolSupport, ProtocolSupportError> {
match self {
Self::Stable => Ok(ProtocolSupport::stable()),
Self::Final => ProtocolSupport::try_new(["2026-07-28"]),
}
}
}
#[derive(Parser)]
#[command(
name = "mcp-repl",
about = "Interactive MCP client REPL",
trailing_var_arg = true
)]
struct Args {
#[arg(long, value_enum, default_value = "stable")]
protocol: ProtocolMode,
#[arg(long)]
http: Option<String>,
#[arg(long, conflicts_with_all = ["http", "command", "server"])]
demo: bool,
#[arg(long, value_name = "NAME")]
server: Option<String>,
#[arg(long, value_name = "PATH")]
config: Option<String>,
#[arg(long)]
list_servers: bool,
#[arg(long, value_enum, default_value = "auto")]
color: style::ColorMode,
#[arg(long)]
bearer: Option<String>,
#[arg(long = "header", value_name = "NAME: VALUE")]
headers: Vec<String>,
#[arg(short = 'e', long = "exec", value_name = "COMMAND")]
exec: Vec<String>,
#[arg(long)]
json: bool,
#[arg(long)]
verbose: bool,
#[arg(long, value_enum, value_name = "STRATEGY")]
sampling: Option<sampling::SamplingMode>,
#[arg(long)]
no_history: bool,
#[arg(long)]
no_reconnect: bool,
#[arg(long)]
trace: bool,
command: Vec<String>,
}
static JSON_OUTPUT: AtomicBool = AtomicBool::new(false);
static HAD_ERROR: AtomicBool = AtomicBool::new(false);
fn json_output() -> bool {
JSON_OUTPUT.load(Ordering::Relaxed)
}
fn note_error() {
HAD_ERROR.store(true, Ordering::Relaxed);
}
fn error_json(message: &str) -> String {
serde_json::json!({ "error": message }).to_string()
}
#[derive(Default)]
pub struct Surface {
pub tools: Vec<ToolDefinition>,
pub prompts: Vec<PromptDefinition>,
pub resources: Vec<ResourceDefinition>,
pub templates: Vec<ResourceTemplateDefinition>,
}
pub const BUILTINS: &[(&str, &str)] = &[
("help", "list built-ins and the server's tools"),
("tools", "list tools"),
("prompts", "list prompts"),
("resources", "list resources"),
("templates", "list resource templates"),
("find", "search the surface by keyword"),
("describe", "show schemas and metadata for a name"),
("read", "read a resource"),
("subscribe", "watch a resource for updates"),
("unsubscribe", "stop watching a resource"),
("subscriptions", "list active resource subscriptions"),
("prompt", "get a prompt"),
("call", "call a tool with raw JSON"),
("bench", "time repeated calls to a tool"),
("jobs", "list background tasks"),
("task", "show a background task"),
("wait", "wait for a background task"),
("cancel", "cancel a background task"),
("alias", "define, list, or show a command alias"),
("unalias", "remove a command alias"),
("refresh", "re-fetch the server surface"),
("info", "replay the connection banner plus capabilities"),
("wire", "toggle raw JSON-RPC frame tracing (on|off)"),
("last", "reprint the previous request and response"),
("vars", "list captured variables"),
("unset", "clear a captured variable"),
("quit", "exit"),
("exit", "exit"),
];
fn coerce_arg(schema: &serde_json::Value, key: &str, raw: &str) -> serde_json::Value {
let ty = schema
.get("properties")
.and_then(|p| p.get(key))
.and_then(|s| s.get("type"))
.and_then(|t| t.as_str());
match ty {
Some("integer") => raw
.parse::<i64>()
.map(Into::into)
.unwrap_or_else(|_| serde_json::Value::String(raw.to_string())),
Some("number") => raw
.parse::<f64>()
.ok()
.and_then(|n| serde_json::Number::from_f64(n).map(serde_json::Value::Number))
.unwrap_or_else(|| serde_json::Value::String(raw.to_string())),
Some("boolean") => raw
.parse::<bool>()
.map(serde_json::Value::Bool)
.unwrap_or_else(|_| serde_json::Value::String(raw.to_string())),
Some("array") | Some("object") => {
serde_json::from_str(raw).unwrap_or_else(|_| serde_json::Value::String(raw.to_string()))
}
_ => {
serde_json::from_str(raw).unwrap_or_else(|_| serde_json::Value::String(raw.to_string()))
}
}
}
fn parse_kv_args(schema: &serde_json::Value, tokens: &[&str]) -> serde_json::Value {
if tokens.len() == 1
&& tokens[0].starts_with('{')
&& let Ok(v) = serde_json::from_str::<serde_json::Value>(tokens[0])
{
return v;
}
let mut map = serde_json::Map::new();
for t in tokens {
if let Some((k, v)) = t.split_once('=') {
map.insert(k.to_string(), coerce_arg(schema, k, v));
}
}
serde_json::Value::Object(map)
}
fn render_content(content: &[Content]) {
for c in content {
match c {
Content::Text { text, .. } => {
if style::colors_enabled() && style::looks_like_markdown(text) {
println!("{}", style::render_markdown(text));
} else {
println!("{text}");
}
}
other => {
let v = serde_json::to_value(other).unwrap_or_default();
let ty = v.get("type").and_then(|t| t.as_str()).unwrap_or("content");
match ty {
"image" | "audio" => {
let mime = v.get("mimeType").and_then(|m| m.as_str()).unwrap_or("?");
let len = v.get("data").and_then(|d| d.as_str()).map_or(0, str::len);
println!(
"{}",
tag(Style::new(), &format!("{ty} {mime}, {len} base64 chars"))
);
}
_ => println!("{}", json_pretty(&v)),
}
}
}
}
}
fn render_task(task: &TaskObject) {
println!(
"task {} status={} {}",
paint(Style::new().bold(), &task.task_id),
paint(task_status_style(task.status), &task.status.to_string()),
task.status_message.as_deref().unwrap_or("")
);
if let Some(result) = &task.result {
render_content(&result.content);
}
if let Some(err) = &task.error {
println!("{} {}: {}", style::error_prefix(), err.code, err.message);
}
}
#[derive(Clone, Debug)]
struct ConnectionInfo {
protocol_version: String,
capabilities: ServerCapabilities,
server_info: Implementation,
instructions: Option<String>,
}
impl From<InitializeResult> for ConnectionInfo {
fn from(info: InitializeResult) -> Self {
Self {
protocol_version: info.protocol_version,
capabilities: info.capabilities,
server_info: info.server_info,
instructions: info.instructions,
}
}
}
impl ConnectionInfo {
fn from_discovery(discovery: DiscoverResult, protocol_version: String) -> Self {
let server_info = discovery
.meta
.as_ref()
.and_then(|meta| meta.server_info.clone())
.unwrap_or_else(|| Implementation {
name: "MCP server".to_string(),
version: "unknown".to_string(),
..Default::default()
});
Self {
protocol_version,
capabilities: discovery.capabilities,
server_info,
instructions: discovery.instructions,
}
}
}
async fn connection_info(client: &McpClient) -> Option<ConnectionInfo> {
if let Some(info) = client.server_info().await {
return Some(info.into());
}
let discovery = client.discovery().await?;
let protocol_version = client.selected_protocol_version().await?;
Some(ConnectionInfo::from_discovery(discovery, protocol_version))
}
async fn establish_connection(
client: &McpClient,
protocol: ProtocolMode,
) -> tower_mcp::Result<ConnectionInfo> {
match protocol {
ProtocolMode::Stable => client
.initialize("mcp-repl", env!("CARGO_PKG_VERSION"))
.await
.map(Into::into),
ProtocolMode::Final => {
let discovery: DiscoverResult = client
.discover("mcp-repl", env!("CARGO_PKG_VERSION"))
.await?;
let protocol_version = client
.selected_protocol_version()
.await
.unwrap_or_else(|| "2026-07-28".to_string());
Ok(ConnectionInfo::from_discovery(discovery, protocol_version))
}
}
}
fn client_builder(protocol: ProtocolMode) -> Result<McpClientBuilder, ProtocolSupportError> {
Ok(McpClient::builder()
.protocol_support(protocol.support()?)
.with_elicitation()
.with_sampling())
}
fn print_banner(info: &ConnectionInfo) {
println!(
"connected: {} v{} {}",
paint(Style::new().bold(), &info.server_info.name),
info.server_info.version,
paint(
Style::new().dimmed(),
&format!("(protocol {})", info.protocol_version)
)
);
if let Some(instructions) = &info.instructions {
if style::colors_enabled() && style::looks_like_markdown(instructions) {
println!("{}", style::render_markdown(instructions));
} else {
println!("{instructions}");
}
}
}
pub fn timing(elapsed: Duration) -> String {
let body = if elapsed.as_millis() < 1000 {
format!("[{}ms]", elapsed.as_millis())
} else {
format!("[{:.2}s]", elapsed.as_secs_f64())
};
paint(Style::new().dimmed(), &body)
}
fn print_tool_overview(surface: &Surface) {
const CAP: usize = 30;
if surface.tools.is_empty() {
return;
}
for t in surface.tools.iter().take(CAP) {
println!(
"{:24} {}",
paint(Style::new().fg(Color::Green), &t.name),
t.description.as_deref().unwrap_or("")
);
}
if surface.tools.len() > CAP {
println!(
"{}",
paint(
Style::new().dimmed(),
&format!("... +{} more, type `tools`", surface.tools.len() - CAP)
)
);
}
}
fn print_find(surface: &Surface, query: &str) {
let hits = find::search(surface, query);
if json_output() {
let v: Vec<serde_json::Value> = hits
.iter()
.map(|h| {
serde_json::json!({
"kind": h.kind.heading(),
"name": h.name,
"description": h.description,
"score": h.score,
})
})
.collect();
println!("{}", json_pretty(&serde_json::Value::Array(v)));
return;
}
if hits.is_empty() {
note_error();
println!("no match for {}", paint(Style::new().fg(Color::Red), query));
return;
}
let total = hits.len();
for (kind, group) in find::grouped(hits) {
println!("{}:", paint(Style::new().bold(), kind.heading()));
for hit in group {
println!(
" {:24} {}",
paint(Style::new().fg(Color::Green), &hit.name),
hit.description
);
}
}
println!(
"{}",
paint(
Style::new().dimmed(),
&format!("{total} match{}", if total == 1 { "" } else { "es" })
)
);
}
fn print_counts(surface: &Surface) {
println!(
"{} tools, {} prompts, {} resources, {} templates. Type `help`.",
surface.tools.len(),
surface.prompts.len(),
surface.resources.len(),
surface.templates.len()
);
}
async fn with_reconnect<T, F, Fut>(
session: &Session,
surface: &Arc<RwLock<Surface>>,
op: F,
) -> Result<T, tower_mcp::Error>
where
F: Fn(Arc<McpClient>) -> Fut,
Fut: Future<Output = Result<T, tower_mcp::Error>>,
{
let seen = session.generation();
let err = match op(session.client()).await {
Ok(value) => return Ok(value),
Err(e) => e,
};
if !session.can_reconnect() || !is_session_lost(&err) {
return Err(err);
}
if let Err(reconnect_err) = session.reconnect(seen).await {
eprintln!("reconnect failed: {reconnect_err}");
return Err(err);
}
*surface.write().unwrap() = fetch_surface(&session.client()).await;
eprintln!("{}", paint(Style::new().dimmed(), "[reconnected]"));
let retried = op(session.client()).await;
if let Err(e) = &retried
&& is_session_lost(e)
{
eprintln!(
"still no session after reconnecting. The server is likely down or \
restart-looping; check its logs, or pass --no-reconnect to see the \
raw errors."
);
}
retried
}
async fn fetch_surface_once(client: &McpClient) -> (Surface, bool) {
fn take<T>(
what: &str,
r: Result<Vec<T>, tower_mcp::Error>,
not_initialized: &mut bool,
) -> Vec<T> {
match r {
Ok(v) => v,
Err(e) => {
if is_not_initialized(&e) {
*not_initialized = true;
} else {
eprintln!("warning: fetching {what} failed: {e}");
}
Vec::new()
}
}
}
let (tools, prompts, resources, templates) = tokio::join!(
client.list_all_tools(),
client.list_all_prompts(),
client.list_all_resources(),
client.list_all_resource_templates(),
);
let mut ni = false;
let surface = Surface {
tools: take("tools", tools, &mut ni),
prompts: take("prompts", prompts, &mut ni),
resources: take("resources", resources, &mut ni),
templates: take("resource templates", templates, &mut ni),
};
(surface, ni)
}
async fn fetch_surface(client: &McpClient) -> Surface {
fetch_surface_once(client).await.0
}
async fn refresh_surface(session: &Session) -> Surface {
let (fresh, not_initialized) = fetch_surface_once(&session.client()).await;
if !not_initialized || !session.can_reconnect() {
return fresh;
}
let seen = session.generation();
match session.reconnect(seen).await {
Ok(()) => {
eprintln!("{}", paint(Style::new().dimmed(), "[reconnected]"));
fetch_surface(&session.client()).await
}
Err(e) => {
eprintln!("reconnect failed: {e}");
fresh
}
}
}
async fn fetch_surface_initial(client: &McpClient) -> Surface {
const ATTEMPTS: usize = 4;
for attempt in 1..=ATTEMPTS {
let (surface, not_initialized) = fetch_surface_once(client).await;
if !not_initialized {
return surface;
}
if attempt == ATTEMPTS {
eprintln!(
"warning: the server kept rejecting surface requests as not-initialized \
after {ATTEMPTS} attempts. The session the handshake established is not \
being recognized on follow-up requests. Two common causes: the server runs \
multiple instances without a shared session store, so requests scatter \
across instances; or a single instance restarted (crash, OOM, or redeploy) \
between requests and lost its in-memory sessions. Try `refresh`. A \
persistent session store or the stateless protocol avoids both; if it is a \
single instance, check its logs and resources (an OOM-looping machine \
flaps like this)."
);
return surface;
}
tokio::time::sleep(Duration::from_millis(200 * attempt as u64)).await;
}
unreachable!()
}
fn build_http_config(
bearer: Option<String>,
headers: &[String],
profile_bearer: Option<String>,
profile_headers: &[(String, String)],
) -> Result<HttpClientConfig, String> {
let mut config = HttpClientConfig::default();
for (name, value) in profile_headers {
config = config.header(name.as_str(), value.as_str());
}
if let Some(token) = bearer
.or(profile_bearer)
.or_else(|| std::env::var("MCP_BEARER").ok())
{
config = config.bearer_token(token);
}
for raw in headers {
let (name, value) = raw
.split_once(':')
.ok_or_else(|| format!("invalid --header {raw:?}: expected `Name: Value`"))?;
config = config.header(name.trim(), value.trim());
}
Ok(config)
}
fn demo_router() -> tower_mcp::McpRouter {
use tower_mcp::extract::RawArgs;
use tower_mcp::protocol::{CompleteResult, CompletionReference, ReadResourceResult};
use tower_mcp::resource::ResourceTemplateBuilder;
use tower_mcp::{CallToolResult, PromptBuilder, TaskSupportMode, ToolBuilder};
const NOTES: &[(&str, &str)] = &[
("groceries", "- eggs\n- coffee"),
("ideas", "# Ideas\n\n- a REPL for MCP servers"),
("todo", "1. ship it"),
];
tower_mcp::McpRouter::new()
.server_info("mcp-repl-demo", env!("CARGO_PKG_VERSION"))
.prompt(
PromptBuilder::new("greet")
.description("Generate a greeting (name tab-completes via the server)")
.required_arg("name", "The person to greet")
.handler(|args| async move {
let name = args.get("name").map(|s| s.as_str()).unwrap_or("World");
Ok(tower_mcp::GetPromptResult::user_message(format!(
"Please greet {name} warmly."
)))
})
.build(),
)
.resource(
tower_mcp::resource::ResourceBuilder::new("note://status")
.name("Status")
.description("A one-line status note (subscribe to it)")
.mime_type("text/plain")
.handler(|| async {
Ok(ReadResourceResult::text(
"note://status",
"all quiet on the demo server",
))
})
.build(),
)
.resource_template(
ResourceTemplateBuilder::new("note://{name}")
.name("Notes")
.description("Tiny in-memory notes (name tab-completes via the server)")
.mime_type("text/markdown")
.handler(
|uri: String, vars: std::collections::HashMap<String, String>| async move {
let name = vars.get("name").cloned().unwrap_or_default();
let text = NOTES
.iter()
.find(|(n, _)| *n == name)
.map(|(_, t)| (*t).to_string())
.unwrap_or_else(|| format!("no note named `{name}`"));
Ok(ReadResourceResult::text(uri, text))
},
),
)
.completion_handler(|params| async move {
let partial = params.argument.value;
let candidates: Vec<String> = match ¶ms.reference {
CompletionReference::Prompt { name } if name == "greet" => {
["Ada", "Alan", "Grace", "Linus"]
.iter()
.map(|s| s.to_string())
.collect()
}
CompletionReference::Resource { uri } if uri == "note://{name}" => {
NOTES.iter().map(|(n, _)| n.to_string()).collect()
}
_ => Vec::new(),
};
Ok(CompleteResult::new(
candidates
.into_iter()
.filter(|c| c.starts_with(&partial))
.collect::<Vec<_>>(),
))
})
.tool(
ToolBuilder::new("echo")
.description("Echo a message back")
.extractor_handler((), |RawArgs(args): RawArgs| async move {
let msg = args.get("message").and_then(|v| v.as_str()).unwrap_or("");
Ok(CallToolResult::text(msg.to_string()))
})
.build(),
)
.tool(
ToolBuilder::new("about")
.description("Markdown-formatted notes about this demo server")
.extractor_handler((), |RawArgs(_): RawArgs| async move {
Ok(CallToolResult::text(
"# mcp-repl demo\n\n\
A tiny in-process router for exploring the REPL.\n\n\
- `echo message=hi` echoes back\n\
- `slow_add a=2 b=3 &` runs **task-augmented**\n\
- `describe slow_add` shows the tool's schemas\n",
))
})
.build(),
)
.tool(
ToolBuilder::new("slow_add")
.description("Add two numbers, slowly (try running with a trailing &)")
.task_support(TaskSupportMode::Optional)
.extractor_handler((), |RawArgs(args): RawArgs| async move {
let a = args.get("a").and_then(|v| v.as_i64()).unwrap_or(0);
let b = args.get("b").and_then(|v| v.as_i64()).unwrap_or(0);
tokio::time::sleep(Duration::from_secs(3)).await;
Ok(CallToolResult::text(format!("{}", a + b)))
})
.build(),
)
}
fn notification_handler(refresh_tx: tokio::sync::mpsc::UnboundedSender<()>) -> NotificationHandler {
let t = refresh_tx.clone();
let r = refresh_tx.clone();
let p = refresh_tx;
NotificationHandler::new()
.on_tools_changed(move || {
let _ = t.send(());
})
.on_resources_changed(move || {
let _ = r.send(());
})
.on_prompts_changed(move || {
let _ = p.send(());
})
.on_progress(|p| {
let pct = match (p.progress, p.total) {
(done, Some(total)) if total > 0.0 => {
format!(" {:.0}%", 100.0 * done / total)
}
_ => String::new(),
};
println!(
"{} {}",
tag(Style::new().fg(Color::Cyan), &format!("progress{pct}")),
p.message.as_deref().unwrap_or("")
);
})
.on_resource_updated(|uri| {
let known = if subscribe::contains(&uri) {
String::new()
} else {
format!(" {}", paint(Style::new().dimmed(), "(not subscribed here)"))
};
println!(
"{} {uri}{known}",
tag(Style::new().fg(Color::Cyan), "resource updated")
);
})
.on_log_message(|m| {
println!(
"{} {}",
tag(log_level_style(m.level), &format!("log {}", m.level)),
m.data
);
})
}
fn http_connector(
url: String,
config: HttpClientConfig,
make_handler: Arc<dyn Fn() -> ReplClientHandler + Send + Sync>,
protocol: ProtocolMode,
) -> Connector {
Box::new(move || {
let (url, config, handler) = (url.clone(), config.clone(), make_handler());
Box::pin(async move {
let client = client_builder(protocol)
.map_err(|error| tower_mcp::Error::Transport(error.to_string()))?
.connect(
TracingTransport::new(HttpClientTransport::with_config(url, config)),
handler,
)
.await?;
establish_connection(&client, protocol).await?;
Ok(client)
})
})
}
fn load_config(explicit: Option<&str>) -> config::Config {
let Some((path, explicit)) = config::config_path(explicit) else {
return config::Config::default();
};
match config::Config::load(&path, explicit) {
Ok(c) => c,
Err(e) => {
eprintln!("error: {e}");
std::process::exit(2);
}
}
}
fn print_servers(config: &config::Config) {
if config.servers.is_empty() {
println!("no server profiles configured");
return;
}
let width = config.names().iter().map(|n| n.len()).max().unwrap_or(0);
for (name, profile) in &config.servers {
println!(
"{:width$} {}",
paint(Style::new().fg(Color::Cyan), name),
paint(Style::new().dimmed(), &profile.summary()),
);
}
}
fn resolve_profile(args: &Args, config: &config::Config) -> Option<(String, config::Connection)> {
let name = args
.server
.clone()
.or_else(|| match args.command.as_slice() {
[only] if config.servers.contains_key(only) => Some(only.clone()),
_ => None,
})?;
let profile = match config.profile(&name) {
Ok(p) => p,
Err(e) => {
eprintln!("error: {e}");
std::process::exit(2);
}
};
if profile.bearer.is_some() {
eprintln!(
"warning: profile {name:?} stores a literal `bearer` token; prefer \
`bearer_env = \"VAR\"` so the token is not kept in the config file"
);
}
match profile.resolve_with(|var| std::env::var(var).ok()) {
Ok(connection) => Some((name, connection)),
Err(e) => {
eprintln!("error: server profile {name:?}: {e}");
std::process::exit(2);
}
}
}
fn log_level_style(level: LogLevel) -> Style {
match level {
LogLevel::Emergency | LogLevel::Alert | LogLevel::Critical | LogLevel::Error => {
Style::new().fg(Color::Red)
}
LogLevel::Warning => Style::new().fg(Color::Yellow),
LogLevel::Notice | LogLevel::Info => Style::new().fg(Color::Green),
_ => Style::new().dimmed(),
}
}
#[tokio::main]
async fn main() -> Result<(), tower_mcp::BoxError> {
tracing_subscriber::fmt()
.with_env_filter(
tracing_subscriber::EnvFilter::try_from_default_env().unwrap_or_else(|_| "warn".into()),
)
.init();
let args = Args::parse();
style::init(args.color);
wire::init(args.trace);
JSON_OUTPUT.store(args.json, Ordering::Relaxed);
let config_file = config::config_path(args.config.as_deref()).map(|(path, _)| path);
let profiles = load_config(args.config.as_deref());
if args.list_servers {
print_servers(&profiles);
return Ok(());
}
let profile = resolve_profile(&args, &profiles);
let one_shot = !args.exec.is_empty();
let quiet = one_shot && !args.verbose;
let at_prompt = Arc::new(AtomicBool::new(false));
let (refresh_tx, mut refresh_rx) = tokio::sync::mpsc::unbounded_channel::<()>();
let make_handler: Arc<dyn Fn() -> ReplClientHandler + Send + Sync> = {
let refresh_tx = refresh_tx.clone();
let at_prompt = at_prompt.clone();
Arc::new(move || {
ReplClientHandler::new(notification_handler(refresh_tx.clone()), at_prompt.clone())
})
};
drop(refresh_tx);
sampling::init(sampling::resolve(args.sampling, one_shot));
let (profile_name, connection) = match profile {
Some((name, c)) => (Some(name), Some(c)),
None => (None, None),
};
let aliases = Arc::new(RwLock::new(Aliases::new(
profiles.aliases.clone(),
profile_name
.as_ref()
.and_then(|name| profiles.servers.get(name))
.map(|p| p.aliases.clone())
.unwrap_or_default(),
profile_name.clone(),
config_file,
)));
let connection = match (args.http.clone(), connection) {
(
Some(url),
Some(config::Connection::Http {
bearer, headers, ..
}),
) => Some(config::Connection::Http {
url,
bearer,
headers,
}),
(Some(url), _) => Some(config::Connection::Http {
url,
bearer: None,
headers: Vec::new(),
}),
(None, Some(c)) => Some(c),
(None, None) if !args.command.is_empty() => Some(config::Connection::Stdio {
command: args.command.clone(),
}),
(None, None) => None,
};
let over_http = matches!(connection, Some(config::Connection::Http { .. }));
if !over_http && (args.bearer.is_some() || !args.headers.is_empty()) {
eprintln!("warning: --bearer/--header apply only to HTTP servers; ignoring them here");
}
if let Some(name) = &profile_name
&& !quiet
{
println!(
"{}",
tag(Style::new().fg(Color::Cyan), &format!("profile {name}"))
);
}
let builder = client_builder(args.protocol)?;
let mut connector: Option<Connector> = None;
let client = if args.demo {
builder
.connect(
TracingTransport::new(ChannelTransport::new(demo_router())),
make_handler(),
)
.await?
} else {
match connection {
Some(config::Connection::Http {
url,
bearer,
headers,
}) => {
let config =
build_http_config(args.bearer.clone(), &args.headers, bearer, &headers)?;
if !args.no_reconnect {
connector = Some(http_connector(
url.clone(),
config.clone(),
make_handler.clone(),
args.protocol,
));
}
builder
.connect(
TracingTransport::new(HttpClientTransport::with_config(url, config)),
make_handler(),
)
.await?
}
Some(config::Connection::Stdio { command }) => {
let cmd_args: Vec<&str> = command[1..].iter().map(|s| s.as_str()).collect();
let transport = StdioClientTransport::spawn(&command[0], &cmd_args).await?;
builder
.connect(TracingTransport::new(transport), make_handler())
.await?
}
None => {
eprintln!(
"usage: mcp-repl <server command...> | --http <url> | --server <name> | --demo"
);
std::process::exit(2);
}
}
};
let info = establish_connection(&client, args.protocol).await?;
let server_name = info.server_info.name.clone();
if !quiet {
print_banner(&info);
}
let session = Arc::new(Session::new(client, connector));
let client = session.client();
let surface = Arc::new(RwLock::new(fetch_surface_initial(&client).await));
if !quiet {
let s = surface.read().unwrap();
print_counts(&s);
let instructions_list_tools = info
.instructions
.as_deref()
.is_some_and(|instr| s.tools.first().is_some_and(|t| instr.contains(&t.name)));
if !instructions_list_tools {
print_tool_overview(&s);
}
}
if one_shot {
let mut jobs: Vec<(String, String)> = Vec::new();
for cmd in &args.exec {
if handle_line(&session, &surface, &aliases, &mut jobs, cmd.trim()).await {
break;
}
}
std::process::exit(if HAD_ERROR.load(Ordering::Relaxed) {
1
} else {
0
});
}
let (line_tx, mut line_rx) = tokio::sync::mpsc::channel::<String>(1);
let (ack_tx, ack_rx) = std::sync::mpsc::channel::<()>();
editor::spawn_readline_thread(
server_name,
surface.clone(),
session.clone(),
aliases.clone(),
tokio::runtime::Handle::current(),
line_tx,
ack_rx,
at_prompt,
!args.no_history,
);
let mut jobs: Vec<(String, String)> = Vec::new();
loop {
tokio::select! {
Some(()) = refresh_rx.recv() => {
let fresh = fetch_surface(&session.client()).await;
println!("{} {} tools, {} prompts, {} resources",
tag(Style::new().fg(Color::Cyan), "surface changed"),
fresh.tools.len(), fresh.prompts.len(), fresh.resources.len());
*surface.write().unwrap() = fresh;
}
maybe_line = line_rx.recv() => {
let Some(line) = maybe_line else { break };
let quit = handle_line(&session, &surface, &aliases, &mut jobs, line.trim()).await;
let _ = ack_tx.send(());
if quit {
break;
}
}
}
}
Ok(())
}
async fn handle_line(
session: &Arc<Session>,
surface: &Arc<RwLock<Surface>>,
aliases: &Arc<RwLock<Aliases>>,
jobs: &mut Vec<(String, String)>,
line: &str,
) -> bool {
if line.is_empty() {
return false;
}
let expanded;
let line = match aliases.read().unwrap().expand(line) {
Ok(None) => line,
Ok(Some(text)) => {
expanded = text;
expanded.trim()
}
Err(e) => {
note_error();
if json_output() {
println!("{}", error_json(&e));
} else {
println!("{}: {e}", style::error_prefix());
}
return false;
}
};
let (output, routed) = vars::route(line);
let command = match vars::substitute(routed) {
Ok(c) => c,
Err(e) => {
note_error();
if json_output() {
println!("{}", error_json(&e));
} else {
println!("{}: {e}", style::error_prefix());
}
return false;
}
};
let line = command.as_str();
let client = session.client();
let mut tokens: Vec<&str> = line.split_whitespace().collect();
let background = tokens.last() == Some(&"&");
if background {
tokens.pop();
}
if tokens.is_empty() {
return false;
}
let cmd = tokens[0];
let rest = &tokens[1..];
match cmd {
"quit" | "exit" => return true,
"help" => {
println!("built-ins:");
println!(" tools | prompts | resources | templates list the server surface");
println!(" find <keyword> search the surface");
println!(" describe <name> schemas and metadata");
println!(" read <uri> read a resource");
println!(" subscribe <uri> | unsubscribe <uri> watch a resource for updates");
println!(" subscriptions list active subscriptions");
println!(" prompt <name> [k=v...] get a prompt");
println!(" call <tool> <json> call a tool with raw JSON");
println!(" bench <tool> [k=v...] [--n N] [--concurrency C] time repeated calls");
println!(" <tool> [k=v...] call a tool (schema-coerced)");
println!(" <tool> [k=v...] & run task-augmented (SEP-2663)");
println!(" jobs | task <id> | wait <id> | cancel <id> manage tasks");
println!(" alias [<name>=<expansion>] | unalias <name> command aliases");
println!(" wire [on|off] trace raw JSON-RPC frames");
println!(" last reprint the previous exchange");
println!(
" vars | unset <name> list or clear captured variables"
);
println!(
" name = <cmd> [| <path>] capture a result (filter with | path)"
);
println!(" $name.path in args reference a captured value");
println!(" refresh | info | quit");
let s = surface.read().unwrap();
if !s.tools.is_empty() {
println!("tools:");
for t in &s.tools {
println!(
" {:24} {}",
paint(Style::new().fg(Color::Green), &t.name),
t.description.as_deref().unwrap_or("")
);
}
}
}
"tools" | "prompts" | "resources" | "templates" => {
let s = surface.read().unwrap();
if json_output() {
let v = match cmd {
"tools" => serde_json::to_value(&s.tools),
"prompts" => serde_json::to_value(&s.prompts),
"resources" => serde_json::to_value(&s.resources),
_ => serde_json::to_value(&s.templates),
}
.unwrap_or_default();
println!("{}", json_pretty(&v));
return false;
}
match cmd {
"tools" => {
for t in &s.tools {
println!(
"{:24} {}",
paint(Style::new().fg(Color::Green), &t.name),
t.description.as_deref().unwrap_or("")
);
}
}
"prompts" => {
for p in &s.prompts {
let args: Vec<String> = p
.arguments
.iter()
.map(|a| {
if a.required {
format!("<{}>", a.name)
} else {
format!("[{}]", a.name)
}
})
.collect();
println!(
"{:24} {} {}",
paint(Style::new().fg(Color::Green), &p.name),
paint(Style::new().fg(Color::Cyan), &args.join(" ")),
p.description.as_deref().unwrap_or("")
);
}
}
"resources" => {
for r in &s.resources {
println!(
"{:40} {}",
paint(Style::new().fg(Color::Green), &r.uri),
r.name
);
}
if !s.templates.is_empty() {
println!(
"{}",
paint(
Style::new().dimmed(),
&format!(
"(+ {} resource template(s) with variables, see `templates`)",
s.templates.len()
)
)
);
}
}
_ => {
for t in &s.templates {
println!(
"{:40} {}",
paint(Style::new().fg(Color::Green), &t.uri_template),
t.name
);
}
if !s.resources.is_empty() {
println!(
"{}",
paint(
Style::new().dimmed(),
&format!(
"(+ {} concrete resource(s), see `resources`)",
s.resources.len()
)
)
);
}
}
}
}
"find" => {
let query = rest.join(" ");
if query.is_empty() {
println!("usage: find <keyword>");
return false;
}
print_find(&surface.read().unwrap(), &query);
}
"describe" => {
let Some(name) = rest.first() else {
println!("usage: describe <tool|prompt|resource|template>");
return false;
};
describe(&surface.read().unwrap(), name);
}
"read" => {
let Some(uri) = rest.first() else {
println!("usage: read <uri>");
return false;
};
let started = std::time::Instant::now();
match with_reconnect(
session,
surface,
|c| async move { c.read_resource(uri).await },
)
.await
{
Ok(result) if json_output() => {
println!(
"{}",
json_pretty(&serde_json::to_value(&result).unwrap_or_default())
);
}
Ok(result) => {
for c in result.contents {
if let Some(text) = c.text {
let is_md = c
.mime_type
.as_deref()
.is_some_and(|m| m.contains("markdown"))
|| style::looks_like_markdown(&text);
if style::colors_enabled() && is_md {
println!("{}", style::render_markdown(&text));
} else {
println!("{text}");
}
} else if let Some(blob) = c.blob {
println!(
"{}",
tag(Style::new(), &format!("binary {} base64 chars", blob.len()))
);
}
}
}
Err(e) => {
note_error();
if json_output() {
println!("{}", error_json(&e.to_string()));
} else {
println!("{}: {e}", style::error_prefix());
}
}
}
if !json_output() {
println!("{}", timing(started.elapsed()));
}
}
"subscribe" | "unsubscribe" => {
let Some(uri) = rest.first() else {
println!("usage: {cmd} <uri>");
return false;
};
handle_subscription(&client, cmd, uri).await;
}
"subscriptions" => {
let active = subscribe::list();
if json_output() {
println!("{}", json_pretty(&serde_json::json!(active)));
return false;
}
if active.is_empty() {
println!("no active subscriptions (try `subscribe <uri>`)");
return false;
}
for uri in &active {
println!("{}", paint(Style::new().fg(Color::Green), uri));
}
}
"prompt" => {
let Some(name) = rest.first() else {
println!("usage: prompt <name> [k=v...]");
return false;
};
let mut prompt_args = HashMap::new();
for t in &rest[1..] {
if let Some((k, v)) = t.split_once('=') {
prompt_args.insert(k.to_string(), v.to_string());
}
}
let started = std::time::Instant::now();
match with_reconnect(session, surface, |c| {
let prompt_args = prompt_args.clone();
async move { c.get_prompt(name, Some(prompt_args)).await }
})
.await
{
Ok(result) if json_output() => {
println!(
"{}",
json_pretty(&serde_json::to_value(&result).unwrap_or_default())
);
}
Ok(result) => {
for m in result.messages {
let v = serde_json::to_value(&m).unwrap_or_default();
let role = v.get("role").and_then(|r| r.as_str()).unwrap_or("?");
let text = v
.pointer("/content/text")
.and_then(|t| t.as_str())
.map(str::to_string)
.unwrap_or_else(|| {
v.get("content").map(|c| c.to_string()).unwrap_or_default()
});
println!("{} {}", tag(Style::new().fg(Color::Cyan), role), text);
}
}
Err(e) => {
note_error();
if json_output() {
println!("{}", error_json(&e.to_string()));
} else {
println!("{}: {e}", style::error_prefix());
}
}
}
if !json_output() {
println!("{}", timing(started.elapsed()));
}
}
"call" => {
let Some(name) = rest.first() else {
println!("usage: call <tool> <json>");
return false;
};
let json = rest[1..].join(" ");
let arguments: serde_json::Value = match serde_json::from_str(&json) {
Ok(v) => v,
Err(e) => {
note_error();
if json_output() {
println!("{}", error_json(&format!("invalid JSON: {e}")));
} else {
println!("invalid JSON: {e}");
}
return false;
}
};
run_tool(session, surface, jobs, name, arguments, background, &output).await;
}
"bench" => {
handle_bench(&client, surface, rest, background).await;
}
"jobs" => {
if jobs.is_empty() {
println!("no background tasks");
}
for (id, tool) in jobs.iter() {
match client.task_get(id).await {
Ok(task) => println!(
"{id} {tool} {}",
paint(task_status_style(task.status), &task.status.to_string())
),
Err(_) => println!("{id} {tool} (gone)"),
}
}
}
"task" | "wait" | "cancel" => {
let Some(id) = rest.first() else {
println!("usage: {cmd} <task-id>");
return false;
};
let outcome = match cmd {
"task" => client.task_get(id).await,
"wait" => client.task_wait(id).await,
_ => match client.task_cancel(id, None).await {
Ok(()) => {
println!("cancel acknowledged");
client.task_get(id).await
}
Err(e) => Err(e),
},
};
match outcome {
Ok(task) if json_output() => {
println!(
"{}",
json_pretty(&serde_json::to_value(&task).unwrap_or_default())
);
}
Ok(task) => render_task(&task),
Err(e) => {
note_error();
if json_output() {
println!("{}", error_json(&e.to_string()));
} else {
println!("{}: {e}", style::error_prefix());
}
}
}
}
"alias" | "unalias" => {
let raw = line.strip_prefix(cmd).unwrap_or("").trim();
handle_alias(aliases, surface, cmd, raw);
}
"wire" => match rest.first().copied() {
Some("on") => {
wire().set_trace(true);
println!("wire tracing on (frames print to stderr)");
}
Some("off") => {
wire().set_trace(false);
println!("wire tracing off");
}
None => println!(
"wire tracing is {}",
if wire().trace_enabled() { "on" } else { "off" }
),
Some(other) => println!("usage: wire [on|off] (got `{other}`)"),
},
"last" => match wire().last_exchange() {
None => {
if json_output() {
println!("{}", error_json("no exchange yet"));
} else {
println!("no request has been sent yet");
}
}
Some((request, response)) => {
if json_output() {
println!(
"{}",
json_pretty(&serde_json::json!({
"request": request.json,
"response": response.map(|r| r.json),
}))
);
} else {
println!("{}", wire::render(wire::Direction::Sent, &request));
match response {
Some(response) => {
println!("{}", wire::render(wire::Direction::Received, &response));
}
None => println!("(no response recorded for it)"),
}
}
}
},
"refresh" => {
let fresh = refresh_surface(session).await;
println!(
"{} tools, {} prompts, {} resources, {} templates",
fresh.tools.len(),
fresh.prompts.len(),
fresh.resources.len(),
fresh.templates.len()
);
*surface.write().unwrap() = fresh;
}
"info" => match connection_info(&client).await {
Some(info) => {
print_banner(&info);
print_counts(&surface.read().unwrap());
let caps = serde_json::to_value(&info.capabilities).unwrap_or_default();
println!("capabilities: {}", json_pretty(&caps));
println!(
"{}",
paint(
Style::new().dimmed(),
&format!("sampling: {}", sampling::mode().as_str())
)
);
}
None => println!("not initialized"),
},
"vars" => {
let all = vars::list();
if json_output() {
let map: serde_json::Map<String, serde_json::Value> = all.into_iter().collect();
println!("{}", json_pretty(&serde_json::Value::Object(map)));
} else if all.is_empty() {
println!("{}", paint(Style::new().dimmed(), "no variables"));
} else {
for (name, value) in all {
println!(
"{} {}",
paint(Style::new().fg(Color::Cyan), &format!("${name} =")),
value_summary(&value)
);
}
}
}
"unset" => match rest.first() {
Some(name) => {
if vars::unset(name) {
if !json_output() {
println!("unset ${name}");
}
} else {
note_error();
command_error(&format!("no such variable `${name}`"));
}
}
None => command_error("usage: unset <name>"),
},
tool_name => {
let schema = {
let s = surface.read().unwrap();
s.tools
.iter()
.find(|t| t.name == tool_name)
.map(|t| t.input_schema.clone())
};
let Some(schema) = schema else {
note_error();
let suggestion = find::did_you_mean(&surface.read().unwrap(), tool_name);
if json_output() {
match &suggestion {
Some(near) => println!(
"{}",
serde_json::json!({
"error": format!("unknown command: {tool_name}"),
"didYouMean": near,
})
),
None => {
println!("{}", error_json(&format!("unknown command: {tool_name}")))
}
}
} else {
let name = paint(Style::new().fg(Color::Red), tool_name);
match suggestion {
Some(near) => println!(
"unknown command: {name}; did you mean `{}`?",
paint(Style::new().fg(Color::Green), &near)
),
None => println!("unknown command: {name} (try `help`)"),
}
}
return false;
};
let arguments = parse_kv_args(&schema, rest);
run_tool(
session, surface, jobs, tool_name, arguments, background, &output,
)
.await;
}
}
false
}
async fn handle_bench(
client: &Arc<McpClient>,
surface: &Arc<RwLock<Surface>>,
rest: &[&str],
background: bool,
) {
if background {
command_error("bench cannot run task-augmented; drop the trailing `&`");
return;
}
let plan = match bench::parse(rest) {
Ok(plan) => plan,
Err(e) => {
command_error(&e);
return;
}
};
let schema = {
let s = surface.read().unwrap();
s.tools
.iter()
.find(|t| t.name == plan.tool)
.map(|t| t.input_schema.clone())
};
let Some(schema) = schema else {
command_error(&format!("no tool named `{}` (try `tools`)", plan.tool));
return;
};
let arg_tokens: Vec<&str> = plan.args.iter().map(String::as_str).collect();
let arguments = parse_kv_args(&schema, &arg_tokens);
let outcome = bench::run(client, &plan.tool, arguments, plan.n, plan.concurrency).await;
if outcome.errors > 0 {
note_error();
}
if json_output() {
println!("{}", json_pretty(&bench::render_json(&plan, &outcome)));
return;
}
println!("{}", bench::render(&plan, &outcome));
if let Some(message) = &outcome.first_error {
println!(
"{} {}",
tag(Style::new().fg(Color::Red), "first error"),
message
);
}
println!("{}", timing(outcome.total));
}
async fn handle_subscription(client: &Arc<McpClient>, cmd: &str, uri: &str) {
if cmd == "subscribe"
&& let Some(info) = connection_info(client).await
&& !subscribe::server_supports(
&serde_json::to_value(&info.capabilities).unwrap_or_default(),
)
{
eprintln!(
"warning: {} does not advertise resources.subscribe; the request will \
probably be rejected",
info.server_info.name
);
}
let started = std::time::Instant::now();
let outcome = if cmd == "subscribe" {
client.subscribe_resource(uri).await
} else {
client.unsubscribe_resource(uri).await
};
match outcome {
Ok(()) => {
let changed = if cmd == "subscribe" {
subscribe::add(uri)
} else {
subscribe::remove(uri)
};
if json_output() {
println!(
"{}",
serde_json::json!({ cmd: uri, "alreadyInEffect": !changed })
);
} else {
let note = if changed {
String::new()
} else {
format!(" {}", paint(Style::new().dimmed(), "(already in effect)"))
};
println!("{cmd}d {}{note}", paint(Style::new().fg(Color::Green), uri));
}
}
Err(e) => {
note_error();
if json_output() {
println!("{}", error_json(&e.to_string()));
} else {
println!("{}: {e}", style::error_prefix());
}
}
}
if !json_output() {
println!("{}", timing(started.elapsed()));
}
}
fn handle_alias(
aliases: &Arc<RwLock<Aliases>>,
surface: &Arc<RwLock<Surface>>,
cmd: &str,
raw: &str,
) {
let (global, rest) = match raw.strip_prefix("--global") {
Some(r) if r.is_empty() || r.starts_with(char::is_whitespace) => (true, r.trim_start()),
_ => (false, raw),
};
let rest = rest.trim();
if cmd == "unalias" {
if rest.is_empty() || rest.contains(char::is_whitespace) {
println!("usage: unalias [--global] <name>");
return;
}
match aliases.write().unwrap().remove(rest, global) {
Ok(applied) => {
report_alias_warning(applied.warning.as_deref());
if json_output() {
println!(
"{}",
serde_json::json!({
"removed": rest,
"expansion": applied.previous,
"scope": applied.scope.label(),
})
);
} else {
println!(
"removed {} {}",
paint(Style::new().fg(Color::Cyan), rest),
paint(
Style::new().dimmed(),
&format!("({})", applied.scope.label())
)
);
}
}
Err(e) => command_error(&e),
}
return;
}
if rest.is_empty() {
let aliases = aliases.read().unwrap();
let entries = aliases.entries();
if json_output() {
let rendered: Vec<serde_json::Value> = entries
.iter()
.map(|e| {
serde_json::json!({
"name": e.name,
"expansion": e.expansion,
"scope": e.scope.label(),
})
})
.collect();
println!("{}", json_pretty(&serde_json::Value::Array(rendered)));
return;
}
if entries.is_empty() {
println!("no aliases defined (try `alias t=tools`)");
return;
}
let width = entries.iter().map(|e| e.name.len()).max().unwrap_or(0);
for e in &entries {
println!(
"{:width$} {} {}",
paint(Style::new().fg(Color::Cyan), &e.name),
e.expansion,
paint(Style::new().dimmed(), &format!("({})", e.scope.label()))
);
}
return;
}
let Some((name, expansion)) = rest.split_once('=') else {
let aliases = aliases.read().unwrap();
match aliases.lookup(rest) {
Some((expansion, scope)) if json_output() => println!(
"{}",
serde_json::json!({
"name": rest,
"expansion": expansion,
"scope": scope.label(),
})
),
Some((expansion, scope)) => println!(
"{} = {} {}",
paint(Style::new().fg(Color::Cyan), rest),
expansion,
paint(Style::new().dimmed(), &format!("({})", scope.label()))
),
None => command_error(&format!(
"no alias named `{rest}` (define one with `alias {rest}=<expansion>`)"
)),
}
return;
};
let name = name.trim();
match aliases
.write()
.unwrap()
.define(name, expansion.trim(), global)
{
Ok(applied) => {
report_alias_warning(applied.warning.as_deref());
if json_output() {
println!(
"{}",
serde_json::json!({
"name": name,
"expansion": expansion.trim(),
"scope": applied.scope.label(),
"replaced": applied.previous,
})
);
return;
}
println!(
"{} = {} {}",
paint(Style::new().fg(Color::Cyan), name),
expansion.trim(),
paint(
Style::new().dimmed(),
&format!("({})", applied.scope.label())
)
);
if surface.read().unwrap().tools.iter().any(|t| t.name == name) {
println!(
"{}",
paint(
Style::new().dimmed(),
&format!("note: this shadows the tool `{name}` on this server")
)
);
}
}
Err(e) => command_error(&e),
}
}
fn report_alias_warning(warning: Option<&str>) {
if let Some(w) = warning {
eprintln!("warning: {w}");
}
}
fn command_error(message: &str) {
note_error();
if json_output() {
println!("{}", error_json(message));
} else {
println!("{}: {message}", style::error_prefix());
}
}
fn describe(surface: &Surface, name: &str) {
if let Some(t) = surface.tools.iter().find(|t| t.name == name) {
println!(
"tool {} {}",
paint(Style::new().fg(Color::Green).bold(), &t.name),
t.description.as_deref().unwrap_or("")
);
if let Some(a) = &t.annotations {
let mut hints = Vec::new();
if a.read_only_hint {
hints.push("read-only");
}
if a.idempotent_hint {
hints.push("idempotent");
}
if a.destructive_hint && !a.read_only_hint {
hints.push("destructive");
}
if a.open_world_hint {
hints.push("open-world");
}
if !hints.is_empty() {
println!(" hints: {}", hints.join(", "));
}
}
if let Some(e) = &t.execution {
let v = serde_json::to_value(e).unwrap_or_default();
if let Some(mode) = v.get("taskSupport").and_then(|m| m.as_str()) {
println!(" task support: {mode}");
}
}
println!("input schema:");
println!("{}", json_pretty(&t.input_schema));
if let Some(out) = &t.output_schema {
println!("output schema:");
println!("{}", json_pretty(out));
}
return;
}
if let Some(p) = surface.prompts.iter().find(|p| p.name == name) {
println!(
"prompt {} {}",
paint(Style::new().fg(Color::Green).bold(), &p.name),
p.description.as_deref().unwrap_or("")
);
if p.arguments.is_empty() {
println!(" (no arguments)");
} else {
println!("arguments:");
for a in &p.arguments {
println!(
" {:20} {:10} {}",
paint(Style::new().fg(Color::Cyan), &a.name),
if a.required { "required" } else { "optional" },
a.description.as_deref().unwrap_or("")
);
}
}
return;
}
if let Some(r) = surface
.resources
.iter()
.find(|r| r.uri == name || r.name == name)
{
println!(
"resource {}",
paint(Style::new().fg(Color::Green).bold(), &r.uri)
);
println!(" name: {}", r.name);
if let Some(t) = &r.title {
println!(" title: {t}");
}
if let Some(d) = &r.description {
println!(" description: {d}");
}
if let Some(m) = &r.mime_type {
println!(" mimeType: {m}");
}
if let Some(s) = r.size {
println!(" size: {s} bytes");
}
return;
}
if let Some(t) = surface
.templates
.iter()
.find(|t| t.uri_template == name || t.name == name)
{
println!(
"template {}",
paint(Style::new().fg(Color::Green).bold(), &t.uri_template)
);
println!(" name: {}", t.name);
if let Some(d) = &t.description {
println!(" description: {d}");
}
if let Some(m) = &t.mime_type {
println!(" mimeType: {m}");
}
if !t.arguments.is_empty() {
println!("arguments:");
for a in &t.arguments {
println!(
" {:20} {:10} {}",
paint(Style::new().fg(Color::Cyan), &a.name),
if a.required { "required" } else { "optional" },
a.description.as_deref().unwrap_or("")
);
}
}
return;
}
println!("nothing on the surface named `{name}` (try `tools`, `prompts`, `resources`)");
}
async fn run_tool(
session: &Arc<Session>,
surface: &Arc<RwLock<Surface>>,
jobs: &mut Vec<(String, String)>,
name: &str,
arguments: serde_json::Value,
background: bool,
output: &vars::Output,
) {
if background {
match with_reconnect(session, surface, |c| {
let arguments = arguments.clone();
async move { c.call_tool_as_task(name, arguments, None).await }
})
.await
{
Ok(created) => {
if json_output() {
println!(
"{}",
json_pretty(&serde_json::to_value(&created).unwrap_or_default())
);
} else {
println!(
"{} started",
tag(
Style::new().fg(Color::Yellow),
&format!("task {}", created.task.task_id)
)
);
}
jobs.push((created.task.task_id, name.to_string()));
}
Err(e) => {
note_error();
if json_output() {
println!("{}", error_json(&e.to_string()));
} else {
println!("{}: {e}", style::error_prefix());
}
}
}
return;
}
let started = std::time::Instant::now();
match with_reconnect(session, surface, |c| {
let arguments = arguments.clone();
async move { c.call_tool(name, arguments).await }
})
.await
{
Ok(result) => {
if result.is_error {
note_error();
}
if output.is_plain() {
if json_output() {
println!(
"{}",
json_pretty(&serde_json::to_value(&result).unwrap_or_default())
);
} else {
if result.is_error {
println!("{}", tag(Style::new().fg(Color::Red), "tool error"));
}
render_content(&result.content);
}
} else {
emit_result(result_value(&result), output);
}
}
Err(e) => {
note_error();
if json_output() {
println!("{}", error_json(&e.to_string()));
} else {
println!("{}: {e}", style::error_prefix());
}
}
}
if !json_output() {
println!("{}", timing(started.elapsed()));
}
}
fn result_value(result: &tower_mcp::CallToolResult) -> serde_json::Value {
if let Some(structured) = &result.structured_content {
return structured.clone();
}
if let [Content::Text { text, .. }] = result.content.as_slice() {
return serde_json::from_str(text)
.unwrap_or_else(|_| serde_json::Value::String(text.clone()));
}
serde_json::to_value(&result.content).unwrap_or_default()
}
fn emit_result(mut value: serde_json::Value, output: &vars::Output) {
if let Some(path) = &output.filter {
match vars::get_path(&value, path) {
Some(selected) => value = selected,
None => {
note_error();
command_error(&format!("path `{path}` not found in result"));
return;
}
}
}
if let Some(name) = &output.capture {
vars::set(name, value.clone());
if json_output() {
println!("{}", json_pretty(&value));
} else {
println!(
"{} {}",
paint(Style::new().fg(Color::Cyan), &format!("${name} =")),
value_summary(&value)
);
}
} else if json_output() {
println!("{}", json_pretty(&value));
} else {
render_value(&value);
}
}
fn value_summary(value: &serde_json::Value) -> String {
match value {
serde_json::Value::String(s) => format!("{s:?}"),
serde_json::Value::Array(a) => format!("[{} items]", a.len()),
serde_json::Value::Object(o) => format!("{{{} fields}}", o.len()),
other => other.to_string(),
}
}
fn render_value(value: &serde_json::Value) {
match value {
serde_json::Value::String(s) => println!("{s}"),
serde_json::Value::Array(_) | serde_json::Value::Object(_) => {
println!("{}", json_pretty(value))
}
other => println!("{other}"),
}
}
#[cfg(test)]
mod tests {
use super::*;
use std::sync::Mutex;
use async_trait::async_trait;
use tower_mcp::client::ClientTransport;
struct DiscoveryTransport {
result: serde_json::Value,
incoming_tx: tokio::sync::mpsc::Sender<String>,
incoming_rx: tokio::sync::mpsc::Receiver<String>,
outgoing: Arc<Mutex<Vec<serde_json::Value>>>,
connected: bool,
}
impl DiscoveryTransport {
fn new(result: serde_json::Value) -> (Self, Arc<Mutex<Vec<serde_json::Value>>>) {
let (incoming_tx, incoming_rx) = tokio::sync::mpsc::channel(4);
let outgoing = Arc::new(Mutex::new(Vec::new()));
(
Self {
result,
incoming_tx,
incoming_rx,
outgoing: outgoing.clone(),
connected: true,
},
outgoing,
)
}
}
#[async_trait]
impl ClientTransport for DiscoveryTransport {
async fn send(&mut self, message: &str) -> tower_mcp::Result<()> {
let request: serde_json::Value = serde_json::from_str(message)?;
self.outgoing.lock().unwrap().push(request.clone());
if let Some(id) = request.get("id") {
self.incoming_tx
.send(
serde_json::json!({
"jsonrpc": "2.0",
"id": id,
"result": self.result,
})
.to_string(),
)
.await
.map_err(|error| tower_mcp::Error::Transport(error.to_string()))?;
}
Ok(())
}
async fn recv(&mut self) -> tower_mcp::Result<Option<String>> {
Ok(self.incoming_rx.recv().await)
}
fn is_connected(&self) -> bool {
self.connected
}
async fn close(&mut self) -> tower_mcp::Result<()> {
self.connected = false;
Ok(())
}
}
fn jsonrpc(code: i32, message: &str) -> tower_mcp::Error {
tower_mcp::Error::JsonRpc(tower_mcp::error::JsonRpcError {
code,
message: message.to_string(),
data: None,
})
}
#[test]
fn protocol_selection_is_stable_by_default_and_final_is_exact() {
let stable = Args::try_parse_from(["mcp-repl", "--demo"]).unwrap();
assert_eq!(stable.protocol, ProtocolMode::Stable);
assert_eq!(
stable.protocol.support().unwrap().versions(),
tower_mcp::protocol::SUPPORTED_PROTOCOL_VERSIONS
);
for value in ["2026-07-28", "final"] {
let final_args =
Args::try_parse_from(["mcp-repl", "--protocol", value, "--demo"]).unwrap();
assert_eq!(final_args.protocol, ProtocolMode::Final);
assert_eq!(
final_args.protocol.support().unwrap().versions(),
["2026-07-28"]
);
}
}
#[tokio::test]
async fn stable_selection_uses_initialize() {
let client = client_builder(ProtocolMode::Stable)
.unwrap()
.connect_simple(ChannelTransport::new(demo_router()))
.await
.unwrap();
let info = establish_connection(&client, ProtocolMode::Stable)
.await
.unwrap();
assert_eq!(info.server_info.name, "mcp-repl-demo");
assert_eq!(
info.protocol_version,
tower_mcp::protocol::LATEST_PROTOCOL_VERSION
);
assert!(client.server_info().await.is_some());
assert!(client.discovery().await.is_none());
}
#[tokio::test]
async fn final_selection_uses_discover_with_required_metadata() {
let (transport, outgoing) = DiscoveryTransport::new(serde_json::json!({
"resultType": "complete",
"supportedVersions": ["2026-07-28"],
"capabilities": {"tools": {}},
"ttlMs": 0,
"cacheScope": "private",
"_meta": {
"io.modelcontextprotocol/serverInfo": {
"name": "final-test-server",
"version": "1.0.0"
}
}
}));
let client = client_builder(ProtocolMode::Final)
.unwrap()
.connect_simple(transport)
.await
.unwrap();
let info = establish_connection(&client, ProtocolMode::Final)
.await
.unwrap();
assert_eq!(info.server_info.name, "final-test-server");
assert_eq!(info.protocol_version, "2026-07-28");
assert!(client.server_info().await.is_none());
assert!(client.discovery().await.is_some());
let sent = outgoing.lock().unwrap();
assert_eq!(sent.len(), 1);
assert_eq!(sent[0]["method"], "server/discover");
assert_eq!(
sent[0]["params"]["_meta"]["io.modelcontextprotocol/protocolVersion"],
"2026-07-28"
);
assert!(
sent[0]["params"]["_meta"]["io.modelcontextprotocol/clientCapabilities"].is_object()
);
assert_eq!(
sent[0]["params"]["_meta"]["io.modelcontextprotocol/clientInfo"]["name"],
"mcp-repl"
);
}
#[test]
fn build_http_config_sets_bearer_and_trims_headers() {
let cfg = build_http_config(
Some("tok".into()),
&["X-Api-Key: abc".into(), "X-Trim : v ".into()],
None,
&[],
)
.unwrap();
assert_eq!(
cfg.headers.get("Authorization").map(String::as_str),
Some("Bearer tok")
);
assert_eq!(
cfg.headers.get("X-Api-Key").map(String::as_str),
Some("abc")
);
assert_eq!(cfg.headers.get("X-Trim").map(String::as_str), Some("v"));
}
#[test]
fn profile_auth_applies_and_flags_override_it() {
let profile_headers = [
("X-Api-Key".to_string(), "from-profile".to_string()),
("X-Kept".to_string(), "profile".to_string()),
];
let cfg =
build_http_config(None, &[], Some("profile-tok".into()), &profile_headers).unwrap();
assert_eq!(
cfg.headers.get("Authorization").map(String::as_str),
Some("Bearer profile-tok")
);
assert_eq!(
cfg.headers.get("X-Api-Key").map(String::as_str),
Some("from-profile")
);
let cfg = build_http_config(
Some("flag-tok".into()),
&["X-Api-Key: from-flag".into()],
Some("profile-tok".into()),
&profile_headers,
)
.unwrap();
assert_eq!(
cfg.headers.get("Authorization").map(String::as_str),
Some("Bearer flag-tok")
);
assert_eq!(
cfg.headers.get("X-Api-Key").map(String::as_str),
Some("from-flag")
);
assert_eq!(
cfg.headers.get("X-Kept").map(String::as_str),
Some("profile")
);
}
#[test]
fn build_http_config_rejects_header_without_colon() {
let err = build_http_config(Some("tok".into()), &["nope".into()], None, &[]).unwrap_err();
assert!(
err.contains("nope"),
"error should name the bad header: {err}"
);
assert!(
err.contains("Name: Value"),
"error should show the format: {err}"
);
}
#[test]
fn timing_formats_sub_second_and_seconds() {
assert!(timing(Duration::from_millis(142)).contains("[142ms]"));
assert!(timing(Duration::from_millis(2500)).contains("[2.50s]"));
}
#[test]
fn bench_is_a_listed_builtin() {
assert!(BUILTINS.iter().any(|(name, _)| *name == "bench"));
}
#[test]
fn find_is_a_completable_builtin() {
assert!(BUILTINS.iter().any(|(name, _)| *name == "find"));
}
#[test]
fn error_json_is_a_valid_object() {
let v: serde_json::Value = serde_json::from_str(&error_json("boom: it broke")).unwrap();
assert_eq!(v["error"], "boom: it broke");
}
#[test]
fn file_backed_history_writes_on_sync() {
use reedline::{FileBackedHistory, History, HistoryItem};
let path = std::env::temp_dir().join(format!("mcp-repl-hist-{}.txt", std::process::id()));
let _ = std::fs::remove_file(&path);
{
let mut h = FileBackedHistory::with_file(10, path.clone()).unwrap();
h.save(HistoryItem::from_command_line("echo persisted"))
.unwrap();
h.sync().unwrap();
}
let contents = std::fs::read_to_string(&path).unwrap();
assert!(
contents.contains("echo persisted"),
"history was not written to disk: {contents:?}"
);
let _ = std::fs::remove_file(&path);
}
async fn demo_client() -> McpClient {
let client = McpClient::builder()
.connect_simple(ChannelTransport::new(demo_router()))
.await
.unwrap();
client.initialize("mcp-repl-test", "0").await.unwrap();
client
}
async fn demo_session() -> (Arc<Session>, Arc<std::sync::atomic::AtomicUsize>) {
let connects = Arc::new(std::sync::atomic::AtomicUsize::new(0));
let counter = connects.clone();
let connector: Connector = Box::new(move || {
let counter = counter.clone();
Box::pin(async move {
counter.fetch_add(1, Ordering::SeqCst);
Ok(demo_client().await)
})
});
(
Arc::new(Session::new(demo_client().await, Some(connector))),
connects,
)
}
#[tokio::test(flavor = "multi_thread")]
async fn dropped_session_is_rebuilt_and_the_command_retried() {
let (session, connects) = demo_session().await;
let surface = Arc::new(RwLock::new(Surface::default()));
let attempts = Arc::new(std::sync::atomic::AtomicUsize::new(0));
let dead = Arc::as_ptr(&session.client()) as usize;
let seen: Arc<RwLock<Vec<usize>>> = Arc::new(RwLock::new(Vec::new()));
let (calls, saw) = (attempts.clone(), seen.clone());
let result = with_reconnect(&session, &surface, |c| {
let (calls, saw) = (calls.clone(), saw.clone());
async move {
saw.write().unwrap().push(Arc::as_ptr(&c) as usize);
if calls.fetch_add(1, Ordering::SeqCst) == 0 {
return Err(jsonrpc(
-32600,
"Client must send notifications/initialized before making requests",
));
}
c.call_tool("echo", serde_json::json!({ "message": "alive" }))
.await
}
})
.await
.expect("the retried call should succeed on the rebuilt session");
assert_eq!(attempts.load(Ordering::SeqCst), 2, "one retry, not a loop");
let seen = seen.read().unwrap();
assert_eq!(seen[0], dead);
assert_ne!(seen[1], dead, "the retry reused the dead client");
assert_eq!(
connects.load(Ordering::SeqCst),
1,
"reconnected exactly once"
);
assert_eq!(session.generation(), 1);
match result.content.first() {
Some(Content::Text { text, .. }) => assert_eq!(text, "alive"),
other => panic!("unexpected content: {other:?}"),
}
assert!(
!surface.read().unwrap().tools.is_empty(),
"surface should be refreshed after reconnect"
);
}
#[tokio::test(flavor = "multi_thread")]
async fn a_still_dead_server_surfaces_the_error_after_one_retry() {
let (session, connects) = demo_session().await;
let surface = Arc::new(RwLock::new(Surface::default()));
let attempts = Arc::new(std::sync::atomic::AtomicUsize::new(0));
let calls = attempts.clone();
let err = with_reconnect(&session, &surface, |_c| {
let calls = calls.clone();
async move {
calls.fetch_add(1, Ordering::SeqCst);
Err::<(), _>(tower_mcp::Error::Transport(
"HTTP 503 Service Unavailable from server: ".into(),
))
}
})
.await
.unwrap_err();
assert!(is_session_lost(&err));
assert_eq!(attempts.load(Ordering::SeqCst), 2, "bounded to one retry");
assert_eq!(connects.load(Ordering::SeqCst), 1);
}
#[tokio::test(flavor = "multi_thread")]
async fn ordinary_errors_do_not_reconnect() {
let (session, connects) = demo_session().await;
let surface = Arc::new(RwLock::new(Surface::default()));
let attempts = Arc::new(std::sync::atomic::AtomicUsize::new(0));
let calls = attempts.clone();
let err = with_reconnect(&session, &surface, |_c| {
let calls = calls.clone();
async move {
calls.fetch_add(1, Ordering::SeqCst);
Err::<(), _>(jsonrpc(-32602, "Invalid params"))
}
})
.await
.unwrap_err();
assert!(matches!(err, tower_mcp::Error::JsonRpc(j) if j.code == -32602));
assert_eq!(attempts.load(Ordering::SeqCst), 1, "no retry");
assert_eq!(connects.load(Ordering::SeqCst), 0, "no reconnect");
}
#[tokio::test(flavor = "multi_thread")]
async fn a_session_without_a_connector_never_retries() {
let session = Arc::new(Session::new(demo_client().await, None));
let surface = Arc::new(RwLock::new(Surface::default()));
let attempts = Arc::new(std::sync::atomic::AtomicUsize::new(0));
assert!(!session.can_reconnect());
let calls = attempts.clone();
let err = with_reconnect(&session, &surface, |_c| {
let calls = calls.clone();
async move {
calls.fetch_add(1, Ordering::SeqCst);
Err::<(), _>(tower_mcp::Error::SessionExpired)
}
})
.await
.unwrap_err();
assert!(matches!(err, tower_mcp::Error::SessionExpired));
assert_eq!(attempts.load(Ordering::SeqCst), 1);
}
}