valknar/llmx
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: leverage elicitations in the MCP server (#1623)

Browse Source 
This updates the MCP server so that if it receives an
`ExecApprovalRequest` from the `Codex` session, it in turn sends an [MCP
elicitation](https://modelcontextprotocol.io/specification/draft/client/elicitation)
to the client to ask for the approval decision. Upon getting a response,
it forwards the client's decision via `Op::ExecApproval`.

Admittedly, we should be doing the same thing for
`ApplyPatchApprovalRequest`, but this is our first time experimenting
with elicitations, so I'm inclined to defer wiring that code path up
until we feel good about how this one works.

---
[//]: # (BEGIN SAPLING FOOTER)
Stack created with [Sapling](https://sapling-scm.com). Best reviewed
with [ReviewStack](https://reviewstack.dev/openai/codex/pull/1623).
* __->__ #1623
* #1622
* #1621
* #1620
This commit is contained in:
Michael Bolin
2025-07-19 01:32:03 -04:00
committed by GitHub
parent 11fd3123be
commit 018003e52f
9 changed files with 149 additions and 22 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
codex-rs/Cargo.lock generated
View File

@@ -799,6 +799,7 @@ dependencies = [
"schemars 0.8.22",
"serde",
"serde_json",
"shlex",
"tokio",
"toml 0.9.1",
"tracing",

5
codex-rs/core/src/mcp_connection_manager.rs
View File

@@ -18,6 +18,7 @@ use mcp_types::ClientCapabilities;
use mcp_types::Implementation;
use mcp_types::Tool;
use serde_json::json;
use sha1::Digest;
use sha1::Sha1;
use tokio::task::JoinSet;
@@ -135,7 +136,9 @@ impl McpConnectionManager {
experimental: None,
roots: None,
sampling: None,
elicitation: None,
// https://modelcontextprotocol.io/specification/2025-06-18/client/elicitation#capabilities
// indicates this should be an empty object.
elicitation: Some(json!({})),
},
client_info: Implementation {
name: "codex-mcp-client".to_owned(),

1
codex-rs/mcp-server/Cargo.toml
View File

@@ -22,6 +22,7 @@ mcp-types = { path = "../mcp-types" }
schemars = "0.8.22"
serde = { version = "1", features = ["derive"] }
serde_json = "1"
shlex = "1.3.0"
toml = "0.9"
tracing = { version = "0.1.41", features = ["log"] }
tracing-subscriber = { version = "0.3", features = ["fmt", "env-filter"] }

107
codex-rs/mcp-server/src/codex_tool_runner.rs
View File

@@ -4,18 +4,27 @@
use std::sync::Arc;
use codex_core::Codex;
use codex_core::codex_wrapper::init_codex;
use codex_core::config::Config as CodexConfig;
use codex_core::protocol::AgentMessageEvent;
use codex_core::protocol::EventMsg;
use codex_core::protocol::ExecApprovalRequestEvent;
use codex_core::protocol::InputItem;
use codex_core::protocol::Op;
use codex_core::protocol::ReviewDecision;
use codex_core::protocol::Submission;
use codex_core::protocol::TaskCompleteEvent;
use mcp_types::CallToolResult;
use mcp_types::ContentBlock;
use mcp_types::ElicitRequest;
use mcp_types::ElicitRequestParamsRequestedSchema;
use mcp_types::ModelContextProtocolRequest;
use mcp_types::RequestId;
use mcp_types::TextContent;
use serde::Deserialize;
use serde_json::json;
use tracing::error;
use crate::outgoing_message::OutgoingMessageSender;
@@ -45,6 +54,7 @@ pub async fn run_codex_tool_session(
return;
}
};
let codex = Arc::new(codex);
// Send initial SessionConfigured event.
outgoing.send_event_as_notification(&first_event).await;
@@ -58,7 +68,7 @@ pub async fn run_codex_tool_session(
};
let submission = Submission {
id: sub_id,
id: sub_id.clone(),
op: Op::UserInput {
items: vec![InputItem::Text {
text: initial_prompt.clone(),
@@ -77,18 +87,50 @@ pub async fn run_codex_tool_session(
Ok(event) => {
outgoing.send_event_as_notification(&event).await;
match &event.msg {
EventMsg::ExecApprovalRequest(_) => {
let result = CallToolResult {
content: vec![ContentBlock::TextContent(TextContent {
r#type: "text".to_string(),
text: "EXEC_APPROVAL_REQUIRED".to_string(),
annotations: None,
})],
is_error: None,
structured_content: None,
};
outgoing.send_response(id.clone(), result.into()).await;
match event.msg {
EventMsg::ExecApprovalRequest(ExecApprovalRequestEvent {
command,
cwd,
reason: _,
}) => {
let escaped_command = shlex::try_join(command.iter().map(|s| s.as_str()))
.unwrap_or_else(|_| command.join(" "));
let message = format!("Allow Codex to run `{escaped_command}` in {cwd:?}?");
let params = json!({
// These fields are required so that `params`
// conforms to ElicitRequestParams.
"message": message,
"requestedSchema": ElicitRequestParamsRequestedSchema {
r#type: "object".to_string(),
properties: json!({}),
required: None,
},
// These are additional fields the client can use to
// correlate the request with the codex tool call.
"codex_elicitation": "exec-approval",
"codex_mcp_tool_call_id": sub_id,
"codex_event_id": event.id,
"codex_command": command,
// Could convert it to base64 encoded bytes if we
// don't want to use to_string_lossy() here?
"codex_cwd": cwd.to_string_lossy().to_string()
});
let on_response = outgoing
.send_request(ElicitRequest::METHOD, Some(params))
.await;
// Listen for the response on a separate task so we do
// not block the main loop of this function.
{
let codex = codex.clone();
let event_id = event.id.clone();
tokio::spawn(async move {
on_exec_approval_response(event_id, on_response, codex).await;
});
}
break;
}
EventMsg::ApplyPatchApprovalRequest(_) => {
@@ -172,3 +214,42 @@ pub async fn run_codex_tool_session(
}
}
}
async fn on_exec_approval_response(
event_id: String,
receiver: tokio::sync::oneshot::Receiver<mcp_types::Result>,
codex: Arc<Codex>,
) {
let response = receiver.await;
let value = match response {
Ok(value) => value,
Err(err) => {
error!("request failed: {err:?}");
return;
}
};
// Try to deserialize `value` and then make the appropriate call to `codex`.
let response = match serde_json::from_value::<ExecApprovalResponse>(value) {
Ok(response) => response,
Err(err) => {
error!("failed to deserialize ExecApprovalResponse: {err}");
return;
}
};
if let Err(err) = codex
.submit(Op::ExecApproval {
id: event_id,
decision: response.decision,
})
.await
{
error!("failed to submit ExecApproval: {err}");
}
}
#[derive(Debug, Deserialize)]
pub struct ExecApprovalResponse {
pub decision: ReviewDecision,
}

2
codex-rs/mcp-server/src/lib.rs
View File

@@ -72,7 +72,7 @@ pub async fn run_main(codex_linux_sandbox_exe: Option<PathBuf>) -> IoResult<()>
while let Some(msg) = incoming_rx.recv().await {
match msg {
JSONRPCMessage::Request(r) => processor.process_request(r).await,
JSONRPCMessage::Response(r) => processor.process_response(r),
JSONRPCMessage::Response(r) => processor.process_response(r).await,
JSONRPCMessage::Notification(n) => processor.process_notification(n),
JSONRPCMessage::Error(e) => processor.process_error(e),
}

4
codex-rs/mcp-server/src/message_processor.rs
View File

@@ -101,8 +101,10 @@ impl MessageProcessor {
}
/// Handle a standalone JSON-RPC response originating from the peer.
pub(crate) fn process_response(&mut self, response: JSONRPCResponse) {
pub(crate) async fn process_response(&mut self, response: JSONRPCResponse) {
tracing::info!("<- response: {:?}", response);
let JSONRPCResponse { id, result, .. } = response;
self.outgoing.notify_client_response(id, result).await
}
/// Handle a fire-and-forget JSON-RPC notification.

42
codex-rs/mcp-server/src/outgoing_message.rs
View File

@@ -1,3 +1,4 @@
use std::collections::HashMap;
use std::sync::atomic::AtomicI64;
use std::sync::atomic::Ordering;
@@ -12,11 +13,15 @@ use mcp_types::JSONRPCResponse;
use mcp_types::RequestId;
use mcp_types::Result;
use serde::Serialize;
use tokio::sync::Mutex;
use tokio::sync::mpsc;
use tokio::sync::oneshot;
use tracing::warn;
pub(crate) struct OutgoingMessageSender {
next_request_id: AtomicI64,
sender: mpsc::Sender<OutgoingMessage>,
request_id_to_callback: Mutex<HashMap<RequestId, oneshot::Sender<Result>>>,
}
impl OutgoingMessageSender {
@@ -24,17 +29,48 @@ impl OutgoingMessageSender {
Self {
next_request_id: AtomicI64::new(0),
sender,
request_id_to_callback: Mutex::new(HashMap::new()),
}
}
#[allow(dead_code)]
pub(crate) async fn send_request(&self, method: &str, params: Option<serde_json::Value>) {
pub(crate) async fn send_request(
&self,
method: &str,
params: Option<serde_json::Value>,
) -> oneshot::Receiver<Result> {
let id = RequestId::Integer(self.next_request_id.fetch_add(1, Ordering::Relaxed));
let outgoing_message_id = id.clone();
let (tx_approve, rx_approve) = oneshot::channel();
{
let mut request_id_to_callback = self.request_id_to_callback.lock().await;
request_id_to_callback.insert(id, tx_approve);
}
let outgoing_message = OutgoingMessage::Request(OutgoingRequest {
id: RequestId::Integer(self.next_request_id.fetch_add(1, Ordering::Relaxed)),
id: outgoing_message_id,
method: method.to_string(),
params,
});
let _ = self.sender.send(outgoing_message).await;
rx_approve
}
pub(crate) async fn notify_client_response(&self, id: RequestId, result: Result) {
let entry = {
let mut request_id_to_callback = self.request_id_to_callback.lock().await;
request_id_to_callback.remove_entry(&id)
};
match entry {
Some((id, sender)) => {
if let Err(err) = sender.send(result) {
warn!("could not notify callback for {id:?} due to: {err:?}");
}
}
None => {
warn!("could not find callback for {id:?}");
}
}
}
pub(crate) async fn send_response(&self, id: RequestId, result: Result) {

5
codex-rs/mcp-types/generate_mcp_types.py
View File

@@ -18,6 +18,9 @@ SCHEMA_VERSION = "2025-06-18"
JSONRPC_VERSION = "2.0"
STANDARD_DERIVE = "#[derive(Debug, Clone, PartialEq, Deserialize, Serialize)]\n"
STANDARD_HASHABLE_DERIVE = (
"#[derive(Debug, Clone, PartialEq, Deserialize, Serialize, Hash, Eq)]\n"
)
# Will be populated with the schema's `definitions` map in `main()` so that
# helper functions (for example `define_any_of`) can perform look-ups while
@@ -391,7 +394,7 @@ def define_string_enum(
def define_untagged_enum(name: str, type_list: list[str], out: list[str]) -> None:
out.append(STANDARD_DERIVE)
out.append(STANDARD_HASHABLE_DERIVE)
out.append("#[serde(untagged)]\n")
out.append(f"pub enum {name} {{\n")
for simple_type in type_list:

4
codex-rs/mcp-types/src/lib.rs
View File

@@ -931,7 +931,7 @@ pub struct ProgressNotificationParams {
pub total: Option<f64>,
}
#[derive(Debug, Clone, PartialEq, Deserialize, Serialize)]
#[derive(Debug, Clone, PartialEq, Deserialize, Serialize, Hash, Eq)]
#[serde(untagged)]
pub enum ProgressToken {
String(String),
@@ -1031,7 +1031,7 @@ pub struct Request {
pub params: Option<serde_json::Value>,
}
#[derive(Debug, Clone, PartialEq, Deserialize, Serialize)]
#[derive(Debug, Clone, PartialEq, Deserialize, Serialize, Hash, Eq)]
#[serde(untagged)]
pub enum RequestId {
String(String),