qualia_client_core/
remote_mcp.rs

1//! Remote-MCP inference backend — reach an external provider (Claude / Google / X, or another
2//! Webizen node) over the Model Context Protocol to run a completion on the person's behalf.
3//!
4//! Local inference is PREFERRED; this is the opt-in, costly path (Timothy's directive: local-first,
5//! external-via-MCP when wanted/needed; future provider credentials slot in behind the same seam).
6//! Native-only — it does network / process I/O and is never part of the wasm bundle.
7//!
8//! It issues an MCP `tools/call` to a configured inference tool (default `llm_infer`) and extracts the
9//! text from the MCP content result. Three transports, mirroring how the rest of the platform speaks
10//! MCP: **TCP** (newline-delimited JSON-RPC, exactly what the Webizen desktop MCP server on `:4245`
11//! serves), **Stdio** (spawn an MCP server command), and **HTTP** (JSON-RPC POST).
12
13use crate::agent_registry::McpTransport;
14use std::io::{BufRead, BufReader, Write};
15use std::time::Duration;
16
17/// Default MCP tool name to call for inference (the Webizen MCP surface exposes `llm_infer`).
18pub const DEFAULT_INFER_TOOL: &str = "llm_infer";
19
20/// Build the JSON-RPC `tools/call` request body for an inference call.
21///
22/// The system prompt (if any) is prepended to the user prompt so the request works against any MCP
23/// inference tool that accepts a single `prompt` string argument; `model` is passed through when set.
24fn build_infer_request(
25    infer_tool: &str,
26    model: Option<&str>,
27    system: Option<&str>,
28    prompt: &str,
29) -> serde_json::Value {
30    let full = match system {
31        Some(sys) if !sys.trim().is_empty() => format!("{sys}\n\n{prompt}"),
32        _ => prompt.to_string(),
33    };
34    let mut args = serde_json::Map::new();
35    args.insert("prompt".into(), serde_json::json!(full));
36    if let Some(m) = model {
37        if !m.is_empty() {
38            args.insert("model".into(), serde_json::json!(m));
39        }
40    }
41    serde_json::json!({
42        "jsonrpc": "2.0",
43        "id": 1,
44        "method": "tools/call",
45        "params": { "name": infer_tool, "arguments": serde_json::Value::Object(args) }
46    })
47}
48
49/// Extract the text output from an MCP `tools/call` JSON-RPC response, tolerant of shape variation.
50fn parse_infer_response(resp: &serde_json::Value) -> Result<String, String> {
51    if let Some(err) = resp.get("error") {
52        let msg = err
53            .get("message")
54            .and_then(|v| v.as_str())
55            .unwrap_or("remote MCP error");
56        return Err(format!("remote MCP error: {msg}"));
57    }
58    let result = resp
59        .get("result")
60        .ok_or_else(|| "remote MCP response missing `result`".to_string())?;
61
62    // Canonical MCP content format: result.content = [{ type:"text", text:"…" }, …]
63    if let Some(content) = result.get("content").and_then(|c| c.as_array()) {
64        let mut out = String::new();
65        for part in content {
66            if let Some(t) = part.get("text").and_then(|v| v.as_str()) {
67                out.push_str(t);
68            }
69        }
70        if !out.is_empty() {
71            return Ok(out);
72        }
73    }
74    // Fallbacks for simpler servers.
75    if let Some(t) = result.as_str() {
76        return Ok(t.to_string());
77    }
78    for k in ["text", "output", "completion", "response"] {
79        if let Some(t) = result.get(k).and_then(|v| v.as_str()) {
80            return Ok(t.to_string());
81        }
82    }
83    if result
84        .get("isError")
85        .and_then(|v| v.as_bool())
86        .unwrap_or(false)
87    {
88        return Err("remote MCP tool reported an error".to_string());
89    }
90    Err("remote MCP response had no text content".to_string())
91}
92
93/// Run one inference over the configured MCP transport and return the completion text.
94///
95/// `infer_tool` defaults to [`DEFAULT_INFER_TOOL`] when `None`. This is a blocking call — the caller
96/// should run it off the UI thread (the desktop command wrapper uses `spawn_blocking`).
97pub fn remote_mcp_infer(
98    transport: &McpTransport,
99    infer_tool: Option<&str>,
100    model: Option<&str>,
101    system: Option<&str>,
102    prompt: &str,
103) -> Result<String, String> {
104    let tool = infer_tool.unwrap_or(DEFAULT_INFER_TOOL);
105    let req = build_infer_request(tool, model, system, prompt);
106    let resp = match transport {
107        McpTransport::Tcp { host, port } => call_tcp(host, *port, &req)?,
108        McpTransport::Stdio { command, args } => call_stdio(command, args, &req)?,
109        McpTransport::Http { url, credential_id } => call_http(url, credential_id.as_deref(), &req)?,
110    };
111    parse_infer_response(&resp)
112}
113
114fn call_tcp(host: &str, port: u16, req: &serde_json::Value) -> Result<serde_json::Value, String> {
115    use std::net::TcpStream;
116    let stream =
117        TcpStream::connect((host, port)).map_err(|e| format!("connect {host}:{port}: {e}"))?;
118    stream.set_read_timeout(Some(Duration::from_secs(120))).ok();
119    stream.set_write_timeout(Some(Duration::from_secs(30))).ok();
120    let mut writer = stream.try_clone().map_err(|e| e.to_string())?;
121    let mut reader = BufReader::new(stream);
122    let line = serde_json::to_string(req).map_err(|e| e.to_string())?;
123    writer
124        .write_all(line.as_bytes())
125        .map_err(|e| e.to_string())?;
126    writer.write_all(b"\n").map_err(|e| e.to_string())?;
127    writer.flush().ok();
128    // Read JSON-RPC response lines until one carries our result/error (skip any notifications).
129    let mut buf = String::new();
130    for _ in 0..100 {
131        buf.clear();
132        let n = reader
133            .read_line(&mut buf)
134            .map_err(|e| format!("read: {e}"))?;
135        if n == 0 {
136            break;
137        }
138        let t = buf.trim();
139        if t.is_empty() {
140            continue;
141        }
142        if let Ok(v) = serde_json::from_str::<serde_json::Value>(t) {
143            if v.get("result").is_some() || v.get("error").is_some() {
144                return Ok(v);
145            }
146        }
147    }
148    Err("no JSON-RPC response from remote MCP (TCP)".into())
149}
150
151fn call_stdio(
152    command: &str,
153    args: &[String],
154    req: &serde_json::Value,
155) -> Result<serde_json::Value, String> {
156    use std::process::{Command, Stdio};
157    let mut child = Command::new(command)
158        .args(args)
159        .stdin(Stdio::piped())
160        .stdout(Stdio::piped())
161        .stderr(Stdio::null())
162        .spawn()
163        .map_err(|e| format!("spawn {command}: {e}"))?;
164    {
165        let stdin = child.stdin.as_mut().ok_or("no stdin on MCP child")?;
166        let line = serde_json::to_string(req).map_err(|e| e.to_string())?;
167        stdin
168            .write_all(line.as_bytes())
169            .map_err(|e| e.to_string())?;
170        stdin.write_all(b"\n").map_err(|e| e.to_string())?;
171        stdin.flush().ok();
172    }
173    let stdout = child.stdout.take().ok_or("no stdout on MCP child")?;
174    let mut reader = BufReader::new(stdout);
175    let mut buf = String::new();
176    let mut found = None;
177    for _ in 0..200 {
178        buf.clear();
179        let n = reader
180            .read_line(&mut buf)
181            .map_err(|e| format!("read: {e}"))?;
182        if n == 0 {
183            break;
184        }
185        let t = buf.trim();
186        if t.is_empty() {
187            continue;
188        }
189        if let Ok(v) = serde_json::from_str::<serde_json::Value>(t) {
190            if v.get("result").is_some() || v.get("error").is_some() {
191                found = Some(v);
192                break;
193            }
194        }
195    }
196    let _ = child.kill();
197    let _ = child.wait();
198    found.ok_or_else(|| "no JSON-RPC response from stdio MCP server".to_string())
199}
200
201fn call_http(
202    url: &str,
203    credential_id: Option<&str>,
204    req: &serde_json::Value,
205) -> Result<serde_json::Value, String> {
206    let client = reqwest::blocking::Client::builder()
207        .timeout(Duration::from_secs(180))
208        .build()
209        .map_err(|e| e.to_string())?;
210    let mut request = client
211        .post(url)
212        .header("content-type", "application/json")
213        .header("accept", "application/json")
214        .json(req);
215    if let Some(connection) = credential_id.filter(|value| !value.trim().is_empty()) {
216        // The credential exists only in the platform keychain and is read at
217        // dispatch time after the caller has obtained consent.
218        let secret = crate::provider_credentials::bearer_credential(connection)?;
219        request = request.bearer_auth(secret);
220    }
221    let resp = request.send()
222        .map_err(|e| format!("http post: {e}"))?;
223    let status = resp.status();
224    let text = resp.text().map_err(|e| e.to_string())?;
225    if !status.is_success() {
226        let snippet: String = text.chars().take(240).collect();
227        return Err(format!("remote MCP HTTP {status}: {snippet}"));
228    }
229    serde_json::from_str(&text).map_err(|e| format!("parse http json: {e}"))
230}
231
232/// Test an MCP endpoint without asking it to generate text.  It sends the
233/// standard `tools/list` request and only reports that a response was received.
234pub fn remote_mcp_probe(transport: &McpTransport) -> Result<usize, String> {
235    let req = serde_json::json!({
236        "jsonrpc": "2.0", "id": 1, "method": "tools/list", "params": {}
237    });
238    let response = match transport {
239        McpTransport::Tcp { host, port } => call_tcp(host, *port, &req)?,
240        McpTransport::Stdio { command, args } => call_stdio(command, args, &req)?,
241        McpTransport::Http { url, credential_id } => call_http(url, credential_id.as_deref(), &req)?,
242    };
243    if let Some(error) = response.get("error") {
244        let message = error.get("message").and_then(|value| value.as_str()).unwrap_or("MCP error");
245        return Err(format!("MCP tools/list failed: {message}"));
246    }
247    let tools = response
248        .get("result")
249        .and_then(|result| result.get("tools"))
250        .and_then(|tools| tools.as_array())
251        .ok_or_else(|| "MCP tools/list response had no tools array".to_string())?;
252    Ok(tools.len())
253}
254
255#[cfg(test)]
256mod tests {
257    use super::*;
258
259    #[test]
260    fn request_is_valid_tools_call() {
261        let req = build_infer_request("llm_chat", Some("phi-3"), Some("Be terse."), "hi");
262        assert_eq!(req["jsonrpc"], "2.0");
263        assert_eq!(req["method"], "tools/call");
264        assert_eq!(req["params"]["name"], "llm_chat");
265        assert_eq!(req["params"]["arguments"]["model"], "phi-3");
266        let prompt = req["params"]["arguments"]["prompt"].as_str().unwrap();
267        assert!(prompt.starts_with("Be terse."));
268        assert!(prompt.ends_with("hi"));
269    }
270
271    #[test]
272    fn request_omits_empty_model_and_system() {
273        let req = build_infer_request(DEFAULT_INFER_TOOL, None, None, "just this");
274        assert!(req["params"]["arguments"].get("model").is_none());
275        assert_eq!(req["params"]["arguments"]["prompt"], "just this");
276    }
277
278    #[test]
279    fn parses_mcp_content_array() {
280        let resp = serde_json::json!({
281            "jsonrpc": "2.0", "id": 1,
282            "result": { "content": [ {"type":"text","text":"Hello"}, {"type":"text","text":", world"} ] }
283        });
284        assert_eq!(parse_infer_response(&resp).unwrap(), "Hello, world");
285    }
286
287    #[test]
288    fn parses_simple_fallbacks() {
289        let a = serde_json::json!({ "result": "plain string" });
290        assert_eq!(parse_infer_response(&a).unwrap(), "plain string");
291        let b = serde_json::json!({ "result": { "text": "keyed text" } });
292        assert_eq!(parse_infer_response(&b).unwrap(), "keyed text");
293    }
294
295    #[test]
296    fn surfaces_errors() {
297        let e = serde_json::json!({ "error": { "code": -32000, "message": "boom" } });
298        assert!(parse_infer_response(&e).unwrap_err().contains("boom"));
299        let ie = serde_json::json!({ "result": { "isError": true, "content": [] } });
300        assert!(parse_infer_response(&ie).is_err());
301    }
302}
qualia_client_core/remote_mcp.rs

qualia_client_core/
remote_mcp.rs
