mirror of
https://github.com/block/goose.git
synced 2026-07-17 12:56:20 +02:00
feat: Agent Client Protocol implementation of goose (#4511)
This commit is contained in:
Michael Neale
Generated
+66
@@ -28,6 +28,22 @@ dependencies = [
|
||||
"cpufeatures",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "agent-client-protocol"
|
||||
version = "0.1.1"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "6b91e5ec3ce05e8effb2a7a3b7b1a587daa6699b9f98bbde6a35e44b8c6c773a"
|
||||
dependencies = [
|
||||
"anyhow",
|
||||
"async-broadcast",
|
||||
"futures",
|
||||
"log",
|
||||
"parking_lot",
|
||||
"schemars",
|
||||
"serde",
|
||||
"serde_json",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "ahash"
|
||||
version = "0.8.11"
|
||||
@@ -427,6 +443,18 @@ dependencies = [
|
||||
"serde_json",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "async-broadcast"
|
||||
version = "0.7.2"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "435a87a52755b8f27fcf321ac4f04b2802e337c8c4872923137471ec39c37532"
|
||||
dependencies = [
|
||||
"event-listener",
|
||||
"event-listener-strategy",
|
||||
"futures-core",
|
||||
"pin-project-lite",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "async-compression"
|
||||
version = "0.4.20"
|
||||
@@ -1525,6 +1553,15 @@ dependencies = [
|
||||
"memchr",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "concurrent-queue"
|
||||
version = "2.5.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "4ca0197aee26d1ae37445ee532fefce43251d24cc7c166799f4d46817f1d3973"
|
||||
dependencies = [
|
||||
"crossbeam-utils",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "config"
|
||||
version = "0.14.1"
|
||||
@@ -2146,6 +2183,27 @@ dependencies = [
|
||||
"windows-sys 0.48.0",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "event-listener"
|
||||
version = "5.4.1"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "e13b66accf52311f30a0db42147dadea9850cb48cd070028831ae5f5d4b856ab"
|
||||
dependencies = [
|
||||
"concurrent-queue",
|
||||
"parking",
|
||||
"pin-project-lite",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "event-listener-strategy"
|
||||
version = "0.5.4"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "8be9f3dfaaffdae2972880079a491a1a8bb7cbed0b8dd7a347f668b4150a3b93"
|
||||
dependencies = [
|
||||
"event-listener",
|
||||
"pin-project-lite",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "eventsource-client"
|
||||
version = "0.12.2"
|
||||
@@ -2644,6 +2702,7 @@ dependencies = [
|
||||
name = "goose-cli"
|
||||
version = "1.7.0"
|
||||
dependencies = [
|
||||
"agent-client-protocol",
|
||||
"anstream",
|
||||
"anyhow",
|
||||
"async-trait",
|
||||
@@ -4573,6 +4632,12 @@ version = "0.1.1"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "b15813163c1d831bf4a13c3610c05c0d03b39feb07f7e09fa234dac9b15aaf39"
|
||||
|
||||
[[package]]
|
||||
name = "parking"
|
||||
version = "2.2.1"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "f38d5652c16fde515bb1ecef450ab0f6a219d619a7274976324d5e377f7dceba"
|
||||
|
||||
[[package]]
|
||||
name = "parking_lot"
|
||||
version = "0.12.3"
|
||||
@@ -6598,6 +6663,7 @@ checksum = "66a539a9ad6d5d281510d5bd368c973d636c02dbf8a67300bfb6b950696ad7df"
|
||||
dependencies = [
|
||||
"bytes",
|
||||
"futures-core",
|
||||
"futures-io",
|
||||
"futures-sink",
|
||||
"pin-project-lite",
|
||||
"tokio",
|
||||
|
||||
@@ -0,0 +1,9 @@
|
||||
# Goose Jokes 🦢
|
||||
|
||||
## Why did the Goose become a developer?
|
||||
|
||||
Because it wanted to help debug all those "fowl" errors in the code!
|
||||
|
||||
---
|
||||
|
||||
*Feel free to add more goose-related programming humor below!*
|
||||
@@ -30,6 +30,12 @@ Designed for maximum flexibility, goose works with any LLM and supports multi-mo
|
||||
- [Documentation](https://block.github.io/goose/docs/category/getting-started)
|
||||
|
||||
|
||||
# A Little Goose Humor 🦢
|
||||
|
||||
> Why did the developer choose goose as their AI agent?
|
||||
>
|
||||
> Because it always helps them "migrate" their code to production! 🚀
|
||||
|
||||
# Goose Around with Us
|
||||
- [Discord](https://discord.gg/block-opensource)
|
||||
- [YouTube](https://www.youtube.com/@blockopensource)
|
||||
|
||||
@@ -22,9 +22,11 @@ mcp-client = { path = "../mcp-client" }
|
||||
mcp-server = { path = "../mcp-server" }
|
||||
mcp-core = { path = "../mcp-core" }
|
||||
rmcp = { workspace = true }
|
||||
agent-client-protocol = "0.1.1"
|
||||
clap = { version = "4.4", features = ["derive"] }
|
||||
cliclack = "0.3.5"
|
||||
console = "0.15.8"
|
||||
uuid = { version = "1.11", features = ["v4"] }
|
||||
dotenvy = "0.15.7"
|
||||
bat = "0.24.0"
|
||||
anyhow = "1.0"
|
||||
@@ -47,7 +49,6 @@ shlex = "1.3.0"
|
||||
async-trait = "0.1.86"
|
||||
base64 = "0.22.1"
|
||||
regex = "1.11.1"
|
||||
uuid = { version = "1.11", features = ["v4"] }
|
||||
nix = { version = "0.30.1", features = ["process", "signal"] }
|
||||
tar = "0.4"
|
||||
# Web server dependencies
|
||||
@@ -56,7 +57,7 @@ tower-http = { version = "0.5", features = ["cors", "fs"] }
|
||||
http = "1.0"
|
||||
webbrowser = "1.0"
|
||||
indicatif = "0.17.11"
|
||||
tokio-util = "0.7.15"
|
||||
tokio-util = { version = "0.7.15", features = ["compat"] }
|
||||
is-terminal = "0.4.16"
|
||||
anstream = "0.6.18"
|
||||
|
||||
|
||||
@@ -3,6 +3,7 @@ use clap::{Args, Parser, Subcommand};
|
||||
|
||||
use goose::config::{Config, ExtensionConfig};
|
||||
|
||||
use crate::commands::acp::run_acp_agent;
|
||||
use crate::commands::bench::agent_generator;
|
||||
use crate::commands::configure::handle_configure;
|
||||
use crate::commands::info::handle_info;
|
||||
@@ -291,6 +292,10 @@ enum Command {
|
||||
#[command(about = "Run one of the mcp servers bundled with goose")]
|
||||
Mcp { name: String },
|
||||
|
||||
/// Run Goose as an ACP (Agent Client Protocol) agent
|
||||
#[command(about = "Run Goose as an ACP agent server on stdio")]
|
||||
Acp {},
|
||||
|
||||
/// Start or resume interactive chat sessions
|
||||
#[command(
|
||||
about = "Start or resume interactive chat sessions",
|
||||
@@ -709,6 +714,7 @@ pub async fn cli() -> Result<()> {
|
||||
Some(Command::Configure {}) => "configure",
|
||||
Some(Command::Info { .. }) => "info",
|
||||
Some(Command::Mcp { .. }) => "mcp",
|
||||
Some(Command::Acp {}) => "acp",
|
||||
Some(Command::Session { .. }) => "session",
|
||||
Some(Command::Project {}) => "project",
|
||||
Some(Command::Projects) => "projects",
|
||||
@@ -739,6 +745,10 @@ pub async fn cli() -> Result<()> {
|
||||
Some(Command::Mcp { name }) => {
|
||||
let _ = run_server(&name).await;
|
||||
}
|
||||
Some(Command::Acp {}) => {
|
||||
let _ = run_acp_agent().await;
|
||||
return Ok(());
|
||||
}
|
||||
Some(Command::Session {
|
||||
command,
|
||||
identifier,
|
||||
|
||||
@@ -0,0 +1,495 @@
|
||||
use agent_client_protocol::{self as acp, Client, SessionNotification};
|
||||
use anyhow::Result;
|
||||
use goose::agents::Agent;
|
||||
use goose::config::{Config, ExtensionConfigManager};
|
||||
use goose::conversation::message::{Message, MessageContent};
|
||||
use goose::conversation::Conversation;
|
||||
use goose::providers::create;
|
||||
use std::collections::{HashMap, HashSet};
|
||||
use std::sync::Arc;
|
||||
use tokio::sync::{mpsc, oneshot, Mutex};
|
||||
use tokio::task::JoinSet;
|
||||
use tokio_util::compat::{TokioAsyncReadCompatExt as _, TokioAsyncWriteCompatExt as _};
|
||||
use tokio_util::sync::CancellationToken;
|
||||
use tracing::{error, info, warn};
|
||||
|
||||
/// Represents a single Goose session for ACP
|
||||
struct GooseSession {
|
||||
agent: Agent,
|
||||
messages: Conversation,
|
||||
tool_call_ids: HashMap<String, String>, // Maps internal tool IDs to ACP tool call IDs
|
||||
cancel_token: Option<CancellationToken>, // Active cancellation token for prompt processing
|
||||
}
|
||||
|
||||
/// Goose ACP Agent implementation that connects to real Goose agents
|
||||
struct GooseAcpAgent {
|
||||
session_update_tx: mpsc::UnboundedSender<(acp::SessionNotification, oneshot::Sender<()>)>,
|
||||
sessions: Arc<Mutex<HashMap<String, GooseSession>>>,
|
||||
provider: Arc<dyn goose::providers::base::Provider>,
|
||||
}
|
||||
|
||||
impl GooseAcpAgent {
|
||||
async fn new(
|
||||
session_update_tx: mpsc::UnboundedSender<(acp::SessionNotification, oneshot::Sender<()>)>,
|
||||
) -> Result<Self> {
|
||||
// Load config and create provider
|
||||
let config = Config::global();
|
||||
|
||||
let provider_name: String = config
|
||||
.get_param("GOOSE_PROVIDER")
|
||||
.map_err(|e| anyhow::anyhow!("No provider configured: {}", e))?;
|
||||
|
||||
let model_name: String = config
|
||||
.get_param("GOOSE_MODEL")
|
||||
.map_err(|e| anyhow::anyhow!("No model configured: {}", e))?;
|
||||
|
||||
let model_config = goose::model::ModelConfig {
|
||||
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)?;
|
||||
|
||||
Ok(Self {
|
||||
session_update_tx,
|
||||
sessions: Arc::new(Mutex::new(HashMap::new())),
|
||||
provider,
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
impl acp::Agent for GooseAcpAgent {
|
||||
async fn initialize(
|
||||
&self,
|
||||
arguments: acp::InitializeRequest,
|
||||
) -> Result<acp::InitializeResponse, acp::Error> {
|
||||
info!("ACP: Received initialize request {:?}", arguments);
|
||||
|
||||
// Advertise Goose's capabilities
|
||||
let agent_capabilities = acp::AgentCapabilities {
|
||||
load_session: false, // TODO: Implement session persistence
|
||||
prompt_capabilities: acp::PromptCapabilities {
|
||||
image: true, // Goose supports image inputs via providers
|
||||
audio: false, // TODO: Add audio support when providers support it
|
||||
embedded_context: true, // Goose can handle embedded context resources
|
||||
},
|
||||
};
|
||||
|
||||
Ok(acp::InitializeResponse {
|
||||
protocol_version: acp::V1,
|
||||
agent_capabilities,
|
||||
auth_methods: Vec::new(),
|
||||
})
|
||||
}
|
||||
|
||||
async fn authenticate(&self, arguments: acp::AuthenticateRequest) -> Result<(), acp::Error> {
|
||||
info!("ACP: Received authenticate request {:?}", arguments);
|
||||
Ok(())
|
||||
}
|
||||
|
||||
async fn new_session(
|
||||
&self,
|
||||
arguments: acp::NewSessionRequest,
|
||||
) -> Result<acp::NewSessionResponse, acp::Error> {
|
||||
info!("ACP: Received new session request {:?}", arguments);
|
||||
|
||||
// Generate a unique session ID
|
||||
let session_id = uuid::Uuid::new_v4().to_string();
|
||||
|
||||
// Create a new Agent and session for this ACP session
|
||||
let agent = Agent::new();
|
||||
agent
|
||||
.update_provider(self.provider.clone())
|
||||
.await
|
||||
.map_err(|_| acp::Error::internal_error())?;
|
||||
|
||||
// Load and add extensions just like the normal CLI
|
||||
// Get all enabled extensions from configuration
|
||||
let extensions_to_run: Vec<_> = ExtensionConfigManager::get_all()
|
||||
.map_err(|e| {
|
||||
error!("Failed to load extensions: {}", e);
|
||||
acp::Error::internal_error()
|
||||
})?
|
||||
.into_iter()
|
||||
.filter(|ext| ext.enabled)
|
||||
.map(|ext| ext.config)
|
||||
.collect();
|
||||
|
||||
// Add extensions to the agent in parallel
|
||||
let agent_ptr = Arc::new(agent);
|
||||
let mut set = JoinSet::new();
|
||||
let mut waiting_on = HashSet::new();
|
||||
|
||||
for extension in extensions_to_run {
|
||||
waiting_on.insert(extension.name());
|
||||
let agent_ptr_clone = agent_ptr.clone();
|
||||
set.spawn(async move {
|
||||
(
|
||||
extension.name(),
|
||||
agent_ptr_clone.add_extension(extension.clone()).await,
|
||||
)
|
||||
});
|
||||
}
|
||||
|
||||
// Wait for all extensions to load
|
||||
while let Some(result) = set.join_next().await {
|
||||
match result {
|
||||
Ok((name, Ok(_))) => {
|
||||
waiting_on.remove(&name);
|
||||
info!("Loaded extension: {}", name);
|
||||
}
|
||||
Ok((name, Err(e))) => {
|
||||
warn!("Failed to load extension '{}': {}", name, e);
|
||||
waiting_on.remove(&name);
|
||||
}
|
||||
Err(e) => {
|
||||
error!("Task error while loading extension: {}", e);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Unwrap the Arc to get the agent back
|
||||
let agent = Arc::try_unwrap(agent_ptr).map_err(|_| {
|
||||
error!("Failed to unwrap agent Arc");
|
||||
acp::Error::internal_error()
|
||||
})?;
|
||||
|
||||
let session = GooseSession {
|
||||
agent,
|
||||
messages: Conversation::new_unvalidated(Vec::new()),
|
||||
tool_call_ids: HashMap::new(),
|
||||
cancel_token: None,
|
||||
};
|
||||
|
||||
// Store the session
|
||||
let mut sessions = self.sessions.lock().await;
|
||||
sessions.insert(session_id.clone(), session);
|
||||
|
||||
info!("Created new session with ID: {}", session_id);
|
||||
|
||||
Ok(acp::NewSessionResponse {
|
||||
session_id: acp::SessionId(session_id.into()),
|
||||
})
|
||||
}
|
||||
|
||||
async fn load_session(&self, arguments: acp::LoadSessionRequest) -> Result<(), acp::Error> {
|
||||
info!("ACP: Received load session request {:?}", arguments);
|
||||
// For now, will start a new session. We could use goose session storage as an enhancement
|
||||
// we would need to map ACP session IDs to goose session ids (which by default are auto generated)
|
||||
// normal goose session restore in CLI doesn't load conversation visually.
|
||||
//
|
||||
// Example flow:
|
||||
// - Load session file by session_id (might need to map ACP session IDs to Goose session paths)
|
||||
// - For each message in history:
|
||||
// - If user message: send user_message_chunk notification
|
||||
// - If assistant message: send agent_message_chunk notification
|
||||
// - If tool calls/responses: send appropriate notifications
|
||||
|
||||
// For now, we don't support loading previous sessions
|
||||
Err(acp::Error::method_not_found())
|
||||
}
|
||||
|
||||
#[allow(clippy::too_many_lines)]
|
||||
async fn prompt(
|
||||
&self,
|
||||
arguments: acp::PromptRequest,
|
||||
) -> Result<acp::PromptResponse, acp::Error> {
|
||||
info!("ACP: Received prompt request {:?}", arguments);
|
||||
|
||||
// Get the session
|
||||
let session_id = arguments.session_id.0.to_string();
|
||||
let mut sessions = self.sessions.lock().await;
|
||||
let session = sessions
|
||||
.get_mut(&session_id)
|
||||
.ok_or_else(acp::Error::invalid_params)?;
|
||||
|
||||
// Convert ACP prompt to Goose message
|
||||
let mut user_message = Message::user();
|
||||
|
||||
// Process all content blocks from the prompt
|
||||
for block in arguments.prompt {
|
||||
match block {
|
||||
acp::ContentBlock::Text(text) => {
|
||||
user_message = user_message.with_text(&text.text);
|
||||
}
|
||||
acp::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);
|
||||
}
|
||||
acp::ContentBlock::Resource(resource) => {
|
||||
// Embed resource content as text with context
|
||||
match &resource.resource {
|
||||
acp::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
|
||||
}
|
||||
}
|
||||
}
|
||||
_ => {
|
||||
// Ignore unsupported content types for now
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Add message to conversation
|
||||
session.messages.push(user_message);
|
||||
|
||||
// Create and store cancellation token for this prompt
|
||||
let cancel_token = CancellationToken::new();
|
||||
session.cancel_token = Some(cancel_token.clone());
|
||||
|
||||
// Get agent's reply through the Goose agent
|
||||
let mut stream = session
|
||||
.agent
|
||||
.reply(session.messages.clone(), None, Some(cancel_token.clone()))
|
||||
.await
|
||||
.map_err(|e| {
|
||||
error!("Error getting agent reply: {}", e);
|
||||
acp::Error::internal_error()
|
||||
})?;
|
||||
|
||||
use futures::StreamExt;
|
||||
|
||||
// Track if we were cancelled
|
||||
let mut was_cancelled = false;
|
||||
|
||||
// Process the agent's response stream
|
||||
while let Some(event) = stream.next().await {
|
||||
// Check if we've been cancelled
|
||||
if cancel_token.is_cancelled() {
|
||||
was_cancelled = true;
|
||||
break;
|
||||
}
|
||||
|
||||
match event {
|
||||
Ok(goose::agents::AgentEvent::Message(message)) => {
|
||||
// Add to conversation
|
||||
session.messages.push(message.clone());
|
||||
|
||||
// Process message content, including tool calls
|
||||
for content_item in &message.content {
|
||||
match content_item {
|
||||
MessageContent::Text(text) => {
|
||||
// Stream text to the client
|
||||
let (tx, rx) = oneshot::channel();
|
||||
self.session_update_tx
|
||||
.send((
|
||||
SessionNotification {
|
||||
session_id: arguments.session_id.clone(),
|
||||
update: acp::SessionUpdate::AgentMessageChunk {
|
||||
content: text.text.clone().into(),
|
||||
},
|
||||
},
|
||||
tx,
|
||||
))
|
||||
.map_err(|_| acp::Error::internal_error())?;
|
||||
rx.await.map_err(|_| acp::Error::internal_error())?;
|
||||
}
|
||||
MessageContent::ToolRequest(tool_request) => {
|
||||
// Generate ACP tool call ID and track mapping
|
||||
let acp_tool_id = format!("tool_{}", uuid::Uuid::new_v4());
|
||||
session
|
||||
.tool_call_ids
|
||||
.insert(tool_request.id.clone(), acp_tool_id.clone());
|
||||
|
||||
// Extract tool name and parameters from the ToolCall if successful
|
||||
let (tool_name, locations) = match &tool_request.tool_call {
|
||||
Ok(tool_call) => {
|
||||
let name = tool_call.name.clone();
|
||||
|
||||
// Extract file locations from certain tools for client tracking
|
||||
let mut locs = Vec::new();
|
||||
if name == "developer__text_editor" {
|
||||
// Try to extract the path from the arguments
|
||||
let args = &tool_call.arguments;
|
||||
if let Some(path_str) =
|
||||
args.get("path").and_then(|p| p.as_str())
|
||||
{
|
||||
locs.push(acp::ToolCallLocation {
|
||||
path: path_str.into(),
|
||||
line: Some(1),
|
||||
});
|
||||
}
|
||||
}
|
||||
(name, locs)
|
||||
}
|
||||
Err(_) => ("unknown".to_string(), Vec::new()),
|
||||
};
|
||||
|
||||
// Send tool call notification
|
||||
let (tx, rx) = oneshot::channel();
|
||||
self.session_update_tx
|
||||
.send((
|
||||
SessionNotification {
|
||||
session_id: arguments.session_id.clone(),
|
||||
update: acp::SessionUpdate::ToolCall(acp::ToolCall {
|
||||
id: acp::ToolCallId(acp_tool_id.clone().into()),
|
||||
title: format!("Calling tool: {}", tool_name),
|
||||
kind: acp::ToolKind::default(),
|
||||
status: acp::ToolCallStatus::Pending,
|
||||
content: Vec::new(),
|
||||
locations,
|
||||
raw_input: None,
|
||||
raw_output: None,
|
||||
}),
|
||||
},
|
||||
tx,
|
||||
))
|
||||
.map_err(|_| acp::Error::internal_error())?;
|
||||
rx.await.map_err(|_| acp::Error::internal_error())?;
|
||||
|
||||
// No need for separate update - status is already set to Pending
|
||||
}
|
||||
MessageContent::ToolResponse(tool_response) => {
|
||||
// Look up the ACP tool call ID
|
||||
if let Some(acp_tool_id) =
|
||||
session.tool_call_ids.get(&tool_response.id)
|
||||
{
|
||||
// Determine if the tool call succeeded or failed
|
||||
let status = if tool_response.tool_result.is_ok() {
|
||||
acp::ToolCallStatus::Completed
|
||||
} else {
|
||||
acp::ToolCallStatus::Failed
|
||||
};
|
||||
|
||||
// Send status update (completed or failed)
|
||||
let (tx, rx) = oneshot::channel();
|
||||
self.session_update_tx
|
||||
.send((
|
||||
SessionNotification {
|
||||
session_id: arguments.session_id.clone(),
|
||||
update: acp::SessionUpdate::ToolCallUpdate(
|
||||
acp::ToolCallUpdate {
|
||||
id: acp::ToolCallId(
|
||||
acp_tool_id.clone().into(),
|
||||
),
|
||||
fields: acp::ToolCallUpdateFields {
|
||||
status: Some(status),
|
||||
..Default::default()
|
||||
},
|
||||
},
|
||||
),
|
||||
},
|
||||
tx,
|
||||
))
|
||||
.map_err(|_| acp::Error::internal_error())?;
|
||||
rx.await.map_err(|_| acp::Error::internal_error())?;
|
||||
}
|
||||
}
|
||||
MessageContent::Thinking(thinking) => {
|
||||
// Stream thinking/reasoning content as thought chunks
|
||||
let (tx, rx) = oneshot::channel();
|
||||
self.session_update_tx
|
||||
.send((
|
||||
SessionNotification {
|
||||
session_id: arguments.session_id.clone(),
|
||||
update: acp::SessionUpdate::AgentThoughtChunk {
|
||||
content: thinking.thinking.clone().into(),
|
||||
},
|
||||
},
|
||||
tx,
|
||||
))
|
||||
.map_err(|_| acp::Error::internal_error())?;
|
||||
rx.await.map_err(|_| acp::Error::internal_error())?;
|
||||
}
|
||||
_ => {
|
||||
// Ignore other content types for now
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
Ok(_) => {
|
||||
// Ignore other events for now
|
||||
}
|
||||
Err(e) => {
|
||||
error!("Error in agent response stream: {}", e);
|
||||
return Err(acp::Error::internal_error());
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Clear the cancel token since we're done
|
||||
session.cancel_token = None;
|
||||
|
||||
Ok(acp::PromptResponse {
|
||||
stop_reason: if was_cancelled {
|
||||
acp::StopReason::Cancelled
|
||||
} else {
|
||||
acp::StopReason::EndTurn
|
||||
},
|
||||
})
|
||||
}
|
||||
|
||||
async fn cancel(&self, args: acp::CancelNotification) -> Result<(), acp::Error> {
|
||||
info!("ACP: Received cancel request {:?}", args);
|
||||
|
||||
// Get the session and cancel its active operation
|
||||
let session_id = args.session_id.0.to_string();
|
||||
let mut sessions = self.sessions.lock().await;
|
||||
|
||||
if let Some(session) = sessions.get_mut(&session_id) {
|
||||
if let Some(ref token) = session.cancel_token {
|
||||
info!("Cancelling active prompt for session {}", session_id);
|
||||
token.cancel();
|
||||
}
|
||||
} else {
|
||||
warn!("Cancel request for non-existent session: {}", session_id);
|
||||
}
|
||||
|
||||
Ok(())
|
||||
}
|
||||
}
|
||||
|
||||
/// Run the ACP agent server
|
||||
pub async fn run_acp_agent() -> Result<()> {
|
||||
info!("Starting Goose ACP agent server on stdio");
|
||||
eprintln!("Goose ACP agent started. Listening on stdio...");
|
||||
|
||||
let outgoing = tokio::io::stdout().compat_write();
|
||||
let incoming = tokio::io::stdin().compat();
|
||||
|
||||
// The AgentSideConnection will spawn futures onto our Tokio runtime.
|
||||
// LocalSet and spawn_local are used because the futures from the
|
||||
// agent-client-protocol crate are not Send.
|
||||
let local_set = tokio::task::LocalSet::new();
|
||||
local_set
|
||||
.run_until(async move {
|
||||
let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel();
|
||||
|
||||
// Start up the GooseAcpAgent connected to stdio.
|
||||
let agent = GooseAcpAgent::new(tx)
|
||||
.await
|
||||
.map_err(|e| anyhow::anyhow!("Failed to create ACP agent: {}", e))?;
|
||||
let (conn, handle_io) =
|
||||
acp::AgentSideConnection::new(agent, outgoing, incoming, |fut| {
|
||||
tokio::task::spawn_local(fut);
|
||||
});
|
||||
|
||||
// Kick off a background task to send the agent's session notifications to the client.
|
||||
tokio::task::spawn_local(async move {
|
||||
while let Some((session_notification, tx)) = rx.recv().await {
|
||||
let result = conn.session_notification(session_notification).await;
|
||||
if let Err(e) = result {
|
||||
error!("ACP session notification error: {}", e);
|
||||
break;
|
||||
}
|
||||
tx.send(()).ok();
|
||||
}
|
||||
});
|
||||
|
||||
// Run until stdin/stdout are closed.
|
||||
handle_io.await
|
||||
})
|
||||
.await?;
|
||||
|
||||
Ok(())
|
||||
}
|
||||
@@ -1,3 +1,4 @@
|
||||
pub mod acp;
|
||||
pub mod bench;
|
||||
pub mod configure;
|
||||
pub mod info;
|
||||
|
||||
Executable
+117
@@ -0,0 +1,117 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
Simple ACP client to test the Goose ACP agent.
|
||||
Connects to goose acp running on stdio.
|
||||
"""
|
||||
|
||||
import subprocess
|
||||
import json
|
||||
import sys
|
||||
import uuid
|
||||
|
||||
class AcpClient:
|
||||
def __init__(self):
|
||||
# Start the goose acp process
|
||||
self.process = subprocess.Popen(
|
||||
['cargo', 'run', '-p', 'goose-cli', '--', 'acp'],
|
||||
stdin=subprocess.PIPE,
|
||||
stdout=subprocess.PIPE,
|
||||
stderr=subprocess.PIPE,
|
||||
text=True,
|
||||
bufsize=0
|
||||
)
|
||||
self.request_id = 0
|
||||
|
||||
def send_request(self, method, params=None):
|
||||
self.request_id += 1
|
||||
request = {
|
||||
"jsonrpc": "2.0",
|
||||
"method": method,
|
||||
"id": self.request_id,
|
||||
}
|
||||
if params:
|
||||
request["params"] = params
|
||||
|
||||
# Send the request
|
||||
request_str = json.dumps(request)
|
||||
print(f">>> Sending: {request_str}")
|
||||
self.process.stdin.write(request_str + '\n')
|
||||
self.process.stdin.flush()
|
||||
|
||||
# Read response
|
||||
response_line = self.process.stdout.readline()
|
||||
if not response_line:
|
||||
return None
|
||||
|
||||
print(f"<<< Response: {response_line}")
|
||||
return json.loads(response_line)
|
||||
|
||||
def initialize(self):
|
||||
return self.send_request("initialize", {
|
||||
"protocolVersion": "v1",
|
||||
"clientCapabilities": {},
|
||||
"clientInfo": {
|
||||
"name": "test-client",
|
||||
"version": "1.0.0"
|
||||
}
|
||||
})
|
||||
|
||||
def new_session(self):
|
||||
return self.send_request("newSession", {
|
||||
"context": {}
|
||||
})
|
||||
|
||||
def prompt(self, session_id, text):
|
||||
return self.send_request("prompt", {
|
||||
"sessionId": session_id,
|
||||
"prompt": [
|
||||
{
|
||||
"type": "text",
|
||||
"text": text
|
||||
}
|
||||
]
|
||||
})
|
||||
|
||||
def close(self):
|
||||
if self.process:
|
||||
self.process.terminate()
|
||||
self.process.wait()
|
||||
|
||||
def main():
|
||||
print("Starting ACP client test...")
|
||||
client = AcpClient()
|
||||
|
||||
try:
|
||||
# Initialize the agent
|
||||
print("\n1. Initializing agent...")
|
||||
init_response = client.initialize()
|
||||
if init_response and 'result' in init_response:
|
||||
print(f" Initialized successfully: {init_response['result']}")
|
||||
else:
|
||||
print(f" Failed to initialize: {init_response}")
|
||||
return
|
||||
|
||||
# Create a new session
|
||||
print("\n2. Creating new session...")
|
||||
session_response = client.new_session()
|
||||
if session_response and 'result' in session_response:
|
||||
session_id = session_response['result']['sessionId']
|
||||
print(f" Created session: {session_id}")
|
||||
else:
|
||||
print(f" Failed to create session: {session_response}")
|
||||
return
|
||||
|
||||
# Send a prompt
|
||||
print("\n3. Sending prompt...")
|
||||
prompt_response = client.prompt(session_id, "Hello! What is 2 + 2?")
|
||||
if prompt_response:
|
||||
print(f" Got response: {prompt_response}")
|
||||
else:
|
||||
print(" Failed to get prompt response")
|
||||
|
||||
finally:
|
||||
client.close()
|
||||
print("\nTest complete.")
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
Reference in New Issue
Block a user