openai/codex
1
0
Fork 0
mirror of https://github.com/openai/codex.git synced 2026-11-06 20:20:34 +00:00
Code Issues Packages Projects Releases Wiki Activity

Add connector metadata to MCP tool call spans

Browse Source 
Propagate connector id and connector name into the mcp.tools.call span and add tracing coverage for Codex Apps and stdio MCP paths.

Co-authored-by: Codex <noreply@openai.com>
This commit is contained in:
nicholasclark-openai
2026-03-25 10:48:13 -07:00
co-authored by Codex
parent 52c7ffc6ee
commit 9d8e825671
5 changed files with 172 additions and 41 deletions
Download Patch File Download Diff File Expand all files Collapse all files
codex-rs/core/src/mcp_tool_call.rs
+20 -4
View File
@@ -122,6 +122,12 @@ pub(crate) async fn handle_mcp_tool_call(
}
let request_meta =
build_mcp_tool_call_request_meta(turn_context.as_ref(), &server, metadata.as_ref());
let connector_id = metadata
.as_ref()
.and_then(|metadata| metadata.connector_id.clone());
let connector_name = metadata
.as_ref()
.and_then(|metadata| metadata.connector_name.clone());
let server_origin = sess
.services
.mcp_connection_manager
@@ -163,6 +169,8 @@ pub(crate) async fn handle_mcp_tool_call(
arguments_value: arguments_value.clone(),
request_meta: request_meta.clone(),
server_origin: server_origin.as_deref(),
connector_id: connector_id.clone(),
connector_name: connector_name.clone(),
},
)
.await;
@@ -248,6 +256,8 @@ pub(crate) async fn handle_mcp_tool_call(
arguments_value: arguments_value.clone(),
request_meta,
server_origin: server_origin.as_deref(),
connector_id,
connector_name,
},
)
.await;
@@ -282,10 +292,14 @@ async fn execute_mcp_tool_call(
let tool_call_span = crate::network_trace::mcp_tool_call_span(
sess.as_ref(),
turn_context.as_ref(),
request.server,
request.tool_name,
request.call_id,
request.server_origin,
crate::network_trace::McpToolCallTrace {
server_name: request.server,
tool_name: request.tool_name,
call_id: request.call_id,
server_origin: request.server_origin,
connector_id: request.connector_id.as_deref(),
connector_name: request.connector_name.as_deref(),
},
);
let result = async {
sess.call_tool(
@@ -319,6 +333,8 @@ struct McpToolCallRequest<'a> {
arguments_value: Option<serde_json::Value>,
request_meta: Option<serde_json::Value>,
server_origin: Option<&'a str>,
connector_id: Option<String>,
connector_name: Option<String>,
}
async fn maybe_mark_thread_memory_mode_polluted(sess: &Session, turn_context: &TurnContext) {
codex-rs/core/src/network_trace.rs
+18 -10
View File
@@ -96,15 +96,21 @@ pub(crate) fn responses_http_request_span(
span
}
pub(crate) struct McpToolCallTrace<'a> {
pub server_name: &'a str,
pub tool_name: &'a str,
pub call_id: &'a str,
pub server_origin: Option<&'a str>,
pub connector_id: Option<&'a str>,
pub connector_name: Option<&'a str>,
}
pub(crate) fn mcp_tool_call_span(
session: &Session,
turn_context: &TurnContext,
server_name: &str,
tool_name: &str,
call_id: &str,
server_origin: Option<&str>,
trace: McpToolCallTrace<'_>,
) -> Span {
let transport = match server_origin {
let transport = match trace.server_origin {
Some("stdio") => "stdio",
Some(_) => "streamable_http",
None => "",
@@ -114,11 +120,13 @@ pub(crate) fn mcp_tool_call_span(
otel.kind = "client",
rpc.system = "jsonrpc",
rpc.method = "tools/call",
mcp.server.name = server_name,
mcp.server.origin = server_origin.unwrap_or(""),
mcp.server.name = trace.server_name,
mcp.server.origin = trace.server_origin.unwrap_or(""),
mcp.transport = transport,
tool.name = tool_name,
tool.call_id = call_id,
mcp.connector.id = trace.connector_id.unwrap_or(""),
mcp.connector.name = trace.connector_name.unwrap_or(""),
tool.name = trace.tool_name,
tool.call_id = trace.call_id,
conversation.id = Empty,
session.id = Empty,
turn.id = Empty,
@@ -126,6 +134,6 @@ pub(crate) fn mcp_tool_call_span(
server.port = Empty,
);
CorrelationFields::from_turn_context(session, turn_context).record_on(&span);
record_server_fields(&span, server_origin);
record_server_fields(&span, trace.server_origin);
span
}
codex-rs/core/tests/suite/mod.rs
+2
View File
@@ -92,6 +92,8 @@ mod model_visible_layout;
mod models_cache_ttl;
mod models_etag_responses;
mod otel;
#[cfg(not(target_os = "windows"))]
mod otel_apps;
mod pending_input;
mod permissions_messages;
mod personality;
codex-rs/core/tests/suite/otel_apps.rs
+104
View File
@@ -0,0 +1,104 @@
#![allow(clippy::unwrap_used, clippy::expect_used)]
use anyhow::Result;
use codex_core::CodexAuth;
use codex_features::Feature;
use codex_protocol::protocol::EventMsg;
use codex_protocol::protocol::Op;
use codex_protocol::user_input::UserInput;
use core_test_support::apps_test_server::AppsTestServer;
use core_test_support::responses::ev_assistant_message;
use core_test_support::responses::ev_completed;
use core_test_support::responses::ev_function_call;
use core_test_support::responses::ev_response_created;
use core_test_support::responses::mount_sse_once;
use core_test_support::responses::sse;
use core_test_support::responses::start_mock_server;
use core_test_support::skip_if_no_network;
use core_test_support::test_codex::test_codex;
use core_test_support::wait_for_event;
use std::sync::Mutex;
use tracing::Level;
use tracing_subscriber::fmt::format::FmtSpan;
use tracing_test::internal::MockWriter;
#[tokio::test(flavor = "current_thread")]
async fn codex_apps_mcp_span_records_connector_metadata() -> Result<()> {
skip_if_no_network!(Ok(()));
let buffer: &'static Mutex<Vec<u8>> = Box::leak(Box::new(Mutex::new(Vec::new())));
let subscriber = tracing_subscriber::fmt()
.with_level(true)
.with_ansi(false)
.with_max_level(Level::TRACE)
.with_span_events(FmtSpan::FULL)
.with_writer(MockWriter::new(buffer))
.finish();
let _guard = tracing::subscriber::set_default(subscriber);
let server = start_mock_server().await;
let apps_server = AppsTestServer::mount(&server).await?;
mount_sse_once(
&server,
sse(vec![
ev_response_created("resp-1"),
ev_function_call(
"calendar-call-1",
"mcp__codex_apps__calendar_create_event",
r#"{"title":"Lunch","starts_at":"2026-03-10T12:00:00Z"}"#,
),
ev_completed("resp-1"),
]),
)
.await;
mount_sse_once(
&server,
sse(vec![
ev_assistant_message("msg-1", "calendar tool completed successfully."),
ev_completed("resp-2"),
]),
)
.await;
let fixture = test_codex()
.with_auth(CodexAuth::create_dummy_chatgpt_auth_for_testing())
.with_config(move |config| {
config
.features
.enable(Feature::Apps)
.expect("test config should allow feature update");
config.chatgpt_base_url = apps_server.chatgpt_base_url;
config.model = Some("gpt-5-codex".to_string());
})
.build(&server)
.await?;
fixture
.codex
.submit(Op::UserInput {
items: vec![UserInput::Text {
text: "create a calendar event".into(),
text_elements: Vec::new(),
}],
final_output_json_schema: None,
})
.await?;
wait_for_event(&fixture.codex, |event| {
matches!(event, EventMsg::TurnComplete(_))
})
.await;
let logs = String::from_utf8(buffer.lock().unwrap().clone()).unwrap();
assert!(
logs.contains("mcp.tools.call{otel.kind=\"client\"")
&& logs.contains("mcp.server.name=\"codex_apps\"")
&& logs.contains("mcp.connector.id=\"calendar\"")
&& logs.contains("mcp.connector.name=\"Calendar\"")
&& logs.contains("tool.name=\"calendar_create_event\""),
"missing connector metadata on mcp.tools.call span\nlogs:\n{logs}"
);
Ok(())
}
codex-rs/core/tests/suite/rmcp_client.rs
+28 -27
View File
@@ -48,16 +48,28 @@ use tokio::process::Child;
use tokio::process::Command;
use tokio::time::Instant;
use tokio::time::sleep;
use tracing_test::traced_test;
use tracing::Level;
use tracing_subscriber::fmt::format::FmtSpan;
use tracing_test::internal::MockWriter;
static OPENAI_PNG: &str = "data:image/png;base64,iVBORw0KGgoAAAANSUhEUgAAAD0AAAA9CAYAAAAeYmHpAAAE6klEQVR4Aeyau44UVxCGx1fZsmRLlm3Zoe0XcGQ5cUiCCIgJeS9CHgAhMkISQnIuGQgJEkBcxLW+nqnZ6uqqc+nuWRC7q/P3qetf9e+MtOwyX25O4Nep6JPyop++0qev9HrfgZ+F6r2DuB/vHOrt/UIkqdDHYvujOW6fO7h/CNEI+a5jc+pBR8uy0jVFsziYu5HtfSUk+Io34q921hLNctFSX0gwww+S8wce8K1LfCU+cYW4888aov8NxqvQILUPPReLOrm6zyLxa4i+6VZuFbJo8d1MOHZm+7VUtB/aIvhPWc/3SWg49JcwFLlHxuXKjtyloo+YNhuW3VS+WPBuUEMvCFKjEDVgFBQHXrnazpqiSxNZCkQ1kYiozsbm9Oz7l4i2Il7vGccGNWAc3XosDrZe/9P3ZnMmzHNEQw4smf8RQ87XEAMsC7Az0Au+dgXerfH4+sHvEc0SYGic8WBBUGqFH2gN7yDrazy7m2pbRTeRmU3+MjZmr1h6LJgPbGy23SI6GlYT0brQ71IY8Us4PNQCm+zepSbaD2BY9xCaAsD9IIj/IzFmKMSdHHonwdZATbTnYREf6/VZGER98N9yCWIvXQwXDoDdhZJoT8jwLnJXDB9w4Sb3e6nK5ndzlkTLnP3JBu4LKkbrYrU69gCVceV0JvpyuW1xlsUVngzhwMetn/XamtTORF9IO5YnWNiyeF9zCAfqR3fUW+vZZKLtgP+ts8BmQRBREAdRDhH3o8QuRh/YucNFz2BEjxbRN6LGzphfKmvP6v6QhqIQyZ8XNJ0W0X83MR1PEcJBNO2KC2Z1TW/v244scp9FwRViZxIOBF0Lctk7ZVSavdLvRlV1hz/ysUi9sr8CIcB3nvWBwA93ykTz18eAYxQ6N/K2DkPA1lv3iXCwmDUT7YkjIby9siXueIJj9H+pzSqJ9oIuJWTUgSSt4WO7o/9GGg0viR4VinNRUDoIj34xoCd6pxD3aK3zfdbnx5v1J3ZNNEJsE0sBG7N27ReDrJc4sFxz7dI/ZAbOmmiKvHBitQXpAdR6+F7v+/ol/tOouUV01EeMZQF2BoQDn6dP4XNr+j9GZEtEK1/L8pFw7bd3a53tsTa7WD+054jOFmPg1XBKPQgnqFfmFcy32ZRvjmiIIQTYFvyDxQ8nH8WIwwGwlyDjDznnilYyFr6njrlZwsKkBpO59A7OwgdzPEWRm+G+oeb7IfyNuzjEEVLrOVxJsxvxwF8kmCM6I2QYmJunz4u4TrADpfl7mlbRTWQ7VmrBzh3+C9f6Grc3YoGN9dg/SXFthpRsT6vobfXRs2VBlgBHXVMLHjDNbIZv1sZ9+X3hB09cXdH1JKViyG0+W9bWZDa/r2f9zAFR71sTzGpMSWz2iI4YssWjWo3REy1MDGjdwe5e0dFSiAC1JakBvu4/CUS8Eh6dqHdU0Or0ioY3W5ClSqDXAy7/6SRfgw8vt4I+tbvvNtFT2kVDhY5+IGb1rCqYaXNF08vSALsXCPmt0kQNqJT1p5eI1mkIV/BxCY1z85lOzeFbPBQHURkkPTlwTYK9gTVE25l84IbFFN+YJDHjdpn0gq6mrHht0dkcjbM4UL9283O5p77GN+SPW/QwVB4IUYg7Or+Kp7naR6qktP98LNF2UxWo9yObPIT9KYg+hK4i56no4rfnM0qeyFf6AwAAAP//trwR3wAAAAZJREFUAwBZ0sR75itw5gAAAABJRU5ErkJggg==";
#[tokio::test(flavor = "multi_thread", worker_threads = 1)]
#[tokio::test(flavor = "current_thread")]
#[serial(mcp_test_value)]
#[traced_test]
async fn stdio_server_round_trip() -> anyhow::Result<()> {
skip_if_no_network!(Ok(()));
let buffer: &'static std::sync::Mutex<Vec<u8>> =
Box::leak(Box::new(std::sync::Mutex::new(Vec::new())));
let subscriber = tracing_subscriber::fmt()
.with_level(true)
.with_ansi(false)
.with_max_level(Level::TRACE)
.with_span_events(FmtSpan::FULL)
.with_writer(MockWriter::new(buffer))
.finish();
let _guard = tracing::subscriber::set_default(subscriber);
let server = responses::start_mock_server().await;
let call_id = "call-123";
@@ -191,30 +203,19 @@ async fn stdio_server_round_trip() -> anyhow::Result<()> {
wait_for_event(&fixture.codex, |ev| matches!(ev, EventMsg::TurnComplete(_))).await;
logs_assert(|lines: &[&str]| {
lines
.iter()
.find(|line| {
line.contains("turn{otel.name=\"session_task.turn\"")
&& line.contains("conversation.id=")
&& line.contains("thread.id=")
&& line.contains("turn.id=")
&& line.contains("model=")
&& line.contains("mcp.tools.call{")
&& line.contains("otel.kind=\"client\"")
&& line.contains("rpc.system=\"jsonrpc\"")
&& line.contains("rpc.method=\"tools/call\"")
&& line.contains("mcp.server.name=\"rmcp\"")
&& line.contains("mcp.transport=\"stdio\"")
&& line.contains("tool.name=\"echo\"")
&& line.contains("tool.call_id=\"call-123\"")
&& line.contains("turn.id=")
})
.map(|_| Ok(()))
.unwrap_or_else(|| {
Err("missing mcp.tools.call span nested under session_task.turn".to_string())
})
});
let logs = String::from_utf8(buffer.lock().unwrap().clone()).unwrap();
assert!(
logs.contains("turn{otel.name=\"session_task.turn\"")
&& logs.contains("mcp.tools.call{otel.kind=\"client\"")
&& logs.contains("rpc.system=\"jsonrpc\"")
&& logs.contains("rpc.method=\"tools/call\"")
&& logs.contains("mcp.server.name=\"rmcp\"")
&& logs.contains("mcp.transport=\"stdio\"")
&& logs.contains("tool.name=\"echo\"")
&& logs.contains("tool.call_id=\"call-123\"")
&& logs.contains("turn.id="),
"missing mcp.tools.call span nested under session_task.turn\nlogs:\n{logs}"
);
server.verify().await;