yapi_core/mcp/
client.rs

1//! An MCP client over one transport: the `initialize` handshake, requests with
2//! timeouts, progress and cancellation, server requests (`ping`,
3//! `roots/list`) and notifications. Port of `client.ts` in pi-mcp `v1.0.0`.
4
5use std::collections::HashMap;
6use std::sync::atomic::{AtomicU64, Ordering};
7use std::sync::{Arc, Mutex};
8use std::time::Duration;
9
10use serde::Deserialize;
11use serde_json::{Map, Value, json};
12use tokio::sync::{Notify, mpsc, oneshot};
13use tokio_util::sync::CancellationToken;
14use yapi_types::sync::lock;
15
16use super::jsonrpc::{Incoming, METHOD_NOT_FOUND, McpError, classify, id_key, is_id};
17use super::transport::{Event, Transport};
18
19/// The protocol version the client asks for.
20pub const LATEST_PROTOCOL_VERSION: &str = "2025-11-25";
21/// Versions accepted from a server. A server that does not support the
22/// requested version answers with its own latest one.
23pub const SUPPORTED_PROTOCOL_VERSIONS: [&str; 4] = [
24    LATEST_PROTOCOL_VERSION,
25    "2025-06-18",
26    "2025-03-26",
27    "2024-11-05",
28];
29const MAX_LIST_PAGES: usize = 1_000;
30
31/// A directory the client exposes to the server.
32#[derive(Clone, Debug, PartialEq)]
33pub struct Root {
34    /// A `file://` URI.
35    pub uri: String,
36    /// A display name.
37    pub name: Option<String>,
38}
39
40/// Who the client is and how it behaves.
41#[derive(Clone, Debug, PartialEq)]
42pub struct ClientOptions {
43    /// Client name, sent in `initialize`.
44    pub name: String,
45    /// Client version.
46    pub version: String,
47    /// Default per-request timeout. Zero waits forever.
48    pub request_timeout: Duration,
49    /// Roots answered to `roots/list`. With any, the client declares the
50    /// `roots` capability.
51    pub roots: Vec<Root>,
52}
53
54/// Receives the parameters of `notifications/progress` for one request.
55pub type ProgressListener = Arc<dyn Fn(&Value) + Send + Sync>;
56
57/// Per-request options.
58#[derive(Clone, Default)]
59pub struct RequestOptions {
60    /// Cancels the request and notifies the server.
61    pub cancel: Option<CancellationToken>,
62    /// Overrides the client's timeout. Progress notifications restart it.
63    pub timeout: Option<Duration>,
64    /// Asks for progress notifications and receives them.
65    pub on_progress: Option<ProgressListener>,
66}
67
68/// A tool a server offers.
69#[derive(Clone, Debug, PartialEq, Deserialize)]
70#[serde(rename_all = "camelCase")]
71pub struct Tool {
72    /// Name as the server knows it.
73    pub name: String,
74    /// Display title.
75    #[serde(default)]
76    pub title: Option<String>,
77    /// What it does.
78    #[serde(default)]
79    pub description: Option<String>,
80    /// JSON schema of the arguments.
81    pub input_schema: Map<String, Value>,
82    /// JSON schema of `structuredContent`.
83    #[serde(default)]
84    pub output_schema: Option<Map<String, Value>>,
85    /// Hints such as `readOnlyHint`.
86    #[serde(default)]
87    pub annotations: Option<Map<String, Value>>,
88}
89
90/// What the server said in `initialize`.
91#[derive(Clone, Debug, PartialEq)]
92pub struct ServerInfo {
93    /// The negotiated protocol version.
94    pub protocol_version: String,
95    /// Server capabilities.
96    pub capabilities: Map<String, Value>,
97    /// Instructions for using the server's tools.
98    pub instructions: Option<String>,
99}
100
101#[derive(Clone, Copy, Debug, PartialEq, Eq)]
102enum State {
103    Connecting,
104    Connected,
105    Closed,
106}
107
108struct Pending {
109    reply: oneshot::Sender<Result<Value, McpError>>,
110    progress: Option<ProgressListener>,
111    touched: Arc<Notify>,
112}
113
114type Listener = Arc<dyn Fn(&Value) + Send + Sync>;
115
116struct Inner {
117    options: ClientOptions,
118    transport: Transport,
119    state: Mutex<State>,
120    next_id: AtomicU64,
121    pending: Mutex<HashMap<String, Pending>>,
122    server: Mutex<Option<ServerInfo>>,
123    listeners: Mutex<Vec<(String, Listener)>>,
124    close_listeners: Mutex<Vec<Arc<dyn Fn() + Send + Sync>>>,
125}
126
127/// A connected client. Cheap to clone, and clones share the connection.
128#[derive(Clone)]
129pub struct McpClient {
130    inner: Arc<Inner>,
131}
132
133fn validate_initialize(value: &Value) -> Result<ServerInfo, McpError> {
134    let invalid = || McpError::invalid("Invalid MCP initialize result");
135    let object = value.as_object().ok_or_else(invalid)?;
136    let version = object
137        .get("protocolVersion")
138        .and_then(Value::as_str)
139        .ok_or_else(invalid)?;
140    let capabilities = object
141        .get("capabilities")
142        .and_then(Value::as_object)
143        .ok_or_else(invalid)?;
144    let info = object
145        .get("serverInfo")
146        .and_then(Value::as_object)
147        .ok_or_else(invalid)?;
148    if !info.get("name").is_some_and(Value::is_string)
149        || !info.get("version").is_some_and(Value::is_string)
150    {
151        return Err(invalid());
152    }
153    let instructions = match object.get("instructions") {
154        None => None,
155        Some(Value::String(text)) => Some(text.clone()),
156        Some(_) => return Err(invalid()),
157    };
158    Ok(ServerInfo {
159        protocol_version: version.to_owned(),
160        capabilities: capabilities.clone(),
161        instructions,
162    })
163}
164
165/// One page of a list result: the items under `key`, each checked by `valid`.
166fn list_page(
167    method: &str,
168    key: &str,
169    value: &Value,
170    valid: fn(&Map<String, Value>) -> bool,
171) -> Result<(Vec<Value>, Option<String>), McpError> {
172    let items = value
173        .get(key)
174        .and_then(Value::as_array)
175        .ok_or_else(|| McpError::invalid(format!("Invalid MCP {method} result")))?;
176    for item in items {
177        if !item.as_object().is_some_and(valid) {
178            return Err(McpError::invalid(format!(
179                "Invalid entry in MCP {method} result"
180            )));
181        }
182    }
183    // Some servers end pagination with `null` or `""` instead of omitting it.
184    let cursor = match value.get("nextCursor") {
185        None | Some(Value::Null) => None,
186        Some(Value::String(text)) if text.is_empty() => None,
187        Some(Value::String(text)) => Some(text.clone()),
188        Some(_) => {
189            return Err(McpError::invalid(format!("Invalid MCP {method} cursor")));
190        }
191    };
192    Ok((items.clone(), cursor))
193}
194
195fn is_tool(tool: &Map<String, Value>) -> bool {
196    tool.get("name").is_some_and(Value::is_string)
197        && tool.get("inputSchema").is_some_and(Value::is_object)
198}
199
200/// The spec requires `name`, but some servers omit it.
201fn is_resource(resource: &Map<String, Value>) -> bool {
202    resource.get("uri").is_some_and(Value::is_string)
203        && resource.get("name").is_none_or(Value::is_string)
204}
205
206fn is_resource_template(template: &Map<String, Value>) -> bool {
207    template.get("uriTemplate").is_some_and(Value::is_string)
208        && template.get("name").is_none_or(Value::is_string)
209}
210
211/// What a resource listing lists.
212#[derive(Clone, Copy, Debug, PartialEq, Eq)]
213pub enum ResourceKind {
214    /// Resources, from `resources/list`.
215    Resources,
216    /// Resource templates, from `resources/templates/list`.
217    Templates,
218}
219
220impl ResourceKind {
221    fn method(self) -> &'static str {
222        match self {
223            ResourceKind::Resources => "resources/list",
224            ResourceKind::Templates => "resources/templates/list",
225        }
226    }
227
228    /// The key of the items in a listing result.
229    pub fn key(self) -> &'static str {
230        match self {
231            ResourceKind::Resources => "resources",
232            ResourceKind::Templates => "resourceTemplates",
233        }
234    }
235
236    /// The field that locates an item, which also names items without a name.
237    fn uri_field(self) -> &'static str {
238        match self {
239            ResourceKind::Resources => "uri",
240            ResourceKind::Templates => "uriTemplate",
241        }
242    }
243
244    fn valid(self) -> fn(&Map<String, Value>) -> bool {
245        match self {
246            ResourceKind::Resources => is_resource,
247            ResourceKind::Templates => is_resource_template,
248        }
249    }
250}
251
252/// Fills a missing `name` from `field`, as pi-mcp does.
253fn with_name(mut item: Value, field: &str) -> Value {
254    if item.get("name").is_none()
255        && let Some(fallback) = item.get(field).cloned()
256        && let Some(object) = item.as_object_mut()
257    {
258        object.insert("name".into(), fallback);
259    }
260    item
261}
262
263/// `items` with names from their `field` where they lack one.
264fn named(items: Vec<Value>, field: &str) -> Vec<Value> {
265    items
266        .into_iter()
267        .map(|item| with_name(item, field))
268        .collect()
269}
270
271impl McpClient {
272    /// Opens `transport` and performs the `initialize` handshake.
273    pub async fn connect(
274        options: ClientOptions,
275        transport: Transport,
276    ) -> Result<McpClient, McpError> {
277        Self::connect_with_stderr(options, transport)
278            .await
279            .map_err(|failure| failure.0)
280    }
281
282    /// [`McpClient::connect`], failing with a stdio server's stderr too, as
283    /// pi reads it for every failed connection.
284    pub async fn connect_with_stderr(
285        options: ClientOptions,
286        transport: Transport,
287    ) -> Result<McpClient, Box<(McpError, Option<String>)>> {
288        let (events, receiver) = mpsc::unbounded_channel();
289        let client = McpClient {
290            inner: Arc::new(Inner {
291                options,
292                transport,
293                state: Mutex::new(State::Connecting),
294                next_id: AtomicU64::new(1),
295                pending: Mutex::new(HashMap::new()),
296                server: Mutex::new(None),
297                listeners: Mutex::new(Vec::new()),
298                close_listeners: Mutex::new(Vec::new()),
299            }),
300        };
301        tokio::spawn(client.clone().dispatch(receiver));
302        match client.handshake(events).await {
303            Ok(()) => Ok(client),
304            Err(error) => {
305                client.close().await;
306                Err(Box::new((error, client.stderr())))
307            }
308        }
309    }
310
311    async fn handshake(&self, events: super::transport::Events) -> Result<(), McpError> {
312        self.inner.transport.start(events).await?;
313        let mut capabilities = Map::new();
314        if !self.inner.options.roots.is_empty() {
315            capabilities.insert("roots".into(), json!({}));
316        }
317        let params = json!({
318            "protocolVersion": LATEST_PROTOCOL_VERSION,
319            "capabilities": capabilities,
320            "clientInfo": {"name": self.inner.options.name, "version": self.inner.options.version},
321        });
322        let result = self
323            .request_internal("initialize", Some(params), RequestOptions::default(), true)
324            .await?;
325        let info = validate_initialize(&result)?;
326        if !SUPPORTED_PROTOCOL_VERSIONS.contains(&info.protocol_version.as_str()) {
327            return Err(McpError::Other(format!(
328                "MCP server selected unsupported protocol version {}",
329                info.protocol_version
330            )));
331        }
332        self.inner
333            .transport
334            .set_protocol_version(&info.protocol_version);
335        *lock(&self.inner.server) = Some(info);
336        self.notify_internal("notifications/initialized", None, true)
337            .await?;
338        *lock(&self.inner.state) = State::Connected;
339        Ok(())
340    }
341
342    /// Whether `other` is a clone of this client.
343    pub fn same(&self, other: &McpClient) -> bool {
344        Arc::ptr_eq(&self.inner, &other.inner)
345    }
346
347    /// Whether the connection is open.
348    pub fn is_connected(&self) -> bool {
349        *lock(&self.inner.state) == State::Connected
350    }
351
352    /// What the server said in `initialize`.
353    pub fn server(&self) -> Option<ServerInfo> {
354        lock(&self.inner.server).clone()
355    }
356
357    /// The end of a stdio server's stderr.
358    pub fn stderr(&self) -> Option<String> {
359        self.inner.transport.stderr()
360    }
361
362    /// Calls `listener` with the parameters of every `method` notification.
363    pub fn on_notification(&self, method: &str, listener: impl Fn(&Value) + Send + Sync + 'static) {
364        lock(&self.inner.listeners).push((method.to_owned(), Arc::new(listener)));
365    }
366
367    /// Calls `listener` once when the connection closes, for any reason.
368    pub fn on_close(&self, listener: impl Fn() + Send + Sync + 'static) {
369        lock(&self.inner.close_listeners).push(Arc::new(listener));
370    }
371
372    /// Sends a request and waits for its result.
373    pub async fn request(
374        &self,
375        method: &str,
376        params: Option<Value>,
377        options: RequestOptions,
378    ) -> Result<Value, McpError> {
379        self.request_internal(method, params, options, false).await
380    }
381
382    /// Every tool, following pagination.
383    pub async fn list_tools(&self, options: RequestOptions) -> Result<Vec<Tool>, McpError> {
384        let items = self
385            .list_all("tools/list", "tools", is_tool, options)
386            .await?;
387        items
388            .into_iter()
389            .map(|item| {
390                serde_json::from_value(item)
391                    .map_err(|_| McpError::invalid("Invalid entry in MCP tools/list result"))
392            })
393            .collect()
394    }
395
396    /// Every resource or resource template, following pagination.
397    pub async fn list_resources(
398        &self,
399        kind: ResourceKind,
400        options: RequestOptions,
401    ) -> Result<Vec<Value>, McpError> {
402        let items = self
403            .list_all(kind.method(), kind.key(), kind.valid(), options)
404            .await?;
405        Ok(named(items, kind.uri_field()))
406    }
407
408    /// One page of resources or resource templates from `cursor`, and the
409    /// next cursor.
410    pub async fn list_resources_page(
411        &self,
412        kind: ResourceKind,
413        cursor: Option<String>,
414        options: RequestOptions,
415    ) -> Result<(Vec<Value>, Option<String>), McpError> {
416        let (items, next) = self
417            .list_page(kind.method(), kind.key(), kind.valid(), cursor, options)
418            .await?;
419        Ok((named(items, kind.uri_field()), next))
420    }
421
422    /// Reads one resource.
423    pub async fn read_resource(
424        &self,
425        uri: &str,
426        options: RequestOptions,
427    ) -> Result<Value, McpError> {
428        let result = self
429            .request("resources/read", Some(json!({ "uri": uri })), options)
430            .await?;
431        let contents = result
432            .get("contents")
433            .and_then(Value::as_array)
434            .ok_or_else(|| McpError::invalid("Invalid MCP resources/read result"))?;
435        for item in contents {
436            let valid = item.get("uri").is_some_and(Value::is_string)
437                && (item.get("text").is_some_and(Value::is_string)
438                    || item.get("blob").is_some_and(Value::is_string));
439            if !valid {
440                return Err(McpError::invalid(
441                    "Invalid contents in MCP resources/read result",
442                ));
443            }
444        }
445        Ok(result)
446    }
447
448    /// Calls a tool. A result without `content` gets an empty one, as the SDKs
449    /// default it.
450    pub async fn call_tool(
451        &self,
452        name: &str,
453        arguments: Value,
454        options: RequestOptions,
455    ) -> Result<Value, McpError> {
456        let mut params = json!({ "name": name });
457        if !arguments.is_null() {
458            params["arguments"] = arguments;
459        }
460        let mut result = self.request("tools/call", Some(params), options).await?;
461        let object = result
462            .as_object_mut()
463            .filter(|object| object.get("content").is_none_or(Value::is_array))
464            .ok_or_else(|| McpError::invalid("Invalid MCP tools/call result"))?;
465        if object
466            .get("structuredContent")
467            .is_some_and(|value| !value.is_object())
468        {
469            return Err(McpError::invalid(
470                "Invalid MCP tools/call structured content",
471            ));
472        }
473        if !object.contains_key("content") {
474            object.insert("content".into(), json!([]));
475        }
476        Ok(result)
477    }
478
479    /// Closes the connection. Requests in flight fail.
480    pub async fn close(&self) {
481        self.mark_closed();
482        self.inner.transport.close().await;
483    }
484
485    async fn list_page(
486        &self,
487        method: &str,
488        key: &str,
489        valid: fn(&Map<String, Value>) -> bool,
490        cursor: Option<String>,
491        options: RequestOptions,
492    ) -> Result<(Vec<Value>, Option<String>), McpError> {
493        let params = cursor.map(|cursor| json!({ "cursor": cursor }));
494        let result = self.request(method, params, options).await?;
495        list_page(method, key, &result, valid)
496    }
497
498    async fn list_all(
499        &self,
500        method: &str,
501        key: &str,
502        valid: fn(&Map<String, Value>) -> bool,
503        options: RequestOptions,
504    ) -> Result<Vec<Value>, McpError> {
505        let mut items = Vec::new();
506        let mut seen = std::collections::HashSet::new();
507        let mut cursor = None;
508        for _ in 0..MAX_LIST_PAGES {
509            let (page, next) = self
510                .list_page(method, key, valid, cursor, options.clone())
511                .await?;
512            items.extend(page);
513            let Some(next) = next else {
514                return Ok(items);
515            };
516            if !seen.insert(next.clone()) {
517                return Err(McpError::Other(format!(
518                    "MCP {method} returned duplicate cursor: {next}"
519                )));
520            }
521            cursor = Some(next);
522        }
523        Err(McpError::Other(format!(
524            "MCP {method} exceeded {MAX_LIST_PAGES} pages"
525        )))
526    }
527
528    fn require(&self, allow_connecting: bool) -> Result<(), McpError> {
529        let state = *lock(&self.inner.state);
530        match state {
531            State::Connected => Ok(()),
532            State::Connecting if allow_connecting => Ok(()),
533            State::Connecting => Err(McpError::Closed("MCP client is connecting".into())),
534            State::Closed => Err(McpError::Closed("MCP client is closed".into())),
535        }
536    }
537
538    async fn request_internal(
539        &self,
540        method: &str,
541        params: Option<Value>,
542        options: RequestOptions,
543        allow_connecting: bool,
544    ) -> Result<Value, McpError> {
545        self.require(allow_connecting)?;
546        let cancel = options.cancel.clone().unwrap_or_default();
547        if cancel.is_cancelled() {
548            return Err(McpError::Aborted);
549        }
550        let id = self.inner.next_id.fetch_add(1, Ordering::SeqCst);
551        let key = id_key(&json!(id));
552        let params = match (&options.on_progress, params) {
553            (Some(_), params) => {
554                let mut params = params.unwrap_or_else(|| json!({}));
555                let mut meta = params
556                    .get("_meta")
557                    .and_then(Value::as_object)
558                    .cloned()
559                    .unwrap_or_default();
560                meta.insert("progressToken".into(), json!(id));
561                params["_meta"] = Value::Object(meta);
562                Some(params)
563            }
564            (None, params) => params,
565        };
566        let mut message = json!({"jsonrpc": "2.0", "id": id, "method": method});
567        if let Some(params) = params {
568            message["params"] = params;
569        }
570        let (reply, mut result) = oneshot::channel();
571        let touched = Arc::new(Notify::new());
572        lock(&self.inner.pending).insert(
573            key.clone(),
574            Pending {
575                reply,
576                progress: options.on_progress.clone(),
577                touched: Arc::clone(&touched),
578            },
579        );
580        // The send runs on its own, as pi's does, so the timeout and
581        // cancellation apply while a server is slow to take the message. A
582        // stdio write is never dropped half way. Its failure fails the request.
583        let client = self.clone();
584        let sent = key.clone();
585        tokio::spawn(async move {
586            if let Err(error) = client.inner.transport.send(&message).await
587                && let Some(entry) = client.remove_pending(&sent)
588            {
589                let _ = entry.reply.send(Err(error));
590            }
591        });
592        let timeout = options
593            .timeout
594            .unwrap_or(self.inner.options.request_timeout);
595        // The spec forbids cancelling `initialize`.
596        let cancellable = method != "initialize";
597        loop {
598            let expired = async {
599                if timeout.is_zero() {
600                    std::future::pending::<()>().await;
601                }
602                tokio::time::sleep(timeout).await;
603            };
604            tokio::select! {
605                outcome = &mut result => {
606                    return outcome.unwrap_or_else(|_| Err(McpError::closed()));
607                }
608                () = touched.notified() => {}
609                () = expired => {
610                    self.remove_pending(&key);
611                    if cancellable {
612                        self.send_cancelled(id, "Request timed out");
613                    }
614                    return Err(McpError::Timeout(timeout.as_millis() as u64));
615                }
616                () = cancel.cancelled() => {
617                    self.remove_pending(&key);
618                    if cancellable {
619                        self.send_cancelled(id, "AbortError: This operation was aborted");
620                    }
621                    return Err(McpError::Aborted);
622                }
623            }
624        }
625    }
626
627    fn send_cancelled(&self, id: u64, reason: &str) {
628        let client = self.clone();
629        let message = json!({
630            "jsonrpc": "2.0",
631            "method": "notifications/cancelled",
632            "params": {"requestId": id, "reason": reason},
633        });
634        tokio::spawn(async move {
635            if !matches!(*lock(&client.inner.state), State::Closed) {
636                let _ = client.inner.transport.send(&message).await;
637            }
638        });
639    }
640
641    async fn notify_internal(
642        &self,
643        method: &str,
644        params: Option<Value>,
645        allow_connecting: bool,
646    ) -> Result<(), McpError> {
647        self.require(allow_connecting)?;
648        let mut message = json!({"jsonrpc": "2.0", "method": method});
649        if let Some(params) = params {
650            message["params"] = params;
651        }
652        self.inner.transport.send(&message).await
653    }
654
655    fn remove_pending(&self, key: &str) -> Option<Pending> {
656        lock(&self.inner.pending).remove(key)
657    }
658
659    /// Rejects requests in flight and reports the close once.
660    fn mark_closed(&self) {
661        let was = std::mem::replace(&mut *lock(&self.inner.state), State::Closed);
662        let pending: Vec<Pending> = lock(&self.inner.pending).drain().map(|(_, p)| p).collect();
663        for entry in pending {
664            let _ = entry.reply.send(Err(McpError::closed()));
665        }
666        if was != State::Closed {
667            let listeners = lock(&self.inner.close_listeners).clone();
668            for listener in listeners {
669                listener();
670            }
671        }
672    }
673
674    async fn dispatch(self, mut events: mpsc::UnboundedReceiver<Event>) {
675        while let Some(event) = events.recv().await {
676            match event {
677                Event::Message(message) => self.handle(&message),
678                // pi-mcp reports these to error listeners, which pi does not add.
679                Event::Error(_) => {}
680                Event::Closed => {
681                    self.mark_closed();
682                    break;
683                }
684            }
685        }
686    }
687
688    fn handle(&self, message: &Value) {
689        match classify(message) {
690            Some(Incoming::Response { id, outcome }) => {
691                let Some(entry) = self.remove_pending(&id_key(id)) else {
692                    return;
693                };
694                let _ = entry.reply.send(outcome.cloned());
695            }
696            Some(Incoming::Request { id, method, .. }) => {
697                let result = match method {
698                    "ping" => Ok(json!({})),
699                    "roots/list" if !self.inner.options.roots.is_empty() => {
700                        let roots: Vec<Value> = self
701                            .inner
702                            .options
703                            .roots
704                            .iter()
705                            .map(|root| match &root.name {
706                                Some(name) => json!({"uri": root.uri, "name": name}),
707                                None => json!({"uri": root.uri}),
708                            })
709                            .collect();
710                        Ok(json!({ "roots": roots }))
711                    }
712                    other => Err(json!({
713                        "code": METHOD_NOT_FOUND,
714                        "message": format!("Method not found: {other}"),
715                    })),
716                };
717                let response = match result {
718                    Ok(result) => json!({"jsonrpc": "2.0", "id": id, "result": result}),
719                    Err(error) => json!({"jsonrpc": "2.0", "id": id, "error": error}),
720                };
721                let client = self.clone();
722                tokio::spawn(async move {
723                    let _ = client.inner.transport.send(&response).await;
724                });
725            }
726            Some(Incoming::Notification { method, params }) => {
727                let params = params.cloned().unwrap_or(Value::Null);
728                if method == "notifications/progress" {
729                    self.progress(&params);
730                }
731                let listeners: Vec<Listener> = lock(&self.inner.listeners)
732                    .iter()
733                    .filter(|(name, _)| name == method)
734                    .map(|(_, listener)| Arc::clone(listener))
735                    .collect();
736                for listener in listeners {
737                    listener(&params);
738                }
739            }
740            None => {}
741        }
742    }
743
744    fn progress(&self, params: &Value) {
745        let Some(token) = params.get("progressToken").filter(|token| is_id(token)) else {
746            return;
747        };
748        if !params.get("progress").is_some_and(Value::is_number) {
749            return;
750        }
751        // The progress token is the id of a request that asked for progress.
752        let (touched, listener) = match lock(&self.inner.pending).get(&id_key(token)) {
753            Some(Pending {
754                touched,
755                progress: Some(listener),
756                ..
757            }) => (Arc::clone(touched), Arc::clone(listener)),
758            _ => return,
759        };
760        touched.notify_one();
761        listener(params);
762    }
763}
yapi_core/mcp/client.rs

yapi_core/mcp/
client.rs
