mirror of
https://github.com/openai/codex.git
synced 2026-10-31 19:28:49 +00:00
mcp ui collision fixes: route resource reads by origin call
This commit is contained in:
Drew Schuster
@@ -1938,6 +1938,7 @@ mod tests {
|
||||
thread_id: Some("thread-1".to_string()),
|
||||
server: "server-a".to_string(),
|
||||
uri: "file:///tmp/resource".to_string(),
|
||||
origin_call_id: None,
|
||||
},
|
||||
};
|
||||
assert_eq!(
|
||||
@@ -2116,6 +2117,7 @@ mod tests {
|
||||
thread_id: None,
|
||||
server: "server-a".to_string(),
|
||||
uri: "file:///tmp/resource".to_string(),
|
||||
origin_call_id: None,
|
||||
},
|
||||
};
|
||||
assert_eq!(mcp_resource_read.serialization_scope(), None);
|
||||
|
||||
@@ -85,6 +85,9 @@ pub struct McpResourceReadParams {
|
||||
pub thread_id: Option<String>,
|
||||
pub server: String,
|
||||
pub uri: String,
|
||||
/// The MCP tool call whose app context identifies the resource provider. Requires `threadId`.
|
||||
#[ts(optional = nullable)]
|
||||
pub origin_call_id: Option<String>,
|
||||
}
|
||||
|
||||
#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)]
|
||||
|
||||
@@ -72,6 +72,19 @@ fn test_absolute_path() -> AbsolutePathBuf {
|
||||
absolute_path("readable")
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn mcp_resource_read_params_deserialize_origin_call_id() {
|
||||
let params: McpResourceReadParams = serde_json::from_value(json!({
|
||||
"threadId": "thread-123",
|
||||
"server": "codex_apps",
|
||||
"uri": "test://codex/resource",
|
||||
"originCallId": "call-123",
|
||||
}))
|
||||
.expect("deserialize resource read params");
|
||||
|
||||
assert_eq!(params.origin_call_id.as_deref(), Some("call-123"));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn thread_sources_round_trip_as_scalar_labels() {
|
||||
for (source, label) in [
|
||||
|
||||
@@ -234,7 +234,7 @@ Example with notification opt-out:
|
||||
- `tool/requestUserInput` — prompt the user with 1–3 short questions for a tool call and return their answers (experimental).
|
||||
- `config/mcpServer/reload` — reload MCP server config from disk and queue a refresh for loaded threads (applied on each thread's next active turn); returns `{}`. Use this after editing `config.toml` without restarting the server.
|
||||
- `mcpServerStatus/list` — enumerate configured MCP servers with their tools, auth status, server info, plus resources/resource templates for `full` detail; supports optional `threadId` and cursor+limit pagination. If `threadId` is omitted, the server reads from the latest global config directly. If `detail` is omitted, the server defaults to `full`.
|
||||
- `mcpServer/resource/read` — read a resource from a configured MCP server by optional `threadId`, `server`, and `uri`, returning text/blob resource `contents`. If `threadId` is omitted, the server reads from the latest MCP config directly.
|
||||
- `mcpServer/resource/read` — read a resource from a configured MCP server by optional `threadId`, `server`, `uri`, and `originCallId`, returning text/blob resource `contents`. When `originCallId` identifies a Codex Apps tool call backed by plugin-runtime, app-server validates the persisted app context and routes the read through that app's `fetch_resource` tool. Other reads continue to use MCP `resources/read`. `originCallId` requires `threadId`; if `threadId` is omitted, the server reads from the latest MCP config directly.
|
||||
- `mcpServer/tool/call` — call a tool on a thread's configured MCP server by `threadId`, `server`, `tool`, optional `arguments`, and optional `_meta`, returning the MCP tool result.
|
||||
- `windowsSandbox/setupStart` — start Windows sandbox setup for the selected mode (`elevated` or `unelevated`); accepts an optional absolute `cwd` to target setup for a specific workspace, returns `{ started: true }` immediately, and later emits `windowsSandbox/setupCompleted`.
|
||||
- `feedback/upload` — submit a feedback report (classification + optional reason/logs, conversation_id, and optional `extraLogFiles` attachments array); returns the tracking thread id.
|
||||
|
||||
@@ -458,6 +458,7 @@ impl MessageProcessor {
|
||||
let mcp_processor = McpRequestProcessor::new(
|
||||
auth_manager.clone(),
|
||||
Arc::clone(&thread_manager),
|
||||
thread_state_manager.clone(),
|
||||
outgoing.clone(),
|
||||
config_manager.clone(),
|
||||
);
|
||||
|
||||
@@ -508,6 +508,7 @@ mod git_processor;
|
||||
mod initialize_processor;
|
||||
mod marketplace_processor;
|
||||
mod mcp_processor;
|
||||
mod mcp_resource_read;
|
||||
mod plugins;
|
||||
mod process_exec_processor;
|
||||
mod remote_control_processor;
|
||||
|
||||
@@ -6,6 +6,7 @@ const MCP_TOOL_THREAD_ID_META_KEY: &str = "threadId";
|
||||
pub(crate) struct McpRequestProcessor {
|
||||
auth_manager: Arc<AuthManager>,
|
||||
thread_manager: Arc<ThreadManager>,
|
||||
thread_state_manager: ThreadStateManager,
|
||||
outgoing: Arc<OutgoingMessageSender>,
|
||||
config_manager: ConfigManager,
|
||||
}
|
||||
@@ -14,12 +15,14 @@ impl McpRequestProcessor {
|
||||
pub(crate) fn new(
|
||||
auth_manager: Arc<AuthManager>,
|
||||
thread_manager: Arc<ThreadManager>,
|
||||
thread_state_manager: ThreadStateManager,
|
||||
outgoing: Arc<OutgoingMessageSender>,
|
||||
config_manager: ConfigManager,
|
||||
) -> Self {
|
||||
Self {
|
||||
auth_manager,
|
||||
thread_manager,
|
||||
thread_state_manager,
|
||||
outgoing,
|
||||
config_manager,
|
||||
}
|
||||
@@ -393,19 +396,34 @@ impl McpRequestProcessor {
|
||||
thread_id,
|
||||
server,
|
||||
uri,
|
||||
origin_call_id,
|
||||
} = params;
|
||||
|
||||
if let Some(thread_id) = thread_id {
|
||||
let (_, thread) = self.load_thread(&thread_id).await?;
|
||||
if let Some(thread_id_string) = thread_id {
|
||||
let (thread_id, thread) = self.load_thread(&thread_id_string).await?;
|
||||
let thread_state_manager = self.thread_state_manager.clone();
|
||||
let request_id = request_id.clone();
|
||||
|
||||
tokio::spawn(async move {
|
||||
let result = thread.read_mcp_resource(&server, &uri).await;
|
||||
let result = mcp_resource_read::read_thread_mcp_resource(
|
||||
&thread_state_manager,
|
||||
thread_id,
|
||||
&thread_id_string,
|
||||
&thread,
|
||||
&server,
|
||||
&uri,
|
||||
origin_call_id.as_deref(),
|
||||
)
|
||||
.await;
|
||||
Self::send_mcp_resource_read_response(outgoing, request_id, result).await;
|
||||
});
|
||||
return Ok(());
|
||||
}
|
||||
|
||||
if origin_call_id.is_some() {
|
||||
return Err(invalid_request("originCallId requires threadId"));
|
||||
}
|
||||
|
||||
let config = self.load_latest_config(/*fallback_cwd*/ None).await?;
|
||||
let mcp_manager = self.thread_manager.mcp_manager();
|
||||
let mcp_config = mcp_manager.runtime_config(&config).await;
|
||||
@@ -475,7 +493,7 @@ impl McpRequestProcessor {
|
||||
}
|
||||
}
|
||||
|
||||
fn with_mcp_tool_call_thread_id_meta(
|
||||
pub(super) fn with_mcp_tool_call_thread_id_meta(
|
||||
meta: Option<serde_json::Value>,
|
||||
thread_id: &str,
|
||||
) -> Option<serde_json::Value> {
|
||||
|
||||
@@ -0,0 +1,300 @@
|
||||
use super::mcp_processor::with_mcp_tool_call_thread_id_meta;
|
||||
use super::*;
|
||||
use codex_mcp::CODEX_APPS_MCP_SERVER_NAME;
|
||||
use codex_mcp::MCP_TOOL_CODEX_APPS_META_KEY;
|
||||
use codex_mcp::ToolInfo;
|
||||
|
||||
const LEGACY_CODEX_APPS_MCP_SERVER_NAME: &str = "codex-connectors-mcp";
|
||||
const PLUGIN_RUNTIME_MCP_SERVER_NAME: &str = "plugin-runtime";
|
||||
const FETCH_RESOURCE_TOOL_NAME: &str = "fetch_resource";
|
||||
|
||||
pub(super) async fn read_thread_mcp_resource(
|
||||
thread_state_manager: &ThreadStateManager,
|
||||
thread_id: ThreadId,
|
||||
thread_id_string: &str,
|
||||
thread: &Arc<CodexThread>,
|
||||
server: &str,
|
||||
uri: &str,
|
||||
origin_call_id: Option<&str>,
|
||||
) -> anyhow::Result<serde_json::Value> {
|
||||
let Some(origin_call_id) = origin_call_id else {
|
||||
return thread.read_mcp_resource(server, uri).await;
|
||||
};
|
||||
if server != CODEX_APPS_MCP_SERVER_NAME {
|
||||
return thread.read_mcp_resource(server, uri).await;
|
||||
}
|
||||
|
||||
let server_info = thread
|
||||
.mcp_server_info(server)
|
||||
.await
|
||||
.ok_or_else(|| anyhow::anyhow!("unable to identify codex_apps MCP server"))?;
|
||||
match server_info.name.as_str() {
|
||||
LEGACY_CODEX_APPS_MCP_SERVER_NAME => return thread.read_mcp_resource(server, uri).await,
|
||||
PLUGIN_RUNTIME_MCP_SERVER_NAME => {}
|
||||
implementation => {
|
||||
anyhow::bail!("unsupported codex_apps MCP server implementation `{implementation}`")
|
||||
}
|
||||
}
|
||||
|
||||
let origin =
|
||||
resolve_mcp_resource_origin(thread_state_manager, thread_id, thread, origin_call_id)
|
||||
.await?;
|
||||
if origin.server != server {
|
||||
anyhow::bail!(
|
||||
"originating MCP tool call server `{}` does not match resource server `{server}`",
|
||||
origin.server
|
||||
);
|
||||
}
|
||||
if origin.resource_uri.as_deref() != Some(uri) {
|
||||
anyhow::bail!(
|
||||
"originating MCP tool call resource URI {:?} does not match requested URI `{uri}`",
|
||||
origin.resource_uri
|
||||
);
|
||||
}
|
||||
|
||||
let tool_info = thread
|
||||
.mcp_tool_info(server, &origin.tool)
|
||||
.await
|
||||
.ok_or_else(|| {
|
||||
anyhow::anyhow!(
|
||||
"originating MCP tool `{}` is not available on server `{server}`",
|
||||
origin.tool
|
||||
)
|
||||
})?;
|
||||
let meta = build_plugin_runtime_fetch_resource_meta(&origin, uri, &tool_info)?;
|
||||
let meta = with_mcp_tool_call_thread_id_meta(Some(meta), thread_id_string);
|
||||
let result = thread
|
||||
.call_mcp_tool(
|
||||
server,
|
||||
FETCH_RESOURCE_TOOL_NAME,
|
||||
Some(serde_json::json!({ "uri": uri })),
|
||||
meta,
|
||||
)
|
||||
.await?;
|
||||
plugin_runtime_fetch_resource_response(result)
|
||||
}
|
||||
|
||||
async fn resolve_mcp_resource_origin(
|
||||
thread_state_manager: &ThreadStateManager,
|
||||
thread_id: ThreadId,
|
||||
thread: &Arc<CodexThread>,
|
||||
origin_call_id: &str,
|
||||
) -> anyhow::Result<McpResourceOrigin> {
|
||||
let active_turn = {
|
||||
let thread_state = thread_state_manager.thread_state(thread_id).await;
|
||||
thread_state.lock().await.active_turn_snapshot()
|
||||
};
|
||||
if let Some(origin) = active_turn
|
||||
.as_ref()
|
||||
.and_then(|turn| find_mcp_resource_origin(&turn.items, origin_call_id))
|
||||
{
|
||||
return origin;
|
||||
}
|
||||
|
||||
let history = thread.load_history(/*include_archived*/ true).await?;
|
||||
let turns = build_api_turns_from_rollout_items(&history.items);
|
||||
turns
|
||||
.iter()
|
||||
.rev()
|
||||
.find_map(|turn| find_mcp_resource_origin(&turn.items, origin_call_id))
|
||||
.unwrap_or_else(|| {
|
||||
Err(anyhow::anyhow!(
|
||||
"originating MCP tool call `{origin_call_id}` was not found"
|
||||
))
|
||||
})
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, PartialEq, Eq)]
|
||||
struct McpResourceOrigin {
|
||||
server: String,
|
||||
tool: String,
|
||||
connector_id: String,
|
||||
link_id: Option<String>,
|
||||
resource_uri: Option<String>,
|
||||
}
|
||||
|
||||
fn find_mcp_resource_origin(
|
||||
items: &[ThreadItem],
|
||||
origin_call_id: &str,
|
||||
) -> Option<anyhow::Result<McpResourceOrigin>> {
|
||||
items.iter().find_map(|item| {
|
||||
let ThreadItem::McpToolCall {
|
||||
id,
|
||||
server,
|
||||
tool,
|
||||
app_context,
|
||||
..
|
||||
} = item
|
||||
else {
|
||||
return None;
|
||||
};
|
||||
if id != origin_call_id {
|
||||
return None;
|
||||
}
|
||||
Some(match app_context {
|
||||
Some(app_context) => Ok(McpResourceOrigin {
|
||||
server: server.clone(),
|
||||
tool: tool.clone(),
|
||||
connector_id: app_context.connector_id.clone(),
|
||||
link_id: app_context.link_id.clone(),
|
||||
resource_uri: app_context.resource_uri.clone(),
|
||||
}),
|
||||
None => Err(anyhow::anyhow!(
|
||||
"originating MCP tool call `{origin_call_id}` has no app context"
|
||||
)),
|
||||
})
|
||||
})
|
||||
}
|
||||
|
||||
fn build_plugin_runtime_fetch_resource_meta(
|
||||
origin: &McpResourceOrigin,
|
||||
uri: &str,
|
||||
tool_info: &ToolInfo,
|
||||
) -> anyhow::Result<serde_json::Value> {
|
||||
let tool_meta = tool_info
|
||||
.tool
|
||||
.meta
|
||||
.as_ref()
|
||||
.ok_or_else(|| anyhow::anyhow!("originating MCP tool is missing metadata"))?;
|
||||
let connector_id = tool_info
|
||||
.connector_id
|
||||
.as_deref()
|
||||
.ok_or_else(|| anyhow::anyhow!("originating MCP tool is missing connector_id"))?;
|
||||
let link_id = tool_meta
|
||||
.0
|
||||
.get("link_id")
|
||||
.and_then(serde_json::Value::as_str)
|
||||
.ok_or_else(|| anyhow::anyhow!("originating MCP tool is missing link_id"))?;
|
||||
let codex_apps_meta = tool_meta
|
||||
.0
|
||||
.get(MCP_TOOL_CODEX_APPS_META_KEY)
|
||||
.and_then(serde_json::Value::as_object)
|
||||
.ok_or_else(|| anyhow::anyhow!("originating MCP tool is missing _codex_apps metadata"))?;
|
||||
let resource_uri = mcp_app_resource_uri(&tool_meta.0)
|
||||
.ok_or_else(|| anyhow::anyhow!("originating MCP tool is missing its app resource URI"))?;
|
||||
if resource_uri != uri {
|
||||
anyhow::bail!(
|
||||
"originating MCP tool resource URI `{resource_uri}` does not match requested URI `{uri}`"
|
||||
);
|
||||
}
|
||||
if connector_id != origin.connector_id {
|
||||
anyhow::bail!(
|
||||
"originating MCP tool connector `{connector_id}` does not match app context connector `{}`",
|
||||
origin.connector_id
|
||||
);
|
||||
}
|
||||
|
||||
let synthetic_link = codex_apps_meta
|
||||
.get("synthetic_link")
|
||||
.and_then(serde_json::Value::as_bool)
|
||||
== Some(true);
|
||||
match origin.link_id.as_deref() {
|
||||
Some(origin_link_id) if origin_link_id == link_id => {}
|
||||
None if synthetic_link => {}
|
||||
_ => anyhow::bail!("originating MCP tool link does not match app context link"),
|
||||
}
|
||||
|
||||
let tool_resource_route = codex_apps_meta
|
||||
.get("resource_uri")
|
||||
.and_then(serde_json::Value::as_str)
|
||||
.ok_or_else(|| anyhow::anyhow!("originating MCP tool is missing its resource route"))?;
|
||||
let route_segments = tool_resource_route
|
||||
.strip_prefix('/')
|
||||
.map(|route| route.split('/').collect::<Vec<_>>())
|
||||
.filter(|segments| segments.len() == 3 && segments.iter().all(|part| !part.is_empty()))
|
||||
.ok_or_else(|| anyhow::anyhow!("originating MCP tool has an invalid resource route"))?;
|
||||
if route_segments[0] != connector_id || route_segments[1] != link_id {
|
||||
anyhow::bail!("originating MCP tool resource route does not match connector and link");
|
||||
}
|
||||
|
||||
let mut fetch_resource_meta = serde_json::Map::from_iter([
|
||||
(
|
||||
"resource_uri".to_string(),
|
||||
serde_json::Value::String(format!(
|
||||
"/{connector_id}/{link_id}/{FETCH_RESOURCE_TOOL_NAME}"
|
||||
)),
|
||||
),
|
||||
(
|
||||
"contains_mcp_source".to_string(),
|
||||
serde_json::Value::Bool(
|
||||
codex_apps_meta
|
||||
.get("contains_mcp_source")
|
||||
.and_then(serde_json::Value::as_bool)
|
||||
== Some(true),
|
||||
),
|
||||
),
|
||||
]);
|
||||
if synthetic_link {
|
||||
fetch_resource_meta.insert("synthetic_link".to_string(), serde_json::Value::Bool(true));
|
||||
}
|
||||
|
||||
let mut meta = serde_json::Map::from_iter([
|
||||
(
|
||||
MCP_TOOL_CODEX_APPS_META_KEY.to_string(),
|
||||
serde_json::Value::Object(fetch_resource_meta),
|
||||
),
|
||||
(
|
||||
"x-codex-turn-metadata".to_string(),
|
||||
serde_json::json!({
|
||||
"mcp_request_meta": {
|
||||
"selected_connector_ids": [connector_id],
|
||||
}
|
||||
}),
|
||||
),
|
||||
]);
|
||||
if let Some(connector_name) = tool_info.connector_name.as_ref() {
|
||||
meta.insert(
|
||||
"connector_name".to_string(),
|
||||
serde_json::Value::String(connector_name.clone()),
|
||||
);
|
||||
}
|
||||
if let Some(connector_description) = tool_info.namespace_description.as_ref() {
|
||||
meta.insert(
|
||||
"connector_description".to_string(),
|
||||
serde_json::Value::String(connector_description.clone()),
|
||||
);
|
||||
}
|
||||
Ok(serde_json::Value::Object(meta))
|
||||
}
|
||||
|
||||
fn mcp_app_resource_uri(meta: &serde_json::Map<String, serde_json::Value>) -> Option<&str> {
|
||||
meta.get("ui")
|
||||
.and_then(serde_json::Value::as_object)
|
||||
.and_then(|ui| ui.get("resourceUri"))
|
||||
.and_then(serde_json::Value::as_str)
|
||||
.or_else(|| {
|
||||
meta.get("ui/resourceUri")
|
||||
.and_then(serde_json::Value::as_str)
|
||||
})
|
||||
.or_else(|| {
|
||||
meta.get("openai/outputTemplate")
|
||||
.and_then(serde_json::Value::as_str)
|
||||
})
|
||||
}
|
||||
|
||||
fn plugin_runtime_fetch_resource_response(
|
||||
result: codex_protocol::mcp::CallToolResult,
|
||||
) -> anyhow::Result<serde_json::Value> {
|
||||
if result.is_error == Some(true) {
|
||||
anyhow::bail!("plugin-runtime fetch_resource returned an error");
|
||||
}
|
||||
let mut structured_content = result
|
||||
.structured_content
|
||||
.ok_or_else(|| anyhow::anyhow!("plugin-runtime fetch_resource returned no contents"))?;
|
||||
let contents = structured_content
|
||||
.get_mut("contents")
|
||||
.and_then(serde_json::Value::as_array_mut)
|
||||
.ok_or_else(|| {
|
||||
anyhow::anyhow!("plugin-runtime fetch_resource returned invalid contents")
|
||||
})?;
|
||||
for content in contents {
|
||||
if let Some(content) = content.as_object_mut()
|
||||
&& !content.contains_key("_meta")
|
||||
&& let Some(meta) = content.remove("meta")
|
||||
{
|
||||
content.insert("_meta".to_string(), meta);
|
||||
}
|
||||
}
|
||||
let response = serde_json::from_value::<McpResourceReadResponse>(structured_content)?;
|
||||
Ok(serde_json::to_value(response)?)
|
||||
}
|
||||
@@ -1,4 +1,5 @@
|
||||
use std::sync::Arc;
|
||||
use std::sync::Mutex;
|
||||
use std::sync::atomic::AtomicUsize;
|
||||
use std::sync::atomic::Ordering;
|
||||
use std::time::Duration;
|
||||
@@ -84,9 +85,9 @@ const SKILLS_READ_CALL_ID: &str = "skills-read";
|
||||
const SKILLS_READ_AGAIN_CALL_ID: &str = "skills-read-again";
|
||||
|
||||
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
|
||||
async fn mcp_resource_read_returns_resource_contents() -> Result<()> {
|
||||
async fn mcp_resource_read_without_origin_uses_resources_read() -> Result<()> {
|
||||
let responses_server = responses::start_mock_server().await;
|
||||
let (apps_server_url, _apps_server_calls, apps_server_handle) =
|
||||
let (apps_server_url, apps_server_calls, apps_server_handle) =
|
||||
start_resource_apps_mcp_server().await?;
|
||||
let responses_server_uri = responses_server.uri();
|
||||
let (_codex_home, mut mcp) =
|
||||
@@ -110,6 +111,7 @@ async fn mcp_resource_read_returns_resource_contents() -> Result<()> {
|
||||
thread_id: Some(thread.id),
|
||||
server: "codex_apps".to_string(),
|
||||
uri: TEST_RESOURCE_URI.to_string(),
|
||||
origin_call_id: None,
|
||||
})
|
||||
.await?;
|
||||
let read_response: JSONRPCResponse = timeout(
|
||||
@@ -121,6 +123,7 @@ async fn mcp_resource_read_returns_resource_contents() -> Result<()> {
|
||||
to_response::<McpResourceReadResponse>(read_response)?,
|
||||
expected_resource_read_response()
|
||||
);
|
||||
assert_eq!(apps_server_calls.resource_read_metas(), vec![Meta::new()]);
|
||||
|
||||
apps_server_handle.abort();
|
||||
let _ = apps_server_handle.await;
|
||||
@@ -526,8 +529,8 @@ enabled = false
|
||||
}
|
||||
|
||||
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
|
||||
async fn mcp_resource_read_returns_resource_contents_without_thread() -> Result<()> {
|
||||
let (apps_server_url, _apps_server_calls, apps_server_handle) =
|
||||
async fn mcp_resource_read_without_thread_uses_resources_read() -> Result<()> {
|
||||
let (apps_server_url, apps_server_calls, apps_server_handle) =
|
||||
start_resource_apps_mcp_server().await?;
|
||||
|
||||
let codex_home = TempDir::new()?;
|
||||
@@ -560,6 +563,7 @@ apps = true
|
||||
thread_id: None,
|
||||
server: "codex_apps".to_string(),
|
||||
uri: TEST_RESOURCE_URI.to_string(),
|
||||
origin_call_id: None,
|
||||
})
|
||||
.await?;
|
||||
let read_response: JSONRPCResponse = timeout(
|
||||
@@ -572,6 +576,7 @@ apps = true
|
||||
to_response::<McpResourceReadResponse>(read_response)?,
|
||||
expected_resource_read_response()
|
||||
);
|
||||
assert_eq!(apps_server_calls.resource_read_metas(), vec![Meta::new()]);
|
||||
|
||||
apps_server_handle.abort();
|
||||
let _ = apps_server_handle.await;
|
||||
@@ -624,6 +629,7 @@ async fn mcp_resource_read_returns_error_for_unknown_thread() -> Result<()> {
|
||||
thread_id: Some("00000000-0000-4000-8000-000000000000".to_string()),
|
||||
server: "codex_apps".to_string(),
|
||||
uri: TEST_RESOURCE_URI.to_string(),
|
||||
origin_call_id: None,
|
||||
},
|
||||
})
|
||||
.await;
|
||||
@@ -746,6 +752,7 @@ struct ResourceAppsMcpCalls {
|
||||
list_resources: AtomicUsize,
|
||||
main_prompt_reads: AtomicUsize,
|
||||
reference_reads: AtomicUsize,
|
||||
resource_read_metas: Mutex<Vec<Meta>>,
|
||||
}
|
||||
|
||||
impl ResourceAppsMcpCalls {
|
||||
@@ -756,6 +763,13 @@ impl ResourceAppsMcpCalls {
|
||||
reference_reads: self.reference_reads.load(Ordering::Relaxed),
|
||||
}
|
||||
}
|
||||
|
||||
fn resource_read_metas(&self) -> Vec<Meta> {
|
||||
self.resource_read_metas
|
||||
.lock()
|
||||
.expect("resource read metadata lock poisoned")
|
||||
.clone()
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Debug, PartialEq, Eq)]
|
||||
@@ -827,8 +841,17 @@ impl ServerHandler for ResourceAppsMcpServer {
|
||||
async fn read_resource(
|
||||
&self,
|
||||
request: ReadResourceRequestParams,
|
||||
_context: RequestContext<RoleServer>,
|
||||
context: RequestContext<RoleServer>,
|
||||
) -> Result<ReadResourceResult, rmcp::ErrorData> {
|
||||
let mut meta = context.meta;
|
||||
// rmcp adds a transport progress token to outgoing requests. Capture
|
||||
// only the client-supplied metadata that this test is exercising.
|
||||
meta.0.remove("progressToken");
|
||||
self.calls
|
||||
.resource_read_metas
|
||||
.lock()
|
||||
.expect("resource read metadata lock poisoned")
|
||||
.push(meta);
|
||||
let uri = request.uri;
|
||||
if uri == SKILL_MAIN_PROMPT_URI {
|
||||
self.calls.main_prompt_reads.fetch_add(1, Ordering::Relaxed);
|
||||
|
||||
@@ -0,0 +1,377 @@
|
||||
use std::borrow::Cow;
|
||||
use std::sync::Arc;
|
||||
use std::sync::Mutex;
|
||||
use std::time::Duration;
|
||||
|
||||
use anyhow::Result;
|
||||
use app_test_support::ChatGptAuthFixture;
|
||||
use app_test_support::TestAppServer;
|
||||
use app_test_support::to_response;
|
||||
use app_test_support::write_chatgpt_auth;
|
||||
use axum::Router;
|
||||
use codex_app_server_protocol::ItemCompletedNotification;
|
||||
use codex_app_server_protocol::JSONRPCResponse;
|
||||
use codex_app_server_protocol::McpResourceContent;
|
||||
use codex_app_server_protocol::McpResourceReadParams;
|
||||
use codex_app_server_protocol::McpResourceReadResponse;
|
||||
use codex_app_server_protocol::RequestId;
|
||||
use codex_app_server_protocol::ThreadItem;
|
||||
use codex_app_server_protocol::ThreadStartParams;
|
||||
use codex_app_server_protocol::ThreadStartResponse;
|
||||
use codex_app_server_protocol::TurnStartParams;
|
||||
use codex_app_server_protocol::UserInput;
|
||||
use codex_config::types::AuthCredentialsStoreMode;
|
||||
use core_test_support::responses;
|
||||
use pretty_assertions::assert_eq;
|
||||
use rmcp::handler::server::ServerHandler;
|
||||
use rmcp::model::CallToolRequestParams;
|
||||
use rmcp::model::CallToolResult;
|
||||
use rmcp::model::Content;
|
||||
use rmcp::model::Implementation;
|
||||
use rmcp::model::JsonObject;
|
||||
use rmcp::model::ListToolsResult;
|
||||
use rmcp::model::Meta;
|
||||
use rmcp::model::ProtocolVersion;
|
||||
use rmcp::model::ServerCapabilities;
|
||||
use rmcp::model::ServerInfo;
|
||||
use rmcp::model::Tool;
|
||||
use rmcp::model::ToolAnnotations;
|
||||
use rmcp::service::RequestContext;
|
||||
use rmcp::service::RoleServer;
|
||||
use rmcp::transport::StreamableHttpServerConfig;
|
||||
use rmcp::transport::StreamableHttpService;
|
||||
use rmcp::transport::streamable_http_server::session::local::LocalSessionManager;
|
||||
use serde_json::Value;
|
||||
use serde_json::json;
|
||||
use tempfile::TempDir;
|
||||
use tokio::net::TcpListener;
|
||||
use tokio::task::JoinHandle;
|
||||
use tokio::time::timeout;
|
||||
|
||||
const DEFAULT_READ_TIMEOUT: Duration = Duration::from_secs(10);
|
||||
const CONNECTOR_ID: &str = "walmart";
|
||||
const CONNECTOR_NAME: &str = "Walmart";
|
||||
const LINK_ID: &str = "link_walmart";
|
||||
const ORIGIN_CALL_ID: &str = "walmart-product-search-call";
|
||||
const PRODUCT_SEARCH_TOOL: &str = "walmart_product_search";
|
||||
const PRODUCT_SEARCH_MODEL_NAMESPACE: &str = "mcp__codex_apps__walmart";
|
||||
const PRODUCT_SEARCH_MODEL_TOOL: &str = "_product_search";
|
||||
const RESOURCE_URI: &str = "ui://widget/product-search.html";
|
||||
|
||||
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
|
||||
async fn mcp_resource_read_with_origin_calls_plugin_runtime_fetch_resource() -> Result<()> {
|
||||
let responses_server = responses::start_mock_server().await;
|
||||
let (apps_server_url, recorded_calls, apps_server_handle) =
|
||||
start_plugin_runtime_mcp_server().await?;
|
||||
let (_codex_home, mut mcp) =
|
||||
start_test_app_server(&apps_server_url, &responses_server.uri()).await?;
|
||||
|
||||
let response_mock = responses::mount_sse_sequence(
|
||||
&responses_server,
|
||||
vec![
|
||||
responses::sse(vec![
|
||||
responses::ev_response_created("resp-product-search"),
|
||||
responses::ev_function_call_with_namespace(
|
||||
ORIGIN_CALL_ID,
|
||||
PRODUCT_SEARCH_MODEL_NAMESPACE,
|
||||
PRODUCT_SEARCH_MODEL_TOOL,
|
||||
&json!({ "query": "bed lamps" }).to_string(),
|
||||
),
|
||||
responses::ev_completed("resp-product-search"),
|
||||
]),
|
||||
responses::sse(vec![
|
||||
responses::ev_response_created("resp-done"),
|
||||
responses::ev_assistant_message("msg-done", "Done"),
|
||||
responses::ev_completed("resp-done"),
|
||||
]),
|
||||
],
|
||||
)
|
||||
.await;
|
||||
|
||||
let thread_start_id = mcp
|
||||
.send_thread_start_request_with_auto_env(ThreadStartParams {
|
||||
model: Some("mock-model".to_string()),
|
||||
..Default::default()
|
||||
})
|
||||
.await?;
|
||||
let thread_start_response: JSONRPCResponse = timeout(
|
||||
DEFAULT_READ_TIMEOUT,
|
||||
mcp.read_stream_until_response_message(RequestId::Integer(thread_start_id)),
|
||||
)
|
||||
.await??;
|
||||
let ThreadStartResponse { thread, .. } = to_response(thread_start_response)?;
|
||||
|
||||
let turn_start_id = mcp
|
||||
.send_turn_start_request(TurnStartParams {
|
||||
thread_id: thread.id.clone(),
|
||||
input: vec![UserInput::Text {
|
||||
text: "Find bed lamps".to_string(),
|
||||
text_elements: Vec::new(),
|
||||
}],
|
||||
..Default::default()
|
||||
})
|
||||
.await?;
|
||||
timeout(
|
||||
DEFAULT_READ_TIMEOUT,
|
||||
mcp.read_stream_until_response_message(RequestId::Integer(turn_start_id)),
|
||||
)
|
||||
.await??;
|
||||
|
||||
let completed = wait_for_mcp_tool_call_completed(&mut mcp, ORIGIN_CALL_ID).await?;
|
||||
let ThreadItem::McpToolCall {
|
||||
app_context: Some(app_context),
|
||||
..
|
||||
} = completed.item
|
||||
else {
|
||||
anyhow::bail!("originating MCP tool call should include app context");
|
||||
};
|
||||
assert_eq!(app_context.connector_id, CONNECTOR_ID);
|
||||
assert_eq!(app_context.link_id.as_deref(), Some(LINK_ID));
|
||||
assert_eq!(app_context.resource_uri.as_deref(), Some(RESOURCE_URI));
|
||||
|
||||
timeout(
|
||||
DEFAULT_READ_TIMEOUT,
|
||||
mcp.read_stream_until_notification_message("turn/completed"),
|
||||
)
|
||||
.await??;
|
||||
|
||||
let read_request_id = mcp
|
||||
.send_mcp_resource_read_request(McpResourceReadParams {
|
||||
thread_id: Some(thread.id.clone()),
|
||||
server: "codex_apps".to_string(),
|
||||
uri: RESOURCE_URI.to_string(),
|
||||
origin_call_id: Some(ORIGIN_CALL_ID.to_string()),
|
||||
})
|
||||
.await?;
|
||||
let read_response: JSONRPCResponse = timeout(
|
||||
DEFAULT_READ_TIMEOUT,
|
||||
mcp.read_stream_until_response_message(RequestId::Integer(read_request_id)),
|
||||
)
|
||||
.await??;
|
||||
|
||||
assert_eq!(
|
||||
to_response::<McpResourceReadResponse>(read_response)?,
|
||||
McpResourceReadResponse {
|
||||
contents: vec![McpResourceContent::Text {
|
||||
uri: RESOURCE_URI.to_string(),
|
||||
mime_type: Some("text/html".to_string()),
|
||||
text: "<html>Walmart</html>".to_string(),
|
||||
meta: Some(json!({ "app": CONNECTOR_ID })),
|
||||
}],
|
||||
}
|
||||
);
|
||||
|
||||
{
|
||||
let calls = recorded_calls
|
||||
.lock()
|
||||
.expect("recorded tool calls lock poisoned");
|
||||
let fetch_resource_call = calls
|
||||
.iter()
|
||||
.find(|call| call["name"] == "fetch_resource")
|
||||
.expect("resource read should call fetch_resource");
|
||||
assert_eq!(
|
||||
fetch_resource_call["arguments"],
|
||||
json!({ "uri": RESOURCE_URI })
|
||||
);
|
||||
assert_eq!(
|
||||
fetch_resource_call.pointer("/meta/_codex_apps"),
|
||||
Some(&json!({
|
||||
"resource_uri": format!("/{CONNECTOR_ID}/{LINK_ID}/fetch_resource"),
|
||||
"contains_mcp_source": true,
|
||||
}))
|
||||
);
|
||||
assert_eq!(
|
||||
fetch_resource_call
|
||||
.pointer("/meta/x-codex-turn-metadata/mcp_request_meta/selected_connector_ids"),
|
||||
Some(&json!([CONNECTOR_ID]))
|
||||
);
|
||||
assert_eq!(fetch_resource_call["meta"]["threadId"], thread.id);
|
||||
}
|
||||
|
||||
assert_eq!(response_mock.requests().len(), 2);
|
||||
apps_server_handle.abort();
|
||||
let _ = apps_server_handle.await;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
async fn start_test_app_server(
|
||||
apps_server_url: &str,
|
||||
responses_server_uri: &str,
|
||||
) -> Result<(TempDir, TestAppServer)> {
|
||||
let codex_home = TempDir::new()?;
|
||||
std::fs::write(
|
||||
codex_home.path().join("config.toml"),
|
||||
format!(
|
||||
r#"
|
||||
model = "mock-model"
|
||||
approval_policy = "untrusted"
|
||||
sandbox_mode = "read-only"
|
||||
|
||||
model_provider = "mock_provider"
|
||||
chatgpt_base_url = "{apps_server_url}"
|
||||
mcp_oauth_credentials_store = "file"
|
||||
|
||||
[features]
|
||||
apps = true
|
||||
|
||||
[model_providers.mock_provider]
|
||||
name = "Mock provider for test"
|
||||
base_url = "{responses_server_uri}/v1"
|
||||
wire_api = "responses"
|
||||
request_max_retries = 0
|
||||
stream_max_retries = 0
|
||||
"#
|
||||
),
|
||||
)?;
|
||||
write_chatgpt_auth(
|
||||
codex_home.path(),
|
||||
ChatGptAuthFixture::new("chatgpt-token")
|
||||
.account_id("account-123")
|
||||
.chatgpt_user_id("user-123")
|
||||
.chatgpt_account_id("account-123"),
|
||||
AuthCredentialsStoreMode::File,
|
||||
)?;
|
||||
|
||||
let mut mcp = TestAppServer::new_with_auto_env(codex_home.path()).await?;
|
||||
timeout(DEFAULT_READ_TIMEOUT, mcp.initialize()).await??;
|
||||
Ok((codex_home, mcp))
|
||||
}
|
||||
|
||||
async fn start_plugin_runtime_mcp_server()
|
||||
-> Result<(String, Arc<Mutex<Vec<Value>>>, JoinHandle<()>)> {
|
||||
let listener = TcpListener::bind("127.0.0.1:0").await?;
|
||||
let addr = listener.local_addr()?;
|
||||
let apps_server_url = format!("http://{addr}");
|
||||
let calls = Arc::new(Mutex::new(Vec::new()));
|
||||
let server_calls = Arc::clone(&calls);
|
||||
let mcp_service = StreamableHttpService::new(
|
||||
move || {
|
||||
Ok(PluginRuntimeMcpServer {
|
||||
calls: Arc::clone(&server_calls),
|
||||
})
|
||||
},
|
||||
Arc::new(LocalSessionManager::default()),
|
||||
StreamableHttpServerConfig::default(),
|
||||
);
|
||||
let router = Router::new().nest_service("/api/codex/ps/mcp", mcp_service);
|
||||
let handle = tokio::spawn(async move {
|
||||
let _ = axum::serve(listener, router).await;
|
||||
});
|
||||
|
||||
Ok((apps_server_url, calls, handle))
|
||||
}
|
||||
|
||||
#[derive(Clone)]
|
||||
struct PluginRuntimeMcpServer {
|
||||
calls: Arc<Mutex<Vec<Value>>>,
|
||||
}
|
||||
|
||||
impl ServerHandler for PluginRuntimeMcpServer {
|
||||
fn get_info(&self) -> ServerInfo {
|
||||
ServerInfo::new(ServerCapabilities::builder().enable_tools().build())
|
||||
.with_protocol_version(ProtocolVersion::V_2025_06_18)
|
||||
.with_server_info(Implementation::new("plugin-runtime", "1.0.0"))
|
||||
}
|
||||
|
||||
async fn list_tools(
|
||||
&self,
|
||||
_request: Option<rmcp::model::PaginatedRequestParams>,
|
||||
_context: RequestContext<RoleServer>,
|
||||
) -> Result<ListToolsResult, rmcp::ErrorData> {
|
||||
let input_schema: JsonObject = serde_json::from_value(json!({
|
||||
"type": "object",
|
||||
"properties": { "query": { "type": "string" } },
|
||||
"required": ["query"],
|
||||
"additionalProperties": false
|
||||
}))
|
||||
.map_err(|error| rmcp::ErrorData::internal_error(error.to_string(), None))?;
|
||||
let mut tool = Tool::new(
|
||||
Cow::Borrowed(PRODUCT_SEARCH_TOOL),
|
||||
Cow::Borrowed("Search Walmart products."),
|
||||
Arc::new(input_schema),
|
||||
);
|
||||
tool.annotations = Some(ToolAnnotations::new().read_only(true));
|
||||
tool.meta = Some(Meta(serde_json::Map::from_iter([
|
||||
("connector_id".to_string(), json!(CONNECTOR_ID)),
|
||||
("connector_name".to_string(), json!(CONNECTOR_NAME)),
|
||||
(
|
||||
"connector_description".to_string(),
|
||||
json!("Search Walmart products."),
|
||||
),
|
||||
("link_id".to_string(), json!(LINK_ID)),
|
||||
("openai/outputTemplate".to_string(), json!(RESOURCE_URI)),
|
||||
(
|
||||
"_codex_apps".to_string(),
|
||||
json!({
|
||||
"resource_uri": format!(
|
||||
"/{CONNECTOR_ID}/{LINK_ID}/{PRODUCT_SEARCH_TOOL}"
|
||||
),
|
||||
"contains_mcp_source": true,
|
||||
}),
|
||||
),
|
||||
])));
|
||||
|
||||
Ok(ListToolsResult {
|
||||
tools: vec![tool],
|
||||
next_cursor: None,
|
||||
meta: None,
|
||||
})
|
||||
}
|
||||
|
||||
async fn call_tool(
|
||||
&self,
|
||||
request: CallToolRequestParams,
|
||||
context: RequestContext<RoleServer>,
|
||||
) -> Result<CallToolResult, rmcp::ErrorData> {
|
||||
let mut meta = context.meta;
|
||||
meta.0.remove("progressToken");
|
||||
self.calls
|
||||
.lock()
|
||||
.expect("recorded tool calls lock poisoned")
|
||||
.push(json!({
|
||||
"name": request.name,
|
||||
"arguments": request.arguments,
|
||||
"meta": meta,
|
||||
}));
|
||||
|
||||
match request.name.as_ref() {
|
||||
PRODUCT_SEARCH_TOOL => Ok(CallToolResult::structured(json!({ "products": [] }))),
|
||||
"fetch_resource" => {
|
||||
let mut result = CallToolResult::structured(json!({
|
||||
"contents": [{
|
||||
"uri": RESOURCE_URI,
|
||||
"mimeType": "text/html",
|
||||
"text": "<html>Walmart</html>",
|
||||
"meta": { "app": CONNECTOR_ID },
|
||||
}]
|
||||
}));
|
||||
result.content = vec![Content::text("Fetched Walmart resource")];
|
||||
Ok(result)
|
||||
}
|
||||
name => Err(rmcp::ErrorData::invalid_params(
|
||||
format!("unknown tool: {name}"),
|
||||
None,
|
||||
)),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
async fn wait_for_mcp_tool_call_completed(
|
||||
mcp: &mut TestAppServer,
|
||||
call_id: &str,
|
||||
) -> Result<ItemCompletedNotification> {
|
||||
loop {
|
||||
let notification = timeout(
|
||||
DEFAULT_READ_TIMEOUT,
|
||||
mcp.read_stream_until_notification_message("item/completed"),
|
||||
)
|
||||
.await??;
|
||||
let Some(params) = notification.params else {
|
||||
continue;
|
||||
};
|
||||
let completed: ItemCompletedNotification = serde_json::from_value(params)?;
|
||||
if matches!(&completed.item, ThreadItem::McpToolCall { id, .. } if id == call_id) {
|
||||
return Ok(completed);
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -31,6 +31,7 @@ mod marketplace_add;
|
||||
mod marketplace_remove;
|
||||
mod marketplace_upgrade;
|
||||
mod mcp_resource;
|
||||
mod mcp_resource_origin;
|
||||
mod mcp_server_elicitation;
|
||||
mod mcp_server_status;
|
||||
mod mcp_tool;
|
||||
|
||||
@@ -856,6 +856,11 @@ impl McpConnectionManager {
|
||||
server_infos
|
||||
}
|
||||
|
||||
/// Returns available presentation metadata for one server.
|
||||
pub async fn server_info(&self, server: &str) -> Option<McpServerInfo> {
|
||||
self.list_available_server_infos().await.remove(server)
|
||||
}
|
||||
|
||||
fn with_server_metadata(&self, mut tool: ToolInfo) -> ToolInfo {
|
||||
let Some(metadata) = self.server_metadata.get(&tool.server_name) else {
|
||||
tool.supports_parallel_tool_calls = false;
|
||||
|
||||
@@ -644,6 +644,29 @@ impl CodexThread {
|
||||
Ok(serde_json::to_value(result)?)
|
||||
}
|
||||
|
||||
pub async fn mcp_server_info(
|
||||
&self,
|
||||
server: &str,
|
||||
) -> Option<codex_protocol::mcp::McpServerInfo> {
|
||||
self.current_mcp_runtime()
|
||||
.await
|
||||
.manager()
|
||||
.server_info(server)
|
||||
.await
|
||||
}
|
||||
|
||||
pub async fn mcp_tool_info(&self, server: &str, tool: &str) -> Option<codex_mcp::ToolInfo> {
|
||||
self.current_mcp_runtime()
|
||||
.await
|
||||
.manager()
|
||||
.list_all_tools()
|
||||
.await
|
||||
.into_iter()
|
||||
.find(|tool_info| {
|
||||
tool_info.server_name == server && tool_info.tool.name.as_ref() == tool
|
||||
})
|
||||
}
|
||||
|
||||
pub async fn call_mcp_tool(
|
||||
&self,
|
||||
server: &str,
|
||||
|
||||
Reference in New Issue
Block a user