perspt_agent/external_tools/
client.rs

1//! Modern MCP 2026-07-28 client built on the official Rust SDK.
2
3#![allow(deprecated)]
4
5use std::collections::HashMap;
6use std::process::Stdio;
7use std::sync::atomic::{AtomicU64, Ordering};
8use std::sync::{Arc, Mutex};
9use std::time::Duration;
10
11use anyhow::{Context, Result};
12use async_trait::async_trait;
13use rmcp::model::{
14    CallToolRequestParams, CallToolResponse, CallToolResult, CancelledNotificationParam,
15    ClientCapabilities, ClientInfo, ConstString, CreateMessageRequestParams, CreateMessageResult,
16    CustomNotification, CustomRequest, CustomResult, ElicitRequestParams, ElicitResult,
17    ElicitationCapability, FormElicitationCapability, GetPromptRequestParams, GetPromptResult,
18    GetTaskParams, Implementation, ListRootsResult, LoggingMessageNotificationParam,
19    ProgressNotificationParam, ProtocolVersion, ReadResourceRequestParams, ReadResourceResult,
20    ResourceUpdatedNotificationParam, Root, SamplingCapability, ServerNotification,
21    SubscriptionFilter, SubscriptionsAcknowledgedNotificationParams, TaskPayload,
22    UrlElicitationCapability, TASKS_EXTENSION_ID,
23};
24use rmcp::service::{NotificationContext, RequestContext, RoleClient, RunningService};
25use rmcp::transport::streamable_http_client::StreamableHttpClientTransportConfig;
26use rmcp::transport::{StreamableHttpClientTransport, TokioChildProcess};
27use rmcp::{ClientHandler, ClientLifecycleMode, ClientServiceExt, ErrorData as McpError};
28use tokio::io::AsyncReadExt;
29
30use perspt_core::{ExternalToolConfig, ExternalToolTransport};
31use perspt_sdk::{
32    Conversation, GenerationOptions, Message, ModelId, ModelTransport, ProviderToolCall,
33    ToolChoicePolicy, ToolSpec, TurnOutput,
34};
35
36pub const MCP_PROTOCOL_VERSION: &str = "2026-07-28";
37
38/// Product-owned model service for server-initiated MCP sampling.
39#[async_trait]
40pub trait McpSamplingProvider: Send + Sync {
41    async fn create_message(
42        &self,
43        server_id: &str,
44        request: CreateMessageRequestParams,
45    ) -> Result<CreateMessageResult>;
46}
47
48/// Product-owned interaction service for server-initiated MCP elicitation.
49#[async_trait]
50pub trait McpElicitationProvider: Send + Sync {
51    async fn elicit(&self, server_id: &str, request: ElicitRequestParams) -> Result<ElicitResult>;
52}
53
54/// Non-interactive agent policy: acknowledge the protocol request and decline
55/// it immediately instead of hanging a run that has no attached input UI.
56#[derive(Debug, Default)]
57pub struct DecliningMcpElicitationProvider;
58
59#[async_trait]
60impl McpElicitationProvider for DecliningMcpElicitationProvider {
61    async fn elicit(
62        &self,
63        _server_id: &str,
64        _request: ElicitRequestParams,
65    ) -> Result<ElicitResult> {
66        Ok(ElicitResult::new(rmcp::model::ElicitationAction::Decline))
67    }
68}
69
70#[derive(Debug, Clone, Copy, PartialEq, Eq)]
71pub enum McpElicitationAction {
72    Accept,
73    Decline,
74    Cancel,
75}
76
77#[derive(Debug, Clone)]
78pub struct McpPendingElicitation {
79    pub id: u64,
80    pub server_id: String,
81    pub request: serde_json::Value,
82}
83
84struct ElicitationBrokerInner {
85    next_id: AtomicU64,
86    pending: Mutex<HashMap<u64, tokio::sync::oneshot::Sender<ElicitResult>>>,
87    requests: tokio::sync::mpsc::UnboundedSender<McpPendingElicitation>,
88}
89
90/// A product-neutral interactive elicitation bridge. The protocol task waits
91/// on a one-shot response while a TUI or SDK host drains requests and answers
92/// them explicitly.
93#[derive(Clone)]
94pub struct McpElicitationBroker {
95    inner: Arc<ElicitationBrokerInner>,
96    receiver: Arc<Mutex<tokio::sync::mpsc::UnboundedReceiver<McpPendingElicitation>>>,
97}
98
99impl Default for McpElicitationBroker {
100    fn default() -> Self {
101        Self::new()
102    }
103}
104
105impl McpElicitationBroker {
106    pub fn new() -> Self {
107        let (requests, receiver) = tokio::sync::mpsc::unbounded_channel();
108        Self {
109            inner: Arc::new(ElicitationBrokerInner {
110                next_id: AtomicU64::new(1),
111                pending: Mutex::new(HashMap::new()),
112                requests,
113            }),
114            receiver: Arc::new(Mutex::new(receiver)),
115        }
116    }
117
118    pub fn try_next(&self) -> Option<McpPendingElicitation> {
119        self.receiver
120            .lock()
121            .expect("MCP elicitation receiver poisoned")
122            .try_recv()
123            .ok()
124    }
125
126    pub fn respond(
127        &self,
128        id: u64,
129        action: McpElicitationAction,
130        content: Option<serde_json::Value>,
131    ) -> Result<()> {
132        let sender = self
133            .inner
134            .pending
135            .lock()
136            .expect("MCP elicitation map poisoned")
137            .remove(&id)
138            .with_context(|| format!("unknown or already answered MCP elicitation {id}"))?;
139        let action = match action {
140            McpElicitationAction::Accept => rmcp::model::ElicitationAction::Accept,
141            McpElicitationAction::Decline => rmcp::model::ElicitationAction::Decline,
142            McpElicitationAction::Cancel => rmcp::model::ElicitationAction::Cancel,
143        };
144        let mut response = ElicitResult::new(action);
145        if let Some(content) = content {
146            response = response.with_content(content);
147        }
148        sender
149            .send(response)
150            .map_err(|_| anyhow::anyhow!("MCP elicitation requester disconnected"))
151    }
152}
153
154#[async_trait]
155impl McpElicitationProvider for McpElicitationBroker {
156    async fn elicit(&self, server_id: &str, request: ElicitRequestParams) -> Result<ElicitResult> {
157        let id = self.inner.next_id.fetch_add(1, Ordering::Relaxed);
158        let request_value = serde_json::to_value(&request)?;
159        let (sender, receiver) = tokio::sync::oneshot::channel();
160        self.inner
161            .pending
162            .lock()
163            .expect("MCP elicitation map poisoned")
164            .insert(id, sender);
165        if self
166            .inner
167            .requests
168            .send(McpPendingElicitation {
169                id,
170                server_id: server_id.to_string(),
171                request: request_value,
172            })
173            .is_err()
174        {
175            self.inner
176                .pending
177                .lock()
178                .expect("MCP elicitation map poisoned")
179                .remove(&id);
180            anyhow::bail!("MCP elicitation UI is unavailable");
181        }
182        receiver
183            .await
184            .context("MCP elicitation was cancelled because the UI closed")
185    }
186}
187
188#[derive(Clone, Default)]
189pub struct McpClientServices {
190    pub sampling: Option<Arc<dyn McpSamplingProvider>>,
191    pub elicitation: Option<Arc<dyn McpElicitationProvider>>,
192}
193
194/// Sampling adapter used by the governed agent. The requesting server may
195/// suggest a model, but Perspt keeps route selection local and uses the model
196/// already selected for the product lifecycle.
197pub struct ModelTransportSamplingProvider {
198    transport: Arc<dyn ModelTransport>,
199    model: ModelId,
200}
201
202impl ModelTransportSamplingProvider {
203    pub fn new(transport: Arc<dyn ModelTransport>, model: ModelId) -> Self {
204        Self { transport, model }
205    }
206}
207
208#[async_trait]
209impl McpSamplingProvider for ModelTransportSamplingProvider {
210    async fn create_message(
211        &self,
212        _server_id: &str,
213        request: CreateMessageRequestParams,
214    ) -> Result<CreateMessageResult> {
215        request.validate().map_err(|error| anyhow::anyhow!(error))?;
216        anyhow::ensure!(
217            request.include_context.is_none()
218                || request.include_context == Some(rmcp::model::ContextInclusion::None),
219            "MCP sampling includeContext is not advertised by Perspt"
220        );
221        let generation = GenerationOptions {
222            max_tokens: Some(request.max_tokens),
223            temperature: request.temperature,
224            stop_sequences: request.stop_sequences.clone().unwrap_or_default(),
225        };
226        let mut conversation = request
227            .system_prompt
228            .as_deref()
229            .map(Conversation::with_system)
230            .unwrap_or_default();
231        for message in request.messages {
232            append_sample(&mut conversation, message)?;
233        }
234        let tools = request
235            .tools
236            .unwrap_or_default()
237            .into_iter()
238            .map(|tool| ToolSpec {
239                name: tool.name.into_owned(),
240                description: tool
241                    .description
242                    .map(|description| description.into_owned())
243                    .unwrap_or_default(),
244                schema: serde_json::Value::Object(tool.input_schema.as_ref().clone()),
245                strict: false,
246            })
247            .collect::<Vec<_>>();
248        let choice = request
249            .tool_choice
250            .and_then(|choice| choice.mode)
251            .map(|mode| match mode {
252                rmcp::model::ToolChoiceMode::Required => ToolChoicePolicy::Required,
253                rmcp::model::ToolChoiceMode::None => ToolChoicePolicy::None,
254                _ => ToolChoicePolicy::Auto,
255            })
256            .unwrap_or(ToolChoicePolicy::Auto);
257        let output = self
258            .transport
259            .chat_turn_with_options(&self.model, &conversation, &tools, choice, generation)
260            .await
261            .map_err(|error| anyhow::anyhow!(error.to_string()))?;
262        sampling_result(output, self.model.to_string())
263    }
264}
265
266fn append_sample(
267    conversation: &mut Conversation,
268    message: rmcp::model::SamplingMessage,
269) -> Result<()> {
270    let mut texts = Vec::new();
271    let mut calls = Vec::new();
272    let mut results = Vec::new();
273    for content in message.content.into_vec() {
274        match content {
275            rmcp::model::SamplingMessageContentBlock::Text(text) => texts.push(text.text),
276            rmcp::model::SamplingMessageContentBlock::ToolUse(tool) => {
277                calls.push(ProviderToolCall {
278                    call_id: tool.id,
279                    name: tool.name,
280                    arguments: serde_json::Value::Object(tool.input),
281                });
282            }
283            rmcp::model::SamplingMessageContentBlock::ToolResult(result) => {
284                results.push((result.tool_use_id, serde_json::to_string(&result.content)?));
285            }
286            rmcp::model::SamplingMessageContentBlock::Image(_)
287            | rmcp::model::SamplingMessageContentBlock::Audio(_) => {
288                anyhow::bail!(
289                    "the Perspt model transport does not accept MCP media sampling content"
290                )
291            }
292            _ => anyhow::bail!("unknown MCP sampling content block"),
293        }
294    }
295    if !texts.is_empty() {
296        conversation.push(match message.role {
297            rmcp::model::Role::User => Message::User {
298                content: texts.join("\n"),
299            },
300            rmcp::model::Role::Assistant => Message::Assistant {
301                content: texts.join("\n"),
302            },
303        });
304    }
305    if !calls.is_empty() {
306        conversation.push_tool_calls(calls);
307    }
308    for (call_id, content) in results {
309        conversation.push_tool_response(call_id, content);
310    }
311    Ok(())
312}
313
314fn sampling_result(output: TurnOutput, model: String) -> Result<CreateMessageResult> {
315    let result = match output {
316        TurnOutput::Text(text) => {
317            CreateMessageResult::new(rmcp::model::SamplingMessage::assistant_text(text), model)
318                .with_stop_reason(CreateMessageResult::STOP_REASON_END_TURN)
319        }
320        TurnOutput::ToolCalls(calls) => {
321            let contents = calls
322                .into_iter()
323                .map(|call| {
324                    let input = call.arguments.as_object().cloned().with_context(|| {
325                        format!(
326                            "sampling tool call {:?} arguments were not an object",
327                            call.name
328                        )
329                    })?;
330                    Ok(rmcp::model::SamplingMessageContentBlock::tool_use(
331                        call.call_id,
332                        call.name,
333                        input,
334                    ))
335                })
336                .collect::<Result<Vec<_>>>()?;
337            CreateMessageResult::new(
338                rmcp::model::SamplingMessage::new_multiple(rmcp::model::Role::Assistant, contents),
339                model,
340            )
341            .with_stop_reason(CreateMessageResult::STOP_REASON_TOOL_USE)
342        }
343    };
344    result.validate().map_err(anyhow::Error::msg)?;
345    Ok(result)
346}
347
348impl std::fmt::Debug for McpClientServices {
349    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
350        formatter
351            .debug_struct("McpClientServices")
352            .field("sampling", &self.sampling.is_some())
353            .field("elicitation", &self.elicitation.is_some())
354            .finish()
355    }
356}
357
358#[derive(Debug, Clone)]
359pub enum McpServerEvent {
360    Progress(serde_json::Value),
361    Log(serde_json::Value),
362    ResourceUpdated(serde_json::Value),
363    ResourcesChanged,
364    ToolsChanged,
365    PromptsChanged,
366    TaskChanged(serde_json::Value),
367    Cancelled(serde_json::Value),
368    SubscriptionAcknowledged(serde_json::Value),
369    CustomNotification {
370        method: String,
371        params: Option<serde_json::Value>,
372    },
373    SubscriptionEnded(String),
374}
375
376#[derive(Clone)]
377struct PersptClientHandler {
378    server_id: String,
379    info: ClientInfo,
380    roots: Vec<Root>,
381    max_sampling_tokens: u32,
382    services: McpClientServices,
383    events: tokio::sync::mpsc::UnboundedSender<McpServerEvent>,
384}
385
386impl PersptClientHandler {
387    fn emit<T: serde::Serialize>(
388        &self,
389        constructor: impl FnOnce(serde_json::Value) -> McpServerEvent,
390        value: &T,
391    ) {
392        if let Ok(value) = serde_json::to_value(value) {
393            let _ = self.events.send(constructor(value));
394        }
395    }
396
397    async fn task_inputs(
398        &self,
399        requests: rmcp::model::InputRequests,
400    ) -> Result<rmcp::model::InputResponses> {
401        let mut responses = rmcp::model::InputResponses::new();
402        for (key, request) in requests {
403            let value = match request {
404                rmcp::model::InputRequest::CreateMessage(request) => {
405                    anyhow::ensure!(
406                        request.params.max_tokens <= self.max_sampling_tokens,
407                        "task sampling request exceeded local token cap"
408                    );
409                    let provider = self
410                        .services
411                        .sampling
412                        .as_ref()
413                        .context("task requested sampling but sampling is disabled")?;
414                    serde_json::to_value(
415                        provider
416                            .create_message(&self.server_id, request.params)
417                            .await?,
418                    )?
419                }
420                rmcp::model::InputRequest::Elicitation(request) => {
421                    let provider = self
422                        .services
423                        .elicitation
424                        .as_ref()
425                        .context("task requested elicitation but elicitation is disabled")?;
426                    serde_json::to_value(provider.elicit(&self.server_id, request.params).await?)?
427                }
428                rmcp::model::InputRequest::ListRoots(_) => {
429                    serde_json::to_value(ListRootsResult::new(self.roots.clone()))?
430                }
431                _ => anyhow::bail!("MCP task requested an unknown input capability"),
432            };
433            responses.insert(key, value);
434        }
435        Ok(responses)
436    }
437}
438
439impl ClientHandler for PersptClientHandler {
440    async fn create_message(
441        &self,
442        request: CreateMessageRequestParams,
443        _context: RequestContext<RoleClient>,
444    ) -> std::result::Result<CreateMessageResult, McpError> {
445        if request.max_tokens > self.max_sampling_tokens {
446            return Err(McpError::invalid_params(
447                format!(
448                    "sampling request maxTokens {} exceeds local cap {}",
449                    request.max_tokens, self.max_sampling_tokens
450                ),
451                None,
452            ));
453        }
454        let provider = self.services.sampling.as_ref().ok_or_else(|| {
455            McpError::internal_error("MCP sampling provider is unavailable", None)
456        })?;
457        provider
458            .create_message(&self.server_id, request)
459            .await
460            .map_err(|error| McpError::internal_error(error.to_string(), None))
461    }
462
463    async fn list_roots(
464        &self,
465        _context: RequestContext<RoleClient>,
466    ) -> std::result::Result<ListRootsResult, McpError> {
467        Ok(ListRootsResult::new(self.roots.clone()))
468    }
469
470    async fn create_elicitation(
471        &self,
472        request: ElicitRequestParams,
473        _context: RequestContext<RoleClient>,
474    ) -> std::result::Result<ElicitResult, McpError> {
475        let provider = self.services.elicitation.as_ref().ok_or_else(|| {
476            McpError::method_not_found::<rmcp::model::ElicitationCreateRequestMethod>()
477        })?;
478        provider
479            .elicit(&self.server_id, request)
480            .await
481            .map_err(|error| McpError::internal_error(error.to_string(), None))
482    }
483
484    async fn on_custom_request(
485        &self,
486        request: CustomRequest,
487        context: RequestContext<RoleClient>,
488    ) -> std::result::Result<CustomResult, McpError> {
489        // rmcp may preserve SEP-2577-deprecated sampling as a custom request
490        // under the 2026 lifecycle. Route that current wire method through the
491        // same typed handler without negotiating another protocol version.
492        if request.method == rmcp::model::CreateMessageRequestMethod::VALUE {
493            let params = serde_json::from_value(request.params.unwrap_or_default())
494                .map_err(|error| McpError::invalid_params(error.to_string(), None))?;
495            let result = self.create_message(params, context).await?;
496            return serde_json::to_value(result)
497                .map(CustomResult::new)
498                .map_err(|error| McpError::internal_error(error.to_string(), None));
499        }
500        Err(McpError::new(
501            rmcp::model::ErrorCode::METHOD_NOT_FOUND,
502            request.method,
503            None,
504        ))
505    }
506
507    async fn on_progress(
508        &self,
509        params: ProgressNotificationParam,
510        _context: NotificationContext<RoleClient>,
511    ) {
512        self.emit(McpServerEvent::Progress, &params);
513    }
514
515    async fn on_cancelled(
516        &self,
517        params: CancelledNotificationParam,
518        _context: NotificationContext<RoleClient>,
519    ) {
520        self.emit(McpServerEvent::Cancelled, &params);
521    }
522
523    async fn on_logging_message(
524        &self,
525        params: LoggingMessageNotificationParam,
526        _context: NotificationContext<RoleClient>,
527    ) {
528        self.emit(McpServerEvent::Log, &params);
529    }
530
531    async fn on_resource_updated(
532        &self,
533        params: ResourceUpdatedNotificationParam,
534        _context: NotificationContext<RoleClient>,
535    ) {
536        self.emit(McpServerEvent::ResourceUpdated, &params);
537    }
538
539    async fn on_resource_list_changed(&self, _context: NotificationContext<RoleClient>) {
540        let _ = self.events.send(McpServerEvent::ResourcesChanged);
541    }
542
543    async fn on_tool_list_changed(&self, _context: NotificationContext<RoleClient>) {
544        let _ = self.events.send(McpServerEvent::ToolsChanged);
545    }
546
547    async fn on_prompt_list_changed(&self, _context: NotificationContext<RoleClient>) {
548        let _ = self.events.send(McpServerEvent::PromptsChanged);
549    }
550
551    async fn on_task_status(
552        &self,
553        params: rmcp::model::TaskStatusNotificationParams,
554        _context: NotificationContext<RoleClient>,
555    ) {
556        self.emit(McpServerEvent::TaskChanged, &params);
557    }
558
559    async fn on_subscriptions_acknowledged(
560        &self,
561        params: SubscriptionsAcknowledgedNotificationParams,
562        _context: NotificationContext<RoleClient>,
563    ) {
564        self.emit(McpServerEvent::SubscriptionAcknowledged, &params);
565    }
566
567    async fn on_custom_notification(
568        &self,
569        notification: CustomNotification,
570        _context: NotificationContext<RoleClient>,
571    ) {
572        let _ = self.events.send(McpServerEvent::CustomNotification {
573            method: notification.method,
574            params: notification.params,
575        });
576    }
577
578    fn get_info(&self) -> ClientInfo {
579        self.info.clone()
580    }
581}
582
583pub struct McpConnection {
584    client: RunningService<RoleClient, PersptClientHandler>,
585    events: tokio::sync::mpsc::UnboundedReceiver<McpServerEvent>,
586    subscription: Option<tokio::task::JoinHandle<()>>,
587    stderr: Arc<Mutex<Vec<u8>>>,
588    timeout: Duration,
589    max_result_bytes: usize,
590    tasks: bool,
591    task_timeout: Duration,
592}
593
594impl std::fmt::Debug for McpConnection {
595    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
596        formatter
597            .debug_struct("McpConnection")
598            .field("peer", &self.client.peer_info())
599            .finish_non_exhaustive()
600    }
601}
602
603impl McpConnection {
604    pub async fn connect(config: &ExternalToolConfig, services: McpClientServices) -> Result<Self> {
605        anyhow::ensure!(
606            !config.sampling || services.sampling.is_some(),
607            "MCP server {:?} enables sampling but this product installed no sampling provider",
608            config.id
609        );
610        anyhow::ensure!(
611            !config.elicitation || services.elicitation.is_some(),
612            "MCP server {:?} enables elicitation but this product installed no interactive provider",
613            config.id
614        );
615
616        let roots = roots_from(config);
617        let (event_tx, events) = tokio::sync::mpsc::unbounded_channel();
618        let handler = PersptClientHandler {
619            server_id: config.id.clone(),
620            info: client_info(config, !roots.is_empty()),
621            roots,
622            max_sampling_tokens: config.max_sampling_tokens,
623            services,
624            events: event_tx.clone(),
625        };
626        let stderr = Arc::new(Mutex::new(Vec::new()));
627        let client = match config.transport {
628            ExternalToolTransport::Stdio => {
629                connect_stdio(config, handler.clone(), stderr.clone()).await?
630            }
631            ExternalToolTransport::StreamableHttp => connect_http(config, handler).await?,
632        };
633
634        let negotiated = client
635            .peer_info()
636            .context("MCP discover omitted peer information")?;
637        anyhow::ensure!(
638            negotiated.protocol_version == ProtocolVersion::V_2026_07_28,
639            "MCP server did not negotiate required protocol {MCP_PROTOCOL_VERSION}"
640        );
641        let tasks = config.tasks
642            && negotiated
643                .capabilities
644                .extensions
645                .as_ref()
646                .is_some_and(|extensions| extensions.contains_key(TASKS_EXTENSION_ID));
647
648        let mut connection = Self {
649            client,
650            events,
651            subscription: None,
652            stderr,
653            timeout: Duration::from_millis(config.timeout_ms),
654            max_result_bytes: config.max_result_bytes,
655            tasks,
656            task_timeout: Duration::from_millis(config.max_task_wait_ms),
657        };
658        if config.subscriptions {
659            connection
660                .start_subscription(&config.resource_subscriptions, event_tx)
661                .await?;
662        }
663        Ok(connection)
664    }
665
666    pub fn peer_info(&self) -> Option<Arc<rmcp::model::ServerPeerInfo>> {
667        self.client.peer_info()
668    }
669
670    pub fn drain_events(&mut self) -> Vec<McpServerEvent> {
671        let mut events = Vec::new();
672        while let Ok(event) = self.events.try_recv() {
673            events.push(event);
674        }
675        events
676    }
677
678    pub async fn list_tools(&self) -> Result<Vec<rmcp::model::Tool>> {
679        self.bounded(self.client.list_all_tools()).await
680    }
681
682    pub async fn list_resources(&self) -> Result<Vec<rmcp::model::Resource>> {
683        self.bounded(self.client.list_all_resources()).await
684    }
685
686    pub async fn list_resource_templates(&self) -> Result<Vec<rmcp::model::ResourceTemplate>> {
687        self.bounded(self.client.list_all_resource_templates())
688            .await
689    }
690
691    pub async fn read_resource(&self, uri: String) -> Result<ReadResourceResult> {
692        self.bounded(
693            self.client
694                .read_resource(ReadResourceRequestParams::new(uri)),
695        )
696        .await
697    }
698
699    pub async fn list_prompts(&self) -> Result<Vec<rmcp::model::Prompt>> {
700        self.bounded(self.client.list_all_prompts()).await
701    }
702
703    pub async fn get_prompt(
704        &self,
705        name: String,
706        arguments: Option<rmcp::model::JsonObject>,
707    ) -> Result<GetPromptResult> {
708        let mut request = GetPromptRequestParams::new(name);
709        if let Some(arguments) = arguments {
710            request = request.with_arguments(arguments);
711        }
712        self.bounded(self.client.get_prompt(request)).await
713    }
714
715    pub async fn complete(
716        &self,
717        request: rmcp::model::CompleteRequestParams,
718    ) -> Result<rmcp::model::CompleteResult> {
719        self.bounded(self.client.complete(request)).await
720    }
721
722    pub async fn set_log_level(&self, level: rmcp::model::LoggingLevel) -> Result<()> {
723        self.request(
724            self.client
725                .set_level(rmcp::model::SetLevelRequestParams::new(level)),
726        )
727        .await
728    }
729
730    pub async fn call_tool(
731        &self,
732        name: String,
733        arguments: rmcp::model::JsonObject,
734    ) -> Result<CallToolResult> {
735        let mut request = CallToolRequestParams::new(name).with_arguments(arguments);
736        if !self.tasks {
737            return self.bounded(self.client.call_tool(request)).await;
738        }
739        let handler = self.client.service().clone();
740        let mut state_only_rounds = 0u32;
741        for _ in 0..rmcp::model::DEFAULT_MRTR_MAX_ROUNDS {
742            match self
743                .request(self.client.call_tool_once(request.clone()))
744                .await?
745            {
746                CallToolResponse::Complete(result) => return self.ensure_size(result),
747                CallToolResponse::InputRequired(input) => {
748                    let had_requests = input
749                        .input_requests
750                        .as_ref()
751                        .is_some_and(|requests| !requests.is_empty());
752                    anyhow::ensure!(
753                        had_requests || input.request_state.is_some(),
754                        "MCP MRTR response contained neither requests nor state"
755                    );
756                    let responses = handler
757                        .task_inputs(input.input_requests.unwrap_or_default())
758                        .await?;
759                    request.input_responses = (!responses.is_empty()).then_some(responses);
760                    request.request_state = input.request_state;
761                    if had_requests {
762                        state_only_rounds = 0;
763                    } else {
764                        let delay = 25u64.saturating_mul(1u64 << state_only_rounds.min(7));
765                        tokio::time::sleep(Duration::from_millis(delay)).await;
766                        state_only_rounds += 1;
767                    }
768                }
769                CallToolResponse::Task(task) => {
770                    let task_id = task.task.task_id;
771                    return match tokio::time::timeout(
772                        self.task_timeout,
773                        self.wait_for_task(task_id.clone()),
774                    )
775                    .await
776                    {
777                        Ok(result) => result,
778                        Err(_) => {
779                            let _ = self
780                                .request(
781                                    self.client
782                                        .cancel_task(rmcp::model::CancelTaskParams::new(task_id)),
783                                )
784                                .await;
785                            anyhow::bail!("MCP asynchronous task exceeded max_task_wait_ms")
786                        }
787                    };
788                }
789                _ => anyhow::bail!("MCP server returned an unknown tools/call response"),
790            }
791        }
792        anyhow::bail!("MCP tool exceeded the bounded MRTR round limit")
793    }
794
795    async fn wait_for_task(&self, task_id: String) -> Result<CallToolResult> {
796        let handler = self.client.service().clone();
797        loop {
798            let state = self
799                .request(self.client.get_task(GetTaskParams::new(task_id.clone())))
800                .await?;
801            let delay = state
802                .task
803                .task
804                .poll_interval_ms
805                .unwrap_or(500)
806                .clamp(50, 30_000);
807            match state.task.payload {
808                TaskPayload::Working => tokio::time::sleep(Duration::from_millis(delay)).await,
809                TaskPayload::InputRequired { input_requests } => {
810                    let responses = handler.task_inputs(input_requests).await?;
811                    self.request(self.client.update_task(rmcp::model::UpdateTaskParams::new(
812                        task_id.clone(),
813                        responses,
814                    )))
815                    .await?;
816                }
817                TaskPayload::Completed { result } => {
818                    return self
819                        .ensure_size(serde_json::from_value(serde_json::Value::Object(result))?);
820                }
821                TaskPayload::Failed { error } => {
822                    anyhow::bail!("MCP task failed: {}", serde_json::Value::Object(error))
823                }
824                TaskPayload::Cancelled => anyhow::bail!("MCP task was cancelled"),
825                _ => anyhow::bail!("MCP task returned an unknown state"),
826            }
827        }
828    }
829
830    async fn start_subscription(
831        &mut self,
832        resources: &[String],
833        event_tx: tokio::sync::mpsc::UnboundedSender<McpServerEvent>,
834    ) -> Result<()> {
835        let capabilities = &self
836            .client
837            .peer_info()
838            .context("MCP peer information unavailable")?
839            .capabilities;
840        let requested = SubscriptionFilter::builder()
841            .tools_list_changed()
842            .prompts_list_changed()
843            .resources_list_changed()
844            .resource_subscriptions(resources.iter().cloned())
845            .build();
846        let supported = requested.supported_by(capabilities);
847        if supported == SubscriptionFilter::default() {
848            return Ok(());
849        }
850        let mut subscription = self.request(self.client.listen(supported)).await?;
851        self.subscription = Some(tokio::spawn(async move {
852            loop {
853                match subscription.next().await {
854                    Ok(Some(notification)) => emit_subscription(&event_tx, notification),
855                    Ok(None) => {
856                        let _ = event_tx.send(McpServerEvent::SubscriptionEnded(format!(
857                            "{:?}",
858                            subscription.end()
859                        )));
860                        break;
861                    }
862                    Err(error) => {
863                        let _ = event_tx.send(McpServerEvent::SubscriptionEnded(error.to_string()));
864                        break;
865                    }
866                }
867            }
868        }));
869        Ok(())
870    }
871
872    async fn request<T, E>(
873        &self,
874        future: impl std::future::Future<Output = std::result::Result<T, E>>,
875    ) -> Result<T>
876    where
877        E: std::error::Error + Send + Sync + 'static,
878    {
879        tokio::time::timeout(self.timeout, future)
880            .await
881            .context("MCP request timed out")?
882            .map_err(anyhow::Error::new)
883    }
884
885    async fn bounded<T, E>(
886        &self,
887        future: impl std::future::Future<Output = std::result::Result<T, E>>,
888    ) -> Result<T>
889    where
890        T: serde::Serialize,
891        E: std::error::Error + Send + Sync + 'static,
892    {
893        let value = self.request(future).await?;
894        self.ensure_size(value)
895    }
896
897    fn ensure_size<T: serde::Serialize>(&self, value: T) -> Result<T> {
898        anyhow::ensure!(
899            serde_json::to_vec(&value)?.len() <= self.max_result_bytes,
900            "MCP result exceeded byte cap"
901        );
902        Ok(value)
903    }
904
905    pub async fn shutdown(&mut self) -> Result<()> {
906        if let Some(subscription) = self.subscription.take() {
907            subscription.abort();
908        }
909        self.client
910            .close_with_timeout(Duration::from_secs(3))
911            .await
912            .context("joining MCP service")?;
913        Ok(())
914    }
915
916    pub fn stderr_tail(&self) -> Vec<u8> {
917        self.stderr.lock().expect("stderr capture poisoned").clone()
918    }
919}
920
921fn roots_from(config: &ExternalToolConfig) -> Vec<Root> {
922    config
923        .roots
924        .iter()
925        .map(|root| match &root.name {
926            Some(name) => Root::new(root.uri.clone()).with_name(name.clone()),
927            None => Root::new(root.uri.clone()),
928        })
929        .collect()
930}
931
932fn client_info(config: &ExternalToolConfig, has_roots: bool) -> ClientInfo {
933    let mut capabilities = ClientCapabilities::default();
934    if has_roots {
935        let mut roots = rmcp::model::RootsCapabilities::default();
936        roots.list_changed = Some(false);
937        capabilities.roots = Some(roots);
938    }
939    if config.sampling {
940        let mut sampling = SamplingCapability::default();
941        sampling.tools = Some(Default::default());
942        capabilities.sampling = Some(sampling);
943    }
944    if config.elicitation {
945        capabilities.elicitation = Some(
946            ElicitationCapability::new()
947                .with_form(FormElicitationCapability::new().with_schema_validation(false))
948                .with_url(UrlElicitationCapability::new()),
949        );
950    }
951    if config.tasks {
952        capabilities
953            .extensions
954            .get_or_insert_default()
955            .insert(TASKS_EXTENSION_ID.to_string(), Default::default());
956    }
957    ClientInfo::new(
958        capabilities,
959        Implementation::new("perspt", env!("CARGO_PKG_VERSION")),
960    )
961    .with_protocol_version(ProtocolVersion::V_2026_07_28)
962}
963
964fn discover_mode() -> ClientLifecycleMode {
965    ClientLifecycleMode::Discover {
966        preferred_versions: vec![ProtocolVersion::V_2026_07_28],
967    }
968}
969
970async fn connect_stdio(
971    config: &ExternalToolConfig,
972    handler: PersptClientHandler,
973    stderr: Arc<Mutex<Vec<u8>>>,
974) -> Result<RunningService<RoleClient, PersptClientHandler>> {
975    let (program, arguments) = config
976        .command
977        .split_first()
978        .context("stdio MCP command is empty")?;
979    let mut command = tokio::process::Command::new(program);
980    command.args(arguments).env_clear().kill_on_drop(true);
981    if let Some(path) = std::env::var_os("PATH") {
982        command.env("PATH", path);
983    }
984    #[cfg(windows)]
985    if let Some(system_root) = std::env::var_os("SystemRoot") {
986        command.env("SystemRoot", system_root);
987    }
988    for (destination, source) in &config.env_from_env {
989        let value = std::env::var(source)
990            .with_context(|| format!("stdio MCP environment source {source:?} is unavailable"))?;
991        command.env(destination, value);
992    }
993    let (transport, stderr_pipe) = TokioChildProcess::builder(command)
994        .stderr(Stdio::piped())
995        .spawn()
996        .with_context(|| format!("starting stdio MCP server {:?}", config.id))?;
997    if let Some(stderr_pipe) = stderr_pipe {
998        capture_stderr(stderr_pipe, stderr, config.max_stderr_bytes);
999    }
1000    Ok(tokio::time::timeout(
1001        Duration::from_millis(config.timeout_ms),
1002        handler.serve_with_lifecycle(transport, discover_mode()),
1003    )
1004    .await
1005    .context("MCP server/discover timed out")??)
1006}
1007
1008async fn connect_http(
1009    config: &ExternalToolConfig,
1010    handler: PersptClientHandler,
1011) -> Result<RunningService<RoleClient, PersptClientHandler>> {
1012    let uri = config.url.as_deref().context("MCP HTTP URL missing")?;
1013    validate_http_uri(uri)?;
1014    let mut headers = HashMap::new();
1015    for (destination, source) in &config.headers_from_env {
1016        let value = std::env::var(source)
1017            .with_context(|| format!("MCP HTTP header source {source:?} is unavailable"))?;
1018        headers.insert(
1019            reqwest::header::HeaderName::from_bytes(destination.as_bytes())?,
1020            reqwest::header::HeaderValue::from_str(&value)?,
1021        );
1022    }
1023    let builder = reqwest::Client::builder().redirect(reqwest::redirect::Policy::none());
1024    let builder = if is_loopback_uri(uri)? {
1025        builder.no_proxy()
1026    } else {
1027        builder
1028    };
1029    let transport = StreamableHttpClientTransport::with_client(
1030        builder.build()?,
1031        StreamableHttpClientTransportConfig::with_uri(uri)
1032            .custom_headers(headers)
1033            .max_sse_event_size(config.max_result_bytes)
1034            .reinit_on_expired_session(false),
1035    );
1036    Ok(tokio::time::timeout(
1037        Duration::from_millis(config.timeout_ms),
1038        handler.serve_with_lifecycle(transport, discover_mode()),
1039    )
1040    .await
1041    .context("MCP server/discover timed out")??)
1042}
1043
1044fn validate_http_uri(uri: &str) -> Result<()> {
1045    let url = reqwest::Url::parse(uri)?;
1046    let loopback = is_loopback_uri(uri)?;
1047    anyhow::ensure!(
1048        url.scheme() == "https" || (url.scheme() == "http" && loopback),
1049        "MCP Streamable HTTP requires HTTPS except for explicit loopback fixtures"
1050    );
1051    Ok(())
1052}
1053
1054fn is_loopback_uri(uri: &str) -> Result<bool> {
1055    Ok(reqwest::Url::parse(uri)?
1056        .host_str()
1057        .is_some_and(|host| matches!(host, "localhost" | "127.0.0.1" | "[::1]" | "::1")))
1058}
1059
1060fn capture_stderr(
1061    mut pipe: tokio::process::ChildStderr,
1062    captured: Arc<Mutex<Vec<u8>>>,
1063    maximum: usize,
1064) {
1065    tokio::spawn(async move {
1066        let mut chunk = [0u8; 4096];
1067        while let Ok(count) = pipe.read(&mut chunk).await {
1068            if count == 0 {
1069                break;
1070            }
1071            let mut output = captured.lock().expect("stderr capture poisoned");
1072            output.extend_from_slice(&chunk[..count]);
1073            if output.len() > maximum {
1074                let excess = output.len() - maximum;
1075                output.drain(..excess);
1076            }
1077        }
1078    });
1079}
1080
1081fn emit_subscription(
1082    sender: &tokio::sync::mpsc::UnboundedSender<McpServerEvent>,
1083    notification: ServerNotification,
1084) {
1085    let event = match notification {
1086        ServerNotification::ToolListChangedNotification(_) => McpServerEvent::ToolsChanged,
1087        ServerNotification::PromptListChangedNotification(_) => McpServerEvent::PromptsChanged,
1088        ServerNotification::ResourceListChangedNotification(_) => McpServerEvent::ResourcesChanged,
1089        ServerNotification::ResourceUpdatedNotification(notification) => {
1090            McpServerEvent::ResourceUpdated(
1091                serde_json::to_value(notification.params).unwrap_or_default(),
1092            )
1093        }
1094        _ => return,
1095    };
1096    let _ = sender.send(event);
1097}
perspt_agent/external_tools/client.rs

perspt_agent/external_tools/
client.rs
