Skip to content

Use command line to run sub agent and sub recipe (in sequence or parallel) #3190

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
lifeizhou-ap merged 16 commits into from
Jul 4, 2025
Merged

Use command line to run sub agent and sub recipe (in sequence or parallel) #3190

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
16 commits
Select commit Hold shift + click to select a range
8320bf9
parallel execution
lifeizhou-ap Jul 1, 2025
8f9392b
fixed task schema
lifeizhou-ap Jul 1, 2025
e63485b
run subrecipe
lifeizhou-ap Jul 1, 2025
bab0b98
print output lively
lifeizhou-ap Jul 1, 2025
daed8a6
run single task
lifeizhou-ap Jul 1, 2025
9d2b4b6
fmt
lifeizhou-ap Jul 3, 2025
d5a45a9
merge main
lifeizhou-ap Jul 3, 2025
614c43d
merge conflicts
lifeizhou-ap Jul 3, 2025
a7cab6e
fixed format
lifeizhou-ap Jul 3, 2025
9b5414e
use the same output pattern as previous sub recipe implementation
lifeizhou-ap Jul 3, 2025
17e04e5
moved the stateless functions to session utils module
lifeizhou-ap Jul 3, 2025
684ba42
Revert "moved the stateless functions to session utils module"
lifeizhou-ap Jul 3, 2025
741007b
Merge branch 'main' into lifei/try-parallel
lifeizhou-ap Jul 4, 2025
0dc0b44
renamed to sub-recipe-execute-tool
lifeizhou-ap Jul 4, 2025
dbb6dbb
cleaned tool description
lifeizhou-ap Jul 4, 2025
197d3d1
added extra property execution mode
lifeizhou-ap Jul 4, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
8 changes: 7 additions & 1 deletion crates/goose/src/agents/agent.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -9,6 +9,9 @@ use futures::{stream, FutureExt, Stream, StreamExt, TryStreamExt};
use mcp_core::protocol::JsonRpcMessage;

