perspt_agent/external_tools/
chat.rs

1//! Chat-mode MCP tools: the same admission and transports as the agent,
2//! behind the interactive chat interface.
3//!
4//! Chat is not kernel-governed, so its admission capability grants
5//! **read-only effects only** — a server tool that declares (or defaults
6//! to) any mutating effect is rejected in chat mode. `simple-chat` never
7//! constructs this session and keeps the plain streaming path.
8
9#![allow(deprecated)]
10
11use std::sync::Arc;
12
13use anyhow::Result;
14use perspt_core::tools_driver::CoreGenerationConfig;
15use perspt_core::tools_driver::{
16    CoreMessage, CoreToolCall, CoreToolChoice, CoreToolSpec, CoreTurnStreamEvent,
17};
18use perspt_core::{Config, ExternalToolMode, GenAIProvider};
19use perspt_sdk::{ActorId, Capability, EffectKind};
20
21use super::{
22    ExternalToolRuntime, McpClientServices, McpElicitationAction, McpElicitationBroker,
23    McpPendingElicitation, McpSamplingProvider,
24};
25
26/// Bounded tool rounds per chat message.
27const MAX_TOOL_ROUNDS: usize = 4;
28
29/// Private channel marker consumed by the chat TUI as transient status.
30pub const MCP_ACTIVITY_PREFIX: &str = "__PERSPT_MCP_ACTIVITY__:";
31
32fn tool_label(name: &str) -> String {
33    name.rsplit('.')
34        .next()
35        .unwrap_or(name)
36        .trim_start_matches("_perspt_")
37        .replace('_', " ")
38}
39
40/// A chat-scoped MCP lifecycle: discovery happened, specs are ready, and
41/// every call goes through the shared runtime.
42#[derive(Clone)]
43pub struct ChatToolSession {
44    runtime: Arc<tokio::sync::Mutex<ExternalToolRuntime>>,
45    specs: Arc<std::sync::RwLock<Vec<CoreToolSpec>>>,
46    elicitation: Option<McpElicitationBroker>,
47}
48
49struct ChatSamplingProvider {
50    provider: Arc<GenAIProvider>,
51    model: String,
52}
53
54#[async_trait::async_trait]
55impl McpSamplingProvider for ChatSamplingProvider {
56    async fn create_message(
57        &self,
58        _server_id: &str,
59        request: rmcp::model::CreateMessageRequestParams,
60    ) -> Result<rmcp::model::CreateMessageResult> {
61        let sample = prepare_sample(request)?;
62        let output = self
63            .provider
64            .chat_turn_configured(
65                &self.model,
66                &sample.messages,
67                &sample.tools,
68                sample.choice,
69                sample.config,
70            )
71            .await?;
72        let result = sample_result(output, &self.model)?;
73        result.validate().map_err(anyhow::Error::msg)?;
74        Ok(result)
75    }
76}
77
78struct ChatSample {
79    messages: Vec<CoreMessage>,
80    tools: Vec<CoreToolSpec>,
81    choice: CoreToolChoice,
82    config: CoreGenerationConfig,
83}
84
85fn prepare_sample(request: rmcp::model::CreateMessageRequestParams) -> Result<ChatSample> {
86    request.validate().map_err(anyhow::Error::msg)?;
87    anyhow::ensure!(
88        request.include_context.is_none()
89            || request.include_context == Some(rmcp::model::ContextInclusion::None),
90        "MCP sampling includeContext is not advertised by Perspt"
91    );
92    let mut messages = request
93        .system_prompt
94        .map(|content| vec![CoreMessage::System { content }])
95        .unwrap_or_default();
96    for message in request.messages {
97        append_sampling(&mut messages, message)?;
98    }
99    let tools = request
100        .tools
101        .unwrap_or_default()
102        .into_iter()
103        .map(core_tool)
104        .collect();
105    let choice = match request.tool_choice.and_then(|choice| choice.mode) {
106        Some(rmcp::model::ToolChoiceMode::Required) => CoreToolChoice::Required,
107        Some(rmcp::model::ToolChoiceMode::None) => CoreToolChoice::None,
108        _ => CoreToolChoice::Auto,
109    };
110    Ok(ChatSample {
111        messages,
112        tools,
113        choice,
114        config: CoreGenerationConfig {
115            max_tokens: Some(request.max_tokens),
116            temperature: request.temperature,
117            stop_sequences: request.stop_sequences.unwrap_or_default(),
118        },
119    })
120}
121
122fn core_tool(tool: rmcp::model::Tool) -> CoreToolSpec {
123    CoreToolSpec {
124        name: tool.name.into_owned(),
125        description: tool
126            .description
127            .map(|value| value.into_owned())
128            .unwrap_or_default(),
129        schema: serde_json::Value::Object(tool.input_schema.as_ref().clone()),
130        strict: false,
131    }
132}
133
134fn sample_result(
135    output: perspt_core::CoreTurnOutput,
136    model: &str,
137) -> Result<rmcp::model::CreateMessageResult> {
138    match output {
139        perspt_core::CoreTurnOutput::Text(text) => Ok(rmcp::model::CreateMessageResult::new(
140            rmcp::model::SamplingMessage::assistant_text(text),
141            model.to_string(),
142        )
143        .with_stop_reason(rmcp::model::CreateMessageResult::STOP_REASON_END_TURN)),
144        perspt_core::CoreTurnOutput::ToolCalls(calls) => {
145            let blocks = calls
146                .into_iter()
147                .map(|call| {
148                    let input = call.arguments.as_object().cloned().ok_or_else(|| {
149                        anyhow::anyhow!("MCP sampling tool arguments must be an object")
150                    })?;
151                    Ok(rmcp::model::SamplingMessageContentBlock::tool_use(
152                        call.call_id,
153                        call.name,
154                        input,
155                    ))
156                })
157                .collect::<Result<Vec<_>>>()?;
158            Ok(rmcp::model::CreateMessageResult::new(
159                rmcp::model::SamplingMessage::new_multiple(rmcp::model::Role::Assistant, blocks),
160                model.to_string(),
161            )
162            .with_stop_reason(rmcp::model::CreateMessageResult::STOP_REASON_TOOL_USE))
163        }
164    }
165}
166
167fn append_sampling(
168    messages: &mut Vec<CoreMessage>,
169    message: rmcp::model::SamplingMessage,
170) -> Result<()> {
171    let mut text = Vec::new();
172    let mut calls = Vec::new();
173    let mut results = Vec::new();
174    for content in message.content.into_vec() {
175        match content {
176            rmcp::model::SamplingMessageContentBlock::Text(value) => text.push(value.text),
177            rmcp::model::SamplingMessageContentBlock::ToolUse(value) => {
178                calls.push(CoreToolCall {
179                    call_id: value.id,
180                    name: value.name,
181                    arguments: serde_json::Value::Object(value.input),
182                });
183            }
184            rmcp::model::SamplingMessageContentBlock::ToolResult(value) => {
185                results.push((value.tool_use_id, serde_json::to_string(&value.content)?));
186            }
187            rmcp::model::SamplingMessageContentBlock::Image(_)
188            | rmcp::model::SamplingMessageContentBlock::Audio(_) => {
189                anyhow::bail!("the Perspt chat provider does not accept MCP media sampling content")
190            }
191            _ => anyhow::bail!("unknown MCP sampling content block"),
192        }
193    }
194    if !text.is_empty() {
195        messages.push(match message.role {
196            rmcp::model::Role::User => CoreMessage::User {
197                content: text.join("\n"),
198            },
199            rmcp::model::Role::Assistant => CoreMessage::Assistant {
200                content: text.join("\n"),
201            },
202        });
203    }
204    if !calls.is_empty() {
205        messages.push(CoreMessage::AssistantToolCalls { calls });
206    }
207    messages.extend(
208        results
209            .into_iter()
210            .map(|(call_id, content)| CoreMessage::ToolResponse { call_id, content }),
211    );
212    Ok(())
213}
214
215/// Every read-only effect, the most a chat-admitted tool may declare.
216fn read_only_effects() -> Vec<EffectKind> {
217    [
218        EffectKind::ReadFile,
219        EffectKind::SystemProbe,
220        EffectKind::DataRead,
221        EffectKind::ToolSearch,
222        EffectKind::ToolProgram,
223        EffectKind::Search,
224        EffectKind::List,
225        EffectKind::LspQuery,
226        EffectKind::GitRead,
227    ]
228    .into()
229}
230
231fn chat_services(
232    config: &Config,
233    provider: Arc<GenAIProvider>,
234    model: &str,
235) -> (McpClientServices, Option<McpElicitationBroker>) {
236    let enabled = |select: fn(&perspt_core::ExternalToolConfig) -> bool| {
237        config
238            .external_tools
239            .iter()
240            .any(|server| server.supports(ExternalToolMode::Chat) && select(server))
241    };
242    let sampling = enabled(|server| server.sampling).then(|| {
243        Arc::new(ChatSamplingProvider {
244            provider,
245            model: model.to_string(),
246        }) as Arc<dyn McpSamplingProvider>
247    });
248    let elicitation = enabled(|server| server.elicitation).then(McpElicitationBroker::new);
249    let services = McpClientServices {
250        sampling,
251        elicitation: elicitation
252            .clone()
253            .map(|broker| Arc::new(broker) as Arc<dyn super::McpElicitationProvider>),
254    };
255    (services, elicitation)
256}
257
258async fn discover_all(runtime: &mut ExternalToolRuntime) -> Vec<String> {
259    let servers: Vec<String> = runtime
260        .server_ids()
261        .into_iter()
262        .map(str::to_string)
263        .collect();
264    let mut notices = Vec::new();
265    for server_id in servers {
266        match runtime.discover_server(&server_id).await {
267            Ok(admitted) => add_discovery_notice(runtime, &server_id, admitted.len(), &mut notices),
268            Err(error) => notices.push(format!("MCP server {server_id} failed: {error:#}")),
269        }
270    }
271    notices
272}
273
274fn add_discovery_notice(
275    runtime: &ExternalToolRuntime,
276    server_id: &str,
277    admitted: usize,
278    notices: &mut Vec<String>,
279) {
280    let rejected = runtime.admission_rejections(server_id);
281    notices.push(format!(
282        "MCP server {server_id}: {admitted} tool(s) admitted, {} rejected by local policy",
283        rejected.len()
284    ));
285    notices.extend(rejected.iter().take(8).map(|item| {
286        format!(
287            "MCP tool {}.{} rejected: {}",
288            item.server_id, item.remote_tool, item.reason
289        )
290    }));
291    if rejected.len() > 8 {
292        notices.push(format!(
293            "MCP server {server_id}: {} additional rejection(s) omitted",
294            rejected.len() - 8
295        ));
296    }
297}
298
299fn core_specs(runtime: &ExternalToolRuntime) -> Vec<CoreToolSpec> {
300    runtime
301        .admitted_entries()
302        .into_iter()
303        .map(|entry| CoreToolSpec {
304            name: entry.name,
305            description: entry.description,
306            schema: entry.schema,
307            strict: false,
308        })
309        .collect()
310}
311
312impl ChatToolSession {
313    /// Build and discover the chat-mode session; `None` when no server is
314    /// configured for chat. Failing servers are skipped with a notice.
315    pub async fn from_config(
316        config: &Config,
317        provider: Arc<GenAIProvider>,
318        model: &str,
319    ) -> Result<Option<(Self, Vec<String>)>> {
320        let has_chat_servers = config
321            .external_tools
322            .iter()
323            .any(|server| server.supports(ExternalToolMode::Chat));
324        if !has_chat_servers {
325            return Ok(None);
326        }
327        let capability = Capability::new(ActorId::new("chat"), read_only_effects());
328        let mut runtime =
329            ExternalToolRuntime::from_config(config, ExternalToolMode::Chat, vec![capability])?;
330        let (services, elicitation) = chat_services(config, provider, model);
331        runtime.set_client_services(services);
332        let notices = discover_all(&mut runtime).await;
333        let specs = core_specs(&runtime);
334        Ok(Some((
335            Self {
336                runtime: Arc::new(tokio::sync::Mutex::new(runtime)),
337                specs: Arc::new(std::sync::RwLock::new(specs)),
338                elicitation,
339            },
340            notices,
341        )))
342    }
343
344    pub fn has_tools(&self) -> bool {
345        !self
346            .specs
347            .read()
348            .expect("MCP tool specs poisoned")
349            .is_empty()
350    }
351
352    /// Namespaced names of the tools admitted for this chat lifecycle.
353    ///
354    /// The TUI uses this for local discovery only (`/mcp`); descriptions and
355    /// schemas remain in the provider request and are never treated as local
356    /// instructions.
357    pub fn tool_names(&self) -> Vec<String> {
358        self.specs
359            .read()
360            .expect("MCP tool specs poisoned")
361            .iter()
362            .map(|spec| spec.name.clone())
363            .collect()
364    }
365
366    pub fn try_next_elicitation(&self) -> Option<McpPendingElicitation> {
367        self.elicitation
368            .as_ref()
369            .and_then(McpElicitationBroker::try_next)
370    }
371
372    pub fn respond_elicitation(
373        &self,
374        id: u64,
375        action: McpElicitationAction,
376        content: Option<serde_json::Value>,
377    ) -> Result<()> {
378        self.elicitation
379            .as_ref()
380            .ok_or_else(|| anyhow::anyhow!("MCP elicitation is not enabled"))?
381            .respond(id, action, content)
382    }
383
384    /// Run one chat message with bounded, model-selected tool rounds.
385    ///
386    /// Genuine model reasoning and final answer text are streamed through
387    /// `updates`. MCP activity uses a separate transient marker; protocol
388    /// notifications stay in logs and never enter the conversation.
389    pub async fn run_turn(
390        &self,
391        provider: &GenAIProvider,
392        model: &str,
393        mut messages: Vec<CoreMessage>,
394        updates: &tokio::sync::mpsc::UnboundedSender<String>,
395    ) -> Result<()> {
396        for _round in 0..MAX_TOOL_ROUNDS {
397            self.sync_notifications().await?;
398            let specs = self.specs.read().expect("MCP tool specs poisoned").clone();
399            let output = provider
400                .stream_tool_turn(model, &messages, &specs, CoreToolChoice::Auto, |event| {
401                    match event {
402                        CoreTurnStreamEvent::Text(text) => {
403                            let _ = updates.send(text);
404                        }
405                        CoreTurnStreamEvent::Reasoning(text) => {
406                            let _ = updates.send(format!("__PERSPT_REASONING__:{text}"));
407                        }
408                    }
409                })
410                .await?;
411            let calls = match output {
412                perspt_core::tools_driver::CoreTurnOutput::Text(_) => return Ok(()),
413                perspt_core::tools_driver::CoreTurnOutput::ToolCalls(calls) => calls,
414            };
415            messages.push(CoreMessage::AssistantToolCalls {
416                calls: calls.clone(),
417            });
418            for call in calls {
419                let _ = updates.send(format!(
420                    "{MCP_ACTIVITY_PREFIX}Using MCP: {}…",
421                    tool_label(&call.name)
422                ));
423                let content = self.execute(&call).await;
424                messages.push(CoreMessage::ToolResponse {
425                    call_id: call.call_id,
426                    content,
427                });
428            }
429        }
430        let _ = updates.send("The MCP tool round limit was reached without a final answer.".into());
431        Ok(())
432    }
433
434    async fn sync_notifications(&self) -> Result<()> {
435        let mut runtime = self.runtime.lock().await;
436        let events = runtime.sync_server_events().await?;
437        for (server, event) in events {
438            log::debug!("MCP {server}: {event:?}");
439        }
440        let specs = runtime
441            .admitted_entries()
442            .into_iter()
443            .map(|entry| CoreToolSpec {
444                name: entry.name,
445                description: entry.description,
446                schema: entry.schema,
447                strict: false,
448            })
449            .collect();
450        *self.specs.write().expect("MCP tool specs poisoned") = specs;
451        Ok(())
452    }
453
454    async fn execute(&self, call: &CoreToolCall) -> String {
455        let mut runtime = self.runtime.lock().await;
456        match runtime.call(&call.name, call.arguments.clone()).await {
457            Ok(result) => format!(
458                "[untrusted MCP result{}] {}",
459                if result.is_error { ", tool error" } else { "" },
460                result.content
461            ),
462            Err(error) => format!("tool failed: {error:#}"),
463        }
464    }
465}
466
467#[cfg(test)]
468mod tests {
469    use super::*;
470
471    #[test]
472    fn activity_labels_hide_protocol_namespaces() {
473        assert_eq!(tool_label("mcp.todos.list_tasks"), "list tasks");
474        assert_eq!(
475            tool_label("mcp.docs._perspt_resource_read"),
476            "resource read"
477        );
478    }
479}
perspt_agent/external_tools/chat.rs

perspt_agent/external_tools/
chat.rs
