blomios/IdeA
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(mcp): ajoute l'outil idea_ask_agents pour la délégation parallèle (#2)

Browse Source 
Wrapper MCP backend pur : dispatche N requêtes idea_ask_agent en parallèle
et retourne un résultat par requête, dans l'ordre, en rejetant les cibles
dupliquées avant dispatch. Aucun changement de domaine/application — pose
la primitive nécessaire au SDK (chantier feature/sdk-integration).

QA vert (ticket #2).

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
This commit is contained in:
Blomios
2026-07-28 16:56:22 +02:00
parent 454f8ad57d
commit 1ef1fd9e40
4 changed files with 456 additions and 10 deletions
Download Patch File Download Diff File Expand all files Collapse all files

5
crates/backend/src/lib.rs
View File

@ -4923,6 +4923,7 @@ mod mcp_serve_peer_tests {
for expected in [
"idea_list_agents",
"idea_ask_agent",
"idea_ask_agents",
"idea_run_in_background",
"idea_launch_agent",
"idea_stop_agent",
@ -4967,8 +4968,8 @@ mod mcp_serve_peer_tests {
assert!(!names.contains(&"idea_reply"));
assert_eq!(
tools.len(),
30,
"exactly the thirty exposed idea_* tools; got {names:?}"
31,
"exactly the thirty-one exposed idea_* tools; got {names:?}"
);
drop(client); // EOF ⇒ serve loop ends

192
crates/infrastructure/src/orchestrator/mcp/server.rs
View File

@ -23,8 +23,10 @@ use std::time::Instant;
use application::OrchestratorService;
use domain::ports::McpToolPermissionStore;
use domain::{
AgentId, AgentToolPolicy, DomainEvent, IssueRef, McpToolPolicy, OrchestrationSource, Project,
AgentId, AgentToolPolicy, DomainEvent, IssueRef, McpToolPolicy, OrchestrationSource,
OrchestratorCommand, Project,
};
use futures_util::future::join_all;
use serde_json::{json, Value};
use tokio::sync::mpsc;
@ -482,6 +484,12 @@ impl McpServer {
task_len={task_len}",
);
if name == "idea_ask_agents" {
return self
.tools_call_ask_agents(arguments, requester_label, arg_target, started)
.await;
}
if tools::is_ticket_tool(&name) {
let result = match &self.ticket_tools {
Some(provider) => {
@ -611,6 +619,65 @@ impl McpServer {
}
}
/// Comfort wrapper over several parallel `idea_ask_agent` dispatches.
///
/// This remains an MCP-adapter-only fan-out/fan-in: each item is validated and
/// translated to the existing [`OrchestratorCommand::AskAgent`] shape, then sent to
/// [`OrchestratorService::dispatch`]. Per-item dispatch failures are folded into
/// the result array so one failed target never cancels the others.
async fn tools_call_ask_agents(
&self,
arguments: Value,
requester_label: String,
arg_target: String,
started: Instant,
) -> Result<Value, JsonRpcError> {
let requests = parse_ask_agents_requests(&arguments, &self.requester)?;
let project = self.project.clone();
let service = Arc::clone(&self.service);
let futures = requests.iter().cloned().map(|request| {
let project = project.clone();
let service = Arc::clone(&service);
async move {
let command = OrchestratorCommand::AskAgent {
target: request.target.clone(),
task: request.task,
requester: request.requester,
};
match service.dispatch(&project, command).await {
Ok(outcome) => AskAgentsItemResult {
target: request.target,
ok: true,
text: outcome.reply.unwrap_or(outcome.detail),
},
Err(err) => AskAgentsItemResult {
target: request.target,
ok: false,
text: err.to_string(),
},
}
}
});
let results = join_all(futures).await;
self.publish_processed("idea_ask_agents", true);
let text = serde_json::to_string(
&results
.iter()
.map(AskAgentsItemResult::to_value)
.collect::<Vec<_>>(),
)
.unwrap_or_else(|_| "[]".to_owned());
application::diag!(
"[mcp] tools_call end tool=idea_ask_agents requester={requester_label} \
target={arg_target} ok=true is_error=false result_len={} elapsed_ms={}",
text.len(),
started.elapsed().as_millis(),
);
Ok(tool_result_text(&text, false))
}
fn enforce_tool_policy(
&self,
policy: &AgentToolPolicy,
@ -690,6 +757,129 @@ fn tool_result_text(text: &str, is_error: bool) -> Value {
})
}
#[derive(Clone)]
struct AskAgentsRequest {
target: String,
task: String,
requester: Option<AgentId>,
}
struct AskAgentsItemResult {
target: String,
ok: bool,
text: String,
}
impl AskAgentsItemResult {
fn to_value(&self) -> Value {
json!({
"target": self.target,
"ok": self.ok,
"isError": !self.ok,
"text": self.text,
})
}
}
fn parse_ask_agents_requests(
arguments: &Value,
requester: &str,
) -> Result<Vec<AskAgentsRequest>, JsonRpcError> {
let args = arguments.as_object().ok_or_else(|| {
JsonRpcError::new(
error_codes::INVALID_PARAMS,
"tool `idea_ask_agents` arguments must be a JSON object",
)
})?;
if args.len() != 1 || !args.contains_key("requests") {
return Err(JsonRpcError::new(
error_codes::INVALID_PARAMS,
"tool `idea_ask_agents` expects exactly one field: requests",
));
}
let requests = args
.get("requests")
.and_then(Value::as_array)
.ok_or_else(|| {
JsonRpcError::new(
error_codes::INVALID_PARAMS,
"tool `idea_ask_agents` field `requests` must be a non-empty array",
)
})?;
if requests.is_empty() {
return Err(JsonRpcError::new(
error_codes::INVALID_PARAMS,
"tool `idea_ask_agents` field `requests` must be a non-empty array",
));
}
let mut seen = std::collections::HashSet::new();
let mut out = Vec::with_capacity(requests.len());
for (index, request) in requests.iter().enumerate() {
let obj = request.as_object().ok_or_else(|| {
JsonRpcError::new(
error_codes::INVALID_PARAMS,
format!("tool `idea_ask_agents` requests[{index}] must be an object"),
)
})?;
if obj.len() != 2 || !obj.contains_key("target") || !obj.contains_key("task") {
return Err(JsonRpcError::new(
error_codes::INVALID_PARAMS,
format!("tool `idea_ask_agents` requests[{index}] expects exactly target and task"),
));
}
let target = obj
.get("target")
.and_then(Value::as_str)
.filter(|v| !v.trim().is_empty())
.ok_or_else(|| {
JsonRpcError::new(
error_codes::INVALID_PARAMS,
format!("tool `idea_ask_agents` requests[{index}].target must be a string"),
)
})?
.to_owned();
let task = obj
.get("task")
.and_then(Value::as_str)
.filter(|v| !v.trim().is_empty())
.ok_or_else(|| {
JsonRpcError::new(
error_codes::INVALID_PARAMS,
format!("tool `idea_ask_agents` requests[{index}].task must be a string"),
)
})?
.to_owned();
let duplicate_key = target.to_lowercase();
if !seen.insert(duplicate_key) {
return Err(JsonRpcError::new(
error_codes::INVALID_PARAMS,
format!("tool `idea_ask_agents` duplicate target `{target}` is not supported"),
));
}
let command = tools::map_tool_call(
"idea_ask_agent",
&json!({ "target": target, "task": task }),
requester,
)
.map_err(map_err_to_jsonrpc)?;
let OrchestratorCommand::AskAgent {
target,
task,
requester,
} = command
else {
unreachable!("idea_ask_agent must map to AskAgent");
};
out.push(AskAgentsRequest {
target,
task,
requester,
});
}
Ok(out)
}
/// Maps a [`ToolMapError`] to the right JSON-RPC error code.
fn map_err_to_jsonrpc(err: ToolMapError) -> JsonRpcError {
match err {

40
crates/infrastructure/src/orchestrator/mcp/tools.rs
View File

@ -61,6 +61,7 @@ pub const READ_ONLY_TOOLS: &[&str] = &[
/// Canonical write/action MCP tools denied by default.
pub const WRITE_ACTION_TOOLS: &[&str] = &[
"idea_ask_agent",
"idea_ask_agents",
"idea_run_in_background",
"idea_launch_agent",
"idea_stop_agent",
@ -125,6 +126,7 @@ pub fn tool_returns_reply(tool: &str) -> bool {
matches!(
tool,
"idea_ask_agent"
| "idea_ask_agents"
| "idea_list_agents"
| "idea_context_read"
| "idea_memory_read"
@ -180,6 +182,32 @@ pub fn catalogue() -> Vec<ToolDef> {
"additionalProperties": false
}),
},
ToolDef {
name: "idea_ask_agents",
description: "Ask several IdeA agents in parallel and return one result per request, \
in the same order. Each request is equivalent to one idea_ask_agent \
call; duplicate targets are rejected before dispatch.",
input_schema: json!({
"type": "object",
"properties": {
"requests": {
"type": "array",
"minItems": 1,
"items": {
"type": "object",
"properties": {
"target": { "type": "string", "description": "Target agent display name." },
"task": { "type": "string", "description": "Task or question to send to the target agent." }
},
"required": ["target", "task"],
"additionalProperties": false
}
}
},
"required": ["requests"],
"additionalProperties": false
}),
},
ToolDef {
name: "idea_run_in_background",
description: "Run a command as a first-class IdeA background task. Returns the task id \
@ -659,6 +687,18 @@ mod tests {
);
}
#[test]
fn ask_agents_is_classified_and_returns_inline_reply() {
assert_eq!(
tool_access("idea_ask_agents"),
Some(McpToolAccess::WriteAction)
);
assert!(catalogue()
.into_iter()
.any(|tool| tool.name == "idea_ask_agents"));
assert!(tool_returns_reply("idea_ask_agents"));
}
#[test]
fn run_in_background_maps_to_background_command_with_handshake_owner() {
let cmd = map_tool_call(

229
crates/infrastructure/tests/mcp_server.rs
View File

@ -32,10 +32,11 @@ use domain::ids::SkillId;
use domain::ids::{AgentId, ProfileId, ProjectId};
use domain::markdown::MarkdownDoc;
use domain::ports::{
AgentContextStore, AgentRuntime, ContextInjectionPlan, DirEntry, EventBus, EventStream,
ExitStatus, FileSystem, FsError, IdGenerator, McpToolPermissionStore, OutputStream,
PreparedContext, ProfileStore, PtyError, PtyHandle, PtyPort, RemotePath, RuntimeError,
SessionPlan, SkillStore, SpawnSpec, StoreError,
AgentContextStore, AgentRuntime, AgentSession, AgentSessionError, ContextInjectionPlan,
DirEntry, EventBus, EventStream, ExitStatus, FileSystem, FsError, IdGenerator,
McpToolPermissionStore, OutputStream, PreparedContext, ProfileStore, PtyError, PtyHandle,
PtyPort, RemotePath, ReplyEvent, ReplyStream, RuntimeError, SessionPlan, SkillStore, SpawnSpec,
StoreError,
};
use domain::profile::{
AgentProfile, ContextInjection, McpCapability, McpConfigStrategy, McpTransport,
@ -53,7 +54,7 @@ use uuid::Uuid;
use application::{
CloseTerminal, CreateAgentFromScratch, CreateSkill, LaunchAgent, ListAgents,
OrchestratorService, TerminalSessions, UpdateAgentContext,
OrchestratorService, StructuredSessions, TerminalSessions, UpdateAgentContext,
};
use infrastructure::orchestrator::mcp::jsonrpc::error_codes;
use infrastructure::orchestrator::mcp::tools::classified_tool_names;
@ -467,6 +468,66 @@ fn build_service_with_mailbox(
(Arc::new(service), mailbox, sessions)
}
#[derive(Clone)]
struct FakeStructuredSession {
id: SessionId,
reply: String,
delay_ms: u64,
}
#[async_trait]
impl AgentSession for FakeStructuredSession {
fn id(&self) -> SessionId {
self.id
}
fn conversation_id(&self) -> Option<String> {
Some(format!("conv-{}", self.id))
}
async fn send(&self, _prompt: &str) -> Result<ReplyStream, AgentSessionError> {
if self.delay_ms > 0 {
tokio::time::sleep(std::time::Duration::from_millis(self.delay_ms)).await;
}
Ok(Box::new(std::iter::once(ReplyEvent::Final {
content: self.reply.clone(),
})))
}
async fn shutdown(&self) -> Result<(), AgentSessionError> {
Ok(())
}
}
fn build_service_with_structured(
contexts: FakeContexts,
) -> (Arc<OrchestratorService>, Arc<StructuredSessions>) {
let (base, _mailbox, _sessions) = build_service_with_mailbox(contexts);
let structured = Arc::new(StructuredSessions::new());
let service = Arc::try_unwrap(base)
.unwrap_or_else(|_| panic!("freshly built service must be uniquely owned"))
.with_structured(Arc::clone(&structured));
(Arc::new(service), structured)
}
fn insert_structured_reply(
structured: &StructuredSessions,
agent_id: AgentId,
reply: &str,
delay_ms: u64,
) {
structured.insert_in_project(
project().id,
Arc::new(FakeStructuredSession {
id: SessionId::from_uuid(Uuid::new_v4()),
reply: reply.to_owned(),
delay_ms,
}),
agent_id,
NodeId::from_uuid(Uuid::new_v4()),
);
}
fn server(service: Arc<OrchestratorService>) -> McpServer {
McpServer::new(service, project())
}
@ -665,6 +726,7 @@ async fn tools_list_advertises_the_idea_tools_with_schemas() {
for expected in [
"idea_list_agents",
"idea_ask_agent",
"idea_ask_agents",
"idea_run_in_background",
"idea_launch_agent",
"idea_stop_agent",
@ -707,8 +769,8 @@ async fn tools_list_advertises_the_idea_tools_with_schemas() {
assert!(!names.contains(&"idea_reply"));
assert_eq!(
tools.len(),
30,
"exactly the thirty exposed idea_* tools; got {names:?}"
31,
"exactly the thirty-one exposed idea_* tools; got {names:?}"
);
// Every tool advertises an object input schema.
@ -1446,6 +1508,159 @@ async fn reply_tool_is_not_exposed_over_mcp() {
assert!(resp.result.is_none());
}
#[tokio::test]
async fn ask_agents_fans_out_in_parallel_and_preserves_input_order() {
let contexts = FakeContexts::new();
let slow_id = contexts.seed_agent("slow");
let fast_id = contexts.seed_agent("fast");
let (service, structured) = build_service_with_structured(contexts);
insert_structured_reply(&structured, slow_id, "slow reply", 60);
insert_structured_reply(&structured, fast_id, "fast reply", 0);
let server = server(service);
let raw = tools_call(
21,
"idea_ask_agents",
json!({
"requests": [
{ "target": "slow", "task": "first" },
{ "target": "fast", "task": "second" }
]
}),
);
let response = server.handle_raw(&raw).await.expect("reply owed");
assert!(
response.error.is_none(),
"transport error: {:?}",
response.error
);
let result = response.result.expect("result");
assert_eq!(result["isError"], json!(false), "got {result}");
let payload: Value =
serde_json::from_str(result_text(&result)).expect("ask_agents result is JSON");
let items = payload.as_array().expect("result array");
assert_eq!(items.len(), 2);
assert_eq!(items[0]["target"], json!("slow"));
assert_eq!(items[0]["ok"], json!(true));
assert_eq!(items[0]["isError"], json!(false));
assert_eq!(items[0]["text"], json!("slow reply"));
assert_eq!(items[1]["target"], json!("fast"));
assert_eq!(items[1]["ok"], json!(true));
assert_eq!(items[1]["text"], json!("fast reply"));
}
#[tokio::test]
async fn ask_agents_allows_partial_success() {
let contexts = FakeContexts::new();
let ok_id = contexts.seed_agent("ok-agent");
let (service, structured) = build_service_with_structured(contexts);
insert_structured_reply(&structured, ok_id, "ok reply", 0);
let server = server(service);
let raw = tools_call(
22,
"idea_ask_agents",
json!({
"requests": [
{ "target": "ok-agent", "task": "works" },
{ "target": "missing-agent", "task": "fails" }
]
}),
);
let response = server.handle_raw(&raw).await.expect("reply owed");
assert!(
response.error.is_none(),
"wrapper should not fail transport"
);
let result = response.result.expect("result");
assert_eq!(
result["isError"],
json!(false),
"partial success is allowed"
);
let payload: Value = serde_json::from_str(result_text(&result)).unwrap();
let items = payload.as_array().unwrap();
assert_eq!(items[0]["target"], json!("ok-agent"));
assert_eq!(items[0]["ok"], json!(true));
assert_eq!(items[0]["text"], json!("ok reply"));
assert_eq!(items[1]["target"], json!("missing-agent"));
assert_eq!(items[1]["ok"], json!(false));
assert_eq!(items[1]["isError"], json!(true));
assert!(
items[1]["text"].as_str().unwrap().contains("missing-agent"),
"per-target error should be surfaced; got {payload}"
);
}
#[tokio::test]
async fn ask_agents_rejects_duplicate_targets_before_dispatch() {
let contexts = FakeContexts::new();
let agent_id = contexts.seed_agent("Architect");
let (service, structured) = build_service_with_structured(contexts);
insert_structured_reply(&structured, agent_id, "must not run", 0);
let server = server(service);
let raw = tools_call(
23,
"idea_ask_agents",
json!({
"requests": [
{ "target": "Architect", "task": "one" },
{ "target": "architect", "task": "two" }
]
}),
);
let response = server.handle_raw(&raw).await.expect("reply owed");
let error = response.error.expect("duplicate validation error expected");
assert_eq!(error.code, error_codes::INVALID_PARAMS);
assert!(
error.message.contains("duplicate target"),
"got {}",
error.message
);
assert!(response.result.is_none());
}
#[tokio::test]
async fn ask_agents_rejects_non_strict_shapes_before_dispatch() {
let (service, _structured) = build_service_with_structured(FakeContexts::new());
let server = server(service);
for (id, arguments, expected) in [
(24, json!({}), "requests"),
(25, json!({ "requests": [] }), "non-empty"),
(
26,
json!({ "requests": [{ "target": "a", "task": "x", "extra": true }] }),
"exactly target and task",
),
(
27,
json!({ "requests": [{ "target": "a" }] }),
"exactly target and task",
),
(
28,
json!({ "requests": [{ "target": "", "task": "x" }] }),
"target",
),
] {
let response = server
.handle_raw(&tools_call(id, "idea_ask_agents", arguments))
.await
.expect("reply owed");
let error = response.error.expect("validation error expected");
assert_eq!(error.code, error_codes::INVALID_PARAMS);
assert!(
error.message.contains(expected),
"expected message to contain `{expected}`, got {}",
error.message
);
assert!(response.result.is_none());
}
}
// ---------------------------------------------------------------------------
// 4. idea_list_agents returns the agent list inline (JSON array)
// ---------------------------------------------------------------------------