Upgrade rmcp to 3.0.0-beta.3 (#35720)

## What changed

- Update the `rmcp` model and transport integrations for `3.0.0-beta.3`.
- Preserve legacy `elicitation/create` form requests, schema defaults, metadata, and wire-format compatibility while adopting the new elicitation types.
- Harden OAuth discovery by using GET-first discovery without starting an MCP session, preventing cross-origin header redirects, validating authorization-server and callback issuers, and retaining compatibility with metadata that omits an issuer.
- Keep stored OAuth credentials readable and avoid requiring reauthorization after transient refresh failures.

## Testing

- Add coverage for legacy elicitation round trips, OAuth discovery variants, callback issuer validation, credential migration, and refresh failures.

GitOrigin-RevId: 756197d26f3fd347c28c400228f6b3d06ed493b7
This commit is contained in:
thomas authored and copyberry committed 2026-07-28 05:17:25 +00:00
1 parent 899539c03a
commit 61de0d8fe8
48 files changed
+1391 -639

No files matched your search

+3 -3
View File
@@ -25,7 +25,7 @@ use codex_protocol::protocol::Submission;
use codex_protocol::protocol::TurnCompleteEvent;
use codex_protocol::user_input::UserInput;
use rmcp::model::CallToolResult;
use rmcp::model::Content;
use rmcp::model::ContentBlock;
use rmcp::model::RequestId;
use serde_json::json;
use tokio::sync::Mutex;
@@ -40,7 +40,7 @@ pub(crate) fn create_call_tool_result_with_thread_id(
is_error: Option<bool>,
) -> CallToolResult {
let content_text = text;
let content = vec![Content::text(content_text.clone())];
let content = vec![ContentBlock::text(content_text.clone())];
let structured_content = json!({
"threadId": thread_id,
"content": content_text,
@@ -73,7 +73,7 @@ pub async fn run_codex_tool_session(
{
Ok(res) => res,
Err(e) => {
let result = CallToolResult::error(vec![Content::text(format!(
let result = CallToolResult::error(vec![ContentBlock::text(format!(
"Failed to start Codex session: {e}"
))]);
outgoing.send_response(id.clone(), result).await;
+29 -30
View File
@@ -147,16 +147,8 @@ impl MessageProcessor {
ClientRequest::CompleteRequest(params) => {
self.handle_complete(params.params);
}
ClientRequest::GetTaskInfoRequest(_) => {
self.handle_unsupported_request(request_id, "tasks/get_info")
.await;
}
ClientRequest::ListTasksRequest(_) => {
self.handle_unsupported_request(request_id, "tasks/list")
.await;
}
ClientRequest::GetTaskResultRequest(_) => {
self.handle_unsupported_request(request_id, "tasks/get_result")
ClientRequest::GetTaskRequest(_) => {
self.handle_unsupported_request(request_id, "tasks/get")
.await;
}
ClientRequest::CancelTaskRequest(_) => {
@@ -176,6 +168,10 @@ impl MessageProcessor {
)
.await;
}
request => {
self.handle_unsupported_request(request_id, request.method())
.await;
}
}
}
@@ -205,6 +201,9 @@ impl MessageProcessor {
ClientNotification::CustomNotification(_) => {
tracing::warn!("ignoring custom client notification");
}
_ => {
tracing::warn!("ignoring unsupported client notification");
}
}
}
@@ -330,14 +329,10 @@ impl MessageProcessor {
params: Option<rmcp::model::PaginatedRequestParams>,
) {
tracing::trace!("tools/list -> {params:?}");
let result = rmcp::model::ListToolsResult {
meta: None,
tools: vec![
create_tool_for_codex_tool_call_param(),
create_tool_for_codex_tool_call_reply_param(),
],
next_cursor: None,
};
let result = rmcp::model::ListToolsResult::with_all_items(vec![
create_tool_for_codex_tool_call_param(),
create_tool_for_codex_tool_call_reply_param(),
]);
self.outgoing.send_response(id, result).await;
}
@@ -355,7 +350,7 @@ impl MessageProcessor {
.await
}
_ => {
let result = CallToolResult::error(vec![rmcp::model::Content::text(format!(
let result = CallToolResult::error(vec![rmcp::model::ContentBlock::text(format!(
"Unknown tool '{name}'"
))]);
self.outgoing.send_response(id, result).await;
@@ -374,7 +369,7 @@ impl MessageProcessor {
Ok(tool_cfg) => match tool_cfg.into_config(self.arg0_paths.clone()).await {
Ok(cfg) => cfg,
Err(e) => {
let result = CallToolResult::error(vec![rmcp::model::Content::text(
let result = CallToolResult::error(vec![rmcp::model::ContentBlock::text(
format!("Failed to load Codex configuration from overrides: {e}"),
)]);
self.outgoing.send_response(id, result).await;
@@ -382,15 +377,15 @@ impl MessageProcessor {
}
},
Err(e) => {
let result = CallToolResult::error(vec![rmcp::model::Content::text(format!(
"Failed to parse configuration for Codex tool: {e}"
))]);
let result = CallToolResult::error(vec![rmcp::model::ContentBlock::text(
format!("Failed to parse configuration for Codex tool: {e}"),
)]);
self.outgoing.send_response(id, result).await;
return;
}
},
None => {
let result = CallToolResult::error(vec![rmcp::model::Content::text(
let result = CallToolResult::error(vec![rmcp::model::ContentBlock::text(
"Missing arguments for codex tool-call; the `prompt` field is required.",
)]);
self.outgoing.send_response(id, result).await;
@@ -433,9 +428,9 @@ impl MessageProcessor {
Ok(params) => params,
Err(e) => {
tracing::error!("Failed to parse Codex tool call reply parameters: {e}");
let result = CallToolResult::error(vec![rmcp::model::Content::text(format!(
"Failed to parse configuration for Codex tool: {e}"
))]);
let result = CallToolResult::error(vec![rmcp::model::ContentBlock::text(
format!("Failed to parse configuration for Codex tool: {e}"),
)]);
self.outgoing.send_response(request_id, result).await;
return;
}
@@ -444,7 +439,7 @@ impl MessageProcessor {
tracing::error!(
"Missing arguments for codex-reply tool-call; the `thread_id` and `prompt` fields are required."
);
let result = CallToolResult::error(vec![rmcp::model::Content::text(
let result = CallToolResult::error(vec![rmcp::model::ContentBlock::text(
"Missing arguments for codex-reply tool-call; the `thread_id` and `prompt` fields are required.",
)]);
self.outgoing.send_response(request_id, result).await;
@@ -456,7 +451,7 @@ impl MessageProcessor {
Ok(id) => id,
Err(e) => {
tracing::error!("Failed to parse thread_id: {e}");
let result = CallToolResult::error(vec![rmcp::model::Content::text(format!(
let result = CallToolResult::error(vec![rmcp::model::ContentBlock::text(format!(
"Failed to parse thread_id: {e}"
))]);
self.outgoing.send_response(request_id, result).await;
@@ -502,6 +497,7 @@ impl MessageProcessor {
});
}
#[allow(deprecated)]
fn handle_set_level(&self, params: rmcp::model::SetLevelRequestParams) {
tracing::info!("logging/setLevel -> params: {:?}", params);
}
@@ -528,7 +524,10 @@ impl MessageProcessor {
// ---------------------------------------------------------------------
async fn handle_cancelled_notification(&self, params: rmcp::model::CancelledNotificationParam) {
let request_id = params.request_id;
let Some(request_id) = params.request_id else {
tracing::warn!("ignoring cancellation without a request id");
return;
};
// Create a stable string form early for logging and submission id.
let request_id_string = request_id.to_string();
+30 -1
View File
@@ -80,7 +80,7 @@ impl OutgoingMessageSender {
}
pub(crate) async fn send_response<T: Serialize>(&self, id: RequestId, response: T) {
let result = match serde_json::to_value(response) {
let mut result = match serde_json::to_value(response) {
Ok(result) => result,
Err(err) => {
self.send_error(
@@ -92,6 +92,16 @@ impl OutgoingMessageSender {
}
};
// rmcp result constructors include the modern discriminator by default.
// This legacy server serializes responses directly, bypassing rmcp's
// protocol-aware response handling, so preserve the historical wire shape:
// https://github.com/modelcontextprotocol/rust-sdk/issues/1036
if let Value::Object(object) = &mut result
&& object.get("resultType").and_then(Value::as_str) == Some("complete")
{
object.remove("resultType");
}
let outgoing_message = OutgoingMessage::Response(OutgoingResponse { id, result });
let _ = self.sender.send(outgoing_message);
}
@@ -286,6 +296,25 @@ mod tests {
);
}
#[tokio::test]
async fn outgoing_tool_response_preserves_legacy_wire_format() {
let (outgoing_tx, mut outgoing_rx) = mpsc::unbounded_channel::<OutgoingMessage>();
let outgoing_message_sender = OutgoingMessageSender::new(outgoing_tx);
outgoing_message_sender
.send_response(
RequestId::Number(1),
rmcp::model::CallToolResult::success(Vec::new()),
)
.await;
let Some(OutgoingMessage::Response(response)) = outgoing_rx.recv().await else {
panic!("expected a tool-call response");
};
assert_eq!(response.id, RequestId::Number(1));
assert_eq!(response.result, json!({ "content": [], "isError": false }));
}
#[tokio::test]
async fn test_send_event_as_notification() -> Result<()> {
let (outgoing_tx, mut outgoing_rx) = mpsc::unbounded_channel::<OutgoingMessage>();