use crate::agents::final_output_tool::{FINAL_OUTPUT_CONTINUATION_MESSAGE, FINAL_OUTPUT_TOOL_NAME};
use crate::agents::sub_recipe_execution_tool::sub_recipe_execute_task_tool::{
self, SUB_RECIPE_EXECUTE_TASK_TOOL_NAME,
};
use crate::agents::sub_recipe_manager::SubRecipeManager;
use crate::config::{Config, ExtensionConfigManager, PermissionManager};
use crate::message::Message;
Expand Down Expand Up @@ -286,11 +289,12 @@ impl Agent {

let extension_manager = self.extension_manager.read().await;
let sub_recipe_manager = self.sub_recipe_manager.lock().await;

let result: ToolCallResult = if sub_recipe_manager.is_sub_recipe_tool(&tool_call.name) {
sub_recipe_manager
.dispatch_sub_recipe_tool_call(&tool_call.name, tool_call.arguments.clone())
.await
} else if tool_call.name == SUB_RECIPE_EXECUTE_TASK_TOOL_NAME {
sub_recipe_execute_task_tool::run_tasks(tool_call.arguments.clone()).await
} else if tool_call.name == PLATFORM_READ_RESOURCE_TOOL_NAME {
// Check if the tool is read_resource and handle it separately
ToolCallResult::from(
Expand Down Expand Up @@ -574,6 +578,8 @@ impl Agent {
if let Some(final_output_tool) = self.final_output_tool.lock().await.as_ref() {
prefixed_tools.push(final_output_tool.tool());
}
prefixed_tools
.push(sub_recipe_execute_task_tool::create_sub_recipe_execute_task_tool());
}

prefixed_tools
Expand Down
1 change: 1 addition & 0 deletions crates/goose/src/agents/mod.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -11,6 +11,7 @@ mod reply_parts;
mod router_tool_selector;
mod router_tools;
mod schedule_tool;
pub mod sub_recipe_execution_tool;
pub mod sub_recipe_manager;
pub mod subagent;
pub mod subagent_handler;
Expand Down
90 changes: 20 additions & 70 deletions crates/goose/src/agents/recipe_tools/sub_recipe_tools.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -3,25 +3,20 @@ use std::{collections::HashMap, fs};
use anyhow::Result;
use mcp_core::tool::{Tool, ToolAnnotations};
use serde_json::{json, Map, Value};
use tokio::io::{AsyncBufReadExt, BufReader};
use tokio::process::Command;

use crate::agents::sub_recipe_execution_tool::lib::Task;
use crate::recipe::{Recipe, RecipeParameter, RecipeParameterRequirement, SubRecipe};

pub const SUB_RECIPE_TOOL_NAME_PREFIX: &str = "subrecipe__run_";
pub const SUB_RECIPE_TASK_TOOL_NAME_PREFIX: &str = "subrecipe__create_task";

pub fn create_sub_recipe_tool(sub_recipe: &SubRecipe) -> Tool {
pub fn create_sub_recipe_task_tool(sub_recipe: &SubRecipe) -> Tool {
let input_schema = get_input_schema(sub_recipe).unwrap();
Tool::new(
format!("{}_{}", SUB_RECIPE_TOOL_NAME_PREFIX, sub_recipe.name),
"Run a sub recipe.
Use this tool when you need to run a sub-recipe.
The sub recipe will be run with the provided parameters
and return the output of the sub recipe."
.to_string(),
format!("{}_{}", SUB_RECIPE_TASK_TOOL_NAME_PREFIX, sub_recipe.name),
"Before running this sub recipe, you should first create a task with this tool and then pass the task to the task executor".to_string(),
Copy link

@wendytang wendytang Jul 2, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

shall we add the recipe description into the tool description?

Copy link
Collaborator Author

@lifeizhou-ap lifeizhou-ap Jul 3, 2025 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Hi @wendytang This is a good idea!

However, I will hold on this change because at this point, the subrecipe does not have the content of the sub recipe file, so it does not know the subrecipe description. Parsing sub recipe file logic is currently in the goose-cli, we are going to move it into the goose package or parsing it before running sub-agent, and then we can get the description.

At the moment, without the sub recipe description in the tool description, it works fine. But definitely we will add this in when the parsing recipe file logic moves to goose package

input_schema,
Some(ToolAnnotations {
title: Some(format!("run sub recipe {}", sub_recipe.name)),
title: Some(format!("create sub recipe task {}", sub_recipe.name)),
read_only_hint: false,
destructive_hint: true,
idempotent_hint: false,
Expand Down Expand Up @@ -99,68 +94,23 @@ fn prepare_command_params(
Ok(sub_recipe_params)
}

pub async fn run_sub_recipe(sub_recipe: &SubRecipe, params: Value) -> Result<String> {
pub async fn create_sub_recipe_task(sub_recipe: &SubRecipe, params: Value) -> Result<String> {
let command_params = prepare_command_params(sub_recipe, params)?;

let mut command = Command::new("goose");
command.arg("run").arg("--recipe").arg(&sub_recipe.path);

for (key, value) in command_params {
command.arg("--params").arg(format!("{}={}", key, value));
}

command.stdout(std::process::Stdio::piped());
command.stderr(std::process::Stdio::piped());

let mut child = command
.spawn()
.map_err(|e| anyhow::anyhow!("Failed to spawn: {}", e))?;

let stdout = child.stdout.take().expect("Failed to capture stdout");
let stderr = child.stderr.take().expect("Failed to capture stderr");

let mut stdout_reader = BufReader::new(stdout).lines();
let mut stderr_reader = BufReader::new(stderr).lines();
let stdout_sub_recipe_name = sub_recipe.name.clone();
let stderr_sub_recipe_name = sub_recipe.name.clone();

// Spawn background tasks to read from stdout and stderr
let stdout_task = tokio::spawn(async move {
let mut buffer = String::new();
while let Ok(Some(line)) = stdout_reader.next_line().await {
println!("[sub-recipe {}] {}", stdout_sub_recipe_name, line);
buffer.push_str(&line);
buffer.push('\n');
let payload = json!({
"sub_recipe": {
"name": sub_recipe.name.clone(),
"command_parameters": command_params,
"recipe_path": sub_recipe.path.clone(),
}
buffer
});

let stderr_task = tokio::spawn(async move {
let mut buffer = String::new();
while let Ok(Some(line)) = stderr_reader.next_line().await {
eprintln!(
"[stderr for sub-recipe {}] {}",
stderr_sub_recipe_name, line
);
buffer.push_str(&line);
buffer.push('\n');
}
buffer
});

let status = child
.wait()
.await
.map_err(|e| anyhow::anyhow!("Failed to wait for process: {}", e))?;

let stdout_output = stdout_task.await.unwrap();
let stderr_output = stderr_task.await.unwrap();

if status.success() {
Ok(stdout_output)
} else {
Err(anyhow::anyhow!("Command failed:\n{}", stderr_output))
}
let task = Task {
id: uuid::Uuid::new_v4().to_string(),
task_type: "sub_recipe".to_string(),
payload,
};
let task_json = serde_json::to_string(&task)
.map_err(|e| anyhow::anyhow!("Failed to serialize Task: {}", e))?;
Ok(task_json)
}

#[cfg(test)]
Expand Down
103 changes: 103 additions & 0 deletions crates/goose/src/agents/sub_recipe_execution_tool/executor.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,103 @@
use std::sync::atomic::{AtomicBool, AtomicUsize};
use std::sync::Arc;
use tokio::sync::mpsc;
use tokio::time::Instant;

use crate::agents::sub_recipe_execution_tool::lib::{
Config, ExecutionResponse, ExecutionStats, Task, TaskResult,
};
use crate::agents::sub_recipe_execution_tool::tasks::process_task;
use crate::agents::sub_recipe_execution_tool::workers::{run_scaler, spawn_worker, SharedState};

pub async fn execute_single_task(task: &Task, config: Config) -> ExecutionResponse {
let start_time = Instant::now();
let result = process_task(task, config.timeout_seconds).await;

let execution_time = start_time.elapsed().as_millis();
let completed = if result.status == "success" { 1 } else { 0 };
let failed = if result.status == "failed" { 1 } else { 0 };

ExecutionResponse {
status: "completed".to_string(),
results: vec![result],
stats: ExecutionStats {
total_tasks: 1,
completed,
failed,
execution_time_ms: execution_time,
},
}
}

// Main parallel execution function
pub async fn parallel_execute(tasks: Vec<Task>, config: Config) -> ExecutionResponse {
let start_time = Instant::now();
let task_count = tasks.len();

// Create channels
let (task_tx, task_rx) = mpsc::channel::<Task>(task_count);
let (result_tx, mut result_rx) = mpsc::channel::<TaskResult>(task_count);

// Initialize shared state
let shared_state = Arc::new(SharedState {
task_receiver: Arc::new(tokio::sync::Mutex::new(task_rx)),
result_sender: result_tx,
active_workers: Arc::new(AtomicUsize::new(0)),
should_stop: Arc::new(AtomicBool::new(false)),
completed_tasks: Arc::new(AtomicUsize::new(0)),
});

// Send all tasks to the queue
for task in tasks.clone() {
let _ = task_tx.send(task).await;
}
// Close sender so workers know when queue is empty
drop(task_tx);

// Start initial workers
let mut worker_handles = Vec::new();
for i in 0..config.initial_workers {
let handle = spawn_worker(shared_state.clone(), i, config.timeout_seconds);
worker_handles.push(handle);
}

// Start the scaler
let scaler_state = shared_state.clone();
let scaler_handle = tokio::spawn(async move {
run_scaler(
scaler_state,
task_count,
config.max_workers,
config.timeout_seconds,
)
.await;
});

// Collect results
let mut results = Vec::new();
while let Some(result) = result_rx.recv().await {
results.push(result);
if results.len() >= task_count {
break;
}
}

// Wait for scaler to finish
let _ = scaler_handle.await;

// Calculate stats
let execution_time = start_time.elapsed().as_millis();
let completed = results.iter().filter(|r| r.status == "success").count();
let failed = results.iter().filter(|r| r.status == "failed").count();

ExecutionResponse {
status: "completed".to_string(),
results,
stats: ExecutionStats {
total_tasks: task_count,
completed,
failed,
execution_time_ms: execution_time,
},
}
}
38 changes: 38 additions & 0 deletions crates/goose/src/agents/sub_recipe_execution_tool/lib.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,38 @@
use crate::agents::sub_recipe_execution_tool::executor::execute_single_task;
pub use crate::agents::sub_recipe_execution_tool::executor::parallel_execute;
pub use crate::agents::sub_recipe_execution_tool::types::{
Config, ExecutionResponse, ExecutionStats, Task, TaskResult,
};

use serde_json::Value;

pub async fn execute_tasks(input: Value, execution_mode: &str) -> Result<Value, String> {
let tasks: Vec<Task> =
serde_json::from_value(input.get("tasks").ok_or("Missing tasks field")?.clone())
.map_err(|e| format!("Failed to parse tasks: {}", e))?;

let config: Config = if let Some(config_value) = input.get("config") {
serde_json::from_value(config_value.clone())
.map_err(|e| format!("Failed to parse config: {}", e))?
} else {
Config::default()
};
let task_count = tasks.len();
match execution_mode {
"sequential" => {
if task_count == 1 {
let response = execute_single_task(&tasks[0], config).await;
serde_json::to_value(response)
.map_err(|e| format!("Failed to serialize response: {}", e))
} else {
Err("Sequential execution mode requires exactly one task".to_string())
}
}
"parallel" => {
let response = parallel_execute(tasks, config).await;
serde_json::to_value(response)
.map_err(|e| format!("Failed to serialize response: {}", e))
}
_ => Err("Invalid execution mode".to_string()),
}
}
6 changes: 6 additions & 0 deletions crates/goose/src/agents/sub_recipe_execution_tool/mod.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,6 @@
mod executor;
pub mod lib;
pub mod sub_recipe_execute_task_tool;
mod tasks;
mod types;
mod workers;
Loading