agent_core/
rpc.rs

1// Versioned newline JSON-RPC plus MCP adapters. No I/O: transports supply complete lines.
2// New shared implementation, kept independent from OS streams and browser MessagePorts.
3
4use crate::{json, ActionError, BatchCall, Dispatcher, ErrorCode, Host, Invocation, Value};
5use alloc::{string::String, vec::Vec};
6
7/// Application protocol revision carried by JSON control responses.
8pub const JSON_VERSION: &str = "1";
9/// Current MCP revision, selected per request through params._meta.
10pub const MCP_VERSION: &str = "2026-07-28";
11/// Explicitly supported initialization-based MCP revision.
12pub const MCP_LEGACY_VERSION: &str = "2025-11-25";
13
14/// One transport chooses its protocol explicitly; there is no guessing from requests.
15#[derive(Clone, Copy, Debug, PartialEq, Eq)]
16pub enum Protocol {
17    /// jfguiLib JSON-RPC control protocol version 1.
18    Json,
19    /// MCP with current per-request metadata or the supported legacy handshake.
20    Mcp,
21}
22
23/// Select a connection's protocol from its first valid JSON-RPC envelope. An explicit
24/// modern MCP version or legacy initialization selects MCP; other requests select JSON v1.
25/// Malformed envelopes return None so transports can report them without locking the session.
26pub fn requested_protocol(line: &[u8]) -> Option<Protocol> {
27    let request = json::parse(line).ok()?;
28    if request.get("jsonrpc").and_then(Value::as_str) != Some("2.0")
29        || request.get("method").and_then(Value::as_str).is_none()
30        || request.get("result").is_some()
31        || request.get("error").is_some()
32        || request
33            .get("params")
34            .is_some_and(|params| params.as_object().is_none())
35        || request.get("id").is_some_and(|id| {
36            !matches!(id, Value::Null | Value::String(_))
37                && id.as_u64().is_none()
38                && id.as_i64().is_none()
39        })
40    {
41        return None;
42    }
43    let is_mcp = request.get("params").is_some_and(|params| {
44        params
45            .get("_meta")
46            .and_then(|meta| meta.get("io.modelcontextprotocol/protocolVersion"))
47            .is_some()
48            || request.get("method").and_then(Value::as_str) == Some("initialize")
49                && params.get("protocolVersion").is_some()
50    });
51    Some(if is_mcp {
52        Protocol::Mcp
53    } else {
54        Protocol::Json
55    })
56}
57
58/// Per-connection protocol framing state; application state stays in the Host.
59pub struct RpcSession {
60    pub(crate) protocol: Protocol,
61    pub(crate) server_name: String,
62    pub(crate) server_version: String,
63    pub(crate) legacy_initialized: bool,
64    pub(crate) legacy_ready: bool,
65}
66
67impl RpcSession {
68    /// Whether initialization-based MCP negotiation must survive between requests.
69    /// Stateless JSON/current-MCP browser sessions may be evicted when an inactive cache fills.
70    pub fn requires_persistent_session(&self) -> bool {
71        self.legacy_initialized
72    }
73    /// Create a protocol adapter without activating application control.
74    pub fn new(protocol: Protocol, server_name: &str, server_version: &str) -> Self {
75        Self {
76            protocol,
77            server_name: server_name.into(),
78            server_version: server_version.into(),
79            legacy_initialized: false,
80            legacy_ready: false,
81        }
82    }
83    /// Handle one UTF-8 JSON line, returning a newline-terminated response or no notification reply.
84    /// This method never logs or writes standard streams. Use one session per connection.
85    pub fn handle(
86        &mut self,
87        dispatcher: &Dispatcher,
88        host: &mut impl Host,
89        line: &[u8],
90    ) -> Option<String> {
91        let request = match json::parse(line) {
92            Ok(value) => value,
93            Err(error) => {
94                return Some(encode(error_response(
95                    Value::Null,
96                    RpcError {
97                        code: -32700,
98                        message: "Invalid JSON".into(),
99                        data: Value::object([
100                            ("offset", Value::u64(error.offset as u64)),
101                            ("reason", Value::string(error.kind)),
102                        ]),
103                    },
104                )))
105            }
106        };
107        let id = request.get("id").cloned();
108        let valid_id = id.as_ref().is_none_or(|id| {
109            matches!(id, Value::String(_))
110                || id.as_i64().is_some()
111                || id.as_u64().is_some()
112                || self.protocol == Protocol::Json && matches!(id, Value::Null)
113        });
114        if request.as_object().is_none()
115            || request.get("jsonrpc").and_then(Value::as_str) != Some("2.0")
116            || request.get("method").and_then(Value::as_str).is_none()
117            || !valid_id
118            || request.get("result").is_some()
119            || request.get("error").is_some()
120        {
121            return Some(encode(error_response(
122                Value::Null,
123                RpcError::new(-32600, "Invalid JSON-RPC request"),
124            )));
125        }
126        let method = request.get("method").and_then(Value::as_str).unwrap_or("");
127        let empty = Value::object([]);
128        let params = request.get("params").unwrap_or(&empty);
129        if params.as_object().is_none() {
130            return id.map(|id| {
131                encode(error_response(
132                    id,
133                    RpcError::new(-32602, "Parameters must be an object"),
134                ))
135            });
136        }
137        if id.is_none() {
138            // Deliberately accept only recognized notifications; semantic writes require IDs.
139            if dispatcher.enabled()
140                && self.protocol == Protocol::Mcp
141                && method == "notifications/initialized"
142                && self.legacy_initialized
143            {
144                self.legacy_ready = true;
145            }
146            // All handlers here finish synchronously or return an explicit app task ID.
147            // An MCP cancellation for an already completed request therefore has no effect.
148            return None;
149        }
150        let id = id.unwrap_or(Value::Null);
151        let result = dispatcher
152            .check_enabled()
153            .map_err(RpcError::action)
154            .and_then(|()| match self.protocol {
155                Protocol::Json => self.json_request(dispatcher, host, method, params),
156                Protocol::Mcp => self.mcp_request(dispatcher, host, method, params),
157            });
158        Some(encode(match result {
159            Ok(value) => Value::object([
160                ("jsonrpc", Value::string("2.0")),
161                ("id", id),
162                ("result", value),
163            ]),
164            Err(error) => error_response(id, error),
165        }))
166    }
167    fn json_request(
168        &self,
169        dispatcher: &Dispatcher,
170        host: &mut impl Host,
171        method: &str,
172        params: &Value,
173    ) -> Result<Value, RpcError> {
174        if params
175            .get("protocolVersion")
176            .is_some_and(|v| v.as_str() != Some(JSON_VERSION))
177        {
178            return Err(RpcError::new(
179                -32602,
180                "Unsupported control protocolVersion; expected 1",
181            ));
182        }
183        let mut clean = params.clone();
184        if let Value::Object(fields) = &mut clean {
185            fields.retain(|(key, _)| key != "protocolVersion");
186        }
187        let mut result = if method == "agent/info" {
188            allowed(&clean, &[]).map_err(RpcError::action)?;
189            Value::object([
190                ("protocolVersion", Value::string(JSON_VERSION)),
191                ("serverInfo", self.server_info()),
192                ("revision", Value::u64(host.revision())),
193                ("capabilities", Value::u64(host.capabilities())),
194                ("controlEnabled", Value::Bool(true)),
195            ])
196        } else {
197            if !matches!(
198                method,
199                "actions/list"
200                    | "actions/describe"
201                    | "actions/execute"
202                    | "actions/batch"
203                    | "state/inspect"
204                    | "tasks/get"
205                    | "tasks/cancel"
206                    | "resources/read"
207                    | "view/render"
208            ) {
209                return Err(RpcError::new(-32601, "Method not found"));
210            }
211            control(dispatcher, host, method, &clean).map_err(RpcError::action)?
212        };
213        result.insert("protocolVersion", Value::string(JSON_VERSION));
214        Ok(result)
215    }
216    pub(crate) fn server_info(&self) -> Value {
217        Value::object([
218            ("name", Value::string(&self.server_name)),
219            ("version", Value::string(&self.server_version)),
220        ])
221    }
222}
223
224/// Execute the version-independent methods used by JSON and the MCP tool catalog.
225pub fn control(
226    dispatcher: &Dispatcher,
227    host: &mut impl Host,
228    method: &str,
229    params: &Value,
230) -> Result<Value, ActionError> {
231    dispatcher.check_enabled()?;
232    if params.as_object().is_none() {
233        return Err(invalid("Parameters must be an object"));
234    }
235    match method {
236        "actions/list" => {
237            allowed(params, &["query", "cursor", "limit"])?;
238            let query = optional_str(params, "query")?.unwrap_or("");
239            let cursor = optional_str(params, "cursor")?;
240            let limit = optional_u64(params, "limit")?.unwrap_or(30);
241            if limit > 100 {
242                return Err(invalid("Discovery limit must be at most 100"));
243            }
244            Ok(dispatcher
245                .discover(host, query, cursor, limit as usize)?
246                .to_json())
247        }
248        "actions/describe" => {
249            allowed(params, &["action"])?;
250            dispatcher.describe(host, required_str(params, "action")?)
251        }
252        "actions/execute" => {
253            allowed(params, &["action", "input", "expectedRevision"])?;
254            let empty = Value::object([]);
255            let call = Invocation {
256                action: required_str(params, "action")?,
257                input: params.get("input").unwrap_or(&empty),
258                expected_revision: optional_u64(params, "expectedRevision")?,
259            };
260            Ok(dispatcher.execute(host, call)?.to_json())
261        }
262        "actions/batch" => {
263            allowed(params, &["calls", "atomic", "stopOnError"])?;
264            let Some(calls) = params.get("calls").and_then(Value::as_array) else {
265                return Err(invalid("calls must be an array"));
266            };
267            if calls.is_empty() || calls.len() > 128 {
268                return Err(invalid("A batch must contain 1 to 128 calls"));
269            }
270            let mut parsed = Vec::with_capacity(calls.len());
271            for call in calls {
272                allowed(call, &["action", "input", "expectedRevision"])?;
273                parsed.push(BatchCall {
274                    action: required_str(call, "action")?.into(),
275                    input: call
276                        .get("input")
277                        .cloned()
278                        .unwrap_or_else(|| Value::object([])),
279                    expected_revision: optional_u64(call, "expectedRevision")?,
280                });
281            }
282            Ok(dispatcher
283                .batch(
284                    host,
285                    &parsed,
286                    optional_bool(params, "atomic")?.unwrap_or(false),
287                    optional_bool(params, "stopOnError")?.unwrap_or(true),
288                )?
289                .to_json())
290        }
291        "state/inspect" => host.inspect(params),
292        "tasks/get" => {
293            allowed(params, &["id"])?;
294            host.task_status(required_u64(params, "id")?)
295        }
296        "tasks/cancel" => {
297            allowed(params, &["id"])?;
298            host.task_cancel(required_u64(params, "id")?)
299        }
300        "resources/read" => {
301            allowed(params, &["uri", "offset", "limit"])?;
302            let uri = required_str(params, "uri")?;
303            let offset = optional_u64(params, "offset")?.unwrap_or(0);
304            let limit = optional_u64(params, "limit")?.unwrap_or(32_768);
305            if limit == 0 || limit > 65_536 {
306                return Err(invalid("Resource limit must be 1 to 65536 bytes"));
307            }
308            let info = host.resource_info(uri)?;
309            if info.size.is_some_and(|size| offset > size) {
310                return Err(invalid("Resource offset exceeds its size"));
311            }
312            let bytes = host.resource_read(uri, offset, limit as usize)?;
313            if bytes.len() > limit as usize {
314                return Err(ActionError::new(
315                    ErrorCode::Internal,
316                    "Resource reader exceeded its limit",
317                ));
318            }
319            let next = offset
320                .checked_add(bytes.len() as u64)
321                .ok_or_else(|| invalid("Resource offset overflow"))?;
322            let eof = info.size.is_some_and(|size| next >= size) || bytes.len() < limit as usize;
323            Ok(Value::object([
324                ("resource", info.to_json()),
325                ("offset", Value::u64(offset)),
326                ("nextOffset", Value::u64(next)),
327                ("eof", Value::Bool(eof)),
328                ("encoding", Value::string("base64")),
329                ("data", Value::string(base64(&bytes))),
330            ]))
331        }
332        "view/render" => {
333            let call = Invocation {
334                action: "ui.render",
335                input: params,
336                expected_revision: None,
337            };
338            Ok(dispatcher.execute(host, call)?.to_json())
339        }
340        _ => Err(ActionError::new(
341            ErrorCode::NotFound,
342            "Control method not found",
343        )),
344    }
345}
346
347pub(crate) fn required_str<'a>(value: &'a Value, key: &str) -> Result<&'a str, ActionError> {
348    optional_str(value, key)?.ok_or_else(|| invalid("Required string parameter is missing"))
349}
350pub(crate) fn optional_str<'a>(
351    value: &'a Value,
352    key: &str,
353) -> Result<Option<&'a str>, ActionError> {
354    match value.get(key) {
355        None => Ok(None),
356        Some(Value::String(text)) if text.len() <= 4096 => Ok(Some(text)),
357        _ => Err(invalid("Expected a string of at most 4096 bytes")),
358    }
359}
360pub(crate) fn required_u64(value: &Value, key: &str) -> Result<u64, ActionError> {
361    optional_u64(value, key)?.ok_or_else(|| invalid("Required integer parameter is missing"))
362}
363pub(crate) fn optional_u64(value: &Value, key: &str) -> Result<Option<u64>, ActionError> {
364    match value.get(key) {
365        None => Ok(None),
366        Some(v) => v
367            .as_u64()
368            .map(Some)
369            .ok_or_else(|| invalid("Expected an unsigned integer")),
370    }
371}
372fn optional_bool(value: &Value, key: &str) -> Result<Option<bool>, ActionError> {
373    match value.get(key) {
374        None => Ok(None),
375        Some(Value::Bool(v)) => Ok(Some(*v)),
376        _ => Err(invalid("Expected a boolean")),
377    }
378}
379pub(crate) fn allowed(value: &Value, names: &[&str]) -> Result<(), ActionError> {
380    let Some(fields) = value.as_object() else {
381        return Err(invalid("Parameters must be an object"));
382    };
383    for (key, _) in fields {
384        if !names.contains(&key.as_str()) {
385            return Err(invalid("Unknown parameter")
386                .with_details(Value::object([("parameter", Value::string(key))])));
387        }
388    }
389    Ok(())
390}
391pub(crate) fn invalid(message: &str) -> ActionError {
392    ActionError::new(ErrorCode::InvalidInput, message)
393}
394
395pub(crate) struct RpcError {
396    pub code: i64,
397    pub message: String,
398    pub data: Value,
399}
400impl RpcError {
401    pub fn new(code: i64, message: &str) -> Self {
402        Self {
403            code,
404            message: message.into(),
405            data: Value::Null,
406        }
407    }
408    pub fn action(error: ActionError) -> Self {
409        Self {
410            code: if error.code == ErrorCode::InvalidInput {
411                -32602
412            } else {
413                -31000
414            },
415            message: error.message.clone(),
416            data: error.to_json(),
417        }
418    }
419}
420fn error_response(id: Value, error: RpcError) -> Value {
421    Value::object([
422        ("jsonrpc", Value::string("2.0")),
423        ("id", id),
424        (
425            "error",
426            Value::object([
427                ("code", Value::i64(error.code)),
428                ("message", Value::string(error.message)),
429                ("data", error.data),
430            ]),
431        ),
432    ])
433}
434fn encode(value: Value) -> String {
435    let mut result = match json::stringify(&value) {
436        Ok(text) => text,
437        Err(_) => {
438            let id = value.get("id").cloned().unwrap_or(Value::Null);
439            json::stringify(&error_response(id, RpcError::new(-32603, "Response exceeds encoding limits"))).unwrap_or_else(|_| String::from("{\"jsonrpc\":\"2.0\",\"id\":null,\"error\":{\"code\":-32603,\"message\":\"Encoding failed\"}}"))
440        }
441    };
442    result.push('\n');
443    result
444}
445
446/// Base64-encode a separately requested resource payload.
447pub fn base64(bytes: &[u8]) -> String {
448    const DIGITS: &[u8; 64] = b"ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789+/";
449    let mut out = String::with_capacity(bytes.len().div_ceil(3) * 4);
450    for part in bytes.chunks(3) {
451        let a = part[0];
452        let b = part.get(1).copied().unwrap_or(0);
453        let c = part.get(2).copied().unwrap_or(0);
454        out.push(DIGITS[(a >> 2) as usize] as char);
455        out.push(DIGITS[(((a & 3) << 4) | (b >> 4)) as usize] as char);
456        out.push(if part.len() > 1 {
457            DIGITS[(((b & 15) << 2) | (c >> 6)) as usize] as char
458        } else {
459            '='
460        });
461        out.push(if part.len() > 2 {
462            DIGITS[(c & 63) as usize] as char
463        } else {
464            '='
465        });
466    }
467    out
468}
agent_core/rpc.rs

agent_core/
rpc.rs
