openai/codex
1
0
Fork 0
mirror of https://github.com/openai/codex.git synced 2026-09-16 12:13:30 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat: add conversationId to MCP server messages

Browse Source
This commit is contained in:
Michael Bolin
2026-01-13 21:10:07 -08:00
parent 31d9b6f4d2
commit 80aabc4232
7 changed files with 275 additions and 57 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -32,6 +32,24 @@ use tokio::sync::Mutex;
pub(crate) const INVALID_PARAMS_ERROR_CODE: i64 = -32602;
fn call_tool_result_with_thread_id(
thread_id: ThreadId,
text: String,
is_error: Option<bool>,
) -> CallToolResult {
CallToolResult {
content: vec![ContentBlock::TextContent(TextContent {
r#type: "text".to_string(),
text,
annotations: None,
})],
is_error,
structured_content: Some(json!({
"threadId": thread_id,
})),
}
}
/// Run a complete Codex session and stream events back to the client.
///
/// On completion (success or error) the function sends the appropriate
@@ -73,7 +91,10 @@ pub async fn run_codex_tool_session(
outgoing
.send_event_as_notification(
&session_configured_event,
Some(OutgoingNotificationMeta::new(Some(id.clone()))),
Some(OutgoingNotificationMeta {
request_id: Some(id.clone()),
thread_id: Some(thread_id),
}),
)
.await;
@@ -100,12 +121,25 @@ pub async fn run_codex_tool_session(
if let Err(e) = thread.submit_with_id(submission).await {
tracing::error!("Failed to submit initial prompt: {e}");
let result = call_tool_result_with_thread_id(
thread_id,
format!("Failed to submit initial prompt: {e}"),
Some(true),
);
outgoing.send_response(id.clone(), result).await;
// unregister the id so we don't keep it in the map
running_requests_id_to_codex_uuid.lock().await.remove(&id);
return;
}
run_codex_tool_session_inner(thread, outgoing, id, running_requests_id_to_codex_uuid).await;
run_codex_tool_session_inner(
thread,
outgoing,
id,
thread_id,
running_requests_id_to_codex_uuid,
)
.await;
}
pub async fn run_codex_tool_session_reply(
@@ -114,12 +148,12 @@ pub async fn run_codex_tool_session_reply(
request_id: RequestId,
prompt: String,
running_requests_id_to_codex_uuid: Arc<Mutex<HashMap<RequestId, ThreadId>>>,
conversation_id: ThreadId,
thread_id: ThreadId,
) {
running_requests_id_to_codex_uuid
.lock()
.await
.insert(request_id.clone(), conversation_id);
.insert(request_id.clone(), thread_id);
if let Err(e) = conversation
.submit(Op::UserInput {
items: vec![UserInput::Text { text: prompt }],
@@ -128,6 +162,12 @@ pub async fn run_codex_tool_session_reply(
.await
{
tracing::error!("Failed to submit user input: {e}");
let result = call_tool_result_with_thread_id(
thread_id,
format!("Failed to submit user input: {e}"),
Some(true),
);
outgoing.send_response(request_id.clone(), result).await;
// unregister the id so we don't keep it in the map
running_requests_id_to_codex_uuid
.lock()
@@ -140,6 +180,7 @@ pub async fn run_codex_tool_session_reply(
conversation,
outgoing,
request_id,
thread_id,
running_requests_id_to_codex_uuid,
)
.await;
@@ -149,6 +190,7 @@ async fn run_codex_tool_session_inner(
codex: Arc<CodexThread>,
outgoing: Arc<OutgoingMessageSender>,
request_id: RequestId,
thread_id: ThreadId,
running_requests_id_to_codex_uuid: Arc<Mutex<HashMap<RequestId, ThreadId>>>,
) {
let request_id_str = match &request_id {
@@ -164,7 +206,10 @@ async fn run_codex_tool_session_inner(
outgoing
.send_event_as_notification(
&event,
Some(OutgoingNotificationMeta::new(Some(request_id.clone()))),
Some(OutgoingNotificationMeta {
request_id: Some(request_id.clone()),
thread_id: Some(thread_id),
}),
)
.await;
@@ -188,15 +233,18 @@ async fn run_codex_tool_session_inner(
event.id.clone(),
call_id,
parsed_cmd,
thread_id,
)
.await;
continue;
}
EventMsg::Error(err_event) => {
// Return a response to conclude the tool call when the Codex session reports an error (e.g., interruption).
let result = json!({
"error": err_event.message,
});
// Always respond in tools/call's expected shape, and include conversationId so the client can resume.
let result = call_tool_result_with_thread_id(
thread_id,
err_event.message,
Some(true),
);
outgoing.send_response(request_id.clone(), result).await;
break;
}
@@ -224,6 +272,7 @@ async fn run_codex_tool_session_inner(
request_id.clone(),
request_id_str.clone(),
event.id.clone(),
thread_id,
)
.await;
continue;
@@ -233,15 +282,7 @@ async fn run_codex_tool_session_inner(
Some(msg) => msg,
None => "".to_string(),
};
let result = CallToolResult {
content: vec![ContentBlock::TextContent(TextContent {
r#type: "text".to_string(),
text,
annotations: None,
})],
is_error: None,
structured_content: None,
};
let result = call_tool_result_with_thread_id(thread_id, text, None);
outgoing.send_response(request_id.clone(), result).await;
// unregister the id so we don't keep it in the map
running_requests_id_to_codex_uuid
@@ -317,20 +358,32 @@ async fn run_codex_tool_session_inner(
}
}
Err(e) => {
let result = CallToolResult {
content: vec![ContentBlock::TextContent(TextContent {
r#type: "text".to_string(),
text: format!("Codex runtime error: {e}"),
annotations: None,
})],
is_error: Some(true),
// TODO(mbolin): Could present the error in a more
// structured way.
structured_content: None,
};
let result = call_tool_result_with_thread_id(
thread_id,
format!("Codex runtime error: {e}"),
Some(true),
);
outgoing.send_response(request_id.clone(), result).await;
break;
}
}
}
}
#[cfg(test)]
mod tests {
use super::*;
use pretty_assertions::assert_eq;
#[test]
fn call_tool_result_includes_thread_id_in_structured_content() {
let thread_id = ThreadId::new();
let result = call_tool_result_with_thread_id(thread_id, "done".to_string(), None);
assert_eq!(
result.structured_content,
Some(json!({
"threadId": thread_id,
}))
);
}
}

5
codex-rs/mcp-server/src/exec_approval.rs
View File

@@ -4,6 +4,7 @@ use std::sync::Arc;
use codex_core::CodexThread;
use codex_core::protocol::Op;
use codex_core::protocol::ReviewDecision;
use codex_protocol::ThreadId;
use codex_protocol::parse_command::ParsedCommand;
use mcp_types::ElicitRequest;
use mcp_types::ElicitRequestParamsRequestedSchema;
@@ -30,6 +31,8 @@ pub struct ExecApprovalElicitRequestParams {
// These are additional fields the client can use to
// correlate the request with the codex tool call.
#[serde(rename = "threadId")]
pub thread_id: ThreadId,
pub codex_elicitation: String,
pub codex_mcp_tool_call_id: String,
pub codex_event_id: String,
@@ -59,6 +62,7 @@ pub(crate) async fn handle_exec_approval_request(
event_id: String,
call_id: String,
codex_parsed_cmd: Vec<ParsedCommand>,
thread_id: ThreadId,
) {
let escaped_command =
shlex::try_join(command.iter().map(String::as_str)).unwrap_or_else(|_| command.join(" "));
@@ -74,6 +78,7 @@ pub(crate) async fn handle_exec_approval_request(
properties: json!({}),
required: None,
},
thread_id,
codex_elicitation: "exec-approval".to_string(),
codex_mcp_tool_call_id: tool_call_id.clone(),
codex_event_id: event_id.clone(),

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

@@ -507,7 +507,9 @@ impl MessageProcessor {
annotations: None,
})],
is_error: Some(true),
structured_content: None,
structured_content: Some(json!({
"conversationId": conversation_id,
})),
};
outgoing.send_response(request_id, result).await;
return;

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

@@ -3,6 +3,7 @@ use std::sync::atomic::AtomicI64;
use std::sync::atomic::Ordering;
use codex_core::protocol::Event;
use codex_protocol::ThreadId;
use mcp_types::JSONRPC_VERSION;
use mcp_types::JSONRPCError;
use mcp_types::JSONRPCErrorError;
@@ -209,12 +210,8 @@ pub(crate) struct OutgoingNotificationParams {
#[serde(rename_all = "camelCase")]
pub(crate) struct OutgoingNotificationMeta {
pub request_id: Option<RequestId>,
}
impl OutgoingNotificationMeta {
pub(crate) fn new(request_id: Option<RequestId>) -> Self {
Self { request_id }
}
#[serde(default, skip_serializing_if = "Option::is_none")]
pub thread_id: Option<ThreadId>,
}
#[derive(Debug, Clone, PartialEq, Serialize)]
@@ -251,12 +248,12 @@ mod tests {
let (outgoing_tx, mut outgoing_rx) = mpsc::unbounded_channel::<OutgoingMessage>();
let outgoing_message_sender = OutgoingMessageSender::new(outgoing_tx);
let conversation_id = ThreadId::new();
let thread_id = ThreadId::new();
let rollout_file = NamedTempFile::new()?;
let event = Event {
id: "1".to_string(),
msg: EventMsg::SessionConfigured(SessionConfiguredEvent {
session_id: conversation_id,
session_id: thread_id,
model: "gpt-4o".to_string(),
model_provider_id: "test-provider".to_string(),
approval_policy: AskForApproval::Never,
@@ -313,6 +310,7 @@ mod tests {
};
let meta = OutgoingNotificationMeta {
request_id: Some(RequestId::String("123".to_string())),
thread_id: None,
};
outgoing_message_sender
@@ -348,4 +346,68 @@ mod tests {
assert_eq!(params.unwrap(), expected_params);
Ok(())
}
#[tokio::test]
async fn test_send_event_as_notification_with_meta_and_conversation_id() -> Result<()> {
let (outgoing_tx, mut outgoing_rx) = mpsc::unbounded_channel::<OutgoingMessage>();
let outgoing_message_sender = OutgoingMessageSender::new(outgoing_tx);
let thread_id = ThreadId::new();
let rollout_file = NamedTempFile::new()?;
let session_configured_event = SessionConfiguredEvent {
session_id: thread_id,
model: "gpt-4o".to_string(),
model_provider_id: "test-provider".to_string(),
approval_policy: AskForApproval::Never,
sandbox_policy: SandboxPolicy::ReadOnly,
cwd: PathBuf::from("/home/user/project"),
reasoning_effort: Some(ReasoningEffort::default()),
history_log_id: 1,
history_entry_count: 1000,
initial_messages: None,
rollout_path: rollout_file.path().to_path_buf(),
};
let event = Event {
id: "1".to_string(),
msg: EventMsg::SessionConfigured(session_configured_event.clone()),
};
let meta = OutgoingNotificationMeta {
request_id: Some(RequestId::String("123".to_string())),
thread_id: Some(thread_id),
};
outgoing_message_sender
.send_event_as_notification(&event, Some(meta))
.await;
let result = outgoing_rx.recv().await.unwrap();
let OutgoingMessage::Notification(OutgoingNotification { method, params }) = result else {
panic!("expected Notification for first message");
};
assert_eq!(method, "codex/event");
let expected_params = json!({
"_meta": {
"requestId": "123",
"threadId": thread_id.to_string(),
},
"id": "1",
"msg": {
"type": "session_configured",
"session_id": session_configured_event.session_id,
"model": "gpt-4o",
"model_provider_id": "test-provider",
"approval_policy": "never",
"sandbox_policy": {
"type": "read-only"
},
"cwd": "/home/user/project",
"reasoning_effort": session_configured_event.reasoning_effort,
"history_log_id": session_configured_event.history_log_id,
"history_entry_count": session_configured_event.history_entry_count,
"rollout_path": rollout_file.path().to_path_buf(),
}
});
assert_eq!(params.unwrap(), expected_params);
Ok(())
}
}

7
codex-rs/mcp-server/src/patch_approval.rs
View File

@@ -6,6 +6,7 @@ use codex_core::CodexThread;
use codex_core::protocol::FileChange;
use codex_core::protocol::Op;
use codex_core::protocol::ReviewDecision;
use codex_protocol::ThreadId;
use mcp_types::ElicitRequest;
use mcp_types::ElicitRequestParamsRequestedSchema;
use mcp_types::JSONRPCErrorError;
@@ -19,11 +20,13 @@ use tracing::error;
use crate::codex_tool_runner::INVALID_PARAMS_ERROR_CODE;
use crate::outgoing_message::OutgoingMessageSender;
#[derive(Debug, Serialize)]
#[derive(Debug, Deserialize, Serialize)]
pub struct PatchApprovalElicitRequestParams {
pub message: String,
#[serde(rename = "requestedSchema")]
pub requested_schema: ElicitRequestParamsRequestedSchema,
#[serde(rename = "threadId")]
pub thread_id: ThreadId,
pub codex_elicitation: String,
pub codex_mcp_tool_call_id: String,
pub codex_event_id: String,
@@ -51,6 +54,7 @@ pub(crate) async fn handle_patch_approval_request(
request_id: RequestId,
tool_call_id: String,
event_id: String,
thread_id: ThreadId,
) {
let mut message_lines = Vec::new();
if let Some(r) = &reason {
@@ -65,6 +69,7 @@ pub(crate) async fn handle_patch_approval_request(
properties: json!({}),
required: None,
},
thread_id,
codex_elicitation: "patch-approval".to_string(),
codex_mcp_tool_call_id: tool_call_id.clone(),
codex_event_id: event_id.clone(),

60
codex-rs/mcp-server/tests/suite/codex_tool.rs
View File

@@ -119,6 +119,7 @@ async fn shell_command_approval_triggers_elicitation() -> anyhow::Result<()> {
workdir_for_shell_function_call.path(),
codex_request_id.to_string(),
params.codex_event_id.clone(),
params.thread_id,
)?;
assert_eq!(expected_elicitation_request, elicitation_request);
@@ -158,7 +159,10 @@ async fn shell_command_approval_triggers_elicitation() -> anyhow::Result<()> {
"text": "File created!",
"type": "text"
}
]
],
"structuredContent": {
"threadId": params.thread_id,
}
}),
},
codex_response
@@ -175,6 +179,7 @@ fn create_expected_elicitation_request(
workdir: &Path,
codex_mcp_tool_call_id: String,
codex_event_id: String,
thread_id: codex_protocol::ThreadId,
) -> anyhow::Result<JSONRPCRequest> {
let expected_message = format!(
"Allow Codex to run `{}` in `{}`?",
@@ -193,6 +198,7 @@ fn create_expected_elicitation_request(
properties: json!({}),
required: None,
},
thread_id,
codex_elicitation: "exec-approval".to_string(),
codex_mcp_tool_call_id,
codex_event_id,
@@ -260,7 +266,13 @@ async fn patch_approval_triggers_elicitation() -> anyhow::Result<()> {
)
.await??;
let elicitation_request_id = RequestId::Integer(0);
let elicitation_request_id = elicitation_request.id.clone();
let params = serde_json::from_value::<PatchApprovalElicitRequestParams>(
elicitation_request
.params
.clone()
.ok_or_else(|| anyhow::anyhow!("elicitation_request.params must be set"))?,
)?;
let mut expected_changes = HashMap::new();
expected_changes.insert(
@@ -277,7 +289,8 @@ async fn patch_approval_triggers_elicitation() -> anyhow::Result<()> {
None, // No grant_root expected
None, // No reason expected
codex_request_id.to_string(),
"1".to_string(),
params.codex_event_id.clone(),
params.thread_id,
)?;
assert_eq!(expected_elicitation_request, elicitation_request);
@@ -307,7 +320,10 @@ async fn patch_approval_triggers_elicitation() -> anyhow::Result<()> {
"text": "Patch has been applied successfully!",
"type": "text"
}
]
],
"structuredContent": {
"threadId": params.thread_id,
}
}),
},
codex_response
@@ -331,7 +347,7 @@ async fn test_codex_tool_passes_base_instructions() {
}
async fn codex_tool_passes_base_instructions() -> anyhow::Result<()> {
#![expect(clippy::unwrap_used)]
#![expect(clippy::expect_used, clippy::unwrap_used)]
let server =
create_mock_chat_completions_server(vec![create_final_assistant_message_sse_response(
@@ -360,20 +376,26 @@ async fn codex_tool_passes_base_instructions() -> anyhow::Result<()> {
mcp_process.read_stream_until_response_message(RequestId::Integer(codex_request_id)),
)
.await??;
assert_eq!(codex_response.jsonrpc, JSONRPC_VERSION);
assert_eq!(codex_response.id, RequestId::Integer(codex_request_id));
assert_eq!(
JSONRPCResponse {
jsonrpc: JSONRPC_VERSION.into(),
id: RequestId::Integer(codex_request_id),
result: json!({
"content": [
{
"text": "Enjoy!",
"type": "text"
}
]
}),
},
codex_response
codex_response.result,
json!({
"content": [
{
"text": "Enjoy!",
"type": "text"
}
],
"structuredContent": {
"threadId": codex_response
.result
.get("structuredContent")
.and_then(|v| v.get("threadId"))
.and_then(serde_json::Value::as_str)
.expect("codex tool response should include structuredContent.threadId"),
}
})
);
let requests = server.received_requests().await.unwrap();
@@ -412,6 +434,7 @@ fn create_expected_patch_approval_elicitation_request(
reason: Option<String>,
codex_mcp_tool_call_id: String,
codex_event_id: String,
thread_id: codex_protocol::ThreadId,
) -> anyhow::Result<JSONRPCRequest> {
let mut message_lines = Vec::new();
if let Some(r) = &reason {
@@ -430,6 +453,7 @@ fn create_expected_patch_approval_elicitation_request(
properties: json!({}),
required: None,
},
thread_id,
codex_elicitation: "patch-approval".to_string(),
codex_mcp_tool_call_id,
codex_event_id,

67
docs/codex-mcp-server.md Normal file
View File

@@ -0,0 +1,67 @@
## Using Codex as an MCP Server
The Codex CLI can also be run as an MCP _server_ via `codex mcp-server`. For example, you can use `codex mcp-server` to make Codex available as a tool inside of a multi-agent framework like the OpenAI [Agents SDK](https://platform.openai.com/docs/guides/agents). Use `codex mcp` separately to add/list/get/remove MCP server launchers in your configuration.
### Codex MCP Server Quickstart
You can launch a Codex MCP server with the [Model Context Protocol Inspector](https://modelcontextprotocol.io/legacy/tools/inspector):
```bash
npx @modelcontextprotocol/inspector codex mcp-server
```
Send a `tools/list` request and you will see that there are two tools available:
**`codex`** - Run a Codex session. Accepts configuration parameters matching the Codex Config struct. The `codex` tool takes the following properties:
| Property | Type | Description |
| ----------------------- | ------ | ------------------------------------------------------------------------------------------------------------------------------------------------------ |
| **`prompt`** (required) | string | The initial user prompt to start the Codex conversation. |
| `approval-policy` | string | Approval policy for shell commands generated by the model: `untrusted`, `on-failure`, `on-request`, `never`. |
| `base-instructions` | string | The set of instructions to use instead of the default ones. |
| `config` | object | Individual [config settings](https://github.com/openai/codex/blob/main/docs/config.md#config) that will override what is in `$CODEX_HOME/config.toml`. |
| `cwd` | string | Working directory for the session. If relative, resolved against the server process's current directory. |
| `model` | string | Optional override for the model name (e.g. `o3`, `o4-mini`). |
| `profile` | string | Configuration profile from `config.toml` to specify default options. |
| `sandbox` | string | Sandbox mode: `read-only`, `workspace-write`, or `danger-full-access`. |
The `tools/call` response for `codex` includes the thread id in `structuredContent` so the client can resume the session later:
```json
{
"content": [{ "type": "text", "text": "..." }],
"structuredContent": { "threadId": "..." }
}
```
While the tool is running, `codex/event` notifications include `_meta.threadId` so clients can correlate events to a conversation:
```json
{
"_meta": { "requestId": 1, "threadId": "..." },
"id": "evt-...",
"msg": { "type": "..." }
}
```
**`codex-reply`** - Continue a Codex session by providing the thread id and prompt. The `codex-reply` tool takes the following properties:
| Property | Type | Description |
| ------------------------- | ------ | -------------------------------------------------------- |
| **`prompt`** (required) | string | The next user prompt to continue the Codex conversation. |
| **`threadId`** (required) | string | The id of the conversation to continue. |
### Trying it Out
> [!TIP]
> Codex often takes a few minutes to run. To accommodate this, adjust the MCP inspector's Request and Total timeouts to 600000ms (10 minutes) under ⛭ Configuration.
Use the MCP inspector and `codex mcp-server` to build a simple tic-tac-toe game with the following settings:
**approval-policy:** never
**prompt:** Implement a simple tic-tac-toe game with HTML, JavaScript, and CSS. Write the game in a single file called index.html.
**sandbox:** workspace-write
Click "Run Tool" and you should see a list of events emitted from the Codex MCP server as it builds the game.