feat: propagate elicitation decline and cancel actions end-to-end (#9437)
Co-authored-by: Matt Van Horn <455140+mvanhorn@users.noreply.github.com> Co-authored-by: Douwe M Osinga <douwe@sidewalklabs.com>
This commit is contained in:
@@ -1,12 +1,15 @@
|
||||
use console::style;
|
||||
use rmcp::model::ElicitationAction;
|
||||
use serde_json::Value;
|
||||
use std::collections::HashMap;
|
||||
use std::io::{self, BufRead, IsTerminal, Write};
|
||||
|
||||
pub fn collect_elicitation_input(
|
||||
message: &str,
|
||||
schema: &Value,
|
||||
) -> io::Result<Option<HashMap<String, Value>>> {
|
||||
pub struct ElicitationInput {
|
||||
pub action: ElicitationAction,
|
||||
pub user_data: HashMap<String, Value>,
|
||||
}
|
||||
|
||||
pub fn collect_elicitation_input(message: &str, schema: &Value) -> io::Result<ElicitationInput> {
|
||||
if !message.is_empty() {
|
||||
println!("\n{}", style(message).cyan());
|
||||
}
|
||||
@@ -24,9 +27,18 @@ pub fn collect_elicitation_input(
|
||||
"Approve?"
|
||||
};
|
||||
return match cliclack::confirm(prompt).initial_value(true).interact() {
|
||||
Ok(true) => Ok(Some(HashMap::new())),
|
||||
Ok(false) => Ok(None),
|
||||
Err(e) if e.kind() == io::ErrorKind::Interrupted => Ok(None),
|
||||
Ok(true) => Ok(ElicitationInput {
|
||||
action: ElicitationAction::Accept,
|
||||
user_data: HashMap::new(),
|
||||
}),
|
||||
Ok(false) => Ok(ElicitationInput {
|
||||
action: ElicitationAction::Decline,
|
||||
user_data: HashMap::new(),
|
||||
}),
|
||||
Err(e) if e.kind() == io::ErrorKind::Interrupted => Ok(ElicitationInput {
|
||||
action: ElicitationAction::Cancel,
|
||||
user_data: HashMap::new(),
|
||||
}),
|
||||
Err(e) => Err(e),
|
||||
};
|
||||
}
|
||||
@@ -65,7 +77,12 @@ pub fn collect_elicitation_input(
|
||||
Ok(v) => {
|
||||
data.insert(name.clone(), Value::Bool(v));
|
||||
}
|
||||
Err(e) if e.kind() == io::ErrorKind::Interrupted => return Ok(None),
|
||||
Err(e) if e.kind() == io::ErrorKind::Interrupted => {
|
||||
return Ok(ElicitationInput {
|
||||
action: ElicitationAction::Cancel,
|
||||
user_data: HashMap::new(),
|
||||
});
|
||||
}
|
||||
Err(e) => return Err(e),
|
||||
}
|
||||
continue;
|
||||
@@ -93,7 +110,10 @@ pub fn collect_elicitation_input(
|
||||
|
||||
// Handle Ctrl+C / EOF for cancellation
|
||||
if input.is_none() {
|
||||
return Ok(None);
|
||||
return Ok(ElicitationInput {
|
||||
action: ElicitationAction::Cancel,
|
||||
user_data: HashMap::new(),
|
||||
});
|
||||
}
|
||||
let input = input.unwrap();
|
||||
|
||||
@@ -114,12 +134,18 @@ pub fn collect_elicitation_input(
|
||||
"{}",
|
||||
style(format!("Required field '{}' is missing", name)).red()
|
||||
);
|
||||
return Ok(None);
|
||||
return Ok(ElicitationInput {
|
||||
action: ElicitationAction::Decline,
|
||||
user_data: HashMap::new(),
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
println!();
|
||||
Ok(Some(data))
|
||||
Ok(ElicitationInput {
|
||||
action: ElicitationAction::Accept,
|
||||
user_data: data,
|
||||
})
|
||||
}
|
||||
|
||||
fn read_line() -> io::Result<Option<String>> {
|
||||
|
||||
@@ -353,7 +353,7 @@ pub fn message_to_markdown(message: &Message, export_all_content: bool) -> Strin
|
||||
message
|
||||
));
|
||||
}
|
||||
ActionRequiredData::ElicitationResponse { id, user_data } => {
|
||||
ActionRequiredData::ElicitationResponse { id, user_data, .. } => {
|
||||
md.push_str(&format!(
|
||||
"**Action Required** (elicitation_response): {}\n```json\n{}\n```\n\n",
|
||||
id,
|
||||
|
||||
@@ -37,8 +37,8 @@ use goose::agents::{Agent, SessionConfig, COMPACT_TRIGGERS};
|
||||
use goose::config::extensions::name_to_key;
|
||||
use goose::config::{Config, GooseMode};
|
||||
use input::InputResult;
|
||||
use rmcp::model::PromptMessage;
|
||||
use rmcp::model::ServerNotification;
|
||||
use rmcp::model::{ElicitationAction, PromptMessage};
|
||||
use rmcp::model::{ErrorCode, ErrorData};
|
||||
use strum::VariantNames;
|
||||
|
||||
@@ -1248,25 +1248,37 @@ impl CliSession {
|
||||
let _ = progress_bars.hide();
|
||||
|
||||
match elicitation::collect_elicitation_input(&elicitation_message, &schema) {
|
||||
Ok(Some(user_data)) => {
|
||||
let user_data_value = serde_json::to_value(user_data)
|
||||
Ok(input) => {
|
||||
match &input.action {
|
||||
ElicitationAction::Decline => {
|
||||
output::render_text("Information request declined.", Some(Color::Yellow), true);
|
||||
}
|
||||
ElicitationAction::Cancel => {
|
||||
output::render_text("Information request cancelled.", Some(Color::Yellow), true);
|
||||
}
|
||||
ElicitationAction::Accept => {}
|
||||
}
|
||||
|
||||
let should_cancel = input.action == ElicitationAction::Cancel;
|
||||
let action = input.action;
|
||||
let user_data_value = serde_json::to_value(input.user_data)
|
||||
.unwrap_or(serde_json::Value::Object(serde_json::Map::new()));
|
||||
let response_message = Message::user()
|
||||
.with_content(MessageContent::action_required_elicitation_response(
|
||||
elicitation_id,
|
||||
user_data_value,
|
||||
action,
|
||||
))
|
||||
.with_visibility(false, true);
|
||||
self.messages.push(response_message.clone());
|
||||
// Elicitation responses return an empty stream - the response
|
||||
// unblocks the waiting tool call via ActionRequiredManager
|
||||
let _ = self.agent.reply(response_message, session_config.clone(), Some(cancel_token.clone())).await?;
|
||||
}
|
||||
Ok(None) => {
|
||||
output::render_text("Information request cancelled.", Some(Color::Yellow), true);
|
||||
cancel_token_clone.cancel();
|
||||
drop(stream);
|
||||
break;
|
||||
if should_cancel {
|
||||
cancel_token_clone.cancel();
|
||||
drop(stream);
|
||||
break;
|
||||
}
|
||||
}
|
||||
Err(e) => {
|
||||
output::render_error(&format!("Failed to collect input: {}", e));
|
||||
|
||||
@@ -3,9 +3,9 @@ use crate::mcp_utils::extract_text_from_resource;
|
||||
use crate::utils::sanitize_unicode_tags;
|
||||
use chrono::Utc;
|
||||
use rmcp::model::{
|
||||
AnnotateAble, CallToolRequestParams, CallToolResult, Content, ImageContent, JsonObject,
|
||||
PromptMessage, PromptMessageContent, PromptMessageRole, RawContent, RawImageContent,
|
||||
RawTextContent, Role, TextContent,
|
||||
AnnotateAble, CallToolRequestParams, CallToolResult, Content, ElicitationAction, ImageContent,
|
||||
JsonObject, PromptMessage, PromptMessageContent, PromptMessageRole, RawContent,
|
||||
RawImageContent, RawTextContent, Role, TextContent,
|
||||
};
|
||||
use serde::{Deserialize, Deserializer, Serialize};
|
||||
use std::collections::HashSet;
|
||||
@@ -212,9 +212,16 @@ pub enum ActionRequiredData {
|
||||
ElicitationResponse {
|
||||
id: String,
|
||||
user_data: serde_json::Value,
|
||||
#[serde(default = "default_elicitation_action")]
|
||||
#[schema(value_type = String)]
|
||||
action: ElicitationAction,
|
||||
},
|
||||
}
|
||||
|
||||
fn default_elicitation_action() -> ElicitationAction {
|
||||
ElicitationAction::Accept
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize, ToSchema)]
|
||||
#[serde(rename_all = "camelCase")]
|
||||
pub struct ActionRequired {
|
||||
@@ -478,11 +485,13 @@ impl MessageContent {
|
||||
pub fn action_required_elicitation_response<S: Into<String>>(
|
||||
id: S,
|
||||
user_data: serde_json::Value,
|
||||
action: ElicitationAction,
|
||||
) -> Self {
|
||||
MessageContent::ActionRequired(ActionRequired {
|
||||
data: ActionRequiredData::ElicitationResponse {
|
||||
id: id.into(),
|
||||
user_data,
|
||||
action,
|
||||
},
|
||||
})
|
||||
}
|
||||
@@ -1087,13 +1096,15 @@ pub struct TokenState {
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use crate::conversation::message::{Message, MessageContent, MessageMetadata};
|
||||
use crate::conversation::message::{
|
||||
ActionRequiredData, Message, MessageContent, MessageMetadata,
|
||||
};
|
||||
use crate::conversation::*;
|
||||
use rmcp::model::{
|
||||
AnnotateAble, CallToolRequestParams, PromptMessage, PromptMessageContent,
|
||||
PromptMessageRole, RawEmbeddedResource, RawImageContent, ResourceContents,
|
||||
};
|
||||
use rmcp::model::{ErrorCode, ErrorData};
|
||||
use rmcp::model::{ElicitationAction, ErrorCode, ErrorData};
|
||||
use rmcp::object;
|
||||
use serde_json::Value;
|
||||
|
||||
@@ -1250,6 +1261,29 @@ mod tests {
|
||||
assert!(thinking.signature.is_empty());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_elicitation_response_defaults_action_to_accept() {
|
||||
let action_required: ActionRequiredData = serde_json::from_value(serde_json::json!({
|
||||
"actionType": "elicitationResponse",
|
||||
"id": "request-123",
|
||||
"user_data": { "name": "goose" }
|
||||
}))
|
||||
.unwrap();
|
||||
|
||||
let ActionRequiredData::ElicitationResponse {
|
||||
id,
|
||||
user_data,
|
||||
action,
|
||||
} = action_required
|
||||
else {
|
||||
panic!("Expected elicitation response");
|
||||
};
|
||||
|
||||
assert_eq!(id, "request-123");
|
||||
assert_eq!(user_data, serde_json::json!({ "name": "goose" }));
|
||||
assert_eq!(action, ElicitationAction::Accept);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_agent_visible_content_preserves_thinking_for_provider() {
|
||||
let message = Message::assistant()
|
||||
|
||||
@@ -67,7 +67,8 @@ use futures::future::BoxFuture;
|
||||
use futures::stream::{self, StreamExt};
|
||||
use futures::FutureExt;
|
||||
use rmcp::model::{
|
||||
AnnotateAble, CallToolResult, RawContent, RawTextContent, ResourceContents, Role,
|
||||
AnnotateAble, CallToolResult, ElicitationAction, RawContent, RawTextContent, ResourceContents,
|
||||
Role,
|
||||
};
|
||||
use serde::Deserialize;
|
||||
use std::collections::{HashMap, HashSet};
|
||||
@@ -2660,7 +2661,11 @@ impl GooseAcpAgent {
|
||||
req: ElicitationRespondRequest,
|
||||
) -> Result<EmptyResponse, agent_client_protocol::Error> {
|
||||
ActionRequiredManager::global()
|
||||
.submit_response(req.elicitation_id.clone(), req.user_data.clone())
|
||||
.submit_response(
|
||||
req.elicitation_id.clone(),
|
||||
req.user_data.clone(),
|
||||
ElicitationAction::Accept,
|
||||
)
|
||||
.await
|
||||
.invalid_params_err_ctx("Failed to submit elicitation response")?;
|
||||
|
||||
@@ -2669,6 +2674,7 @@ impl GooseAcpAgent {
|
||||
.with_content(MessageContent::action_required_elicitation_response(
|
||||
req.elicitation_id.clone(),
|
||||
req.user_data,
|
||||
ElicitationAction::Accept,
|
||||
))
|
||||
.agent_only();
|
||||
|
||||
|
||||
@@ -1,4 +1,5 @@
|
||||
use anyhow::Result;
|
||||
use rmcp::model::ElicitationAction;
|
||||
use serde_json::Value;
|
||||
use std::collections::HashMap;
|
||||
use std::sync::Arc;
|
||||
@@ -10,8 +11,14 @@ use uuid::Uuid;
|
||||
|
||||
use crate::conversation::message::{Message, MessageContent};
|
||||
|
||||
#[derive(Debug, Clone, PartialEq)]
|
||||
pub struct ActionRequiredResponse {
|
||||
pub action: ElicitationAction,
|
||||
pub user_data: Value,
|
||||
}
|
||||
|
||||
struct PendingRequest {
|
||||
response_tx: Option<tokio::sync::oneshot::Sender<Value>>,
|
||||
response_tx: Option<tokio::sync::oneshot::Sender<ActionRequiredResponse>>,
|
||||
}
|
||||
|
||||
pub struct ActionRequiredManager {
|
||||
@@ -41,7 +48,7 @@ impl ActionRequiredManager {
|
||||
message: String,
|
||||
schema: Value,
|
||||
timeout_duration: Duration,
|
||||
) -> Result<Value> {
|
||||
) -> Result<ActionRequiredResponse> {
|
||||
let id = Uuid::new_v4().to_string();
|
||||
let (tx, rx) = tokio::sync::oneshot::channel();
|
||||
let pending_request = PendingRequest {
|
||||
@@ -62,7 +69,7 @@ impl ActionRequiredManager {
|
||||
}
|
||||
|
||||
let result = match timeout(timeout_duration, rx).await {
|
||||
Ok(Ok(user_data)) => Ok(user_data),
|
||||
Ok(Ok(response)) => Ok(response),
|
||||
Ok(Err(_)) => {
|
||||
warn!("Response channel closed for request: {}", id);
|
||||
Err(anyhow::anyhow!("Response channel closed"))
|
||||
@@ -78,7 +85,12 @@ impl ActionRequiredManager {
|
||||
result
|
||||
}
|
||||
|
||||
pub async fn submit_response(&self, request_id: String, user_data: Value) -> Result<()> {
|
||||
pub async fn submit_response(
|
||||
&self,
|
||||
request_id: String,
|
||||
user_data: Value,
|
||||
action: ElicitationAction,
|
||||
) -> Result<()> {
|
||||
let pending_arc = {
|
||||
let pending = self.pending.read().await;
|
||||
pending
|
||||
@@ -89,7 +101,10 @@ impl ActionRequiredManager {
|
||||
|
||||
let mut pending = pending_arc.lock().await;
|
||||
if let Some(tx) = pending.response_tx.take() {
|
||||
if tx.send(user_data).is_err() {
|
||||
if tx
|
||||
.send(ActionRequiredResponse { action, user_data })
|
||||
.is_err()
|
||||
{
|
||||
warn!("Failed to send response through oneshot channel");
|
||||
}
|
||||
}
|
||||
@@ -97,3 +112,51 @@ impl ActionRequiredManager {
|
||||
Ok(())
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
use crate::conversation::message::{ActionRequiredData, MessageContent};
|
||||
use serde_json::json;
|
||||
use std::sync::Arc;
|
||||
use std::time::Duration;
|
||||
|
||||
#[tokio::test]
|
||||
async fn request_and_wait_returns_submitted_action() {
|
||||
let manager = Arc::new(ActionRequiredManager::new());
|
||||
let waiter = {
|
||||
let manager = manager.clone();
|
||||
tokio::spawn(async move {
|
||||
manager
|
||||
.request_and_wait(
|
||||
"Need information".to_string(),
|
||||
json!({ "type": "object" }),
|
||||
Duration::from_secs(5),
|
||||
)
|
||||
.await
|
||||
.unwrap()
|
||||
})
|
||||
};
|
||||
|
||||
let message = manager.request_rx.lock().await.recv().await.unwrap();
|
||||
let MessageContent::ActionRequired(action_required) = &message.content[0] else {
|
||||
panic!("Expected action required message");
|
||||
};
|
||||
let ActionRequiredData::Elicitation { id, .. } = &action_required.data else {
|
||||
panic!("Expected elicitation request");
|
||||
};
|
||||
|
||||
manager
|
||||
.submit_response(
|
||||
id.clone(),
|
||||
json!({ "reason": "not needed" }),
|
||||
ElicitationAction::Decline,
|
||||
)
|
||||
.await
|
||||
.unwrap();
|
||||
|
||||
let response = waiter.await.unwrap();
|
||||
assert_eq!(response.action, ElicitationAction::Decline);
|
||||
assert_eq!(response.user_data, json!({ "reason": "not needed" }));
|
||||
}
|
||||
}
|
||||
|
||||
@@ -1450,17 +1450,20 @@ impl Agent {
|
||||
|
||||
for content in &user_message.content {
|
||||
if let MessageContent::ActionRequired(action_required) = content {
|
||||
if let ActionRequiredData::ElicitationResponse { id, user_data } =
|
||||
&action_required.data
|
||||
if let ActionRequiredData::ElicitationResponse {
|
||||
id,
|
||||
user_data,
|
||||
action,
|
||||
} = &action_required.data
|
||||
{
|
||||
// Surface stale/cancelled/timed-out elicitations as a hard
|
||||
// error so callers (e.g. the HTTP handler) can propagate
|
||||
// failure to the client instead of silently reporting
|
||||
// success while the blocked tool call stays unblocked.
|
||||
// The success path returns an empty stream; an Err here
|
||||
// makes the contract: Ok(empty) on accept, Err on reject.
|
||||
// The success path returns an empty stream after the MCP
|
||||
// server receives the user's accept/decline/cancel action.
|
||||
ActionRequiredManager::global()
|
||||
.submit_response(id.clone(), user_data.clone())
|
||||
.submit_response(id.clone(), user_data.clone(), action.clone())
|
||||
.await
|
||||
.map_err(|e| {
|
||||
error!("Failed to submit elicitation response: {}", e);
|
||||
|
||||
@@ -396,8 +396,13 @@ impl ClientHandler for GooseClient {
|
||||
ActionRequiredManager::global()
|
||||
.request_and_wait(message, schema_value, Duration::from_secs(300))
|
||||
.await
|
||||
.map(|user_data| {
|
||||
CreateElicitationResult::new(ElicitationAction::Accept).with_content(user_data)
|
||||
.map(|response| {
|
||||
let result = CreateElicitationResult::new(response.action.clone());
|
||||
if response.action == ElicitationAction::Accept {
|
||||
result.with_content(response.user_data)
|
||||
} else {
|
||||
result
|
||||
}
|
||||
})
|
||||
.map_err(|e| {
|
||||
ErrorData::new(
|
||||
|
||||
@@ -3918,6 +3918,9 @@
|
||||
"actionType"
|
||||
],
|
||||
"properties": {
|
||||
"action": {
|
||||
"type": "string"
|
||||
},
|
||||
"actionType": {
|
||||
"type": "string",
|
||||
"enum": [
|
||||
|
||||
@@ -20,6 +20,7 @@ export type ActionRequiredData = {
|
||||
message: string;
|
||||
requested_schema: unknown;
|
||||
} | {
|
||||
action?: string;
|
||||
actionType: 'elicitationResponse';
|
||||
id: string;
|
||||
user_data: unknown;
|
||||
|
||||
Reference in New Issue
Block a user