clawft_kernel/workload_runtime/
fleet_mcp.rs

1//! MCP-over-HTTP transport for [`super::fleet_inventory`]: the Cognitum
2//! cloud MCP (`POST https://api.cognitum.one/v1/mcp`, streamable HTTP,
3//! OAuth bearer). Live use is owner-run: no test or agent calls the real
4//! cloud. Tests run against a local stub that speaks the same JSON-RPC.
5//!
6//! The OAuth token comes from the operator secret store
7//! ([`OAuthTokens`]) per call and is only placed in an `Authorization`
8//! header; errors are rendered without the request.
9
10use std::time::Duration;
11
12use async_trait::async_trait;
13use clawft_types::secret::SecretString;
14use serde_json::{Value, json};
15
16use super::fleet_inventory::{FleetError, FleetMcp};
17
18/// The Cognitum cloud MCP endpoint.
19pub const COGNITUM_MCP_URL: &str = "https://api.cognitum.one/v1/mcp";
20const TIMEOUT: Duration = Duration::from_secs(30);
21const MAX_REPLY_BYTES: usize = 4 * 1024 * 1024;
22
23/// Where the OAuth token lives (the operator secret store).
24pub trait OAuthTokens: Send + Sync {
25    /// The current access token.
26    fn access_token(&self) -> Result<SecretString, FleetError>;
27}
28
29/// reqwest-backed MCP client. Crate-private on purpose: it can call any
30/// tool, so the only public way to use it is through
31/// [`ReadOnlyFleet::http`](super::fleet_inventory::ReadOnlyFleet::http),
32/// which refuses everything but the read tools.
33pub(crate) struct HttpFleetMcp {
34    url: String,
35    client: reqwest::Client,
36    tokens: Box<dyn OAuthTokens>,
37}
38
39impl HttpFleetMcp {
40    /// Client for `url`: `https://`, or plain `http://` to a loopback stub.
41    pub(crate) fn new(url: &str, tokens: Box<dyn OAuthTokens>) -> Result<Self, FleetError> {
42        let bad = || {
43            FleetError::Transport(
44                "the fleet MCP URL must be https:// (or http:// to localhost, 127.0.0.1 or ::1)"
45                    .into(),
46            )
47        };
48        let parsed = reqwest::Url::parse(url).map_err(|_| bad())?;
49        // `Url` normalises the host, so the comparison is exact: a prefix
50        // such as `127.0.0.1.evil.com` is a different host.
51        let local = matches!(parsed.host_str(), Some("localhost" | "127.0.0.1" | "[::1]"));
52        let ok = match parsed.scheme() {
53            "https" => parsed.host_str().is_some_and(|h| !h.is_empty()),
54            "http" => local && parsed.username().is_empty(),
55            _ => false,
56        };
57        if !ok {
58            return Err(bad());
59        }
60        let client = reqwest::Client::builder()
61            .redirect(reqwest::redirect::Policy::none())
62            .timeout(TIMEOUT)
63            .build()
64            .map_err(|e| FleetError::Transport(format!("http client: {e}")))?;
65        Ok(Self {
66            url: url.to_string(),
67            client,
68            tokens,
69        })
70    }
71
72    async fn rpc(
73        &self,
74        session: Option<&str>,
75        body: Value,
76    ) -> Result<(Option<String>, Value), FleetError> {
77        let token = self.tokens.access_token()?;
78        let mut req = self
79            .client
80            .post(&self.url)
81            .header("accept", "application/json, text/event-stream")
82            .bearer_auth(token.expose())
83            .json(&body);
84        if let Some(s) = session {
85            req = req.header("mcp-session-id", s);
86        }
87        let fail = |e: reqwest::Error| FleetError::Transport(e.without_url().to_string());
88        let resp = req.send().await.map_err(fail)?;
89        let status = resp.status().as_u16();
90        if status == 401 || status == 403 {
91            return Err(FleetError::Unauthorized(status));
92        }
93        if !(200..300).contains(&status) {
94            return Err(FleetError::Transport(format!("HTTP {status}")));
95        }
96        let sid = resp
97            .headers()
98            .get("mcp-session-id")
99            .and_then(|v| v.to_str().ok())
100            .map(str::to_string);
101        let sse = resp
102            .headers()
103            .get("content-type")
104            .and_then(|v| v.to_str().ok())
105            .is_some_and(|t| t.starts_with("text/event-stream"));
106        let too_large = || FleetError::Transport("reply too large".into());
107        if resp
108            .content_length()
109            .is_some_and(|n| n > MAX_REPLY_BYTES as u64)
110        {
111            return Err(too_large());
112        }
113        // Chunk by chunk, so the cap bounds memory even without a (or with
114        // a lying) Content-Length.
115        let mut resp = resp;
116        let mut bytes = Vec::new();
117        while let Some(chunk) = resp.chunk().await.map_err(fail)? {
118            if bytes.len() + chunk.len() > MAX_REPLY_BYTES {
119                return Err(too_large());
120            }
121            bytes.extend_from_slice(&chunk);
122        }
123        let text = String::from_utf8_lossy(&bytes);
124        let payload = if sse {
125            text.lines()
126                .filter_map(|l| l.strip_prefix("data:"))
127                .map(str::trim)
128                .find(|d| d.starts_with('{'))
129                .unwrap_or("")
130                .to_string()
131        } else {
132            text.into_owned()
133        };
134        let v = if payload.trim().is_empty() {
135            Value::Null
136        } else {
137            serde_json::from_str(&payload)
138                .map_err(|_| FleetError::Transport("reply is not JSON-RPC".into()))?
139        };
140        Ok((sid, v))
141    }
142}
143
144#[async_trait]
145impl FleetMcp for HttpFleetMcp {
146    async fn call_tool(&self, name: &str, args: Value) -> Result<Value, FleetError> {
147        let (sid, init) = self
148            .rpc(
149                None,
150                json!({"jsonrpc": "2.0", "id": 1, "method": "initialize", "params": {
151                    "protocolVersion": "2025-03-26", "capabilities": {},
152                    "clientInfo": {"name": "weftos-fleet-inventory", "version": "1"}}}),
153            )
154            .await?;
155        if let Some(e) = init.get("error") {
156            return Err(FleetError::Transport(format!("initialize: {}", brief(e))));
157        }
158        self.rpc(
159            sid.as_deref(),
160            json!({"jsonrpc": "2.0", "method": "notifications/initialized"}),
161        )
162        .await?;
163        let (_, reply) = self
164            .rpc(
165                sid.as_deref(),
166                json!({"jsonrpc": "2.0", "id": 2, "method": "tools/call",
167                       "params": {"name": name, "arguments": args}}),
168            )
169            .await?;
170        if let Some(e) = reply.get("error") {
171            return Err(FleetError::Transport(format!("{name}: {}", brief(e))));
172        }
173        let result = reply
174            .get("result")
175            .ok_or_else(|| FleetError::Schema("no result".into()))?;
176        if result.get("isError").and_then(Value::as_bool) == Some(true) {
177            return Err(FleetError::Transport(format!("{name} reported an error")));
178        }
179        if let Some(s) = result.get("structuredContent") {
180            return Ok(s.clone());
181        }
182        let text = result
183            .get("content")
184            .and_then(Value::as_array)
185            .and_then(|c| c.iter().find_map(|p| p.get("text").and_then(Value::as_str)))
186            .ok_or_else(|| FleetError::Schema("no content text".into()))?;
187        serde_json::from_str(text).map_err(|_| FleetError::Schema("content is not JSON".into()))
188    }
189}
190
191fn brief(e: &Value) -> String {
192    e.get("message")
193        .and_then(Value::as_str)
194        .unwrap_or("error")
195        .chars()
196        .take(200)
197        .collect()
198}
clawft_kernel/workload_runtime/fleet_mcp.rs

clawft_kernel/workload_runtime/
fleet_mcp.rs
