M5: rmcp stdio client and tool adapters
Adds an rmcp-based stdio MCP client in harness-mcp plus adapters exposing a configured server's tools as namespaced harness-tools, with an echo-server fixture and integration test, and wiring through harness-app.
This commit is contained in:
@@ -1 +1,261 @@
|
||||
// MCP stdio client → Tool adapters land here in M5.
|
||||
//! MCP stdio client → `Tool` adapters (M5). For each configured server we spawn the child
|
||||
//! over rmcp's `TokioChildProcess` transport, `initialize`, `list_tools`, and wrap every
|
||||
//! remote tool as an [`McpTool`] named `{server}_{tool}`. Servers are gated behind the `mcp`
|
||||
//! permission key. See `docs/09-integrations.md`.
|
||||
//!
|
||||
//! Out of scope for v1 (matching the doc): resources, prompts, sampling, and non-stdio
|
||||
//! transports.
|
||||
|
||||
use std::collections::HashMap;
|
||||
use std::path::PathBuf;
|
||||
use std::sync::Arc;
|
||||
|
||||
use async_trait::async_trait;
|
||||
use harness_core::tool::{Tool, ToolCtx, ToolError, ToolOutput};
|
||||
use rmcp::model::{CallToolRequestParam, RawContent};
|
||||
use rmcp::service::RunningService;
|
||||
use rmcp::transport::TokioChildProcess;
|
||||
use rmcp::{RoleClient, ServiceExt};
|
||||
use tokio::process::Command;
|
||||
|
||||
/// Sanitized-name cap so a `{server}_{tool}` name stays a legal tool identifier.
|
||||
const MAX_NAME_LEN: usize = 64;
|
||||
|
||||
/// One configured MCP server: the child command plus its environment.
|
||||
#[derive(Debug, Clone)]
|
||||
pub struct ServerConfig {
|
||||
pub command: String,
|
||||
pub args: Vec<String>,
|
||||
pub env: HashMap<String, String>,
|
||||
}
|
||||
|
||||
#[derive(Debug, thiserror::Error)]
|
||||
enum ConnectError {
|
||||
#[error("spawn/transport failed: {0}")]
|
||||
Transport(#[from] std::io::Error),
|
||||
#[error("MCP service error: {0}")]
|
||||
Service(#[from] rmcp::service::ServiceError),
|
||||
}
|
||||
|
||||
/// Connects every configured server and returns ready tool adapters. A server that fails to
|
||||
/// start (or list its tools) is logged and skipped — the rest are unaffected, and the session
|
||||
/// still runs with whatever connected. `named_servers` is `{server_name: config}`.
|
||||
pub async fn connect_all(named_servers: HashMap<String, ServerConfig>) -> Vec<Arc<dyn Tool>> {
|
||||
let mut tools: Vec<Arc<dyn Tool>> = Vec::new();
|
||||
for (name, config) in named_servers {
|
||||
match connect(&name, &config).await {
|
||||
Ok(mut server_tools) => {
|
||||
tracing::info!(server = %name, count = server_tools.len(), "MCP server connected");
|
||||
tools.append(&mut server_tools);
|
||||
}
|
||||
Err(e) => {
|
||||
tracing::warn!(server = %name, error = %e, "MCP server failed to start; skipping");
|
||||
}
|
||||
}
|
||||
}
|
||||
tools
|
||||
}
|
||||
|
||||
async fn connect(name: &str, config: &ServerConfig) -> Result<Vec<Arc<dyn Tool>>, ConnectError> {
|
||||
let mut command = Command::new(&config.command);
|
||||
command.args(&config.args);
|
||||
for (key, value) in &config.env {
|
||||
command.env(key, value);
|
||||
}
|
||||
// rmcp sets stdin/stdout to piped and kill-on-drop; the child dies with the `RunningService`.
|
||||
let transport = TokioChildProcess::new(&mut command)?;
|
||||
let service = Arc::new(().serve(transport).await?);
|
||||
|
||||
let remote_tools = service.peer().list_all_tools().await?;
|
||||
let adapters = remote_tools
|
||||
.into_iter()
|
||||
.map(|tool| {
|
||||
let full_name = qualified_name(name, &tool.name);
|
||||
let parameters = serde_json::Value::Object((*tool.input_schema).clone());
|
||||
Arc::new(McpTool {
|
||||
full_name,
|
||||
remote_name: tool.name.to_string(),
|
||||
description: tool.description.to_string(),
|
||||
parameters,
|
||||
service: service.clone(),
|
||||
}) as Arc<dyn Tool>
|
||||
})
|
||||
.collect();
|
||||
Ok(adapters)
|
||||
}
|
||||
|
||||
/// `{server}_{tool}` sanitized to `[a-zA-Z0-9_-]` and capped at [`MAX_NAME_LEN`] chars.
|
||||
fn qualified_name(server: &str, tool: &str) -> String {
|
||||
let mut name: String = format!("{server}_{tool}")
|
||||
.chars()
|
||||
.map(|c| {
|
||||
if c.is_ascii_alphanumeric() || c == '_' || c == '-' {
|
||||
c
|
||||
} else {
|
||||
'_'
|
||||
}
|
||||
})
|
||||
.collect();
|
||||
name.truncate(MAX_NAME_LEN);
|
||||
name
|
||||
}
|
||||
|
||||
/// A single remote MCP tool exposed to the engine as a `Tool`. Holds a shared handle to the
|
||||
/// server's `RunningService` (kept alive for the whole session so the child stays up).
|
||||
struct McpTool {
|
||||
/// Engine-facing name: sanitized `{server}_{tool}`; also the permission pattern.
|
||||
full_name: String,
|
||||
/// The server's own tool name, sent back verbatim in `call_tool`.
|
||||
remote_name: String,
|
||||
description: String,
|
||||
parameters: serde_json::Value,
|
||||
service: Arc<RunningService<RoleClient, ()>>,
|
||||
}
|
||||
|
||||
#[async_trait]
|
||||
impl Tool for McpTool {
|
||||
fn name(&self) -> &str {
|
||||
&self.full_name
|
||||
}
|
||||
|
||||
fn description(&self) -> &str {
|
||||
&self.description
|
||||
}
|
||||
|
||||
fn parameters(&self) -> serde_json::Value {
|
||||
self.parameters.clone()
|
||||
}
|
||||
|
||||
async fn execute(
|
||||
&self,
|
||||
input: serde_json::Value,
|
||||
ctx: ToolCtx,
|
||||
) -> Result<ToolOutput, ToolError> {
|
||||
ctx.ask
|
||||
.ask(
|
||||
"mcp",
|
||||
self.full_name.clone(),
|
||||
self.full_name.clone(),
|
||||
input.clone(),
|
||||
)
|
||||
.await?;
|
||||
|
||||
let arguments = match input {
|
||||
serde_json::Value::Object(map) => Some(map),
|
||||
serde_json::Value::Null => None,
|
||||
other => {
|
||||
return Err(ToolError::Invalid(format!(
|
||||
"MCP tool arguments must be a JSON object, got {other}"
|
||||
)))
|
||||
}
|
||||
};
|
||||
|
||||
let result = self
|
||||
.service
|
||||
.peer()
|
||||
.call_tool(CallToolRequestParam {
|
||||
name: self.remote_name.clone().into(),
|
||||
arguments,
|
||||
})
|
||||
.await
|
||||
.map_err(|e| ToolError::Other(e.to_string()))?;
|
||||
|
||||
let mut text = String::new();
|
||||
let mut image_index = 0;
|
||||
for content in &result.content {
|
||||
match &content.raw {
|
||||
RawContent::Text(t) => {
|
||||
if !text.is_empty() {
|
||||
text.push('\n');
|
||||
}
|
||||
text.push_str(&t.text);
|
||||
}
|
||||
RawContent::Image(image) => {
|
||||
let note = save_image(&ctx.data_dir, &self.full_name, image_index, image).await;
|
||||
if !text.is_empty() {
|
||||
text.push('\n');
|
||||
}
|
||||
text.push_str(¬e);
|
||||
image_index += 1;
|
||||
}
|
||||
RawContent::Resource(resource) => {
|
||||
let embedded = resource_text(resource);
|
||||
if !embedded.is_empty() {
|
||||
if !text.is_empty() {
|
||||
text.push('\n');
|
||||
}
|
||||
text.push_str(&embedded);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// MCP surfaces tool-level failures as `is_error` with the message in `content`; map
|
||||
// that to a tool error so the model sees it as a failed call rather than a result.
|
||||
if result.is_error.unwrap_or(false) {
|
||||
return Err(ToolError::Other(if text.is_empty() {
|
||||
"MCP tool reported an error".to_string()
|
||||
} else {
|
||||
text
|
||||
}));
|
||||
}
|
||||
|
||||
Ok(ToolOutput::new(self.full_name.clone(), text))
|
||||
}
|
||||
}
|
||||
|
||||
/// Writes an image payload to the session data dir and returns a one-line note for the tool
|
||||
/// output. Best-effort: a write failure still yields a note (without a path).
|
||||
async fn save_image(
|
||||
data_dir: &PathBuf,
|
||||
tool_name: &str,
|
||||
index: usize,
|
||||
image: &rmcp::model::RawImageContent,
|
||||
) -> String {
|
||||
let ext = image.mime_type.rsplit('/').next().unwrap_or("bin");
|
||||
let file_name = format!("{tool_name}-image-{index}.{ext}.b64");
|
||||
let path = data_dir.join(&file_name);
|
||||
let saved = tokio::fs::create_dir_all(data_dir).await.is_ok()
|
||||
&& tokio::fs::write(&path, &image.data).await.is_ok();
|
||||
if saved {
|
||||
format!(
|
||||
"[image: {} ({} base64 bytes) saved to {}]",
|
||||
image.mime_type,
|
||||
image.data.len(),
|
||||
path.display()
|
||||
)
|
||||
} else {
|
||||
format!(
|
||||
"[image: {} ({} base64 bytes, not saved)]",
|
||||
image.mime_type,
|
||||
image.data.len()
|
||||
)
|
||||
}
|
||||
}
|
||||
|
||||
/// Best-effort text extraction from an embedded resource (text resources only in v1).
|
||||
fn resource_text(resource: &rmcp::model::RawEmbeddedResource) -> String {
|
||||
match &resource.resource {
|
||||
rmcp::model::ResourceContents::TextResourceContents { text, .. } => text.clone(),
|
||||
_ => String::new(),
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
|
||||
#[test]
|
||||
fn qualified_name_prefixes_and_sanitizes() {
|
||||
assert_eq!(qualified_name("fs", "read_file"), "fs_read_file");
|
||||
assert_eq!(qualified_name("my.server", "do/thing"), "my_server_do_thing");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn qualified_name_caps_length() {
|
||||
let long_tool = "t".repeat(100);
|
||||
let name = qualified_name("srv", &long_tool);
|
||||
assert_eq!(name.len(), MAX_NAME_LEN);
|
||||
assert!(name.starts_with("srv_t"));
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user