Move plugin install suggestion handler to app-server

This commit is contained in:
pakrym-oai
2026-05-06 11:53:26 -07:00
parent 39c707410b
commit 7c1a3455fa
24 changed files with 523 additions and 572 deletions
+1
View File
@@ -90,6 +90,7 @@ mod mcp_refresh;
mod message_processor;
mod models;
mod outgoing_message;
mod plugin_install_suggest;
mod request_processors;
mod request_serialization;
mod server_request_error;
@@ -304,6 +304,9 @@ impl MessageProcessor {
agent_graph_store.clone(),
installation_id,
)
.register_extension(Arc::new(
crate::plugin_install_suggest::PluginInstallSuggestToolProvider,
) as Arc<dyn codex_core::ToolProvider>)
.session_source(session_source)
.analytics_events_client(analytics_events_client.clone())
.build());
@@ -0,0 +1,303 @@
use std::collections::HashSet;
use codex_app_server_protocol::AppInfo;
use codex_config::types::ToolSuggestDisabledTool;
use codex_core::config::Config;
use codex_core::config::edit::ConfigEdit;
use codex_core::config::edit::ConfigEditsBuilder;
use codex_core::connectors;
use codex_core::extensibility::FunctionCallError;
use codex_core::extensibility::FunctionToolOutput;
use codex_core::extensibility::ToolHandler as ExtensionToolHandler;
use codex_core::extensibility::ToolInvocation;
use codex_core::extensibility::ToolProvider;
use codex_core_plugins::PluginsManager;
use codex_login::CodexAuth;
use codex_mcp::CODEX_APPS_MCP_SERVER_NAME;
use codex_rmcp_client::ElicitationAction;
use codex_rmcp_client::ElicitationResponse;
use codex_tools::DiscoverableTool;
use codex_tools::DiscoverableToolAction;
use codex_tools::DiscoverableToolType;
use codex_tools::REQUEST_PLUGIN_INSTALL_PERSIST_ALWAYS_VALUE;
use codex_tools::REQUEST_PLUGIN_INSTALL_PERSIST_KEY;
use codex_tools::REQUEST_PLUGIN_INSTALL_TOOL_NAME;
use codex_tools::RequestPluginInstallArgs;
use codex_tools::RequestPluginInstallResult;
use codex_tools::ToolName;
use codex_tools::all_requested_connectors_picked_up;
use codex_tools::build_request_plugin_install_elicitation_request;
use codex_tools::filter_request_plugin_install_discoverable_tools_for_client;
use codex_tools::verified_connector_install_completed;
use codex_utils_absolute_path::AbsolutePathBuf;
use serde_json::Value;
use tracing::warn;
pub(crate) struct PluginInstallSuggestToolProvider;
impl ToolProvider for PluginInstallSuggestToolProvider {
fn handlers(&self) -> Vec<ExtensionToolHandler> {
vec![ExtensionToolHandler::function(
ToolName::plain(REQUEST_PLUGIN_INSTALL_TOOL_NAME),
handle_request_plugin_install,
)]
}
}
async fn handle_request_plugin_install(
invocation: ToolInvocation,
) -> Result<FunctionToolOutput, FunctionCallError> {
let arguments = invocation.function_arguments(REQUEST_PLUGIN_INSTALL_TOOL_NAME)?;
let args: RequestPluginInstallArgs = serde_json::from_str(arguments).map_err(|err| {
FunctionCallError::RespondToModel(format!("failed to parse function arguments: {err}"))
})?;
let suggest_reason = args.suggest_reason.trim();
if suggest_reason.is_empty() {
return Err(FunctionCallError::RespondToModel(
"suggest_reason must not be empty".to_string(),
));
}
if args.action_type != DiscoverableToolAction::Install {
return Err(FunctionCallError::RespondToModel(
"plugin install requests currently support only action_type=\"install\"".to_string(),
));
}
if args.tool_type == DiscoverableToolType::Plugin
&& invocation.app_server_client_name() == Some("codex-tui")
{
return Err(FunctionCallError::RespondToModel(
"plugin install requests are not available in codex-tui yet".to_string(),
));
}
let auth = invocation.auth().await;
let mcp_tools = invocation.list_mcp_tools().await;
let accessible_connectors = connectors::with_app_enabled_state(
connectors::accessible_connectors_from_mcp_tools(&mcp_tools),
invocation.config(),
);
let discoverable_tools = connectors::list_tool_suggest_discoverable_tools_with_auth(
invocation.config(),
auth.as_ref(),
&accessible_connectors,
)
.await
.map(|discoverable_tools| {
filter_request_plugin_install_discoverable_tools_for_client(
discoverable_tools,
invocation.app_server_client_name(),
)
})
.map_err(|err| {
FunctionCallError::RespondToModel(format!(
"plugin install requests are unavailable right now: {err}"
))
})?;
let tool = discoverable_tools
.into_iter()
.find(|tool| tool.tool_type() == args.tool_type && tool.id() == args.tool_id)
.ok_or_else(|| {
FunctionCallError::RespondToModel(format!(
"tool_id must match one of the discoverable tools exposed by {REQUEST_PLUGIN_INSTALL_TOOL_NAME}"
))
})?;
let request_id = format!("request_plugin_install_{}", invocation.call_id());
let params = build_request_plugin_install_elicitation_request(
CODEX_APPS_MCP_SERVER_NAME,
invocation.conversation_id_string(),
invocation.turn_id(),
&args,
suggest_reason,
&tool,
);
let response = invocation
.request_mcp_server_elicitation(request_id, params)
.await;
if let Some(response) = response.as_ref() {
maybe_persist_disabled_install_request(&invocation, &tool, response).await;
}
let user_confirmed = response
.as_ref()
.is_some_and(|response| response.action == ElicitationAction::Accept);
let completed = if user_confirmed {
verify_request_plugin_install_completed(&invocation, &tool, auth.as_ref()).await
} else {
false
};
if completed && let DiscoverableTool::Connector(connector) = &tool {
invocation
.merge_connector_selection(HashSet::from([connector.id.clone()]))
.await;
}
let content = serde_json::to_string(&RequestPluginInstallResult {
completed,
user_confirmed,
tool_type: args.tool_type,
action_type: args.action_type,
tool_id: tool.id().to_string(),
tool_name: tool.name().to_string(),
suggest_reason: suggest_reason.to_string(),
})
.map_err(|err| {
FunctionCallError::Fatal(format!(
"failed to serialize {REQUEST_PLUGIN_INSTALL_TOOL_NAME} response: {err}"
))
})?;
Ok(FunctionToolOutput::from_text(content, Some(true)))
}
async fn maybe_persist_disabled_install_request(
invocation: &ToolInvocation,
tool: &DiscoverableTool,
response: &ElicitationResponse,
) {
if !request_plugin_install_response_requests_persistent_disable(response) {
return;
}
let codex_home = invocation.codex_home().clone();
if let Err(err) = persist_disabled_install_request(&codex_home, tool).await {
warn!(
error = %err,
tool_id = tool.id(),
"failed to persist disabled tool suggestion"
);
return;
}
invocation.reload_user_config_layer().await;
}
fn request_plugin_install_response_requests_persistent_disable(
response: &ElicitationResponse,
) -> bool {
if response.action != ElicitationAction::Decline {
return false;
}
response
.meta
.as_ref()
.and_then(Value::as_object)
.and_then(|meta| meta.get(REQUEST_PLUGIN_INSTALL_PERSIST_KEY))
.and_then(Value::as_str)
== Some(REQUEST_PLUGIN_INSTALL_PERSIST_ALWAYS_VALUE)
}
async fn persist_disabled_install_request(
codex_home: &AbsolutePathBuf,
tool: &DiscoverableTool,
) -> anyhow::Result<()> {
ConfigEditsBuilder::new(codex_home)
.with_edits([ConfigEdit::AddToolSuggestDisabledTool(
disabled_install_request(tool),
)])
.apply()
.await
}
fn disabled_install_request(tool: &DiscoverableTool) -> ToolSuggestDisabledTool {
match tool {
DiscoverableTool::Connector(connector) => {
ToolSuggestDisabledTool::connector(connector.id.as_str())
}
DiscoverableTool::Plugin(plugin) => ToolSuggestDisabledTool::plugin(plugin.id.as_str()),
}
}
async fn verify_request_plugin_install_completed(
invocation: &ToolInvocation,
tool: &DiscoverableTool,
auth: Option<&CodexAuth>,
) -> bool {
match tool {
DiscoverableTool::Connector(connector) => refresh_missing_requested_connectors(
invocation,
auth,
std::slice::from_ref(&connector.id),
connector.id.as_str(),
)
.await
.is_some_and(|accessible_connectors| {
verified_connector_install_completed(connector.id.as_str(), &accessible_connectors)
}),
DiscoverableTool::Plugin(plugin) => {
invocation.reload_user_config_layer().await;
let completed = verified_plugin_install_completed(
plugin.id.as_str(),
invocation.config(),
invocation.plugins_manager(),
);
let _ = refresh_missing_requested_connectors(
invocation,
auth,
&plugin.app_connector_ids,
plugin.id.as_str(),
)
.await;
completed
}
}
}
async fn refresh_missing_requested_connectors(
invocation: &ToolInvocation,
auth: Option<&CodexAuth>,
expected_connector_ids: &[String],
tool_id: &str,
) -> Option<Vec<AppInfo>> {
if expected_connector_ids.is_empty() {
return Some(Vec::new());
}
let mcp_tools = invocation.list_mcp_tools().await;
let accessible_connectors = connectors::with_app_enabled_state(
connectors::accessible_connectors_from_mcp_tools(&mcp_tools),
invocation.config(),
);
if all_requested_connectors_picked_up(expected_connector_ids, &accessible_connectors) {
return Some(accessible_connectors);
}
match invocation.hard_refresh_codex_apps_tools_cache().await {
Ok(mcp_tools) => {
let accessible_connectors = connectors::with_app_enabled_state(
connectors::accessible_connectors_from_mcp_tools(&mcp_tools),
invocation.config(),
);
connectors::refresh_accessible_connectors_cache_from_mcp_tools(
invocation.config(),
auth,
&mcp_tools,
);
Some(accessible_connectors)
}
Err(err) => {
warn!(
"failed to refresh codex apps tools cache after plugin install request for {tool_id}: {err:#}"
);
None
}
}
}
fn verified_plugin_install_completed(
tool_id: &str,
config: &Config,
plugins_manager: &PluginsManager,
) -> bool {
let plugins_input = config.plugins_config_input();
plugins_manager
.list_marketplaces_for_config(&plugins_input, &[])
.ok()
.into_iter()
.flat_map(|outcome| outcome.marketplaces)
.flat_map(|marketplace| marketplace.plugins.into_iter())
.any(|plugin| plugin.id == tool_id && plugin.installed)
}
+4
View File
@@ -33,6 +33,10 @@ pub use codex_core::StateDbHandle;
pub use codex_core::ThreadManager;
pub use codex_core::ThreadManagerBuilder;
pub use codex_core::ThreadShutdownReport;
pub use codex_core::FunctionCallError;
pub use codex_core::FunctionToolOutput;
pub use codex_core::ToolHandler;
pub use codex_core::ToolInvocation;
pub use codex_core::ToolProvider;
pub use codex_core::agent_graph_store_from_state_db;
pub use codex_core::config::Config;
+1
View File
@@ -100,6 +100,7 @@ pub(crate) async fn run_codex_thread_interactive(
analytics_events_client: Some(parent_session.services.analytics_events_client.clone()),
state_db: parent_session.services.state_db.clone(),
thread_store: Arc::clone(&parent_session.services.thread_store),
extensions: parent_session.services.extensions.clone(),
}))
.or_cancel(&cancel_token)
.await??;
+3 -3
View File
@@ -115,7 +115,7 @@ pub(crate) async fn list_accessible_and_enabled_connectors_from_manager(
.collect()
}
pub(crate) async fn list_tool_suggest_discoverable_tools_with_auth(
pub async fn list_tool_suggest_discoverable_tools_with_auth(
config: &Config,
auth: Option<&CodexAuth>,
accessible_connectors: &[AppInfo],
@@ -162,7 +162,7 @@ pub async fn list_cached_accessible_connectors_from_mcp_tools(
})
}
pub(crate) fn refresh_accessible_connectors_cache_from_mcp_tools(
pub fn refresh_accessible_connectors_cache_from_mcp_tools(
config: &Config,
auth: Option<&CodexAuth>,
mcp_tools: &HashMap<String, ToolInfo>,
@@ -515,7 +515,7 @@ async fn chatgpt_get_request_with_auth_provider<T: DeserializeOwned>(
}
}
pub(crate) fn accessible_connectors_from_mcp_tools(
pub fn accessible_connectors_from_mcp_tools(
mcp_tools: &HashMap<String, ToolInfo>,
) -> Vec<AppInfo> {
// ToolInfo already carries plugin provenance, so app-level plugin sources
+62 -3
View File
@@ -3,8 +3,24 @@
use std::any::Any;
use std::any::TypeId;
use std::collections::HashMap;
use std::future::Future;
use std::pin::Pin;
use std::sync::Arc;
use crate::tools::registry::ToolHandler as CoreToolHandler;
use crate::tools::registry::ToolKind;
use codex_tools::ToolName;
pub use crate::function_tool::FunctionCallError;
pub use crate::tools::context::FunctionToolOutput;
pub use crate::tools::context::ToolInvocation;
type FunctionHandler = dyn Fn(
ToolInvocation,
) -> Pin<Box<dyn Future<Output = Result<FunctionToolOutput, FunctionCallError>> + Send>>
+ Send
+ Sync;
/// Stores registered implementations of codex-core extension traits.
///
/// Registrations are keyed by the trait object type used at insertion time. To
@@ -49,6 +65,49 @@ impl ExtensionRegistry {
/// Provides tools through codex-core extensibility.
///
/// Implementations are markers for now; tool-surface methods will be added as
/// the extensibility API takes shape.
pub trait ToolProvider: Send + Sync + 'static {}
/// Implementations are expected to return handlers owned by the provider. Tool
/// specs may still be provided by the existing built-in plan while handlers
/// migrate behind this extension point.
pub trait ToolProvider: Send + Sync + 'static {
/// Return tool handlers owned by this provider.
fn handlers(&self) -> Vec<ToolHandler>;
}
/// A tool handler supplied by an extension provider.
pub struct ToolHandler {
tool_name: ToolName,
function: Arc<FunctionHandler>,
}
impl ToolHandler {
/// Wrap a function tool handler for registration with codex-core.
pub fn function<F, Fut>(tool_name: ToolName, handler: F) -> Self
where
F: Fn(ToolInvocation) -> Fut + Send + Sync + 'static,
Fut: Future<Output = Result<FunctionToolOutput, FunctionCallError>> + Send + 'static,
{
Self {
tool_name,
function: Arc::new(move |invocation| Box::pin(handler(invocation))),
}
}
}
impl CoreToolHandler for ToolHandler {
type Output = FunctionToolOutput;
fn tool_name(&self) -> ToolName {
self.tool_name.clone()
}
fn kind(&self) -> ToolKind {
ToolKind::Function
}
fn handle(
&self,
invocation: ToolInvocation,
) -> impl Future<Output = Result<Self::Output, FunctionCallError>> + Send {
(self.function)(invocation)
}
}
+4
View File
@@ -35,6 +35,10 @@ pub mod exec;
pub mod exec_env;
pub mod extensibility;
pub use extensibility::ExtensionRegistry;
pub use extensibility::FunctionCallError;
pub use extensibility::FunctionToolOutput;
pub use extensibility::ToolHandler;
pub use extensibility::ToolInvocation;
pub use extensibility::ToolProvider;
mod exec_policy;
pub mod file_watcher;
+4
View File
@@ -32,6 +32,7 @@ use crate::context::PersonalitySpecInstructions;
use crate::default_skill_metadata_budget;
use crate::environment_selection::ResolvedTurnEnvironments;
use crate::exec_policy::ExecPolicyManager;
use crate::extensibility::ExtensionRegistry;
use crate::parse_turn_item;
use crate::path_utils::normalize_for_native_workdir;
use crate::realtime_conversation::RealtimeConversationManager;
@@ -411,6 +412,7 @@ pub(crate) struct CodexSpawnArgs {
pub(crate) analytics_events_client: Option<AnalyticsEventsClient>,
pub(crate) state_db: Option<state_db::StateDbHandle>,
pub(crate) thread_store: Arc<dyn ThreadStore>,
pub(crate) extensions: ExtensionRegistry,
}
pub(crate) const INITIAL_SUBMIT_ID: &str = "";
@@ -471,6 +473,7 @@ impl Codex {
analytics_events_client,
state_db,
thread_store,
extensions,
} = args;
let (tx_sub, rx_sub) = async_channel::bounded(SUBMISSION_CHANNEL_CAPACITY);
let (tx_event, rx_event) = async_channel::unbounded();
@@ -648,6 +651,7 @@ impl Codex {
analytics_events_client,
state_db,
thread_store,
extensions,
parent_rollout_thread_trace,
)
.await
+2
View File
@@ -370,6 +370,7 @@ impl Session {
analytics_events_client: Option<AnalyticsEventsClient>,
state_db: Option<state_db::StateDbHandle>,
thread_store: Arc<dyn ThreadStore>,
extensions: ExtensionRegistry,
parent_rollout_thread_trace: ThreadTraceContext,
) -> anyhow::Result<Arc<Self>> {
debug!(
@@ -838,6 +839,7 @@ impl Session {
state_db: state_db_ctx.clone(),
live_thread: live_thread_init.as_ref().cloned(),
thread_store: Arc::clone(&thread_store),
extensions,
model_client: ModelClient::new(
Some(Arc::clone(&auth_manager)),
session_id,
+4
View File
@@ -545,6 +545,7 @@ fn test_tool_runtime(session: Arc<Session>, turn_context: Arc<TurnContext>) -> T
parallel_mcp_server_names: HashSet::new(),
discoverable_tools: None,
dynamic_tools: turn_context.dynamic_tools.as_slice(),
tool_providers: Vec::new(),
},
));
let tracker = Arc::new(tokio::sync::Mutex::new(TurnDiffTracker::new()));
@@ -3762,6 +3763,7 @@ pub(crate) async fn make_session_and_context() -> (Session, TurnContext) {
.await
.expect("state db should initialize"),
)),
extensions: ExtensionRegistry::default(),
model_client: ModelClient::new(
Some(auth_manager.clone()),
thread_id.into(),
@@ -5448,6 +5450,7 @@ where
codex_thread_store::LocalThreadStoreConfig::from_config(config.as_ref()),
state_db,
)),
extensions: ExtensionRegistry::default(),
model_client: ModelClient::new(
Some(Arc::clone(&auth_manager)),
thread_id.into(),
@@ -8287,6 +8290,7 @@ async fn fatal_tool_error_stops_turn_and_reports_error() {
parallel_mcp_server_names: HashSet::new(),
discoverable_tools: None,
dynamic_tools: turn_context.dynamic_tools.as_slice(),
tool_providers: Vec::new(),
},
);
let item = ResponseItem::CustomToolCall {
@@ -769,6 +769,7 @@ async fn guardian_subagent_does_not_inherit_parent_exec_policy_rules() {
analytics_events_client: None,
state_db: None,
thread_store,
extensions: ExtensionRegistry::default(),
})
.await
.expect("spawn guardian subagent");
+2
View File
@@ -19,6 +19,7 @@ use crate::compact_remote::run_inline_remote_auto_compact_task;
use crate::compact_remote_v2::run_inline_remote_auto_compact_task as run_inline_remote_auto_compact_task_v2;
use crate::connectors;
use crate::context::ContextualUserFragment;
use crate::extensibility::ToolProvider;
use crate::feedback_tags;
use crate::hook_runtime::PendingInputHookDisposition;
use crate::hook_runtime::emit_hook_completed_events;
@@ -1274,6 +1275,7 @@ pub(crate) async fn built_tools(
parallel_mcp_server_names,
discoverable_tools,
dynamic_tools: turn_context.dynamic_tools.as_slice(),
tool_providers: sess.services.extensions.get::<dyn ToolProvider>(),
},
)))
}
+2
View File
@@ -6,6 +6,7 @@ use crate::agent::AgentControl;
use crate::client::ModelClient;
use crate::config::StartedNetworkProxy;
use crate::exec_policy::ExecPolicyManager;
use crate::extensibility::ExtensionRegistry;
use crate::guardian::GuardianRejection;
use crate::guardian::GuardianRejectionCircuitBreaker;
use crate::mcp::McpManager;
@@ -66,6 +67,7 @@ pub(crate) struct SessionServices {
pub(crate) state_db: Option<StateDbHandle>,
pub(crate) live_thread: Option<LiveThread>,
pub(crate) thread_store: Arc<dyn ThreadStore>,
pub(crate) extensions: ExtensionRegistry,
/// Session-scoped model client shared across turns.
pub(crate) model_client: ModelClient,
pub(crate) code_mode_service: CodeModeService,
+1
View File
@@ -1356,6 +1356,7 @@ impl ThreadManagerState {
analytics_events_client: self.analytics_events_client.clone(),
state_db: Some(self.state_db.clone()),
thread_store: Arc::clone(&self.thread_store),
extensions: self.extensions.clone(),
})
.await?;
let new_thread = self
+1
View File
@@ -302,6 +302,7 @@ async fn build_nested_router(exec: &ExecContext) -> ToolRouter {
parallel_mcp_server_names,
discoverable_tools: None,
dynamic_tools: exec.turn.dynamic_tools.as_slice(),
tool_providers: Vec::new(),
},
)
}
+111 -8
View File
@@ -1,4 +1,5 @@
use crate::context_manager::truncate_function_output_payload;
use crate::function_tool::FunctionCallError;
use crate::original_image_detail::sanitize_original_image_detail;
use crate::session::session::Session;
use crate::session::turn_context::TurnContext;
@@ -7,7 +8,12 @@ use crate::tools::TELEMETRY_PREVIEW_MAX_LINES;
use crate::tools::TELEMETRY_PREVIEW_TRUNCATION_NOTICE;
use crate::turn_diff_tracker::TurnDiffTracker;
use crate::unified_exec::resolve_max_tokens;
use codex_app_server_protocol::McpServerElicitationRequestParams;
use codex_login::CodexAuth;
use codex_mcp::ToolInfo;
use codex_protocol::mcp::CallToolResult;
use codex_rmcp_client::ElicitationResponse;
use codex_core_plugins::PluginsManager;
use codex_protocol::models::DEFAULT_IMAGE_DETAIL;
use codex_protocol::models::FunctionCallOutputBody;
use codex_protocol::models::FunctionCallOutputContentItem;
@@ -24,6 +30,8 @@ use codex_utils_string::take_bytes_at_char_boundary;
use serde::Serialize;
use serde_json::Value as JsonValue;
use std::borrow::Cow;
use std::collections::HashMap;
use std::collections::HashSet;
use std::sync::Arc;
use std::time::Duration;
use tokio::sync::Mutex;
@@ -46,14 +54,109 @@ pub enum ToolCallSource {
#[derive(Clone)]
pub struct ToolInvocation {
pub session: Arc<Session>,
pub turn: Arc<TurnContext>,
pub cancellation_token: CancellationToken,
pub tracker: SharedTurnDiffTracker,
pub call_id: String,
pub tool_name: ToolName,
pub source: ToolCallSource,
pub payload: ToolPayload,
pub(crate) session: Arc<Session>,
pub(crate) turn: Arc<TurnContext>,
pub(crate) cancellation_token: CancellationToken,
pub(crate) tracker: SharedTurnDiffTracker,
pub(crate) call_id: String,
pub(crate) tool_name: ToolName,
pub(crate) source: ToolCallSource,
pub(crate) payload: ToolPayload,
}
impl ToolInvocation {
pub fn function_arguments(&self, tool_name: &str) -> Result<&str, FunctionCallError> {
match &self.payload {
ToolPayload::Function { arguments } => Ok(arguments),
_ => Err(FunctionCallError::Fatal(format!(
"{tool_name} handler received unsupported payload"
))),
}
}
pub fn call_id(&self) -> &str {
self.call_id.as_str()
}
pub fn app_server_client_name(&self) -> Option<&str> {
self.turn.app_server_client_name.as_deref()
}
pub fn config(&self) -> &crate::config::Config {
self.turn.config.as_ref()
}
pub fn codex_home(&self) -> &codex_utils_absolute_path::AbsolutePathBuf {
&self.turn.config.codex_home
}
pub fn conversation_id_string(&self) -> String {
self.session.conversation_id.to_string()
}
pub fn turn_id(&self) -> String {
self.turn.sub_id.clone()
}
pub async fn auth(&self) -> Option<CodexAuth> {
self.session.services.auth_manager.auth().await
}
#[expect(
clippy::await_holding_invalid_type,
reason = "tool extensions read through the session-owned MCP manager guard"
)]
pub async fn list_mcp_tools(&self) -> HashMap<String, ToolInfo> {
self.session
.services
.mcp_connection_manager
.read()
.await
.list_all_tools()
.await
}
#[expect(
clippy::await_holding_invalid_type,
reason = "tool extensions refresh through the session-owned MCP manager guard"
)]
pub async fn hard_refresh_codex_apps_tools_cache(
&self,
) -> anyhow::Result<HashMap<String, ToolInfo>> {
self.session
.services
.mcp_connection_manager
.read()
.await
.hard_refresh_codex_apps_tools_cache()
.await
}
pub async fn request_mcp_server_elicitation(
&self,
request_id: String,
params: McpServerElicitationRequestParams,
) -> Option<ElicitationResponse> {
self.session
.request_mcp_server_elicitation(
self.turn.as_ref(),
rmcp::model::RequestId::String(request_id.into()),
params,
)
.await
}
pub async fn reload_user_config_layer(&self) {
self.session.reload_user_config_layer().await;
}
pub async fn merge_connector_selection(&self, connector_ids: HashSet<String>) {
self.session.merge_connector_selection(connector_ids).await;
}
pub fn plugins_manager(&self) -> &PluginsManager {
self.session.services.plugins_manager.as_ref()
}
}
#[derive(Clone, Debug)]
-2
View File
@@ -9,7 +9,6 @@ pub(crate) mod multi_agents_common;
pub(crate) mod multi_agents_v2;
mod plan;
mod request_permissions;
mod request_plugin_install;
mod request_user_input;
mod shell;
mod test_sync;
@@ -47,7 +46,6 @@ pub use mcp_resource::ListMcpResourcesHandler;
pub use mcp_resource::ReadMcpResourceHandler;
pub use plan::PlanHandler;
pub use request_permissions::RequestPermissionsHandler;
pub use request_plugin_install::RequestPluginInstallHandler;
pub use request_user_input::RequestUserInputHandler;
pub use shell::ContainerExecHandler;
pub use shell::LocalShellHandler;
@@ -1,339 +0,0 @@
use std::collections::HashSet;
use codex_app_server_protocol::AppInfo;
use codex_config::types::ToolSuggestDisabledTool;
use codex_mcp::CODEX_APPS_MCP_SERVER_NAME;
use codex_rmcp_client::ElicitationAction;
use codex_rmcp_client::ElicitationResponse;
use codex_tools::DiscoverableTool;
use codex_tools::DiscoverableToolAction;
use codex_tools::DiscoverableToolType;
use codex_tools::REQUEST_PLUGIN_INSTALL_PERSIST_ALWAYS_VALUE;
use codex_tools::REQUEST_PLUGIN_INSTALL_PERSIST_KEY;
use codex_tools::REQUEST_PLUGIN_INSTALL_TOOL_NAME;
use codex_tools::RequestPluginInstallArgs;
use codex_tools::RequestPluginInstallResult;
use codex_tools::ToolName;
use codex_tools::all_requested_connectors_picked_up;
use codex_tools::build_request_plugin_install_elicitation_request;
use codex_tools::filter_request_plugin_install_discoverable_tools_for_client;
use codex_tools::verified_connector_install_completed;
use rmcp::model::RequestId;
use serde_json::Value;
use tracing::warn;
use crate::config::edit::ConfigEdit;
use crate::config::edit::ConfigEditsBuilder;
use crate::connectors;
use crate::function_tool::FunctionCallError;
use crate::tools::context::FunctionToolOutput;
use crate::tools::context::ToolInvocation;
use crate::tools::context::ToolPayload;
use crate::tools::handlers::parse_arguments;
use crate::tools::registry::ToolHandler;
use crate::tools::registry::ToolKind;
pub struct RequestPluginInstallHandler;
impl ToolHandler for RequestPluginInstallHandler {
type Output = FunctionToolOutput;
fn tool_name(&self) -> ToolName {
ToolName::plain(REQUEST_PLUGIN_INSTALL_TOOL_NAME)
}
fn kind(&self) -> ToolKind {
ToolKind::Function
}
#[expect(
clippy::await_holding_invalid_type,
reason = "plugin install discovery reads through the session-owned manager guard"
)]
async fn handle(&self, invocation: ToolInvocation) -> Result<Self::Output, FunctionCallError> {
let ToolInvocation {
payload,
session,
turn,
call_id,
..
} = invocation;
let arguments = match payload {
ToolPayload::Function { arguments } => arguments,
_ => {
return Err(FunctionCallError::Fatal(format!(
"{REQUEST_PLUGIN_INSTALL_TOOL_NAME} handler received unsupported payload"
)));
}
};
let args: RequestPluginInstallArgs = parse_arguments(&arguments)?;
let suggest_reason = args.suggest_reason.trim();
if suggest_reason.is_empty() {
return Err(FunctionCallError::RespondToModel(
"suggest_reason must not be empty".to_string(),
));
}
if args.action_type != DiscoverableToolAction::Install {
return Err(FunctionCallError::RespondToModel(
"plugin install requests currently support only action_type=\"install\""
.to_string(),
));
}
if args.tool_type == DiscoverableToolType::Plugin
&& turn.app_server_client_name.as_deref() == Some("codex-tui")
{
return Err(FunctionCallError::RespondToModel(
"plugin install requests are not available in codex-tui yet".to_string(),
));
}
let auth = session.services.auth_manager.auth().await;
let manager = session.services.mcp_connection_manager.read().await;
let mcp_tools = manager.list_all_tools().await;
drop(manager);
let accessible_connectors = connectors::with_app_enabled_state(
connectors::accessible_connectors_from_mcp_tools(&mcp_tools),
&turn.config,
);
let discoverable_tools = connectors::list_tool_suggest_discoverable_tools_with_auth(
&turn.config,
auth.as_ref(),
&accessible_connectors,
)
.await
.map(|discoverable_tools| {
filter_request_plugin_install_discoverable_tools_for_client(
discoverable_tools,
turn.app_server_client_name.as_deref(),
)
})
.map_err(|err| {
FunctionCallError::RespondToModel(format!(
"plugin install requests are unavailable right now: {err}"
))
})?;
let tool = discoverable_tools
.into_iter()
.find(|tool| tool.tool_type() == args.tool_type && tool.id() == args.tool_id)
.ok_or_else(|| {
FunctionCallError::RespondToModel(format!(
"tool_id must match one of the discoverable tools exposed by {REQUEST_PLUGIN_INSTALL_TOOL_NAME}"
))
})?;
let request_id = RequestId::String(format!("request_plugin_install_{call_id}").into());
let params = build_request_plugin_install_elicitation_request(
CODEX_APPS_MCP_SERVER_NAME,
session.conversation_id.to_string(),
turn.sub_id.clone(),
&args,
suggest_reason,
&tool,
);
let response = session
.request_mcp_server_elicitation(turn.as_ref(), request_id, params)
.await;
if let Some(response) = response.as_ref() {
maybe_persist_disabled_install_request(&session, &turn, &tool, response).await;
}
let user_confirmed = response
.as_ref()
.is_some_and(|response| response.action == ElicitationAction::Accept);
let completed = if user_confirmed {
verify_request_plugin_install_completed(&session, &turn, &tool, auth.as_ref()).await
} else {
false
};
if completed && let DiscoverableTool::Connector(connector) = &tool {
session
.merge_connector_selection(HashSet::from([connector.id.clone()]))
.await;
}
let content = serde_json::to_string(&RequestPluginInstallResult {
completed,
user_confirmed,
tool_type: args.tool_type,
action_type: args.action_type,
tool_id: tool.id().to_string(),
tool_name: tool.name().to_string(),
suggest_reason: suggest_reason.to_string(),
})
.map_err(|err| {
FunctionCallError::Fatal(format!(
"failed to serialize {REQUEST_PLUGIN_INSTALL_TOOL_NAME} response: {err}"
))
})?;
Ok(FunctionToolOutput::from_text(content, Some(true)))
}
}
async fn maybe_persist_disabled_install_request(
session: &crate::session::session::Session,
turn: &crate::session::turn_context::TurnContext,
tool: &DiscoverableTool,
response: &ElicitationResponse,
) {
if !request_plugin_install_response_requests_persistent_disable(response) {
return;
}
if let Err(err) = persist_disabled_install_request(&turn.config.codex_home, tool).await {
warn!(
error = %err,
tool_id = tool.id(),
"failed to persist disabled tool suggestion"
);
return;
}
session.reload_user_config_layer().await;
}
fn request_plugin_install_response_requests_persistent_disable(
response: &ElicitationResponse,
) -> bool {
if response.action != ElicitationAction::Decline {
return false;
}
response
.meta
.as_ref()
.and_then(Value::as_object)
.and_then(|meta| meta.get(REQUEST_PLUGIN_INSTALL_PERSIST_KEY))
.and_then(Value::as_str)
== Some(REQUEST_PLUGIN_INSTALL_PERSIST_ALWAYS_VALUE)
}
async fn persist_disabled_install_request(
codex_home: &codex_utils_absolute_path::AbsolutePathBuf,
tool: &DiscoverableTool,
) -> anyhow::Result<()> {
ConfigEditsBuilder::new(codex_home)
.with_edits([ConfigEdit::AddToolSuggestDisabledTool(
disabled_install_request(tool),
)])
.apply()
.await
}
fn disabled_install_request(tool: &DiscoverableTool) -> ToolSuggestDisabledTool {
match tool {
DiscoverableTool::Connector(connector) => {
ToolSuggestDisabledTool::connector(connector.id.as_str())
}
DiscoverableTool::Plugin(plugin) => ToolSuggestDisabledTool::plugin(plugin.id.as_str()),
}
}
async fn verify_request_plugin_install_completed(
session: &crate::session::session::Session,
turn: &crate::session::turn_context::TurnContext,
tool: &DiscoverableTool,
auth: Option<&codex_login::CodexAuth>,
) -> bool {
match tool {
DiscoverableTool::Connector(connector) => refresh_missing_requested_connectors(
session,
turn,
auth,
std::slice::from_ref(&connector.id),
connector.id.as_str(),
)
.await
.is_some_and(|accessible_connectors| {
verified_connector_install_completed(connector.id.as_str(), &accessible_connectors)
}),
DiscoverableTool::Plugin(plugin) => {
session.reload_user_config_layer().await;
let config = session.get_config().await;
let completed = verified_plugin_install_completed(
plugin.id.as_str(),
config.as_ref(),
session.services.plugins_manager.as_ref(),
);
let _ = refresh_missing_requested_connectors(
session,
turn,
auth,
&plugin.app_connector_ids,
plugin.id.as_str(),
)
.await;
completed
}
}
}
#[expect(
clippy::await_holding_invalid_type,
reason = "connector cache refresh reads through the session-owned manager guard"
)]
async fn refresh_missing_requested_connectors(
session: &crate::session::session::Session,
turn: &crate::session::turn_context::TurnContext,
auth: Option<&codex_login::CodexAuth>,
expected_connector_ids: &[String],
tool_id: &str,
) -> Option<Vec<AppInfo>> {
if expected_connector_ids.is_empty() {
return Some(Vec::new());
}
let manager = session.services.mcp_connection_manager.read().await;
let mcp_tools = manager.list_all_tools().await;
let accessible_connectors = connectors::with_app_enabled_state(
connectors::accessible_connectors_from_mcp_tools(&mcp_tools),
&turn.config,
);
if all_requested_connectors_picked_up(expected_connector_ids, &accessible_connectors) {
return Some(accessible_connectors);
}
match manager.hard_refresh_codex_apps_tools_cache().await {
Ok(mcp_tools) => {
let accessible_connectors = connectors::with_app_enabled_state(
connectors::accessible_connectors_from_mcp_tools(&mcp_tools),
&turn.config,
);
connectors::refresh_accessible_connectors_cache_from_mcp_tools(
&turn.config,
auth,
&mcp_tools,
);
Some(accessible_connectors)
}
Err(err) => {
warn!(
"failed to refresh codex apps tools cache after plugin install request for {tool_id}: {err:#}"
);
None
}
}
}
fn verified_plugin_install_completed(
tool_id: &str,
config: &crate::config::Config,
plugins_manager: &codex_core_plugins::PluginsManager,
) -> bool {
let plugins_input = config.plugins_config_input();
plugins_manager
.list_marketplaces_for_config(&plugins_input, &[])
.ok()
.into_iter()
.flat_map(|outcome| outcome.marketplaces)
.flat_map(|marketplace| marketplace.plugins.into_iter())
.any(|plugin| plugin.id == tool_id && plugin.installed)
}
#[cfg(test)]
#[path = "request_plugin_install_tests.rs"]
mod tests;
@@ -1,214 +0,0 @@
use super::*;
use crate::plugins::test_support::load_plugins_config;
use crate::plugins::test_support::write_curated_plugin_sha;
use crate::plugins::test_support::write_openai_curated_marketplace;
use crate::plugins::test_support::write_plugins_feature_config;
use codex_config::CONFIG_TOML_FILE;
use codex_config::config_toml::ConfigToml;
use codex_config::types::ToolSuggestConfig;
use codex_config::types::ToolSuggestDisabledTool;
use codex_config::types::ToolSuggestDiscoverable;
use codex_config::types::ToolSuggestDiscoverableType;
use codex_core_plugins::PluginInstallRequest;
use codex_core_plugins::PluginsManager;
use codex_core_plugins::startup_sync::curated_plugins_repo_path;
use codex_rmcp_client::ElicitationResponse;
use codex_tools::DiscoverablePluginInfo;
use codex_utils_absolute_path::AbsolutePathBuf;
use core_test_support::PathExt;
use pretty_assertions::assert_eq;
use rmcp::model::ElicitationAction;
use serde_json::json;
use tempfile::tempdir;
#[tokio::test]
async fn verified_plugin_install_completed_requires_installed_plugin() {
let codex_home = tempdir().expect("tempdir should succeed");
let curated_root = curated_plugins_repo_path(codex_home.path());
write_openai_curated_marketplace(&curated_root, &["sample"]);
write_curated_plugin_sha(codex_home.path());
write_plugins_feature_config(codex_home.path());
let config = load_plugins_config(codex_home.path()).await;
let plugins_manager = PluginsManager::new(codex_home.path().to_path_buf());
assert!(!verified_plugin_install_completed(
"sample@openai-curated",
&config,
&plugins_manager,
));
plugins_manager
.install_plugin(PluginInstallRequest {
plugin_name: "sample".to_string(),
marketplace_path: AbsolutePathBuf::try_from(
curated_root.join(".agents/plugins/marketplace.json"),
)
.expect("marketplace path"),
})
.await
.expect("plugin should install");
let refreshed_config = load_plugins_config(codex_home.path()).await;
assert!(verified_plugin_install_completed(
"sample@openai-curated",
&refreshed_config,
&plugins_manager,
));
}
#[test]
fn request_plugin_install_response_persists_only_decline_always_mode() {
assert!(request_plugin_install_response_requests_persistent_disable(
&ElicitationResponse {
action: ElicitationAction::Decline,
content: None,
meta: Some(json!({
REQUEST_PLUGIN_INSTALL_PERSIST_KEY: REQUEST_PLUGIN_INSTALL_PERSIST_ALWAYS_VALUE
})),
}
));
assert!(
!request_plugin_install_response_requests_persistent_disable(&ElicitationResponse {
action: ElicitationAction::Accept,
content: None,
meta: Some(json!({
REQUEST_PLUGIN_INSTALL_PERSIST_KEY: REQUEST_PLUGIN_INSTALL_PERSIST_ALWAYS_VALUE
})),
})
);
assert!(
!request_plugin_install_response_requests_persistent_disable(&ElicitationResponse {
action: ElicitationAction::Decline,
content: None,
meta: Some(json!({ REQUEST_PLUGIN_INSTALL_PERSIST_KEY: "session" })),
})
);
assert!(
!request_plugin_install_response_requests_persistent_disable(&ElicitationResponse {
action: ElicitationAction::Decline,
content: None,
meta: None,
})
);
}
#[tokio::test]
async fn persist_disabled_install_request_writes_connector_config() {
let codex_home = tempdir().expect("tempdir should succeed");
let tool = connector_tool("connector_calendar", "Google Calendar");
persist_disabled_install_request(&codex_home.path().abs(), &tool)
.await
.expect("persist connector disable");
let contents =
std::fs::read_to_string(codex_home.path().join(CONFIG_TOML_FILE)).expect("read config");
let parsed: ConfigToml = toml::from_str(&contents).expect("parse config");
assert_eq!(
parsed.tool_suggest,
Some(ToolSuggestConfig {
discoverables: Vec::new(),
disabled_tools: vec![ToolSuggestDisabledTool::connector("connector_calendar")],
})
);
}
#[tokio::test]
async fn persist_disabled_install_request_writes_plugin_config() {
let codex_home = tempdir().expect("tempdir should succeed");
let tool = DiscoverableTool::Plugin(Box::new(DiscoverablePluginInfo {
id: "slack@openai-curated".to_string(),
name: "Slack".to_string(),
description: None,
has_skills: true,
mcp_server_names: Vec::new(),
app_connector_ids: Vec::new(),
}));
persist_disabled_install_request(&codex_home.path().abs(), &tool)
.await
.expect("persist plugin disable");
let contents =
std::fs::read_to_string(codex_home.path().join(CONFIG_TOML_FILE)).expect("read config");
let parsed: ConfigToml = toml::from_str(&contents).expect("parse config");
assert_eq!(
parsed.tool_suggest,
Some(ToolSuggestConfig {
discoverables: Vec::new(),
disabled_tools: vec![ToolSuggestDisabledTool::plugin("slack@openai-curated")],
})
);
}
#[tokio::test]
async fn persist_disabled_install_request_dedupes_existing_disabled_tools() {
let codex_home = tempdir().expect("tempdir should succeed");
let tool = connector_tool("connector_calendar", "Google Calendar");
std::fs::write(
codex_home.path().join(CONFIG_TOML_FILE),
r#"
[tool_suggest]
discoverables = [
{ type = "plugin", id = "sample@openai-curated" }
]
[[tool_suggest.disabled_tools]]
type = "connector"
id = " connector_calendar "
[[tool_suggest.disabled_tools]]
type = "connector"
id = "connector_calendar"
[[tool_suggest.disabled_tools]]
type = "connector"
id = " "
[[tool_suggest.disabled_tools]]
type = "plugin"
id = "slack@openai-curated"
"#,
)
.expect("write config");
persist_disabled_install_request(&codex_home.path().abs(), &tool)
.await
.expect("persist connector disable");
let contents =
std::fs::read_to_string(codex_home.path().join(CONFIG_TOML_FILE)).expect("read config");
let parsed: ConfigToml = toml::from_str(&contents).expect("parse config");
assert_eq!(
parsed.tool_suggest,
Some(ToolSuggestConfig {
discoverables: vec![ToolSuggestDiscoverable {
kind: ToolSuggestDiscoverableType::Plugin,
id: "sample@openai-curated".to_string(),
}],
disabled_tools: vec![
ToolSuggestDisabledTool::connector("connector_calendar"),
ToolSuggestDisabledTool::plugin("slack@openai-curated"),
],
})
);
}
fn connector_tool(id: &str, name: &str) -> DiscoverableTool {
DiscoverableTool::Connector(Box::new(AppInfo {
id: id.to_string(),
name: name.to_string(),
description: None,
logo_url: None,
logo_url_dark: None,
distribution_channel: None,
branding: None,
app_metadata: None,
labels: None,
install_url: None,
is_accessible: false,
is_enabled: true,
plugin_display_names: Vec::new(),
}))
}
+9 -1
View File
@@ -1,3 +1,4 @@
use crate::extensibility::ToolProvider;
use crate::function_tool::FunctionCallError;
use crate::sandboxing::SandboxPermissions;
use crate::session::session::Session;
@@ -50,6 +51,7 @@ pub(crate) struct ToolRouterParams<'a> {
pub(crate) parallel_mcp_server_names: HashSet<String>,
pub(crate) discoverable_tools: Option<Vec<DiscoverableTool>>,
pub(crate) dynamic_tools: &'a [DynamicToolSpec],
pub(crate) tool_providers: Vec<Arc<dyn ToolProvider>>,
}
impl ToolRouter {
@@ -61,8 +63,9 @@ impl ToolRouter {
parallel_mcp_server_names,
discoverable_tools,
dynamic_tools,
tool_providers,
} = params;
let builder = build_specs_with_discoverable_tools(
let mut builder = build_specs_with_discoverable_tools(
config,
mcp_tools,
deferred_mcp_tools,
@@ -70,6 +73,11 @@ impl ToolRouter {
discoverable_tools,
dynamic_tools,
);
for provider in tool_providers {
for handler in provider.handlers() {
builder.register_handler(Arc::new(handler));
}
}
let (specs, registry) = builder.build();
let deferred_dynamic_tools = dynamic_tools
.iter()
+3
View File
@@ -38,6 +38,7 @@ async fn parallel_support_does_not_match_namespaced_local_tool_names() -> anyhow
parallel_mcp_server_names: HashSet::new(),
discoverable_tools: None,
dynamic_tools: turn.dynamic_tools.as_slice(),
tool_providers: Vec::new(),
},
);
@@ -111,6 +112,7 @@ async fn mcp_parallel_support_uses_exact_payload_server() -> anyhow::Result<()>
parallel_mcp_server_names: HashSet::from(["echo".to_string()]),
discoverable_tools: None,
dynamic_tools: turn.dynamic_tools.as_slice(),
tool_providers: Vec::new(),
},
);
@@ -178,6 +180,7 @@ async fn model_visible_specs_filter_deferred_dynamic_tools() -> anyhow::Result<(
parallel_mcp_server_names: HashSet::new(),
discoverable_tools: None,
dynamic_tools: &dynamic_tools,
tool_providers: Vec::new(),
},
);
+1 -2
View File
@@ -89,7 +89,6 @@ pub(crate) fn build_specs_with_discoverable_tools(
use crate::tools::handlers::PlanHandler;
use crate::tools::handlers::ReadMcpResourceHandler;
use crate::tools::handlers::RequestPermissionsHandler;
use crate::tools::handlers::RequestPluginInstallHandler;
use crate::tools::handlers::RequestUserInputHandler;
use crate::tools::handlers::ShellCommandHandler;
use crate::tools::handlers::ShellHandler;
@@ -285,7 +284,7 @@ pub(crate) fn build_specs_with_discoverable_tools(
builder.register_handler(Arc::new(ToolSearchHandler::new(entries)));
}
ToolHandlerKind::RequestPluginInstall => {
builder.register_handler(Arc::new(RequestPluginInstallHandler));
// Registered by ToolProvider extensions.
}
ToolHandlerKind::UpdateGoal => {
builder.register_handler(Arc::new(UpdateGoalHandler));
+1
View File
@@ -354,6 +354,7 @@ async fn assert_model_tools(
parallel_mcp_server_names: std::collections::HashSet::new(),
discoverable_tools: None,
dynamic_tools: &[],
tool_providers: Vec::new(),
},
);
let model_visible_specs = router.model_visible_specs();