isb_core/
serve_client.rs

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

isb_core/
serve_client.rs
