feat(agent): orchestration v3 — surface MCP model-agnostic (M0→M4) — §14.3
Expose l'orchestration IdeA comme serveur MCP par-dessus le même OrchestratorService::dispatch, avec repli fichier .ideai/requests pour les CLI sans MCP. v3 réduite à la surface MCP : la messagerie inter-agents et la corrélation requête↔réponse étaient déjà résolues par §17 (send_blocking). - M0 capacité MCP sur le profil (McpCapability/McpConfigStrategy/McpTransport) - M1 injection conf MCP au LaunchAgent + prose adaptée selon la surface - M2 serveur/adapter MCP (JSON-RPC 2.0 maison ; outils idea_*) + ListAgents - M3 câblage par projet (registre mcp_servers jumeau du watcher) - M4 observabilité UI : OrchestratorRequestProcessed.source = file|mcp + badge Trois portes d'entrée (fichier, MCP, UI) → un seul dispatch ; aucun nouveau port applicatif ; MCP confiné à l'adapter infra. Tous lots verts (cycle §3). Cadrage : .ideai/briefs/orchestration-v3-cadrage.md ; ARCHITECTURE.md §14.3. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
246
crates/infrastructure/src/orchestrator/mcp/server.rs
Normal file
246
crates/infrastructure/src/orchestrator/mcp/server.rs
Normal file
@ -0,0 +1,246 @@
|
||||
//! [`McpServer`] — the IdeA MCP **driving adapter** (cadrage Décision 4).
|
||||
//!
|
||||
//! This is the **pair of [`FsOrchestratorWatcher`](super::super::FsOrchestratorWatcher)**:
|
||||
//! another entry door onto the *same* [`OrchestratorService::dispatch`]. Where the
|
||||
//! watcher reads a JSON file, this server reads a JSON-RPC `tools/call`; both build
|
||||
//! an [`OrchestratorCommand`] and return its [`OrchestratorOutcome`] unchanged. It
|
||||
//! invents **no semantics** and **never re-routes** the reply — for `idea_ask_agent`
|
||||
//! it returns `outcome.reply` inline, exactly what `dispatch` produced.
|
||||
//!
|
||||
//! It implements the strict MCP minimum over JSON-RPC 2.0:
|
||||
//! - `initialize` → advertise protocol version + tool capability,
|
||||
//! - `tools/list` → the [`tools::catalogue`],
|
||||
//! - `tools/call` → map → `dispatch` → MCP tool result,
|
||||
//! - notifications (e.g. `notifications/initialized`) → accepted, no reply.
|
||||
//!
|
||||
//! It receives `Arc<OrchestratorService>` and the target [`Project`] by injection
|
||||
//! at the composition root (M3), exactly like the watcher — no application logic is
|
||||
//! duplicated here.
|
||||
|
||||
use std::sync::Arc;
|
||||
|
||||
use application::OrchestratorService;
|
||||
use domain::{DomainEvent, OrchestrationSource, Project};
|
||||
use serde_json::{json, Value};
|
||||
|
||||
use super::jsonrpc::{
|
||||
error_codes, JsonRpcError, JsonRpcRequest, JsonRpcResponse, Transport, TransportError,
|
||||
JSONRPC_VERSION,
|
||||
};
|
||||
use super::tools::{self, ToolMapError};
|
||||
|
||||
/// The MCP protocol version this server speaks (advertised on `initialize`).
|
||||
const MCP_PROTOCOL_VERSION: &str = "2024-11-05";
|
||||
|
||||
/// The IdeA MCP server: an entry adapter over [`OrchestratorService::dispatch`].
|
||||
///
|
||||
/// Cheap to clone the dependencies it holds; one instance serves one project's
|
||||
/// agents over one transport (M3 owns one server per open project, beside the
|
||||
/// watcher).
|
||||
pub struct McpServer {
|
||||
service: Arc<OrchestratorService>,
|
||||
project: Project,
|
||||
/// Optional event sink (a domain [`EventBus`](domain::ports::EventBus) publish
|
||||
/// closure), the twin of the file watcher's. When set, a processed `tools/call`
|
||||
/// publishes [`DomainEvent::OrchestratorRequestProcessed`] tagged
|
||||
/// [`OrchestrationSource::Mcp`] so the presentation layer can surface that this
|
||||
/// delegation arrived through the MCP door. `None` ⇒ no publication (the server
|
||||
/// still works), so existing call sites stay valid.
|
||||
events: Option<Arc<dyn Fn(DomainEvent) + Send + Sync>>,
|
||||
}
|
||||
|
||||
impl McpServer {
|
||||
/// Builds the server from the injected application service and target project.
|
||||
/// No event sink — use [`with_events`](Self::with_events) to surface processed
|
||||
/// requests on the bus.
|
||||
#[must_use]
|
||||
pub fn new(service: Arc<OrchestratorService>, project: Project) -> Self {
|
||||
Self {
|
||||
service,
|
||||
project,
|
||||
events: None,
|
||||
}
|
||||
}
|
||||
|
||||
/// Attaches an event sink so each processed `tools/call` is republished on the
|
||||
/// bus tagged [`OrchestrationSource::Mcp`] — the MCP twin of the file watcher's
|
||||
/// `events` closure. Additive: callers that do not need observability keep using
|
||||
/// [`new`](Self::new).
|
||||
#[must_use]
|
||||
pub fn with_events(mut self, events: Arc<dyn Fn(DomainEvent) + Send + Sync>) -> Self {
|
||||
self.events = Some(events);
|
||||
self
|
||||
}
|
||||
|
||||
/// Serves JSON-RPC messages from `transport` until it closes (clean EOF).
|
||||
///
|
||||
/// Every inbound line is handled in isolation: a malformed line or an unknown
|
||||
/// method yields a JSON-RPC error response, **never a panic** and never a
|
||||
/// dropped connection. Notifications (no `id`) are processed without a reply.
|
||||
pub async fn serve<T: Transport>(&self, transport: &mut T) {
|
||||
loop {
|
||||
let raw = match transport.recv().await {
|
||||
Ok(bytes) => bytes,
|
||||
Err(TransportError::Closed) => break,
|
||||
Err(TransportError::Io(_)) => break,
|
||||
};
|
||||
if let Some(response) = self.handle_raw(&raw).await {
|
||||
let Ok(bytes) = serde_json::to_vec(&response) else {
|
||||
continue;
|
||||
};
|
||||
if transport.send(&bytes).await.is_err() {
|
||||
break;
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Parses one raw message and produces the response to send back, or `None`
|
||||
/// for a notification (no `id`) that owes no reply.
|
||||
///
|
||||
/// Kept standalone (no transport) so the whole request→response behaviour is
|
||||
/// unit-testable over plain bytes, with no I/O.
|
||||
pub async fn handle_raw(&self, raw: &[u8]) -> Option<JsonRpcResponse> {
|
||||
let request: JsonRpcRequest = match serde_json::from_slice(raw) {
|
||||
Ok(r) => r,
|
||||
Err(e) => {
|
||||
// Could not correlate (no id parsed) ⇒ null id per JSON-RPC.
|
||||
return Some(JsonRpcResponse::error(
|
||||
Value::Null,
|
||||
JsonRpcError::new(error_codes::PARSE_ERROR, format!("invalid json: {e}")),
|
||||
));
|
||||
}
|
||||
};
|
||||
|
||||
if request.jsonrpc != JSONRPC_VERSION {
|
||||
let id = request.id.unwrap_or(Value::Null);
|
||||
return Some(JsonRpcResponse::error(
|
||||
id,
|
||||
JsonRpcError::new(
|
||||
error_codes::INVALID_REQUEST,
|
||||
format!("unsupported jsonrpc version: {}", request.jsonrpc),
|
||||
),
|
||||
));
|
||||
}
|
||||
|
||||
// Notification (no id): never reply (the `?` short-circuits to `None`).
|
||||
let id = request.id.clone()?;
|
||||
|
||||
let result = self.dispatch_method(&request.method, request.params).await;
|
||||
Some(match result {
|
||||
Ok(value) => JsonRpcResponse::success(id, value),
|
||||
Err(error) => JsonRpcResponse::error(id, error),
|
||||
})
|
||||
}
|
||||
|
||||
/// Routes a method name to its handler.
|
||||
async fn dispatch_method(
|
||||
&self,
|
||||
method: &str,
|
||||
params: Option<Value>,
|
||||
) -> Result<Value, JsonRpcError> {
|
||||
match method {
|
||||
"initialize" => Ok(self.initialize_result()),
|
||||
"tools/list" => Ok(self.tools_list_result()),
|
||||
"tools/call" => self.tools_call(params.unwrap_or(Value::Null)).await,
|
||||
other => Err(JsonRpcError::new(
|
||||
error_codes::METHOD_NOT_FOUND,
|
||||
format!("method not found: {other}"),
|
||||
)),
|
||||
}
|
||||
}
|
||||
|
||||
/// The `initialize` result: protocol version, server identity, and the fact
|
||||
/// that we expose tools.
|
||||
fn initialize_result(&self) -> Value {
|
||||
json!({
|
||||
"protocolVersion": MCP_PROTOCOL_VERSION,
|
||||
"capabilities": { "tools": {} },
|
||||
"serverInfo": { "name": "idea-orchestrator", "version": env!("CARGO_PKG_VERSION") }
|
||||
})
|
||||
}
|
||||
|
||||
/// The `tools/list` result: the catalogue as MCP tool descriptors.
|
||||
fn tools_list_result(&self) -> Value {
|
||||
let tools: Vec<Value> = tools::catalogue()
|
||||
.into_iter()
|
||||
.map(|t| {
|
||||
json!({
|
||||
"name": t.name,
|
||||
"description": t.description,
|
||||
"inputSchema": t.input_schema,
|
||||
})
|
||||
})
|
||||
.collect();
|
||||
json!({ "tools": tools })
|
||||
}
|
||||
|
||||
/// `tools/call`: map the tool to an [`OrchestratorCommand`], `dispatch` it, and
|
||||
/// fold the [`OrchestratorOutcome`](application::OrchestratorOutcome) into an MCP
|
||||
/// tool result. The `idea_ask_agent` reply is returned **inline** — never
|
||||
/// re-routed.
|
||||
async fn tools_call(&self, params: Value) -> Result<Value, JsonRpcError> {
|
||||
let name = params
|
||||
.get("name")
|
||||
.and_then(Value::as_str)
|
||||
.ok_or_else(|| {
|
||||
JsonRpcError::new(error_codes::INVALID_PARAMS, "missing tool `name`")
|
||||
})?
|
||||
.to_owned();
|
||||
let arguments = params.get("arguments").cloned().unwrap_or(json!({}));
|
||||
|
||||
let command = tools::map_tool_call(&name, &arguments).map_err(map_err_to_jsonrpc)?;
|
||||
|
||||
let result = self.service.dispatch(&self.project, command).await;
|
||||
// Surface the processed delegation on the bus, tagged as the MCP door — the
|
||||
// twin of the file watcher's publish. `ok` mirrors the dispatch outcome; the
|
||||
// action is the tool name. No-op when no sink is attached.
|
||||
self.publish_processed(&name, result.is_ok());
|
||||
match result {
|
||||
Ok(outcome) => {
|
||||
// The text the agent sees: the reply for an `ask`, else the detail.
|
||||
let text = outcome.reply.clone().unwrap_or_else(|| outcome.detail.clone());
|
||||
Ok(tool_result_text(&text, false))
|
||||
}
|
||||
// A failed IdeA command is reported as a tool execution error
|
||||
// (`isError: true`) rather than a protocol error: the agent can read
|
||||
// it and decide, and the connection stays healthy.
|
||||
Err(e) => Ok(tool_result_text(&e.to_string(), true)),
|
||||
}
|
||||
}
|
||||
|
||||
/// Publishes [`DomainEvent::OrchestratorRequestProcessed`] for a handled
|
||||
/// `tools/call`, tagged [`OrchestrationSource::Mcp`]. The requester id is the
|
||||
/// stable `"mcp"` label until the per-session transport bind (open point S-MCP)
|
||||
/// carries the connected agent's identity. No-op without an event sink.
|
||||
fn publish_processed(&self, action: &str, ok: bool) {
|
||||
if let Some(publish) = &self.events {
|
||||
publish(DomainEvent::OrchestratorRequestProcessed {
|
||||
requester_id: "mcp".to_owned(),
|
||||
action: action.to_owned(),
|
||||
ok,
|
||||
source: OrchestrationSource::Mcp,
|
||||
});
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Builds an MCP `tools/call` result with a single text content block.
|
||||
fn tool_result_text(text: &str, is_error: bool) -> Value {
|
||||
json!({
|
||||
"content": [ { "type": "text", "text": text } ],
|
||||
"isError": is_error
|
||||
})
|
||||
}
|
||||
|
||||
/// Maps a [`ToolMapError`] to the right JSON-RPC error code.
|
||||
fn map_err_to_jsonrpc(err: ToolMapError) -> JsonRpcError {
|
||||
match err {
|
||||
ToolMapError::UnknownTool(_) => {
|
||||
JsonRpcError::new(error_codes::METHOD_NOT_FOUND, err.to_string())
|
||||
}
|
||||
ToolMapError::BadArguments(_) | ToolMapError::Invalid(_) => {
|
||||
JsonRpcError::new(error_codes::INVALID_PARAMS, err.to_string())
|
||||
}
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user