mirror of
https://github.com/openai/codex.git
synced 2026-11-13 21:30:47 +00:00
Add MCP inventory support for statusline providers (#61248)
## What changed Add a TUI module for reading MCP server identities and tool catalogs from the connected app server's global inventory. Support paginated discovery and targeted server lookup through `mcpServerStatus/list`. Retain plugin identities and tool errors, filter entries to those with an explicit null `httpOrigin`, and reject repeated pagination cursors or mismatched targeted responses. This adds inventory infrastructure without enabling provider discovery yet. ## Testing Add tests for global inventory independence from startup project overrides, preservation of tool catalogs and errors across pages, transport filtering, and rejection of a targeted response naming another server. GitOrigin-RevId: f429991deed38eb106641b1780f24c701626b4e1
This commit is contained in:
Joe Savona
@@ -229,6 +229,7 @@ mod startup_presentation;
|
||||
mod startup_recovery;
|
||||
mod status;
|
||||
mod status_indicator_widget;
|
||||
mod statusline_provider;
|
||||
mod streaming;
|
||||
mod style;
|
||||
mod subscription;
|
||||
|
||||
@@ -0,0 +1,21 @@
|
||||
//! MCP server identity and inventory for statusline providers.
|
||||
|
||||
#[cfg_attr(not(test), expect(dead_code, reason = "staged provider discovery"))]
|
||||
mod inventory;
|
||||
#[cfg(test)]
|
||||
mod test_support;
|
||||
|
||||
#[derive(Clone, Debug, PartialEq, Eq, PartialOrd, Ord)]
|
||||
pub(crate) struct McpServerName(String);
|
||||
|
||||
impl McpServerName {
|
||||
pub(super) fn as_str(&self) -> &str {
|
||||
&self.0
|
||||
}
|
||||
}
|
||||
|
||||
impl From<&str> for McpServerName {
|
||||
fn from(name: &str) -> Self {
|
||||
Self(name.to_owned())
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,148 @@
|
||||
//! Read MCP tool catalogs from the connected app server’s global inventory.
|
||||
|
||||
use super::McpServerName;
|
||||
use codex_app_server_client::AppServerRequestHandle;
|
||||
use codex_app_server_protocol::ClientRequest;
|
||||
use codex_app_server_protocol::ListMcpServerStatusParams;
|
||||
use codex_app_server_protocol::McpServerConnectionStatus;
|
||||
use codex_app_server_protocol::McpServerStatus;
|
||||
use codex_app_server_protocol::McpServerStatusDetail;
|
||||
use codex_app_server_protocol::RequestId;
|
||||
use codex_plugin::PluginId;
|
||||
use codex_plugin::PluginIdError;
|
||||
use codex_protocol::mcp::Tool;
|
||||
use color_eyre::Result;
|
||||
use color_eyre::eyre::bail;
|
||||
use futures::Stream;
|
||||
use futures::stream;
|
||||
use serde::Deserialize;
|
||||
use serde_json::Value;
|
||||
use std::collections::HashMap;
|
||||
use std::collections::HashSet;
|
||||
use uuid::Uuid;
|
||||
|
||||
#[derive(Debug, PartialEq)]
|
||||
pub(super) struct StatuslineServer {
|
||||
pub(super) name: McpServerName,
|
||||
pub(super) runtime_status: Option<McpServerConnectionStatus>,
|
||||
pub(super) plugin_id: Option<PluginId>,
|
||||
pub(super) tools: Result<HashMap<String, Tool>, String>,
|
||||
}
|
||||
|
||||
impl TryFrom<McpServerStatus> for StatuslineServer {
|
||||
type Error = PluginIdError;
|
||||
|
||||
fn try_from(row: McpServerStatus) -> Result<Self, Self::Error> {
|
||||
let tools = match row.tools_error {
|
||||
Some(error) => Err(error),
|
||||
None if row.runtime_status == Some(McpServerConnectionStatus::Failed) => {
|
||||
Err(format!("MCP server {} is not connected", row.name))
|
||||
}
|
||||
None => Ok(row.tools),
|
||||
};
|
||||
Ok(Self {
|
||||
name: McpServerName(row.name),
|
||||
runtime_status: row.runtime_status,
|
||||
plugin_id: row.plugin_id.as_deref().map(PluginId::parse).transpose()?,
|
||||
tools,
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Deserialize)]
|
||||
#[serde(rename_all = "camelCase")]
|
||||
struct InventoryPage {
|
||||
data: Vec<Value>,
|
||||
next_cursor: Option<String>,
|
||||
}
|
||||
|
||||
async fn read_page(
|
||||
client: &AppServerRequestHandle,
|
||||
params: ListMcpServerStatusParams,
|
||||
) -> Result<InventoryPage> {
|
||||
Ok(client
|
||||
.request_typed(ClientRequest::McpServerStatusList {
|
||||
request_id: RequestId::String(Uuid::new_v4().to_string()),
|
||||
params,
|
||||
})
|
||||
.await?)
|
||||
}
|
||||
|
||||
/// Request the whole inventory. Some hosts paginate by offset into a newly rebuilt catalog.
|
||||
pub(super) fn global_pages(
|
||||
client: &AppServerRequestHandle,
|
||||
) -> impl Stream<Item = Result<Vec<StatuslineServer>>> + '_ {
|
||||
stream::try_unfold(Some((None, HashSet::new())), move |next| async move {
|
||||
let Some((cursor, mut seen)) = next else {
|
||||
return Ok(None);
|
||||
};
|
||||
if let Some(cursor) = &cursor
|
||||
&& !seen.insert(String::clone(cursor))
|
||||
{
|
||||
bail!("MCP inventory repeated a cursor");
|
||||
}
|
||||
let page = read_page(
|
||||
client,
|
||||
ListMcpServerStatusParams {
|
||||
thread_id: None,
|
||||
server_name: None,
|
||||
detail: Some(McpServerStatusDetail::ToolsAndAuthOnly),
|
||||
cursor,
|
||||
limit: None,
|
||||
},
|
||||
)
|
||||
.await?;
|
||||
let mut servers = Vec::new();
|
||||
for value in page.data {
|
||||
if let Some(row) = local_server(value)? {
|
||||
servers.push(row);
|
||||
}
|
||||
}
|
||||
Ok(Some((
|
||||
servers,
|
||||
page.next_cursor.map(|next| (Some(next), seen)),
|
||||
)))
|
||||
})
|
||||
}
|
||||
|
||||
fn local_server(value: Value) -> Result<Option<StatuslineServer>> {
|
||||
// Stdio servers are eligible. The app server sets `httpOrigin` to null for stdio
|
||||
// and for invalid or unsupported HTTP URLs. Skip entries from older hosts that
|
||||
// omit the field because their transport is unknown.
|
||||
if value.get("httpOrigin") != Some(&Value::Null) {
|
||||
return Ok(None);
|
||||
}
|
||||
let status = serde_json::from_value::<McpServerStatus>(value)?;
|
||||
Ok(StatuslineServer::try_from(status).ok())
|
||||
}
|
||||
|
||||
pub(super) async fn server(
|
||||
client: &AppServerRequestHandle,
|
||||
name: &McpServerName,
|
||||
) -> Result<Option<StatuslineServer>> {
|
||||
let mut page = read_page(
|
||||
client,
|
||||
ListMcpServerStatusParams {
|
||||
thread_id: None,
|
||||
server_name: Some(name.0.clone()),
|
||||
detail: Some(McpServerStatusDetail::ToolsAndAuthOnly),
|
||||
cursor: None,
|
||||
limit: None,
|
||||
},
|
||||
)
|
||||
.await?;
|
||||
if page.next_cursor.is_some() || page.data.len() > 1 {
|
||||
bail!("targeted MCP inventory returned more than one server")
|
||||
}
|
||||
let Some(value) = page.data.pop() else {
|
||||
return Ok(None);
|
||||
};
|
||||
if value["name"] != name.as_str() {
|
||||
bail!("targeted MCP inventory did not identify the requested server")
|
||||
}
|
||||
local_server(value)
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
#[path = "inventory_tests.rs"]
|
||||
mod tests;
|
||||
@@ -0,0 +1,188 @@
|
||||
#![cfg(test)]
|
||||
//! Verify global inventory against the connected app server and its configuration.
|
||||
|
||||
use super::super::test_support::reply;
|
||||
use super::super::test_support::row;
|
||||
use super::super::test_support::test_client;
|
||||
use super::*;
|
||||
use crate::AppServerTarget;
|
||||
use crate::backend_id::BackendId;
|
||||
use crate::legacy_core::config::ConfigBuilder;
|
||||
use crate::legacy_core::config::ConfigOverrides;
|
||||
use codex_config::LoaderOverrides;
|
||||
use codex_exec_server::EnvironmentManager;
|
||||
use futures::TryStreamExt;
|
||||
use pretty_assertions::assert_eq;
|
||||
use serde_json::json;
|
||||
use std::sync::Arc;
|
||||
|
||||
#[tokio::test]
|
||||
async fn global_inventory_ignores_the_startup_projects_overrides() -> Result<()> {
|
||||
let temp = tempfile::tempdir()?;
|
||||
let home = temp.path().join("home");
|
||||
let project = temp.path().join("project");
|
||||
std::fs::create_dir_all(&home)?;
|
||||
std::fs::create_dir_all(project.join(".git"))?;
|
||||
std::fs::write(project.join(".git/HEAD"), "ref: refs/heads/main\n")?;
|
||||
std::fs::create_dir(project.join(".codex"))?;
|
||||
let package = home.join("plugins/cache/local/daily/local");
|
||||
std::fs::create_dir_all(package.join(".codex-plugin"))?;
|
||||
std::fs::write(
|
||||
package.join(".codex-plugin/plugin.json"),
|
||||
r#"{"name":"daily"}"#,
|
||||
)?;
|
||||
let missing = temp.path().join("missing-mcp-fixture");
|
||||
std::fs::write(
|
||||
package.join(".mcp.json"),
|
||||
json!({"mcpServers":{
|
||||
"implicit-calendar":{"command":missing}
|
||||
}})
|
||||
.to_string(),
|
||||
)?;
|
||||
let command = toml::Value::String(missing.to_string_lossy().into_owned());
|
||||
let replacement = toml::Value::String(
|
||||
temp.path()
|
||||
.join("missing-project-fixture")
|
||||
.to_string_lossy()
|
||||
.into_owned(),
|
||||
);
|
||||
let trusted = toml::Value::String(project.to_string_lossy().into_owned());
|
||||
std::fs::write(
|
||||
home.join("config.toml"),
|
||||
format!(
|
||||
r#"
|
||||
[projects.{trusted}]
|
||||
trust_level = "trusted"
|
||||
[features]
|
||||
plugins = true
|
||||
remote_plugin = false
|
||||
[plugins."daily@local"]
|
||||
enabled = true
|
||||
[mcp_servers.calendar]
|
||||
command = {command}
|
||||
"#
|
||||
),
|
||||
)?;
|
||||
std::fs::write(
|
||||
project.join(".codex/config.toml"),
|
||||
format!(
|
||||
r#"
|
||||
[mcp_servers.calendar]
|
||||
command = {replacement}
|
||||
[mcp_servers.project_only]
|
||||
command = {replacement}
|
||||
[mcp_servers.implicit-calendar]
|
||||
command = {replacement}
|
||||
"#
|
||||
),
|
||||
)?;
|
||||
let config = ConfigBuilder::default()
|
||||
.codex_home(home.clone())
|
||||
.harness_overrides(ConfigOverrides {
|
||||
cwd: Some(project.clone()),
|
||||
..Default::default()
|
||||
})
|
||||
.build()
|
||||
.await?;
|
||||
assert!(config.mcp_servers.get().contains_key("project_only"));
|
||||
let session = Box::pin(crate::start_app_server_for_picker(
|
||||
BackendId::new(),
|
||||
&config,
|
||||
&AppServerTarget::Embedded,
|
||||
vec![("mcp_servers.cli-weather.command".into(), command)],
|
||||
LoaderOverrides::without_managed_config_for_tests(),
|
||||
/*state_db*/ None,
|
||||
Arc::new(EnvironmentManager::default_for_tests()),
|
||||
))
|
||||
.await?;
|
||||
let client = session.request_handle();
|
||||
let servers = global_pages(&client).try_concat().await?;
|
||||
let daily = PluginId::parse("daily@local")?;
|
||||
let identities = servers
|
||||
.iter()
|
||||
.map(|s| (s.name.as_str(), s.plugin_id.as_ref()))
|
||||
.collect::<Vec<_>>();
|
||||
assert_eq!(
|
||||
identities,
|
||||
[
|
||||
("calendar", None),
|
||||
("cli-weather", None),
|
||||
("implicit-calendar", Some(&daily))
|
||||
]
|
||||
);
|
||||
session.shutdown().await?;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn global_pages_preserve_catalogs_and_skip_hosted_or_unidentified_transports() -> Result<()> {
|
||||
let (client, mut commands) = test_client();
|
||||
let host = async {
|
||||
let tool = json!({"name":"forecast","inputSchema":{}});
|
||||
let direct = row("cli-weather", vec![tool.clone()]);
|
||||
let mut plugin = row("planner", vec![tool]);
|
||||
plugin["pluginId"] = json!("work@local");
|
||||
let mut hosted = row("hosted", vec![]);
|
||||
hosted["httpOrigin"] = json!("https://example.test");
|
||||
let mut older = row("missing-origin", vec![]);
|
||||
older.as_object_mut().unwrap().remove("httpOrigin");
|
||||
let mut failed = row("failed", vec![]);
|
||||
failed["runtimeStatus"] = Value::Null;
|
||||
failed["toolsError"] = json!("connection refused");
|
||||
for (cursor, rows, next) in [
|
||||
(None, vec![direct, hosted, older], Some("next")),
|
||||
(Some("next"), vec![plugin, failed], None),
|
||||
] {
|
||||
reply(
|
||||
&mut commands,
|
||||
"mcpServerStatus/list",
|
||||
json!({"threadId":null,"serverName":null,"detail":"toolsAndAuthOnly","cursor":cursor,"limit":null}),
|
||||
json!({"data":rows,"nextCursor":next}),
|
||||
).await;
|
||||
}
|
||||
};
|
||||
let (servers, ()) = tokio::join!(global_pages(&client).try_concat(), host);
|
||||
let catalogs = servers?
|
||||
.into_iter()
|
||||
.map(|s| {
|
||||
(
|
||||
s.name,
|
||||
s.plugin_id,
|
||||
s.tools.map(|tools| tools.into_keys().collect::<Vec<_>>()),
|
||||
)
|
||||
})
|
||||
.collect::<Vec<_>>();
|
||||
assert_eq!(
|
||||
catalogs,
|
||||
[
|
||||
("cli-weather".into(), None, Ok(vec!["forecast".to_owned()])),
|
||||
(
|
||||
"planner".into(),
|
||||
Some(PluginId::parse("work@local")?),
|
||||
Ok(vec!["forecast".to_owned()])
|
||||
),
|
||||
("failed".into(), None, Err("connection refused".to_owned())),
|
||||
]
|
||||
);
|
||||
Ok(())
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn targeted_inventory_rejects_another_server() {
|
||||
let (client, mut commands) = test_client();
|
||||
let name = "calendar".into();
|
||||
let other = row("other", vec![json!({"name":"footer","inputSchema":{}})]);
|
||||
let (result, ()) = tokio::join!(
|
||||
server(&client, &name),
|
||||
reply(
|
||||
&mut commands,
|
||||
"mcpServerStatus/list",
|
||||
json!({"threadId":null,"serverName":"calendar","detail":"toolsAndAuthOnly","cursor":null,"limit":null}),
|
||||
json!({"data":[other],"nextCursor":null}),
|
||||
)
|
||||
);
|
||||
assert_eq!(
|
||||
result.unwrap_err().to_string(),
|
||||
"targeted MCP inventory did not identify the requested server"
|
||||
);
|
||||
}
|
||||
@@ -0,0 +1,38 @@
|
||||
#![cfg(test)]
|
||||
//! Drive statusline RPCs through the same request handle a remote TUI uses.
|
||||
|
||||
use codex_app_server_client::AppServerClient;
|
||||
use codex_app_server_client::AppServerRequestHandle;
|
||||
use codex_app_server_client::CloudClient;
|
||||
use codex_app_server_client::CloudCommand;
|
||||
use codex_app_server_client::CloudOperation;
|
||||
use pretty_assertions::assert_eq;
|
||||
use serde_json::Value;
|
||||
use serde_json::json;
|
||||
use tokio::sync::mpsc;
|
||||
|
||||
pub(super) type Requests = mpsc::Receiver<CloudCommand>;
|
||||
|
||||
pub(super) fn test_client() -> (AppServerRequestHandle, Requests) {
|
||||
let (cloud, commands, _events) = CloudClient::channel(/*capacity*/ 8);
|
||||
(AppServerClient::Cloud(cloud).request_handle(), commands)
|
||||
}
|
||||
|
||||
pub(super) async fn reply(commands: &mut Requests, method: &str, params: Value, result: Value) {
|
||||
let command = commands.recv().await.unwrap();
|
||||
let CloudOperation::Request(request) = command.operation else {
|
||||
panic!("expected {method}");
|
||||
};
|
||||
let wire = serde_json::to_value(request).unwrap();
|
||||
assert_eq!(wire["method"], method);
|
||||
assert_eq!(wire["params"], params);
|
||||
command.response.send(Ok(Ok(result))).unwrap();
|
||||
}
|
||||
|
||||
pub(super) fn row(name: &str, tools: Vec<Value>) -> Value {
|
||||
let tools = tools
|
||||
.into_iter()
|
||||
.map(|tool| (tool["name"].as_str().unwrap().to_owned(), tool))
|
||||
.collect::<serde_json::Map<_, _>>();
|
||||
json!({"name":name,"runtimeStatus":"connected","httpOrigin":null,"tools":tools,"authStatus":"unsupported","resources":[],"resourceTemplates":[]})
|
||||
}
|
||||
Reference in New Issue
Block a user