isb_core/
serve_client.rs

1//! Calling `isb serve` tools from the CLI over its unix socket.
2//!
3//! One `tools/call` per connection, in MCP 2026-07-28: no `initialize`,
4//! and the version and capabilities ride in each request's `_meta`.
5
6use std::io::{ErrorKind, Read, Write};
7use std::net::{TcpStream, ToSocketAddrs};
8use std::os::unix::net::UnixStream;
9use std::path::Path;
10use std::time::{Duration, Instant};
11
12use serde_json::{Value, json};
13
14use crate::error::{Error, Result};
15
16/// The MCP protocol versions `isb serve` speaks, newest first.
17pub const PROTOCOL_VERSIONS: [&str; 4] = ["2026-07-28", "2025-11-25", "2025-06-18", "2025-03-26"];
18/// The stateless revisions: version and capabilities in every request's
19/// `_meta`, no `initialize`.
20pub const MODERN_VERSIONS: [&str; 1] = ["2026-07-28"];
21/// The revisions that open with `initialize`; an unknown version asked
22/// there is answered with the first.
23pub const LEGACY_VERSIONS: [&str; 3] = ["2025-11-25", "2025-06-18", "2025-03-26"];
24
25const MAX_RESPONSE_BYTES: usize = 64 * 1024 * 1024;
26
27/// Call a tool and return its result. A tool that fails comes back as
28/// [`Error::Remote`] carrying the server-side error's code and data; a
29/// protocol error (unknown tool, bad arguments) as [`Error::Invalid`] or
30/// [`Error::Protocol`].
31pub fn call_tool(socket: &Path, name: &str, args: Value, timeout: Duration) -> Result<Value> {
32    let r = rpc(
33        socket,
34        "tools/call",
35        json!({"name": name, "arguments": args}),
36        timeout,
37    )?;
38    unwrap_result(r)
39}
40
41/// A `tools/call` result as the tool's value, or its error.
42fn unwrap_result(r: Value) -> Result<Value> {
43    let structured = r.get("structuredContent").cloned();
44    if r.get("isError").and_then(Value::as_bool) == Some(true) {
45        let s = structured.unwrap_or(Value::Null);
46        let text = r
47            .pointer("/content/0/text")
48            .and_then(Value::as_str)
49            .unwrap_or("tool failed");
50        return Err(Error::Remote {
51            code: s
52                .get("code")
53                .and_then(Value::as_str)
54                .unwrap_or("tool_error")
55                .to_string(),
56            message: s
57                .get("message")
58                .and_then(Value::as_str)
59                .unwrap_or(text)
60                .to_string(),
61            data: s.get("data").cloned().unwrap_or(Value::Null),
62        });
63    }
64    // Undo the server's wrapping of non-object results.
65    Ok(match structured {
66        Some(Value::Object(mut o)) if o.len() == 1 && o.contains_key("result") => {
67            o.remove("result").unwrap_or(Value::Null)
68        }
69        Some(v) => v,
70        None => r
71            .pointer("/content/0/text")
72            .and_then(Value::as_str)
73            .map(|t| serde_json::from_str(t).unwrap_or_else(|_| json!(t)))
74            .unwrap_or(Value::Null),
75    })
76}
77
78/// The tools the socket's listener exposes, as `tools/list` describes them.
79pub fn list_tools(socket: &Path, timeout: Duration) -> Result<Vec<Value>> {
80    let r = rpc(socket, "tools/list", json!({}), timeout)?;
81    match r.get("tools") {
82        Some(Value::Array(a)) => Ok(a.clone()),
83        _ => Err(Error::Protocol("tools/list: no tools array".into())),
84    }
85}
86
87fn rpc(socket: &Path, method: &str, mut params: Value, timeout: Duration) -> Result<Value> {
88    let version = MODERN_VERSIONS[0];
89    params["_meta"] = json!({
90        "io.modelcontextprotocol/protocolVersion": version,
91        "io.modelcontextprotocol/clientCapabilities": {},
92        "io.modelcontextprotocol/clientInfo": {"name": "isb", "version": env!("CARGO_PKG_VERSION")},
93    });
94    let body = serde_json::to_vec(
95        &json!({"jsonrpc": "2.0", "id": 1, "method": method, "params": params}),
96    )?;
97    // Mcp-Name repeats the tool's name; tool names are plain ASCII.
98    let name = params.get("name").and_then(Value::as_str);
99    let mut headers = vec![("MCP-Protocol-Version", version), ("Mcp-Method", method)];
100    headers.extend(name.map(|n| ("Mcp-Name", n)));
101    let stream = UnixStream::connect(socket).map_err(|e| {
102        std::io::Error::new(
103            e.kind(),
104            format!("cannot connect to isb serve at {}: {e}", socket.display()),
105        )
106    })?;
107    let what = format!("{method} on {}", socket.display());
108    let (status, bytes) = exchange_with(
109        Stream::Unix(stream),
110        "POST",
111        "/mcp",
112        &body,
113        timeout,
114        &what,
115        &headers,
116    )?;
117    let v: Value = serde_json::from_slice(&bytes).map_err(|e| {
118        Error::Protocol(format!(
119            "{what}: HTTP {status}, undecodable body ({e}): {}",
120            String::from_utf8_lossy(&bytes[..bytes.len().min(200)])
121        ))
122    })?;
123    if let Some(err) = v.get("error") {
124        let code = err.get("code").and_then(Value::as_i64).unwrap_or(0);
125        let msg = err
126            .get("message")
127            .and_then(Value::as_str)
128            .unwrap_or("error")
129            .to_string();
130        return Err(if code == -32602 {
131            Error::Invalid(msg)
132        } else {
133            Error::Protocol(format!("{method}: {msg} ({code})"))
134        });
135    }
136    if status != 200 {
137        return Err(Error::Protocol(format!("{what}: HTTP {status}")));
138    }
139    v.get("result")
140        .cloned()
141        .ok_or_else(|| Error::Protocol(format!("{what}: response has no result")))
142}
143
144/// `GET /healthz` on a TCP address: the status and the decoded body.
145pub fn healthz(addr: &str, timeout: Duration) -> Result<(u16, Value)> {
146    let target = addr
147        .to_socket_addrs()?
148        .next()
149        .ok_or_else(|| Error::invalid(format!("{addr} does not resolve")))?;
150    let s = TcpStream::connect_timeout(&target, timeout)?;
151    let (status, body) = exchange(
152        Stream::Tcp(s),
153        "GET",
154        "/healthz",
155        b"",
156        timeout,
157        &format!("GET http://{addr}/healthz"),
158    )?;
159    Ok((status, serde_json::from_slice(&body).unwrap_or(Value::Null)))
160}
161
162enum Stream {
163    Tcp(TcpStream),
164    Unix(UnixStream),
165}
166
167impl Stream {
168    fn set_timeouts(&self, d: Duration) -> std::io::Result<()> {
169        match self {
170            Stream::Tcp(s) => {
171                s.set_read_timeout(Some(d))?;
172                s.set_write_timeout(Some(d))
173            }
174            Stream::Unix(s) => {
175                s.set_read_timeout(Some(d))?;
176                s.set_write_timeout(Some(d))
177            }
178        }
179    }
180
181    fn io(&mut self) -> &mut dyn ReadWrite {
182        match self {
183            Stream::Tcp(s) => s,
184            Stream::Unix(s) => s,
185        }
186    }
187}
188
189trait ReadWrite: Read + Write {}
190impl<T: Read + Write> ReadWrite for T {}
191
192/// One request on a fresh connection, bounded by `timeout` overall. The
193/// server closes after answering, so the response ends at EOF.
194fn exchange(
195    s: Stream,
196    method: &str,
197    path: &str,
198    body: &[u8],
199    timeout: Duration,
200    what: &str,
201) -> Result<(u16, Vec<u8>)> {
202    exchange_with(s, method, path, body, timeout, what, &[])
203}
204
205/// [`exchange`] with extra request headers.
206fn exchange_with(
207    mut s: Stream,
208    method: &str,
209    path: &str,
210    body: &[u8],
211    timeout: Duration,
212    what: &str,
213    extra: &[(&str, &str)],
214) -> Result<(u16, Vec<u8>)> {
215    let started = Instant::now();
216    let timed_out = || {
217        Error::Io(std::io::Error::new(
218            ErrorKind::TimedOut,
219            format!("isb serve did not answer {what} within {timeout:?}"),
220        ))
221    };
222    let to_err = |e: std::io::Error| {
223        if matches!(e.kind(), ErrorKind::WouldBlock | ErrorKind::TimedOut) {
224            timed_out()
225        } else {
226            Error::Io(e)
227        }
228    };
229    s.set_timeouts(timeout)?;
230    let mut more = String::new();
231    for (k, v) in extra {
232        more.push_str(&format!("{k}: {v}\r\n"));
233    }
234    let head = format!(
235        "{method} {path} HTTP/1.1\r\nHost: localhost\r\nUser-Agent: isb/{}\r\n\
236         Content-Type: application/json\r\nAccept: application/json, text/event-stream\r\n\
237         Content-Length: {}\r\n{more}Connection: close\r\n\r\n",
238        env!("CARGO_PKG_VERSION"),
239        body.len()
240    );
241    s.io().write_all(head.as_bytes()).map_err(to_err)?;
242    s.io().write_all(body).map_err(to_err)?;
243    s.io().flush().map_err(to_err)?;
244
245    let mut buf = Vec::with_capacity(8192);
246    let mut chunk = [0u8; 16384];
247    loop {
248        let remaining = timeout.saturating_sub(started.elapsed());
249        if remaining.is_zero() {
250            return Err(timed_out());
251        }
252        // macOS refuses setsockopt (EINVAL) once the server has closed; the
253        // read cannot block then, so the previous timeout is as good.
254        let _ = s.set_timeouts(remaining);
255        match s.io().read(&mut chunk) {
256            Ok(0) => break,
257            Ok(n) => buf.extend_from_slice(&chunk[..n]),
258            Err(e) if e.kind() == ErrorKind::Interrupted => continue,
259            Err(e) => return Err(to_err(e)),
260        }
261        if buf.len() > MAX_RESPONSE_BYTES {
262            return Err(Error::Protocol(format!("{what}: response too large")));
263        }
264    }
265    let mut hs = [httparse::EMPTY_HEADER; 64];
266    let mut r = httparse::Response::new(&mut hs);
267    let n = match r.parse(&buf) {
268        Ok(httparse::Status::Complete(n)) => n,
269        Ok(httparse::Status::Partial) => {
270            return Err(Error::Protocol(format!("{what}: truncated response")));
271        }
272        Err(e) => return Err(Error::Protocol(format!("{what}: bad response: {e}"))),
273    };
274    let status = r.code.unwrap_or(0);
275    let length = r
276        .headers
277        .iter()
278        .find(|h| h.name.eq_ignore_ascii_case("content-length"))
279        .and_then(|h| {
280            std::str::from_utf8(h.value)
281                .ok()?
282                .trim()
283                .parse::<usize>()
284                .ok()
285        });
286    let mut body = buf.split_off(n);
287    if let Some(len) = length {
288        if body.len() < len {
289            return Err(Error::Protocol(format!("{what}: truncated response")));
290        }
291        body.truncate(len);
292    }
293    Ok((status, body))
294}
isb_core/serve_client.rs

isb_core/
serve_client.rs
