prospero_api/
mcp.rs

1//! The fleet as an MCP server (#218).
2//!
3//! prospero exposed REST + SSE only, so an agentic tool could not drive the
4//! fleet the way OpenClaw (`openclaw mcp serve`) and Vibe Kanban already allow.
5//! This is a thin adapter over the same [`FleetProvider`] seam the HTTP handlers
6//! use — no second control path, so both backends (`LocalFleet`, `K8sFleet`)
7//! work through it unchanged.
8//!
9//! **Auth.** The transport is mounted inside the same router as the REST API, so
10//! the `route_layer` scope check runs first: `/mcp` requires `operate` (#2), the
11//! scope a spawn or kill needs. Read-only tools are reachable with that same
12//! credential rather than a lesser one — the surface is a driving surface, and
13//! splitting per-tool scopes here would put a second, divergent authorization
14//! model behind the one the route already enforces.
15
16use std::sync::Arc;
17
18use rmcp::handler::server::router::tool::ToolRouter;
19use rmcp::handler::server::wrapper::Parameters;
20use rmcp::model::{CallToolResult, Content, ServerCapabilities, ServerInfo};
21use rmcp::{ErrorData, ServerHandler, tool, tool_handler, tool_router};
22use serde::Deserialize;
23use serde_json::{Value, json};
24
25use prospero_core::fleet_provider::FleetProvider;
26use prospero_core::model::{AgentId, DrainPolicy, TaskSpec};
27use prospero_core::store::Store;
28
29/// Mount the MCP server as an axum service at `/mcp` (#218).
30///
31/// Stateful sessions (the transport's default): a client initializes once and
32/// keeps the session for its subsequent calls, which is what an editor-shaped
33/// client expects. The service is built per connection from the shared seams,
34/// so every session sees the same fleet.
35#[must_use]
36pub fn service(
37    fleet: Arc<dyn FleetProvider>,
38    store: Arc<dyn Store>,
39) -> rmcp::transport::streamable_http_server::StreamableHttpService<
40    FleetMcp,
41    rmcp::transport::streamable_http_server::session::local::LocalSessionManager,
42> {
43    rmcp::transport::streamable_http_server::StreamableHttpService::new(
44        move || Ok(FleetMcp::new(fleet.clone(), store.clone())),
45        Arc::new(
46            rmcp::transport::streamable_http_server::session::local::LocalSessionManager::default(),
47        ),
48        rmcp::transport::streamable_http_server::StreamableHttpServerConfig::default(),
49    )
50}
51
52/// The MCP server over one fleet.
53#[derive(Clone)]
54pub struct FleetMcp {
55    fleet: Arc<dyn FleetProvider>,
56    store: Arc<dyn Store>,
57    tool_router: ToolRouter<Self>,
58}
59
60impl FleetMcp {
61    /// Build a server over the same seams the HTTP handlers use.
62    #[must_use]
63    pub fn new(fleet: Arc<dyn FleetProvider>, store: Arc<dyn Store>) -> Self {
64        Self {
65            fleet,
66            store,
67            tool_router: Self::tool_router(),
68        }
69    }
70
71    /// A tool failure the *model* should see and can act on — a bad agent id,
72    /// an unreachable backend — as opposed to a protocol error.
73    fn err(msg: impl Into<String>) -> Result<CallToolResult, ErrorData> {
74        Ok(CallToolResult::error(vec![Content::text(msg.into())]))
75    }
76
77    fn ok(value: Value) -> Result<CallToolResult, ErrorData> {
78        Ok(CallToolResult::success(vec![Content::json(value)?]))
79    }
80}
81
82/// Arguments naming one agent.
83#[derive(Debug, Deserialize, schemars::JsonSchema)]
84pub struct AgentArgs {
85    /// The agent's id, as returned by `prospero_list_agents`.
86    pub agent_id: String,
87}
88
89/// Arguments for a spawn.
90#[derive(Debug, Deserialize, schemars::JsonSchema)]
91pub struct SpawnArgs {
92    /// Workspace to spawn under.
93    pub workspace: String,
94    /// The task for the agent.
95    pub prompt: String,
96    /// Optional human-readable label.
97    #[serde(default)]
98    pub label: Option<String>,
99    /// Optional model override.
100    #[serde(default)]
101    pub model: Option<String>,
102    /// Await operator input instead of finishing at the end of a run.
103    #[serde(default)]
104    pub interactive: bool,
105    /// Kill the agent after this many seconds of wall-clock time (#221).
106    #[serde(default)]
107    pub timeout_secs: Option<u64>,
108}
109
110/// Arguments for reading an agent's history.
111#[derive(Debug, Deserialize, schemars::JsonSchema)]
112pub struct EventsArgs {
113    /// The agent's id.
114    pub agent_id: String,
115    /// Return events with `seq >= from`. Defaults to 0.
116    #[serde(default)]
117    pub from: u64,
118    /// Cap on how many events to return. Defaults to 100, max 500 — a model
119    /// context is finite, and a long run's history is not.
120    #[serde(default)]
121    pub limit: Option<usize>,
122}
123
124/// Arguments for sending input to an interactive agent.
125#[derive(Debug, Deserialize, schemars::JsonSchema)]
126pub struct InputArgs {
127    /// The agent's id.
128    pub agent_id: String,
129    /// Message text to deliver.
130    pub text: String,
131}
132
133/// Default and ceiling for `prospero_agent_events`.
134const EVENTS_DEFAULT: usize = 100;
135const EVENTS_MAX: usize = 500;
136
137#[tool_router]
138impl FleetMcp {
139    #[tool(description = "List the workspaces prospero supervises, with health and agent counts.")]
140    async fn prospero_list_workspaces(&self) -> Result<CallToolResult, ErrorData> {
141        let snap = self.fleet.snapshot().await;
142        let workspaces: Vec<Value> = snap
143            .workspaces
144            .iter()
145            .map(|w| {
146                json!({
147                    "name": w.name,
148                    "healthy": matches!(w.health, prospero_core::model::WorkspaceHealth::Healthy),
149                    "agents": w.agents.len(),
150                })
151            })
152            .collect();
153        Self::ok(json!({ "workspaces": workspaces }))
154    }
155
156    #[tool(description = "List every agent in the fleet with its status and workspace.")]
157    async fn prospero_list_agents(&self) -> Result<CallToolResult, ErrorData> {
158        let snap = self.fleet.snapshot().await;
159        let agents: Vec<Value> = snap
160            .workspaces
161            .iter()
162            .flat_map(|w| w.agents.iter())
163            .map(agent_json)
164            .collect();
165        Self::ok(json!({ "agents": agents }))
166    }
167
168    #[tool(description = "Launch an agent in a workspace; returns its id.")]
169    async fn prospero_spawn_agent(
170        &self,
171        Parameters(args): Parameters<SpawnArgs>,
172    ) -> Result<CallToolResult, ErrorData> {
173        let mut request = prospero_core::fleet::SpawnRequest::new(args.prompt);
174        request.label = args.label;
175        request.model = args.model;
176        request.interactive = args.interactive;
177        request.timeout_secs = args.timeout_secs;
178
179        match self
180            .fleet
181            .ensure_agent(TaskSpec {
182                workspace: args.workspace.clone(),
183                request,
184            })
185            .await
186        {
187            Ok(handle) => Self::ok(json!({
188                "agent_id": handle.id.to_string(),
189                "workspace": args.workspace,
190                "created": handle.created,
191            })),
192            Err(e) => Self::err(format!("spawn failed: {e}")),
193        }
194    }
195
196    #[tool(description = "Read one agent's current state.")]
197    async fn prospero_agent_status(
198        &self,
199        Parameters(args): Parameters<AgentArgs>,
200    ) -> Result<CallToolResult, ErrorData> {
201        let snap = self.fleet.snapshot().await;
202        match snap.find_agent(&args.agent_id) {
203            Some((_, agent)) => Self::ok(agent_json(agent)),
204            None => Self::err(format!("unknown agent: {}", args.agent_id)),
205        }
206    }
207
208    #[tool(description = "Read an agent's recorded events from `from` onward (bounded).")]
209    async fn prospero_agent_events(
210        &self,
211        Parameters(args): Parameters<EventsArgs>,
212    ) -> Result<CallToolResult, ErrorData> {
213        let limit = args.limit.unwrap_or(EVENTS_DEFAULT).min(EVENTS_MAX);
214        let events = match self.store.replay(&args.agent_id, args.from).await {
215            Ok(events) => events,
216            Err(e) => return Self::err(format!("reading history failed: {e}")),
217        };
218        let total = events.len();
219        let page: Vec<&prospero_core::FleetEvent> = events.iter().take(limit).collect();
220        let next = page.last().map(|e| e.seq + 1);
221        Self::ok(json!({
222            "events": page,
223            // A model cannot tell a short page from the end of history without
224            // being told, and would stop reading early.
225            "truncated": total > limit,
226            "next_from": next,
227        }))
228    }
229
230    #[tool(description = "Send a message to an interactive agent, resuming its run.")]
231    async fn prospero_send_input(
232        &self,
233        Parameters(args): Parameters<InputArgs>,
234    ) -> Result<CallToolResult, ErrorData> {
235        let frame = prospero_core::AttachInbound::UserMessage { text: args.text };
236        match self
237            .fleet
238            .send_input(&AgentId::from(args.agent_id.clone()), frame)
239            .await
240        {
241            Ok(()) => Self::ok(json!({ "delivered": true })),
242            Err(e) => Self::err(format!("send failed: {e}")),
243        }
244    }
245
246    #[tool(description = "Signal end-of-input to an interactive agent; it finishes after.")]
247    async fn prospero_end_input(
248        &self,
249        Parameters(args): Parameters<AgentArgs>,
250    ) -> Result<CallToolResult, ErrorData> {
251        match self
252            .fleet
253            .send_input(
254                &AgentId::from(args.agent_id.clone()),
255                prospero_core::AttachInbound::EndInput,
256            )
257            .await
258        {
259            Ok(()) => Self::ok(json!({ "delivered": true })),
260            Err(e) => Self::err(format!("end-input failed: {e}")),
261        }
262    }
263
264    #[tool(description = "Kill a running agent.")]
265    async fn prospero_kill_agent(
266        &self,
267        Parameters(args): Parameters<AgentArgs>,
268    ) -> Result<CallToolResult, ErrorData> {
269        match self
270            .fleet
271            .stop_agent(&AgentId::from(args.agent_id.clone()), DrainPolicy::Kill)
272            .await
273        {
274            Ok(()) => Self::ok(json!({ "killed": args.agent_id })),
275            Err(e) => Self::err(format!("kill failed: {e}")),
276        }
277    }
278
279    #[tool(description = "Kill and respawn an agent with the same spec; returns the new id.")]
280    async fn prospero_respawn_agent(
281        &self,
282        Parameters(args): Parameters<AgentArgs>,
283    ) -> Result<CallToolResult, ErrorData> {
284        match self
285            .fleet
286            .restart_agent(&AgentId::from(args.agent_id.clone()))
287            .await
288        {
289            Ok(id) => Self::ok(json!({ "agent_id": id.to_string() })),
290            Err(e) => Self::err(format!("respawn failed: {e}")),
291        }
292    }
293}
294
295/// One agent as the tools report it.
296fn agent_json(agent: &prospero_core::model::Agent) -> Value {
297    json!({
298        "agent_id": agent.id,
299        "name": agent.name,
300        "workspace": agent.workspace,
301        "status": agent.status,
302        "started_at": agent.started_at,
303        "interactive": agent.interactive,
304        "isolated": agent.isolated,
305        // #241: why the backend put it in this state, when something else did.
306        "reason": agent.reason,
307        "detail": agent.detail,
308    })
309}
310
311#[tool_handler(router = self.tool_router)]
312impl ServerHandler for FleetMcp {
313    fn get_info(&self) -> ServerInfo {
314        // `ServerInfo` is `#[non_exhaustive]`, so build from its default and
315        // set what we mean rather than naming every field.
316        let mut info = ServerInfo::default();
317        info.instructions = Some(
318            "Drive a prospero fleet: list workspaces and agents, spawn, read history, \
319             steer an interactive agent, kill or respawn."
320                .into(),
321        );
322        info.capabilities = ServerCapabilities::builder().enable_tools().build();
323        info
324    }
325}
326
327#[cfg(test)]
328mod tests {
329    use super::*;
330    use prospero_core::LocalFleet;
331    use prospero_core::discovery::{DiscoveryEnv, EnsureConfig};
332    use prospero_core::fleet::{FleetConfig, FleetManager};
333    use prospero_core::store::JsonlStore;
334    use prospero_core::testkit::FakeCaliband;
335
336    /// A server over a real `LocalFleet` talking to a fake caliband — the same
337    /// seam the HTTP handlers use, so a passing tool here works in the daemon.
338    async fn harness() -> (FleetMcp, FleetManager, FakeCaliband, tempfile::TempDir) {
339        let dir = tempfile::tempdir().unwrap();
340        let mut config = FleetConfig::new("mcp-host", dir.path());
341        config.discovery_env = DiscoveryEnv {
342            caliban_daemon_runtime_dir: Some(dir.path().to_path_buf()),
343            xdg_runtime_dir: None,
344            tmpdir: None,
345        };
346        config.ensure = EnsureConfig {
347            autostart: false,
348            ..EnsureConfig::default()
349        };
350        let root = dir.path().join("repo");
351        std::fs::create_dir_all(&root).unwrap();
352        let socket =
353            prospero_core::discovery::resolve_socket(&root, &config.discovery_env).unwrap();
354        let fake = FakeCaliband::start_at(&socket).await.unwrap();
355
356        let store = Arc::new(JsonlStore::open(dir.path()).unwrap());
357        let manager = FleetManager::new(config, store.clone()).await.unwrap();
358        manager.add_repo("repo", &root).await.unwrap();
359
360        let fleet: Arc<dyn FleetProvider> = Arc::new(LocalFleet::new(manager.clone()));
361        (FleetMcp::new(fleet, store), manager, fake, dir)
362    }
363
364    /// The JSON a tool returned, or a panic naming the error the model saw.
365    fn payload(result: CallToolResult) -> Value {
366        assert!(
367            !result.is_error.unwrap_or(false),
368            "tool reported an error: {:?}",
369            result.content
370        );
371        let raw = result
372            .content
373            .first()
374            .and_then(|c| c.as_text().map(|t| t.text.clone()))
375            .expect("a tool result carries content");
376        serde_json::from_str(&raw).expect("tool content is JSON")
377    }
378
379    #[tokio::test]
380    async fn listing_reports_the_workspaces_prospero_supervises() {
381        let (mcp, _manager, _fake, _dir) = harness().await;
382        let v = payload(mcp.prospero_list_workspaces().await.unwrap());
383        let names: Vec<&str> = v["workspaces"]
384            .as_array()
385            .unwrap()
386            .iter()
387            .map(|w| w["name"].as_str().unwrap())
388            .collect();
389        assert_eq!(names, vec!["repo"]);
390    }
391
392    /// The round trip that matters: a tool spawns an agent, and the fleet tools
393    /// then see it — proving the adapter drives the same fleet the REST API does.
394    #[tokio::test]
395    async fn a_spawned_agent_is_visible_to_the_other_tools() {
396        let (mcp, manager, _fake, _dir) = harness().await;
397
398        let spawned = payload(
399            mcp.prospero_spawn_agent(Parameters(SpawnArgs {
400                workspace: "repo".into(),
401                prompt: "do the thing".into(),
402                label: Some("mcp-spawn".into()),
403                model: None,
404                interactive: false,
405                timeout_secs: None,
406            }))
407            .await
408            .unwrap(),
409        );
410        let id = spawned["agent_id"].as_str().expect("an id").to_string();
411        assert_eq!(spawned["workspace"], "repo");
412
413        // The fleet view is the poll snapshot, exactly as `/api/fleet` is: a
414        // spawn is visible from the next poll, not the instant it returns.
415        manager.poll_repo_once("repo").await;
416
417        let listed = payload(mcp.prospero_list_agents().await.unwrap());
418        assert!(
419            listed["agents"]
420                .as_array()
421                .unwrap()
422                .iter()
423                .any(|a| a["agent_id"] == id.as_str()),
424            "the spawned agent must appear in the fleet: {listed}"
425        );
426
427        let status = payload(
428            mcp.prospero_agent_status(Parameters(AgentArgs {
429                agent_id: id.clone(),
430            }))
431            .await
432            .unwrap(),
433        );
434        assert_eq!(status["agent_id"], id.as_str());
435    }
436
437    /// An unknown id is a *tool* error the model can read and recover from, not
438    /// a protocol failure that kills the session.
439    #[tokio::test]
440    async fn an_unknown_agent_is_a_readable_tool_error() {
441        let (mcp, _manager, _fake, _dir) = harness().await;
442        let result = mcp
443            .prospero_agent_status(Parameters(AgentArgs {
444                agent_id: "nope".into(),
445            }))
446            .await
447            .unwrap();
448        assert!(result.is_error.unwrap_or(false), "{result:?}");
449        let text = result
450            .content
451            .first()
452            .and_then(|c| c.as_text().map(|t| t.text.clone()))
453            .unwrap_or_default();
454        assert!(
455            text.contains("nope"),
456            "the message must name the id: {text}"
457        );
458    }
459
460    /// History is capped, and says when it was — a model that could not tell a
461    /// short page from the end of the log would stop reading early.
462    #[tokio::test]
463    async fn agent_events_are_bounded_and_say_so() {
464        let (mcp, _manager, _fake, dir) = harness().await;
465        let store = JsonlStore::open(dir.path()).unwrap();
466        for seq in 1..=5u64 {
467            store
468                .append(&prospero_core::FleetEvent {
469                    seq,
470                    ts: "2026-09-20T00:00:00Z".into(),
471                    repo: "repo".into(),
472                    agent_id: "a1".into(),
473                    kind: prospero_core::EventKind::Output {
474                        stream: prospero_core::event::OutputStream::Stdout,
475                        chunk: format!("chunk {seq}"),
476                    },
477                    actor: None,
478                    on_behalf_of: None,
479                })
480                .await
481                .unwrap();
482        }
483
484        let page = payload(
485            mcp.prospero_agent_events(Parameters(EventsArgs {
486                agent_id: "a1".into(),
487                from: 0,
488                limit: Some(2),
489            }))
490            .await
491            .unwrap(),
492        );
493        assert_eq!(page["events"].as_array().unwrap().len(), 2);
494        assert_eq!(page["truncated"], true);
495        assert_eq!(page["next_from"], 3, "resume after the last one delivered");
496
497        let rest = payload(
498            mcp.prospero_agent_events(Parameters(EventsArgs {
499                agent_id: "a1".into(),
500                from: 3,
501                limit: None,
502            }))
503            .await
504            .unwrap(),
505        );
506        assert_eq!(rest["events"].as_array().unwrap().len(), 3);
507        assert_eq!(rest["truncated"], false);
508    }
509}
prospero_api/mcp.rs

prospero_api/
mcp.rs
