ACP over MCP server impl (#35196)

Release Notes:

- N/A

---------

Co-authored-by: Ben Brandt <benjamin.j.brandt@gmail.com>

crates/agent_servers/Cargo.toml

@@ -41,6 +41,7 @@ ui.workspace = true
 util.workspace = true
 uuid.workspace = true
 watch.workspace = true
+indoc.workspace = true
 which.workspace = true
 workspace-hack.workspace = true
 

crates/agent_servers/src/agent_servers.rs

@@ -1,11 +1,14 @@
 mod claude;
+mod codex;
 mod gemini;
+mod mcp_server;
 mod settings;
 
 #[cfg(test)]
 mod e2e_tests;
 
 pub use claude::*;
+pub use codex::*;
 pub use gemini::*;
 pub use settings::*;
 

crates/agent_servers/src/claude.rs

@@ -44,7 +44,7 @@ impl AgentServer for ClaudeCode {
     }
 
     fn empty_state_message(&self) -> &'static str {
-        ""
+        "How can I help you today?"
     }
 
     fn logo(&self) -> ui::IconName {
@@ -190,7 +190,7 @@ impl AgentConnection for ClaudeAgentConnection {
         Task::ready(Err(anyhow!("Authentication not supported")))
     }
 
-    fn prompt(&self, params: acp::PromptToolArguments, cx: &mut App) -> Task<Result<()>> {
+    fn prompt(&self, params: acp::PromptArguments, cx: &mut App) -> Task<Result<()>> {
         let sessions = self.sessions.borrow();
         let Some(session) = sessions.get(&params.session_id) else {
             return Task::ready(Err(anyhow!(
@@ -350,7 +350,7 @@ impl ClaudeAgentSession {
                         ContentChunk::Text { text } | ContentChunk::UntaggedText(text) => {
                             thread
                                 .update(cx, |thread, cx| {
-                                    thread.push_assistant_chunk(text.into(), false, cx)
+                                    thread.push_assistant_content_block(text.into(), false, cx)
                                 })
                                 .log_err();
                         }
@@ -387,9 +387,15 @@ impl ClaudeAgentSession {
                             thread
                                 .update(cx, |thread, cx| {
                                     thread.update_tool_call(
-                                        acp::ToolCallId(tool_use_id.into()),
-                                        acp::ToolCallStatus::Completed,
-                                        (!content.is_empty()).then(|| vec![content.into()]),
+                                        acp::ToolCallUpdate {
+                                            id: acp::ToolCallId(tool_use_id.into()),
+                                            fields: acp::ToolCallUpdateFields {
+                                                status: Some(acp::ToolCallStatus::Completed),
+                                                content: (!content.is_empty())
+                                                    .then(|| vec![content.into()]),
+                                                ..Default::default()
+                                            },
+                                        },
                                         cx,
                                     )
                                 })
@@ -402,7 +408,7 @@ impl ClaudeAgentSession {
                         | ContentChunk::WebSearchToolResult => {
                             thread
                                 .update(cx, |thread, cx| {
-                                    thread.push_assistant_chunk(
+                                    thread.push_assistant_content_block(
                                         format!("Unsupported content: {:?}", chunk).into(),
                                         false,
                                         cx,

crates/agent_servers/src/claude/tools.rs

@@ -311,6 +311,7 @@ impl ClaudeTool {
             label: self.label(),
             content: self.content(),
             locations: self.locations(),
+            raw_input: None,
         }
     }
 }

crates/agent_servers/src/codex.rs

@@ -0,0 +1,317 @@
+use agent_client_protocol as acp;
+use anyhow::anyhow;
+use collections::HashMap;
+use context_server::listener::McpServerTool;
+use context_server::types::requests;
+use context_server::{ContextServer, ContextServerCommand, ContextServerId};
+use futures::channel::{mpsc, oneshot};
+use project::Project;
+use settings::SettingsStore;
+use smol::stream::StreamExt as _;
+use std::cell::RefCell;
+use std::rc::Rc;
+use std::{path::Path, sync::Arc};
+use util::ResultExt;
+
+use anyhow::{Context, Result};
+use gpui::{App, AppContext as _, AsyncApp, Entity, Task, WeakEntity};
+
+use crate::mcp_server::ZedMcpServer;
+use crate::{AgentServer, AgentServerCommand, AllAgentServersSettings, mcp_server};
+use acp_thread::{AcpThread, AgentConnection};
+
+#[derive(Clone)]
+pub struct Codex;
+
+impl AgentServer for Codex {
+    fn name(&self) -> &'static str {
+        "Codex"
+    }
+
+    fn empty_state_headline(&self) -> &'static str {
+        "Welcome to Codex"
+    }
+
+    fn empty_state_message(&self) -> &'static str {
+        "What can I help with?"
+    }
+
+    fn logo(&self) -> ui::IconName {
+        ui::IconName::AiOpenAi
+    }
+
+    fn connect(
+        &self,
+        _root_dir: &Path,
+        project: &Entity<Project>,
+        cx: &mut App,
+    ) -> Task<Result<Rc<dyn AgentConnection>>> {
+        let project = project.clone();
+        cx.spawn(async move |cx| {
+            let settings = cx.read_global(|settings: &SettingsStore, _| {
+                settings.get::<AllAgentServersSettings>(None).codex.clone()
+            })?;
+
+            let Some(command) =
+                AgentServerCommand::resolve("codex", &["mcp"], settings, &project, cx).await
+            else {
+                anyhow::bail!("Failed to find codex binary");
+            };
+
+            let client: Arc<ContextServer> = ContextServer::stdio(
+                ContextServerId("codex-mcp-server".into()),
+                ContextServerCommand {
+                    path: command.path,
+                    args: command.args,
+                    env: command.env,
+                },
+            )
+            .into();
+            ContextServer::start(client.clone(), cx).await?;
+
+            let (notification_tx, mut notification_rx) = mpsc::unbounded();
+            client
+                .client()
+                .context("Failed to subscribe")?
+                .on_notification(acp::SESSION_UPDATE_METHOD_NAME, {
+                    move |notification, _cx| {
+                        let notification_tx = notification_tx.clone();
+                        log::trace!(
+                            "ACP Notification: {}",
+                            serde_json::to_string_pretty(&notification).unwrap()
+                        );
+
+                        if let Some(notification) =
+                            serde_json::from_value::<acp::SessionNotification>(notification)
+                                .log_err()
+                        {
+                            notification_tx.unbounded_send(notification).ok();
+                        }
+                    }
+                });
+
+            let sessions = Rc::new(RefCell::new(HashMap::default()));
+
+            let notification_handler_task = cx.spawn({
+                let sessions = sessions.clone();
+                async move |cx| {
+                    while let Some(notification) = notification_rx.next().await {
+                        CodexConnection::handle_session_notification(
+                            notification,
+                            sessions.clone(),
+                            cx,
+                        )
+                    }
+                }
+            });
+
+            let connection = CodexConnection {
+                client,
+                sessions,
+                _notification_handler_task: notification_handler_task,
+            };
+            Ok(Rc::new(connection) as _)
+        })
+    }
+}
+
+struct CodexConnection {
+    client: Arc<context_server::ContextServer>,
+    sessions: Rc<RefCell<HashMap<acp::SessionId, CodexSession>>>,
+    _notification_handler_task: Task<()>,
+}
+
+struct CodexSession {
+    thread: WeakEntity<AcpThread>,
+    cancel_tx: Option<oneshot::Sender<()>>,
+    _mcp_server: ZedMcpServer,
+}
+
+impl AgentConnection for CodexConnection {
+    fn name(&self) -> &'static str {
+        "Codex"
+    }
+
+    fn new_thread(
+        self: Rc<Self>,
+        project: Entity<Project>,
+        cwd: &Path,
+        cx: &mut AsyncApp,
+    ) -> Task<Result<Entity<AcpThread>>> {
+        let client = self.client.client();
+        let sessions = self.sessions.clone();
+        let cwd = cwd.to_path_buf();
+        cx.spawn(async move |cx| {
+            let client = client.context("MCP server is not initialized yet")?;
+            let (mut thread_tx, thread_rx) = watch::channel(WeakEntity::new_invalid());
+
+            let mcp_server = ZedMcpServer::new(thread_rx, cx).await?;
+
+            let response = client
+                .request::<requests::CallTool>(context_server::types::CallToolParams {
+                    name: acp::NEW_SESSION_TOOL_NAME.into(),
+                    arguments: Some(serde_json::to_value(acp::NewSessionArguments {
+                        mcp_servers: [(
+                            mcp_server::SERVER_NAME.to_string(),
+                            mcp_server.server_config()?,
+                        )]
+                        .into(),
+                        client_tools: acp::ClientTools {
+                            request_permission: Some(acp::McpToolId {
+                                mcp_server: mcp_server::SERVER_NAME.into(),
+                                tool_name: mcp_server::RequestPermissionTool::NAME.into(),
+                            }),
+                            read_text_file: Some(acp::McpToolId {
+                                mcp_server: mcp_server::SERVER_NAME.into(),
+                                tool_name: mcp_server::ReadTextFileTool::NAME.into(),
+                            }),
+                            write_text_file: Some(acp::McpToolId {
+                                mcp_server: mcp_server::SERVER_NAME.into(),
+                                tool_name: mcp_server::WriteTextFileTool::NAME.into(),
+                            }),
+                        },
+                        cwd,
+                    })?),
+                    meta: None,
+                })
+                .await?;
+
+            if response.is_error.unwrap_or_default() {
+                return Err(anyhow!(response.text_contents()));
+            }
+
+            let result = serde_json::from_value::<acp::NewSessionOutput>(
+                response.structured_content.context("Empty response")?,
+            )?;
+
+            let thread =
+                cx.new(|cx| AcpThread::new(self.clone(), project, result.session_id.clone(), cx))?;
+
+            thread_tx.send(thread.downgrade())?;
+
+            let session = CodexSession {
+                thread: thread.downgrade(),
+                cancel_tx: None,
+                _mcp_server: mcp_server,
+            };
+            sessions.borrow_mut().insert(result.session_id, session);
+
+            Ok(thread)
+        })
+    }
+
+    fn authenticate(&self, _cx: &mut App) -> Task<Result<()>> {
+        Task::ready(Err(anyhow!("Authentication not supported")))
+    }
+
+    fn prompt(
+        &self,
+        params: agent_client_protocol::PromptArguments,
+        cx: &mut App,
+    ) -> Task<Result<()>> {
+        let client = self.client.client();
+        let sessions = self.sessions.clone();
+
+        cx.foreground_executor().spawn(async move {
+            let client = client.context("MCP server is not initialized yet")?;
+
+            let (new_cancel_tx, cancel_rx) = oneshot::channel();
+            {
+                let mut sessions = sessions.borrow_mut();
+                let session = sessions
+                    .get_mut(&params.session_id)
+                    .context("Session not found")?;
+                session.cancel_tx.replace(new_cancel_tx);
+            }
+
+            let result = client
+                .request_with::<requests::CallTool>(
+                    context_server::types::CallToolParams {
+                        name: acp::PROMPT_TOOL_NAME.into(),
+                        arguments: Some(serde_json::to_value(params)?),
+                        meta: None,
+                    },
+                    Some(cancel_rx),
+                    None,
+                )
+                .await;
+
+            if let Err(err) = &result
+                && err.is::<context_server::client::RequestCanceled>()
+            {
+                return Ok(());
+            }
+
+            let response = result?;
+
+            if response.is_error.unwrap_or_default() {
+                return Err(anyhow!(response.text_contents()));
+            }
+
+            Ok(())
+        })
+    }
+
+    fn cancel(&self, session_id: &agent_client_protocol::SessionId, _cx: &mut App) {
+        let mut sessions = self.sessions.borrow_mut();
+
+        if let Some(cancel_tx) = sessions
+            .get_mut(session_id)
+            .and_then(|session| session.cancel_tx.take())
+        {
+            cancel_tx.send(()).ok();
+        }
+    }
+}
+
+impl CodexConnection {
+    pub fn handle_session_notification(
+        notification: acp::SessionNotification,
+        threads: Rc<RefCell<HashMap<acp::SessionId, CodexSession>>>,
+        cx: &mut AsyncApp,
+    ) {
+        let threads = threads.borrow();
+        let Some(thread) = threads
+            .get(&notification.session_id)
+            .and_then(|session| session.thread.upgrade())
+        else {
+            log::error!(
+                "Thread not found for session ID: {}",
+                notification.session_id
+            );
+            return;
+        };
+
+        thread
+            .update(cx, |thread, cx| {
+                thread.handle_session_update(notification.update, cx)
+            })
+            .log_err();
+    }
+}
+
+impl Drop for CodexConnection {
+    fn drop(&mut self) {
+        self.client.stop().log_err();
+    }
+}
+
+#[cfg(test)]
+pub(crate) mod tests {
+    use super::*;
+    use crate::AgentServerCommand;
+    use std::path::Path;
+
+    crate::common_e2e_tests!(Codex);
+
+    pub fn local_command() -> AgentServerCommand {
+        let cli_path = Path::new(env!("CARGO_MANIFEST_DIR"))
+            .join("../../../codex/codex-rs/target/debug/codex");
+
+        AgentServerCommand {
+            path: cli_path,
+            args: vec!["mcp".into()],
+            env: None,
+        }
+    }
+}

crates/agent_servers/src/e2e_tests.rs

@@ -351,6 +351,9 @@ pub async fn init_test(cx: &mut TestAppContext) -> Arc<FakeFs> {
                 gemini: Some(AgentServerSettings {
                     command: crate::gemini::tests::local_command(),
                 }),
+                codex: Some(AgentServerSettings {
+                    command: crate::codex::tests::local_command(),
+                }),
             },
             cx,
         );

crates/agent_servers/src/mcp_server.rs

@@ -0,0 +1,201 @@
+use acp_thread::AcpThread;
+use agent_client_protocol as acp;
+use anyhow::{Context, Result};
+use context_server::listener::{McpServerTool, ToolResponse};
+use context_server::types::{
+    Implementation, InitializeParams, InitializeResponse, ProtocolVersion, ServerCapabilities,
+    ToolsCapabilities, requests,
+};
+use futures::channel::oneshot;
+use gpui::{App, AsyncApp, Task, WeakEntity};
+use indoc::indoc;
+
+pub struct ZedMcpServer {
+    server: context_server::listener::McpServer,
+}
+
+pub const SERVER_NAME: &str = "zed";
+
+impl ZedMcpServer {
+    pub async fn new(
+        thread_rx: watch::Receiver<WeakEntity<AcpThread>>,
+        cx: &AsyncApp,
+    ) -> Result<Self> {
+        let mut mcp_server = context_server::listener::McpServer::new(cx).await?;
+        mcp_server.handle_request::<requests::Initialize>(Self::handle_initialize);
+
+        mcp_server.add_tool(RequestPermissionTool {
+            thread_rx: thread_rx.clone(),
+        });
+        mcp_server.add_tool(ReadTextFileTool {
+            thread_rx: thread_rx.clone(),
+        });
+        mcp_server.add_tool(WriteTextFileTool {
+            thread_rx: thread_rx.clone(),
+        });
+
+        Ok(Self { server: mcp_server })
+    }
+
+    pub fn server_config(&self) -> Result<acp::McpServerConfig> {
+        let zed_path = std::env::current_exe()
+            .context("finding current executable path for use in mcp_server")?;
+
+        Ok(acp::McpServerConfig {
+            command: zed_path,
+            args: vec![
+                "--nc".into(),
+                self.server.socket_path().display().to_string(),
+            ],
+            env: None,
+        })
+    }
+
+    fn handle_initialize(_: InitializeParams, cx: &App) -> Task<Result<InitializeResponse>> {
+        cx.foreground_executor().spawn(async move {
+            Ok(InitializeResponse {
+                protocol_version: ProtocolVersion("2025-06-18".into()),
+                capabilities: ServerCapabilities {
+                    experimental: None,
+                    logging: None,
+                    completions: None,
+                    prompts: None,
+                    resources: None,
+                    tools: Some(ToolsCapabilities {
+                        list_changed: Some(false),
+                    }),
+                },
+                server_info: Implementation {
+                    name: SERVER_NAME.into(),
+                    version: "0.1.0".into(),
+                },
+                meta: None,
+            })
+        })
+    }
+}
+
+// Tools
+
+#[derive(Clone)]
+pub struct RequestPermissionTool {
+    thread_rx: watch::Receiver<WeakEntity<AcpThread>>,
+}
+
+impl McpServerTool for RequestPermissionTool {
+    type Input = acp::RequestPermissionArguments;
+    type Output = acp::RequestPermissionOutput;
+
+    const NAME: &'static str = "Confirmation";
+
+    fn description(&self) -> &'static str {
+        indoc! {"
+            Request permission for tool calls.
+
+            This tool is meant to be called programmatically by the agent loop, not the LLM.
+        "}
+    }
+
+    async fn run(
+        &self,
+        input: Self::Input,
+        cx: &mut AsyncApp,
+    ) -> Result<ToolResponse<Self::Output>> {
+        let mut thread_rx = self.thread_rx.clone();
+        let Some(thread) = thread_rx.recv().await?.upgrade() else {
+            anyhow::bail!("Thread closed");
+        };
+
+        let result = thread
+            .update(cx, |thread, cx| {
+                thread.request_tool_call_permission(input.tool_call, input.options, cx)
+            })?
+            .await;
+
+        let outcome = match result {
+            Ok(option_id) => acp::RequestPermissionOutcome::Selected { option_id },
+            Err(oneshot::Canceled) => acp::RequestPermissionOutcome::Canceled,
+        };
+
+        Ok(ToolResponse {
+            content: vec![],
+            structured_content: acp::RequestPermissionOutput { outcome },
+        })
+    }
+}
+
+#[derive(Clone)]
+pub struct ReadTextFileTool {
+    thread_rx: watch::Receiver<WeakEntity<AcpThread>>,
+}
+
+impl McpServerTool for ReadTextFileTool {
+    type Input = acp::ReadTextFileArguments;
+    type Output = acp::ReadTextFileOutput;
+
+    const NAME: &'static str = "Read";
+
+    fn description(&self) -> &'static str {
+        "Reads the content of the given file in the project including unsaved changes."
+    }
+
+    async fn run(
+        &self,
+        input: Self::Input,
+        cx: &mut AsyncApp,
+    ) -> Result<ToolResponse<Self::Output>> {
+        let mut thread_rx = self.thread_rx.clone();
+        let Some(thread) = thread_rx.recv().await?.upgrade() else {
+            anyhow::bail!("Thread closed");
+        };
+
+        let content = thread
+            .update(cx, |thread, cx| {
+                thread.read_text_file(input.path, input.line, input.limit, false, cx)
+            })?
+            .await?;
+
+        Ok(ToolResponse {
+            content: vec![],
+            structured_content: acp::ReadTextFileOutput { content },
+        })
+    }
+}
+
+#[derive(Clone)]
+pub struct WriteTextFileTool {
+    thread_rx: watch::Receiver<WeakEntity<AcpThread>>,
+}
+
+impl McpServerTool for WriteTextFileTool {
+    type Input = acp::WriteTextFileArguments;
+    type Output = ();
+
+    const NAME: &'static str = "Write";
+
+    fn description(&self) -> &'static str {
+        "Write to a file replacing its contents"
+    }
+
+    async fn run(
+        &self,
+        input: Self::Input,
+        cx: &mut AsyncApp,
+    ) -> Result<ToolResponse<Self::Output>> {
+        let mut thread_rx = self.thread_rx.clone();
+        let Some(thread) = thread_rx.recv().await?.upgrade() else {
+            anyhow::bail!("Thread closed");
+        };
+
+        thread
+            .update(cx, |thread, cx| {
+                thread.write_text_file(input.path, input.content, cx)
+            })?
+            .await?;
+
+        Ok(ToolResponse {
+            content: vec![],
+            structured_content: (),
+        })
+    }
+}

crates/agent_servers/src/settings.rs

@@ -13,6 +13,7 @@ pub fn init(cx: &mut App) {
 pub struct AllAgentServersSettings {
     pub gemini: Option<AgentServerSettings>,
     pub claude: Option<AgentServerSettings>,
+    pub codex: Option<AgentServerSettings>,
 }
 
 #[derive(Deserialize, Serialize, Clone, JsonSchema, Debug)]
@@ -29,13 +30,21 @@ impl settings::Settings for AllAgentServersSettings {
     fn load(sources: SettingsSources<Self::FileContent>, _: &mut App) -> Result<Self> {
         let mut settings = AllAgentServersSettings::default();
 
-        for AllAgentServersSettings { gemini, claude } in sources.defaults_and_customizations() {
+        for AllAgentServersSettings {
+            gemini,
+            claude,
+            codex,
+        } in sources.defaults_and_customizations()
+        {
             if gemini.is_some() {
                 settings.gemini = gemini.clone();
             }
             if claude.is_some() {
                 settings.claude = claude.clone();
             }
+            if codex.is_some() {
+                settings.codex = codex.clone();
+            }
         }
 
         Ok(settings)