feat: Streamable HTTP transport for ACP + goose-acp usage (#6741)

crates/goose-acp/Cargo.toml

@@ -7,6 +7,10 @@ license.workspace = true
 repository.workspace = true
 description.workspace = true
 
+[[bin]]
+name = "goose-acp-server"
+path = "src/bin/server.rs"
+
 [lints]
 workspace = true
 
@@ -24,6 +28,16 @@ regex = { workspace = true }
 fs-err = "3"
 url = { workspace = true }
 
+# HTTP server dependencies
+axum = "0.8"
+clap = { version = "4", features = ["derive"] }
+serde = { version = "1.0", features = ["derive"] }
+tower-http = { version = "0.6", features = ["cors"] }
+tracing-subscriber = { version = "0.3", features = ["env-filter", "json"] }
+async-stream = "0.3.6"
+bytes = "1.11.0"
+http-body-util = "0.1.3"
+
 [dev-dependencies]
 assert-json-diff = "2.0.2"
 async-trait = "0.1.89"

crates/goose-acp/src/bin/server.rs

@@ -0,0 +1,56 @@
+use anyhow::Result;
+use clap::Parser;
+use goose_acp::{
+    http::{self, HttpState},
+    server_factory::{AcpServer, AcpServerFactoryConfig},
+};
+use std::net::SocketAddr;
+use std::sync::Arc;
+use tracing::info;
+use tracing_subscriber::{layer::SubscriberExt, util::SubscriberInitExt, EnvFilter};
+
+#[derive(Parser)]
+#[command(name = "goose-acp-server")]
+#[command(about = "ACP server for goose over streamable HTTP")]
+struct Cli {
+    #[arg(long, default_value = "127.0.0.1")]
+    host: String,
+
+    #[arg(long, default_value = "3284")]
+    port: u16,
+
+    #[arg(long = "builtin", action = clap::ArgAction::Append)]
+    builtins: Vec<String>,
+}
+
+#[tokio::main]
+async fn main() -> Result<()> {
+    let filter = EnvFilter::try_from_default_env().unwrap_or_else(|_| EnvFilter::new("info"));
+    tracing_subscriber::registry()
+        .with(filter)
+        .with(tracing_subscriber::fmt::layer().with_target(true))
+        .init();
+
+    let cli = Cli::parse();
+
+    let builtins = if cli.builtins.is_empty() {
+        vec!["developer".to_string()]
+    } else {
+        cli.builtins
+    };
+
+    let config = AcpServerFactoryConfig {
+        builtins,
+        ..Default::default()
+    };
+
+    let server = Arc::new(AcpServer::new(config));
+    let state = Arc::new(HttpState::new(server));
+
+    let addr: SocketAddr = format!("{}:{}", cli.host, cli.port).parse()?;
+    info!("Starting goose-acp-server on {}", addr);
+
+    http::serve(state, addr).await?;
+
+    Ok(())
+}

crates/goose-acp/src/http.rs

@@ -0,0 +1,530 @@
+use anyhow::Result;
+use axum::{
+    body::Body,
+    extract::State,
+    http::{header, Method, Request, StatusCode},
+    response::{IntoResponse, Response, Sse},
+    routing::{delete, get, post},
+    Router,
+};
+use http_body_util::BodyExt;
+use serde_json::Value;
+use std::{
+    collections::HashMap,
+    convert::Infallible,
+    pin::Pin,
+    sync::Arc,
+    task::{Context, Poll},
+    time::Duration,
+};
+use tokio::sync::{mpsc, Mutex, RwLock};
+use tokio_util::compat::{TokioAsyncReadCompatExt, TokioAsyncWriteCompatExt};
+use tower_http::cors::{Any, CorsLayer};
+use tracing::{error, info};
+
+use crate::server_factory::AcpServer;
+
+// ACP header constants
+const HEADER_SESSION_ID: &str = "Acp-Session-Id";
+const EVENT_STREAM_MIME_TYPE: &str = "text/event-stream";
+const JSON_MIME_TYPE: &str = "application/json";
+
+struct HttpSession {
+    to_agent_tx: mpsc::Sender<String>,
+    from_agent_rx: Arc<Mutex<mpsc::Receiver<String>>>,
+    handle: tokio::task::JoinHandle<()>,
+}
+
+pub struct HttpState {
+    server: Arc<AcpServer>,
+    sessions: RwLock<HashMap<String, HttpSession>>,
+}
+
+impl HttpState {
+    pub fn new(server: Arc<AcpServer>) -> Self {
+        Self {
+            server,
+            sessions: RwLock::new(HashMap::new()),
+        }
+    }
+
+    async fn create_session(&self) -> Result<String, StatusCode> {
+        let (to_agent_tx, to_agent_rx) = mpsc::channel::<String>(256);
+        let (from_agent_tx, from_agent_rx) = mpsc::channel::<String>(256);
+
+        let agent = self.server.create_agent().await.map_err(|e| {
+            error!("Failed to create agent: {}", e);
+            StatusCode::INTERNAL_SERVER_ERROR
+        })?;
+
+        let session_id = agent.create_session().await.map_err(|e| {
+            error!("Failed to create ACP session: {}", e);
+            StatusCode::INTERNAL_SERVER_ERROR
+        })?;
+
+        let handle = tokio::spawn(async move {
+            let read_stream = ReceiverToAsyncRead::new(to_agent_rx);
+            let write_stream = SenderToAsyncWrite::new(from_agent_tx);
+
+            if let Err(e) =
+                crate::server::serve(agent, read_stream.compat(), write_stream.compat_write()).await
+            {
+                error!("ACP session error: {}", e);
+            }
+        });
+
+        self.sessions.write().await.insert(
+            session_id.clone(),
+            HttpSession {
+                to_agent_tx,
+                from_agent_rx: Arc::new(Mutex::new(from_agent_rx)),
+                handle,
+            },
+        );
+
+        info!(session_id = %session_id, "Session created");
+        Ok(session_id)
+    }
+
+    async fn has_session(&self, session_id: &str) -> bool {
+        self.sessions.read().await.contains_key(session_id)
+    }
+
+    async fn remove_session(&self, session_id: &str) {
+        if let Some(session) = self.sessions.write().await.remove(session_id) {
+            session.handle.abort();
+            info!(session_id = %session_id, "Session removed");
+        }
+    }
+
+    async fn send_message(&self, session_id: &str, message: String) -> Result<(), StatusCode> {
+        let sessions = self.sessions.read().await;
+        let session = sessions.get(session_id).ok_or(StatusCode::NOT_FOUND)?;
+        session
+            .to_agent_tx
+            .send(message)
+            .await
+            .map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)
+    }
+
+    async fn get_receiver(
+        &self,
+        session_id: &str,
+    ) -> Result<Arc<Mutex<mpsc::Receiver<String>>>, StatusCode> {
+        let sessions = self.sessions.read().await;
+        let session = sessions.get(session_id).ok_or(StatusCode::NOT_FOUND)?;
+        Ok(session.from_agent_rx.clone())
+    }
+}
+
+struct ReceiverToAsyncRead {
+    rx: mpsc::Receiver<String>,
+    buffer: Vec<u8>,
+    pos: usize,
+}
+
+impl ReceiverToAsyncRead {
+    fn new(rx: mpsc::Receiver<String>) -> Self {
+        Self {
+            rx,
+            buffer: Vec::new(),
+            pos: 0,
+        }
+    }
+}
+
+impl tokio::io::AsyncRead for ReceiverToAsyncRead {
+    fn poll_read(
+        mut self: Pin<&mut Self>,
+        cx: &mut Context<'_>,
+        buf: &mut tokio::io::ReadBuf<'_>,
+    ) -> Poll<std::io::Result<()>> {
+        if self.pos < self.buffer.len() {
+            let remaining = &self.buffer[self.pos..];
+            let to_copy = remaining.len().min(buf.remaining());
+            buf.put_slice(&remaining[..to_copy]);
+            self.pos += to_copy;
+            if self.pos >= self.buffer.len() {
+                self.buffer.clear();
+                self.pos = 0;
+            }
+            return Poll::Ready(Ok(()));
+        }
+
+        match Pin::new(&mut self.rx).poll_recv(cx) {
+            Poll::Ready(Some(msg)) => {
+                let bytes = format!("{}\n", msg).into_bytes();
+                let to_copy = bytes.len().min(buf.remaining());
+                buf.put_slice(&bytes[..to_copy]);
+                if to_copy < bytes.len() {
+                    self.buffer = bytes[to_copy..].to_vec();
+                    self.pos = 0;
+                }
+                Poll::Ready(Ok(()))
+            }
+            Poll::Ready(None) => Poll::Ready(Ok(())),
+            Poll::Pending => Poll::Pending,
+        }
+    }
+}
+
+struct SenderToAsyncWrite {
+    tx: mpsc::Sender<String>,
+    buffer: Vec<u8>,
+}
+
+impl SenderToAsyncWrite {
+    fn new(tx: mpsc::Sender<String>) -> Self {
+        Self {
+            tx,
+            buffer: Vec::new(),
+        }
+    }
+}
+
+impl tokio::io::AsyncWrite for SenderToAsyncWrite {
+    fn poll_write(
+        mut self: Pin<&mut Self>,
+        _cx: &mut Context<'_>,
+        buf: &[u8],
+    ) -> Poll<std::io::Result<usize>> {
+        self.buffer.extend_from_slice(buf);
+
+        while let Some(pos) = self.buffer.iter().position(|&b| b == b'\n') {
+            let line = String::from_utf8_lossy(&self.buffer[..pos]).to_string();
+            self.buffer.drain(..=pos);
+
+            if !line.is_empty() {
+                if let Err(e) = self.tx.try_send(line.clone()) {
+                    match e {
+                        mpsc::error::TrySendError::Full(_) => {
+                            let truncated: String = line.chars().take(100).collect();
+                            error!(
+                                "Channel full, dropping message (backpressure): {}",
+                                truncated
+                            );
+                        }
+                        mpsc::error::TrySendError::Closed(_) => {
+                            return Poll::Ready(Err(std::io::Error::new(
+                                std::io::ErrorKind::BrokenPipe,
+                                "Channel closed",
+                            )));
+                        }
+                    }
+                }
+            }
+        }
+
+        Poll::Ready(Ok(buf.len()))
+    }
+
+    fn poll_flush(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<std::io::Result<()>> {
+        Poll::Ready(Ok(()))
+    }
+
+    fn poll_shutdown(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<std::io::Result<()>> {
+        Poll::Ready(Ok(()))
+    }
+}
+
+fn accepts_mime_type(request: &Request<Body>, mime_type: &str) -> bool {
+    request
+        .headers()
+        .get(header::ACCEPT)
+        .and_then(|v| v.to_str().ok())
+        .is_some_and(|accept| accept.contains(mime_type))
+}
+
+fn accepts_json_and_sse(request: &Request<Body>) -> bool {
+    request
+        .headers()
+        .get(header::ACCEPT)
+        .and_then(|v| v.to_str().ok())
+        .is_some_and(|accept| {
+            accept.contains(JSON_MIME_TYPE) && accept.contains(EVENT_STREAM_MIME_TYPE)
+        })
+}
+
+fn content_type_is_json(request: &Request<Body>) -> bool {
+    request
+        .headers()
+        .get(header::CONTENT_TYPE)
+        .and_then(|v| v.to_str().ok())
+        .is_some_and(|ct| ct.starts_with(JSON_MIME_TYPE))
+}
+
+fn get_session_id(request: &Request<Body>) -> Option<String> {
+    request
+        .headers()
+        .get(HEADER_SESSION_ID)
+        .and_then(|v| v.to_str().ok())
+        .map(|s| s.to_string())
+}
+
+fn is_jsonrpc_request(value: &Value) -> bool {
+    value.get("method").is_some() && value.get("id").is_some()
+}
+
+fn is_jsonrpc_notification(value: &Value) -> bool {
+    value.get("method").is_some() && value.get("id").is_none()
+}
+
+fn is_jsonrpc_response(value: &Value) -> bool {
+    value.get("id").is_some() && (value.get("result").is_some() || value.get("error").is_some())
+}
+
+fn is_initialize_request(value: &Value) -> bool {
+    value.get("method").is_some_and(|m| m == "initialize") && value.get("id").is_some()
+}
+
+fn create_sse_stream(
+    receiver: Arc<Mutex<mpsc::Receiver<String>>>,
+    cleanup: Option<(Arc<HttpState>, String)>,
+) -> Sse<impl futures::Stream<Item = Result<axum::response::sse::Event, Infallible>>> {
+    let stream = async_stream::stream! {
+        let mut rx = receiver.lock().await;
+        while let Some(msg) = rx.recv().await {
+            yield Ok::<_, Infallible>(axum::response::sse::Event::default().data(msg));
+        }
+        if let Some((state, session_id)) = cleanup {
+            state.remove_session(&session_id).await;
+        }
+    };
+
+    Sse::new(stream).keep_alive(
+        axum::response::sse::KeepAlive::new()
+            .interval(Duration::from_secs(15))
+            .text(""),
+    )
+}
+
+async fn handle_initialize(state: Arc<HttpState>, json_message: &Value) -> Response {
+    let new_session_id = match state.create_session().await {
+        Ok(id) => id,
+        Err(status) => return status.into_response(),
+    };
+
+    let message_str = serde_json::to_string(json_message).unwrap();
+    if let Err(status) = state.send_message(&new_session_id, message_str).await {
+        state.remove_session(&new_session_id).await;
+        return status.into_response();
+    }
+
+    let receiver = match state.get_receiver(&new_session_id).await {
+        Ok(r) => r,
+        Err(status) => {
+            state.remove_session(&new_session_id).await;
+            return status.into_response();
+        }
+    };
+
+    let sse = create_sse_stream(receiver, Some((state.clone(), new_session_id.clone())));
+    let mut response = sse.into_response();
+    response
+        .headers_mut()
+        .insert(HEADER_SESSION_ID, new_session_id.parse().unwrap());
+    response
+}
+
+async fn handle_request(
+    state: Arc<HttpState>,
+    session_id: String,
+    json_message: &Value,
+) -> Response {
+    if !state.has_session(&session_id).await {
+        return (StatusCode::NOT_FOUND, "Session not found").into_response();
+    }
+
+    let message_str = serde_json::to_string(json_message).unwrap();
+    if let Err(status) = state.send_message(&session_id, message_str).await {
+        return status.into_response();
+    }
+
+    let receiver = match state.get_receiver(&session_id).await {
+        Ok(r) => r,
+        Err(status) => return status.into_response(),
+    };
+
+    create_sse_stream(receiver, None).into_response()
+}
+
+async fn handle_notification_or_response(
+    state: Arc<HttpState>,
+    session_id: String,
+    json_message: &Value,
+) -> Response {
+    if !state.has_session(&session_id).await {
+        return (StatusCode::NOT_FOUND, "Session not found").into_response();
+    }
+
+    let message_str = serde_json::to_string(json_message).unwrap();
+    if let Err(status) = state.send_message(&session_id, message_str).await {
+        return status.into_response();
+    }
+
+    StatusCode::ACCEPTED.into_response()
+}
+
+async fn handle_post(State(state): State<Arc<HttpState>>, request: Request<Body>) -> Response {
+    if !accepts_json_and_sse(&request) {
+        return (
+            StatusCode::NOT_ACCEPTABLE,
+            "Not Acceptable: Client must accept both application/json and text/event-stream",
+        )
+            .into_response();
+    }
+
+    if !content_type_is_json(&request) {
+        return (
+            StatusCode::UNSUPPORTED_MEDIA_TYPE,
+            "Unsupported Media Type: Content-Type must be application/json",
+        )
+            .into_response();
+    }
+
+    let session_id = get_session_id(&request);
+
+    let body_bytes = match request.into_body().collect().await {
+        Ok(collected) => collected.to_bytes(),
+        Err(e) => {
+            error!("Failed to read request body: {}", e);
+            return (StatusCode::BAD_REQUEST, "Failed to read request body").into_response();
+        }
+    };
+
+    let json_message: Value = match serde_json::from_slice(&body_bytes) {
+        Ok(v) => v,
+        Err(e) => {
+            error!("Failed to parse JSON: {}", e);
+            return (StatusCode::BAD_REQUEST, format!("Invalid JSON: {}", e)).into_response();
+        }
+    };
+
+    if json_message.is_array() {
+        return (
+            StatusCode::NOT_IMPLEMENTED,
+            "Batch requests are not supported",
+        )
+            .into_response();
+    }
+
+    if is_initialize_request(&json_message) {
+        handle_initialize(state, &json_message).await
+    } else if is_jsonrpc_request(&json_message) {
+        let Some(id) = session_id else {
+            return (
+                StatusCode::BAD_REQUEST,
+                "Bad Request: Acp-Session-Id header required",
+            )
+                .into_response();
+        };
+        handle_request(state, id, &json_message).await
+    } else if is_jsonrpc_notification(&json_message) || is_jsonrpc_response(&json_message) {
+        let Some(id) = session_id else {
+            return (
+                StatusCode::BAD_REQUEST,
+                "Bad Request: Acp-Session-Id header required",
+            )
+                .into_response();
+        };
+        handle_notification_or_response(state, id, &json_message).await
+    } else {
+        (StatusCode::BAD_REQUEST, "Invalid JSON-RPC message").into_response()
+    }
+}
+
+async fn handle_get(State(state): State<Arc<HttpState>>, request: Request<Body>) -> Response {
+    if !accepts_mime_type(&request, EVENT_STREAM_MIME_TYPE) {
+        return (
+            StatusCode::NOT_ACCEPTABLE,
+            "Not Acceptable: Client must accept text/event-stream",
+        )
+            .into_response();
+    }
+
+    let session_id = match get_session_id(&request) {
+        Some(id) => id,
+        None => {
+            return (
+                StatusCode::BAD_REQUEST,
+                "Bad Request: Acp-Session-Id header required",
+            )
+                .into_response();
+        }
+    };
+
+    if !state.has_session(&session_id).await {
+        return (StatusCode::NOT_FOUND, "Session not found").into_response();
+    }
+
+    let receiver = match state.get_receiver(&session_id).await {
+        Ok(r) => r,
+        Err(status) => return status.into_response(),
+    };
+
+    let stream = async_stream::stream! {
+        let mut rx = receiver.lock().await;
+        while let Some(msg) = rx.recv().await {
+            yield Ok::<_, Infallible>(axum::response::sse::Event::default().data(msg));
+        }
+    };
+
+    Sse::new(stream)
+        .keep_alive(
+            axum::response::sse::KeepAlive::new()
+                .interval(Duration::from_secs(15))
+                .text(""),
+        )
+        .into_response()
+}
+
+async fn handle_delete(State(state): State<Arc<HttpState>>, request: Request<Body>) -> Response {
+    let session_id = match get_session_id(&request) {
+        Some(id) => id,
+        None => {
+            return (
+                StatusCode::BAD_REQUEST,
+                "Bad Request: Acp-Session-Id header required",
+            )
+                .into_response();
+        }
+    };
+
+    if !state.has_session(&session_id).await {
+        return (StatusCode::NOT_FOUND, "Session not found").into_response();
+    }
+
+    state.remove_session(&session_id).await;
+    StatusCode::ACCEPTED.into_response()
+}
+
+async fn health() -> &'static str {
+    "ok"
+}
+
+pub fn create_router(state: Arc<HttpState>) -> Router {
+    let cors = CorsLayer::new()
+        .allow_origin(Any)
+        .allow_methods([Method::GET, Method::POST, Method::DELETE, Method::OPTIONS])
+        .allow_headers([
+            header::CONTENT_TYPE,
+            header::ACCEPT,
+            HEADER_SESSION_ID.parse().unwrap(),
+        ]);
+
+    Router::new()
+        .route("/health", get(health))
+        .route("/acp", post(handle_post))
+        .route("/acp", get(handle_get))
+        .route("/acp", delete(handle_delete))
+        .layer(cors)
+        .with_state(state)
+}
+
+pub async fn serve(state: Arc<HttpState>, addr: std::net::SocketAddr) -> Result<()> {
+    let router = create_router(state);
+    let listener = tokio::net::TcpListener::bind(addr).await?;
+    info!("ACP HTTP server listening on {}", addr);
+    axum::serve(listener, router).await?;
+    Ok(())
+}

crates/goose-acp/src/lib.rs

@@ -1 +1,5 @@
+#![recursion_limit = "256"]
+
+pub mod http;
 pub mod server;
+pub mod server_factory;

crates/goose-acp/src/server.rs

@@ -103,14 +103,11 @@ fn extract_tool_locations(
 ) -> Vec<ToolCallLocation> {
     let mut locations = Vec::new();
 
-    // Get the tool call details
     if let Ok(tool_call) = &tool_request.tool_call {
-        // Only process text_editor tool
         if tool_call.name != "developer__text_editor" {
             return locations;
         }
 
-        // Extract the path from arguments
         let path_str = tool_call
             .arguments
             .as_ref()
@@ -118,50 +115,42 @@ fn extract_tool_locations(
             .and_then(|p| p.as_str());
 
         if let Some(path_str) = path_str {
-            // Get the command type
             let command = tool_call
                 .arguments
                 .as_ref()
                 .and_then(|args| args.get("command"))
                 .and_then(|c| c.as_str());
 
-            // Extract line numbers from the response content
             if let Ok(result) = &tool_response.tool_result {
                 for content in &result.content {
                     if let RawContent::Text(text_content) = &content.raw {
                         let text = &text_content.text;
 
-                        // Parse line numbers based on command type and response format
                         match command {
                             Some("view") => {
-                                // For view command, look for "lines X-Y" pattern in header
                                 let line = extract_view_line_range(text)
                                     .map(|range| range.0 as u32)
                                     .or(Some(1));
                                 locations.push(create_tool_location(path_str, line));
                             }
                             Some("str_replace") | Some("insert") => {
-                                // For edits, extract the first line number from the snippet
                                 let line = extract_first_line_number(text)
                                     .map(|l| l as u32)
                                     .or(Some(1));
                                 locations.push(create_tool_location(path_str, line));
                             }
                             Some("write") => {
-                                // For write, just point to the beginning of the file
                                 locations.push(create_tool_location(path_str, Some(1)));
                             }
                             _ => {
-                                // For other commands or unknown, default to line 1
                                 locations.push(create_tool_location(path_str, Some(1)));
                             }
                         }
-                        break; // Only process first text content
+                        break;
                     }
                 }
             }
 
-            // If we didn't find any locations yet, add a default one
             if locations.is_empty() {
                 locations.push(create_tool_location(path_str, Some(1)));
             }
@@ -172,12 +161,11 @@ fn extract_tool_locations(
 }
 
 fn extract_view_line_range(text: &str) -> Option<(usize, usize)> {
-    // Pattern: "(lines X-Y)" or "(lines X-end)"
     let re = regex::Regex::new(r"\(lines (\d+)-(\d+|end)\)").ok()?;
     if let Some(caps) = re.captures(text) {
         let start = caps.get(1)?.as_str().parse::<usize>().ok()?;
         let end = if caps.get(2)?.as_str() == "end" {
-            start // Use start as a reasonable default
+            start
         } else {
             caps.get(2)?.as_str().parse::<usize>().ok()?
         };
@@ -187,7 +175,6 @@ fn extract_view_line_range(text: &str) -> Option<(usize, usize)> {
 }
 
 fn extract_first_line_number(text: &str) -> Option<usize> {
-    // Pattern: "123: " at the start of a line within a code block
     let re = regex::Regex::new(r"```[^\n]*\n(\d+):").ok()?;
     if let Some(caps) = re.captures(text) {
         return caps.get(1)?.as_str().parse::<usize>().ok();
@@ -212,34 +199,8 @@ fn read_resource_link(link: ResourceLink) -> Option<String> {
 }
 
 fn format_tool_name(tool_name: &str) -> String {
-    if let Some((extension, tool)) = tool_name.split_once("__") {
-        let formatted_extension = extension.replace('_', " ");
-        let formatted_tool = tool.replace('_', " ");
-
-        // Capitalize first letter of each word
-        let capitalize = |s: &str| {
-            s.split_whitespace()
-                .map(|word| {
-                    let mut chars = word.chars();
-                    match chars.next() {
-                        None => String::new(),
-                        Some(first) => first.to_uppercase().collect::<String>() + chars.as_str(),
-                    }
-                })
-                .collect::<Vec<_>>()
-                .join(" ")
-        };
-
-        format!(
-            "{}: {}",
-            capitalize(&formatted_extension),
-            capitalize(&formatted_tool)
-        )
-    } else {
-        // Fallback for tools without double underscore
-        let formatted = tool_name.replace('_', " ");
-        formatted
-            .split_whitespace()
+    let capitalize = |s: &str| {
+        s.split_whitespace()
             .map(|word| {
                 let mut chars = word.chars();
                 match chars.next() {
@@ -249,6 +210,19 @@ fn format_tool_name(tool_name: &str) -> String {
             })
             .collect::<Vec<_>>()
             .join(" ")
+    };
+
+    if let Some((extension, tool)) = tool_name.split_once("__") {
+        let formatted_extension = extension.replace('_', " ");
+        let formatted_tool = tool.replace('_', " ");
+        format!(
+            "{}: {}",
+            capitalize(&formatted_extension),
+            capitalize(&formatted_tool)
+        )
+    } else {
+        let formatted = tool_name.replace('_', " ");
+        capitalize(&formatted)
     }
 }
 
@@ -366,31 +340,60 @@ impl GooseAcpAgent {
         })
     }
 
+    pub async fn create_session(&self) -> Result<String> {
+        let manager = self.agent.config.session_manager.clone();
+        let goose_session = manager
+            .create_session(
+                std::env::current_dir().unwrap_or_default(),
+                "ACP Session".to_string(),
+                SessionType::User,
+            )
+            .await?;
+
+        self.agent
+            .update_provider(self.provider.clone(), &goose_session.id)
+            .await?;
+
+        let session = GooseAcpSession {
+            messages: Conversation::new_unvalidated(Vec::new()),
+            tool_requests: HashMap::new(),
+            cancel_token: None,
+        };
+
+        let mut sessions = self.sessions.lock().await;
+        sessions.insert(goose_session.id.clone(), session);
+
+        info!(
+            session_id = %goose_session.id,
+            session_type = "acp",
+            "Session created"
+        );
+
+        Ok(goose_session.id)
+    }
+
+    pub async fn has_session(&self, session_id: &str) -> bool {
+        self.sessions.lock().await.contains_key(session_id)
+    }
+
     fn convert_acp_prompt_to_message(&self, prompt: Vec<ContentBlock>) -> Message {
         let mut user_message = Message::user();
 
-        // Process all content blocks from the prompt
         for block in prompt {
             match block {
                 ContentBlock::Text(text) => {
                     user_message = user_message.with_text(&text.text);
                 }
                 ContentBlock::Image(image) => {
-                    // Goose supports images via base64 encoded data
-                    // The ACP ImageContent has data as a String directly
                     user_message = user_message.with_image(&image.data, &image.mime_type);
                 }
                 ContentBlock::Resource(resource) => {
-                    // Embed resource content as text with context
-                    match &resource.resource {
-                        EmbeddedResourceResource::TextResourceContents(text_resource) => {
-                            let header = format!("--- Resource: {} ---\n", text_resource.uri);
-                            let content = format!("{}{}\n---\n", header, text_resource.text);
-                            user_message = user_message.with_text(&content);
-                        }
-                        _ => {
-                            // Ignore non-text resources for now
-                        }
+                    if let EmbeddedResourceResource::TextResourceContents(text_resource) =
+                        &resource.resource
+                    {
+                        let header = format!("--- Resource: {} ---\n", text_resource.uri);
+                        let content = format!("{}{}\n---\n", header, text_resource.text);
+                        user_message = user_message.with_text(&content);
                     }
                 }
                 ContentBlock::ResourceLink(link) => {
@@ -398,8 +401,7 @@ impl GooseAcpAgent {
                         user_message = user_message.with_text(text)
                     }
                 }
-                ContentBlock::Audio(..) => (),
-                _ => (), // Handle any future ContentBlock variants
+                ContentBlock::Audio(..) | _ => (),
             }
         }
 
@@ -415,7 +417,6 @@ impl GooseAcpAgent {
     ) -> Result<(), sacp::Error> {
         match content_item {
             MessageContent::Text(text) => {
-                // Stream text to the client
                 cx.send_notification(SessionNotification::new(
                     session_id.clone(),
                     SessionUpdate::AgentMessageChunk(ContentChunk::new(ContentBlock::Text(
@@ -432,7 +433,6 @@ impl GooseAcpAgent {
                     .await?;
             }
             MessageContent::Thinking(thinking) => {
-                // Stream thinking/reasoning content as thought chunks
                 cx.send_notification(SessionNotification::new(
                     session_id.clone(),
                     SessionUpdate::AgentThoughtChunk(ContentChunk::new(ContentBlock::Text(
@@ -458,9 +458,7 @@ impl GooseAcpAgent {
                     )?;
                 }
             }
-            _ => {
-                // Ignore other content types for now
-            }
+            _ => {}
         }
         Ok(())
     }
@@ -472,18 +470,15 @@ impl GooseAcpAgent {
         session: &mut GooseAcpSession,
         cx: &JrConnectionCx<AgentToClient>,
     ) -> Result<(), sacp::Error> {
-        // Store the tool request for later use in response handling
         session
             .tool_requests
             .insert(tool_request.id.clone(), tool_request.clone());
 
-        // Extract tool name from the ToolCall if successful
         let tool_name = match &tool_request.tool_call {
             Ok(tool_call) => tool_call.name.to_string(),
             Err(_) => "error".to_string(),
         };
 
-        // Send tool call notification using the provider's tool call ID directly
         cx.send_notification(SessionNotification::new(
             session_id.clone(),
             SessionUpdate::ToolCall(
@@ -513,14 +508,12 @@ impl GooseAcpAgent {
 
         let content = build_tool_call_content(&tool_response.tool_result);
 
-        // Extract locations from the tool request and response
         let locations = if let Some(tool_request) = session.tool_requests.get(&tool_response.id) {
             extract_tool_locations(tool_request, tool_response)
         } else {
             Vec::new()
         };
 
-        // Send status update using provider's tool call ID directly
         let mut fields = ToolCallUpdateFields::new().status(status).content(content);
         if !locations.is_empty() {
             fields = fields.locations(locations);
@@ -551,7 +544,6 @@ impl GooseAcpAgent {
 
         let formatted_name = format_tool_name(&tool_name);
 
-        // Use the request_id (provider's tool call ID) directly
         let mut fields = ToolCallUpdateFields::new()
             .title(formatted_name)
             .kind(ToolKind::default())
@@ -625,11 +617,10 @@ fn outcome_to_confirmation(outcome: &RequestPermissionOutcome) -> PermissionConf
                 Ok(PermissionOptionKind::AllowOnce) => Permission::AllowOnce,
                 Ok(PermissionOptionKind::RejectOnce) => Permission::DenyOnce,
                 Ok(PermissionOptionKind::RejectAlways) => Permission::AlwaysDeny,
-                Ok(_) => Permission::Cancel, // Handle any future permission kinds
-                Err(_) => Permission::Cancel,
+                _ => Permission::Cancel,
             }
         }
-        _ => Permission::Cancel, // Handle any future variants
+        _ => Permission::Cancel,
     };
     PermissionConfirmation {
         principal_type: PrincipalType::Tool,
@@ -674,14 +665,7 @@ fn build_tool_call_content(tool_result: &ToolResult<CallToolResult>) -> Vec<Tool
                         ContentBlock::Resource(EmbeddedResource::new(resource)),
                     )))
                 }
-                RawContent::Audio(_) => {
-                    // Audio content is not supported in ACP ContentBlock, skip it
-                    None
-                }
-                RawContent::ResourceLink(_) => {
-                    // ResourceLink content is not supported in ACP ContentBlock, skip it
-                    None
-                }
+                RawContent::Audio(_) | RawContent::ResourceLink(_) => None,
             })
             .collect(),
         Err(_) => Vec::new(),
@@ -695,7 +679,6 @@ impl GooseAcpAgent {
     ) -> Result<InitializeResponse, sacp::Error> {
         debug!(?args, "initialize request");
 
-        // Advertise Goose's capabilities
         let capabilities = AgentCapabilities::new()
             .load_session(true)
             .prompt_capabilities(
@@ -718,7 +701,7 @@ impl GooseAcpAgent {
         let goose_session = manager
             .create_session(
                 args.cwd.clone(),
-                "ACP Session".to_string(), // just an initial name - may be replaced by maybe_update_name
+                "ACP Session".to_string(),
                 SessionType::User,
             )
             .await
@@ -727,7 +710,6 @@ impl GooseAcpAgent {
             })?;
         self.update_session_with_provider(&goose_session).await?;
 
-        // Add MCP servers specified in the session request
         for mcp_server in args.mcp_servers {
             let config = match mcp_server_to_extension_config(mcp_server) {
                 Ok(c) => c,
@@ -810,9 +792,7 @@ impl GooseAcpAgent {
             cancel_token: None,
         };
 
-        // Replay conversation history to client
         for message in conversation.messages() {
-            // Only replay user-visible messages
             if !message.metadata.user_visible {
                 continue;
             }
@@ -853,9 +833,7 @@ impl GooseAcpAgent {
                             )),
                         ))?;
                     }
-                    _ => {
-                        // Ignore other content types
-                    }
+                    _ => {}
                 }
             }
         }
@@ -1015,8 +993,6 @@ impl JrMessageHandler for GooseAcpHandler {
             .await
             .if_request(
                 |req: PromptRequest, req_cx: JrRequestCx<PromptResponse>| async {
-                    // Spawn the prompt processing in a task so we don't block the event loop.
-                    // This allows permission responses to be processed while the agent is working.
                     let agent = self.agent.clone();
                     let cx_clone = cx.clone();
                     cx.spawn(async move {
@@ -1042,7 +1018,6 @@ impl JrMessageHandler for GooseAcpHandler {
     }
 }
 
-/// Serve ACP on a given transport (for in-process testing)
 pub async fn serve<R, W>(agent: Arc<GooseAcpAgent>, read: R, write: W) -> Result<()>
 where
     R: futures::AsyncRead + Unpin + Send + 'static,
@@ -1188,14 +1163,6 @@ print(\"hello, world\")
         assert_eq!(format_tool_name("single"), "Single");
     }
 
-    #[test]
-    fn test_format_tool_name_edge_cases() {
-        assert_eq!(format_tool_name(""), "");
-        assert_eq!(format_tool_name("__"), ": ");
-        assert_eq!(format_tool_name("extension__"), "Extension: ");
-        assert_eq!(format_tool_name("__tool"), ": Tool");
-    }
-
     #[test_case(
         RequestPermissionOutcome::Selected(SelectedPermissionOutcome::new(PermissionOptionId::from("allow_once".to_string()))),
         PermissionConfirmation { principal_type: PrincipalType::Tool, permission: Permission::AllowOnce };

crates/goose-acp/src/server_factory.rs

@@ -0,0 +1,76 @@
+use anyhow::Result;
+use goose::config::paths::Paths;
+use goose::config::Config;
+use goose::model::ModelConfig;
+use goose::providers::create;
+use std::sync::Arc;
+use tracing::info;
+
+use crate::server::{AcpServerConfig, GooseAcpAgent};
+
+pub struct AcpServerFactoryConfig {
+    pub builtins: Vec<String>,
+    pub data_dir: std::path::PathBuf,
+    pub config_dir: std::path::PathBuf,
+}
+
+impl Default for AcpServerFactoryConfig {
+    fn default() -> Self {
+        Self {
+            builtins: vec!["developer".to_string()],
+            data_dir: Paths::data_dir(),
+            config_dir: Paths::config_dir(),
+        }
+    }
+}
+
+pub struct AcpServer {
+    config: AcpServerFactoryConfig,
+}
+
+impl AcpServer {
+    pub fn new(config: AcpServerFactoryConfig) -> Self {
+        Self { config }
+    }
+
+    pub async fn create_agent(&self) -> Result<Arc<GooseAcpAgent>> {
+        let global_config = Config::global();
+
+        let provider_name: String = global_config
+            .get_goose_provider()
+            .map_err(|e| anyhow::anyhow!("No provider configured: {}", e))?;
+
+        let model_name: String = global_config
+            .get_goose_model()
+            .map_err(|e| anyhow::anyhow!("No model configured: {}", e))?;
+
+        let model_config = ModelConfig {
+            request_params: None,
+            model_name: model_name.clone(),
+            context_limit: None,
+            temperature: None,
+            max_tokens: None,
+            toolshim: false,
+            toolshim_model: None,
+            fast_model: None,
+        };
+
+        let provider = create(&provider_name, model_config).await?;
+        let goose_mode = global_config
+            .get_goose_mode()
+            .unwrap_or(goose::config::GooseMode::Auto);
+
+        let acp_config = AcpServerConfig {
+            provider,
+            builtins: self.config.builtins.clone(),
+            data_dir: self.config.data_dir.clone(),
+            config_dir: self.config.config_dir.clone(),
+            goose_mode,
+        };
+
+        let agent = GooseAcpAgent::with_config(acp_config).await?;
+        info!("Created new ACP agent");
+
+        Ok(Arc::new(agent))
+    }
+}