origin_mcp_http/
transport.rs

1//! The loopback HTTP server and a small client for it.
2//!
3//! One request per connection: read a complete request, answer it, close. The MCP
4//! client sends its next request on a fresh connection. That keeps the server tiny and
5//! stateless without needing a connection pool.
6
7use crate::activity::Activity;
8use crate::discovery::Discovery;
9use crate::http::{HttpRequest, HttpResponse, parse_request};
10use crate::{MCP_PATH, Token};
11use origin_domain::{AppError, Result};
12use origin_mcp_core::McpServer;
13use std::net::SocketAddr;
14use std::sync::Arc;
15use tokio::io::{AsyncReadExt, AsyncWriteExt};
16use tokio::net::{TcpListener, TcpStream};
17use tokio_util::sync::CancellationToken;
18
19/// A bound loopback endpoint, ready to serve.
20#[derive(Debug)]
21pub struct HttpTransport {
22    listener: TcpListener,
23    addr: SocketAddr,
24    token: Option<Token>,
25    /// When the endpoint last served a request (G18).
26    activity: Activity,
27}
28
29impl HttpTransport {
30    /// Bind `127.0.0.1:0`. The OS picks the port, so two instances never collide.
31    pub async fn bind(token: Option<Token>) -> Result<Self> {
32        let listener = TcpListener::bind(("127.0.0.1", 0))
33            .await
34            .map_err(|error| AppError::internal(format!("cannot bind loopback: {error}")))?;
35        let addr = listener
36            .local_addr()
37            .map_err(|error| AppError::internal(format!("cannot read local address: {error}")))?;
38
39        tracing::debug!(%addr, "mcp http endpoint bound");
40        Ok(Self {
41            listener,
42            addr,
43            token,
44            activity: Activity::default(),
45        })
46    }
47
48    pub fn addr(&self) -> SocketAddr {
49        self.addr
50    }
51
52    /// Activity of this endpoint, for a host that shows an "AI connected" indicator
53    /// (G18). Cloneable and safe to read from another task.
54    pub fn activity(&self) -> Activity {
55        self.activity.clone()
56    }
57
58    pub fn url(&self) -> String {
59        format!("http://{}{}", self.addr, MCP_PATH)
60    }
61
62    pub fn token(&self) -> Option<&Token> {
63        self.token.as_ref()
64    }
65
66    /// The record a client reads to find this endpoint.
67    pub fn discovery(&self) -> Discovery {
68        Discovery {
69            url: self.url(),
70            token: self
71                .token
72                .as_ref()
73                .map(|token| token.expose().to_owned())
74                .unwrap_or_default(),
75        }
76    }
77
78    /// Serve until `stop` is cancelled.
79    ///
80    /// The server is shared by `Arc` rather than cloned per connection: an MCP
81    /// session spans several HTTP requests, so the lifecycle state must survive
82    /// between them. (Stdio clones per session because there the connection *is* the
83    /// session.)
84    pub async fn serve(self, server: Arc<McpServer>, stop: CancellationToken) -> Result<()> {
85        tracing::info!(addr = %self.addr, "mcp http endpoint serving");
86
87        loop {
88            tokio::select! {
89                _ = stop.cancelled() => break,
90                accepted = self.listener.accept() => {
91                    match accepted {
92                        Ok((stream, _peer)) => {
93                            let server = server.clone();
94                            let token = self.token.clone();
95                            let activity = self.activity.clone();
96                            tokio::spawn(async move {
97                                if let Err(error) = serve_connection(stream, &server, token.as_ref(), &activity).await {
98                                    tracing::warn!(%error, "mcp http connection failed");
99                                }
100                            });
101                        }
102                        Err(error) => tracing::warn!(%error, "mcp http accept failed"),
103                    }
104                }
105            }
106        }
107
108        tracing::debug!("mcp http endpoint stopped");
109        Ok(())
110    }
111}
112
113async fn serve_connection(
114    mut stream: TcpStream,
115    server: &McpServer,
116    token: Option<&Token>,
117    activity: &Activity,
118) -> Result<()> {
119    let mut buffer = Vec::new();
120    let mut chunk = [0u8; 4096];
121
122    loop {
123        let read = stream
124            .read(&mut chunk)
125            .await
126            .map_err(|error| AppError::internal(format!("cannot read request: {error}")))?;
127
128        // Client closed before sending a complete request: nothing to answer.
129        if read == 0 {
130            return Ok(());
131        }
132
133        buffer.extend_from_slice(&chunk[..read]);
134
135        if let Some((request, _consumed)) = parse_request(&buffer)? {
136            // A complete request arrived: this is what "a session is active" means.
137            activity.touch();
138            let response = handle(server, token, &request).await;
139            stream
140                .write_all(&response.encode())
141                .await
142                .map_err(|error| AppError::internal(format!("cannot write response: {error}")))?;
143            stream
144                .flush()
145                .await
146                .map_err(|error| AppError::internal(format!("cannot flush response: {error}")))?;
147            return Ok(());
148        }
149    }
150}
151
152/// Answer one request. Pure apart from the server call, so it is unit-testable.
153pub async fn handle(
154    server: &McpServer,
155    token: Option<&Token>,
156    request: &HttpRequest,
157) -> HttpResponse {
158    if request.path != MCP_PATH {
159        return HttpResponse::text(404, format!("unknown path `{}`", request.path));
160    }
161
162    if request.method != "POST" {
163        return HttpResponse::text(405, "MCP uses POST");
164    }
165
166    if let Some(expected) = token {
167        match request.bearer_token() {
168            Some(candidate) if expected.matches(candidate) => {}
169            _ => {
170                tracing::warn!("mcp http request rejected: missing or wrong bearer token");
171                return HttpResponse::text(401, "unauthorized");
172            }
173        }
174    }
175
176    let body = match std::str::from_utf8(&request.body) {
177        Ok(body) => body,
178        Err(_) => return HttpResponse::text(400, "request body is not utf-8"),
179    };
180
181    match server.handle_line(body).await {
182        Some(response) => {
183            let value = serde_json::to_value(&response).unwrap_or(serde_json::Value::Null);
184            HttpResponse::json(200, &value)
185        }
186        // A notification: MCP expects no response body.
187        None => HttpResponse {
188            status: 202,
189            body: Vec::new(),
190        },
191    }
192}
193
194/// Send one JSON-RPC message to an endpoint and return the response body.
195///
196/// Used by the headless start to proxy to a running GUI (G17), and by tests.
197pub async fn post(url: &str, token: Option<&str>, body: &str) -> Result<String> {
198    let (host, port, path) = split_url(url)?;
199
200    let mut stream = TcpStream::connect((host.as_str(), port))
201        .await
202        .map_err(|error| AppError::internal(format!("cannot connect to {url}: {error}")))?;
203
204    let mut request = format!(
205        "POST {path} HTTP/1.1\r\nHost: {host}:{port}\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n",
206        body.len()
207    );
208    if let Some(value) = token {
209        request.push_str("Authorization: Bearer ");
210        request.push_str(value);
211        request.push_str("\r\n");
212    }
213    request.push_str("\r\n");
214
215    stream
216        .write_all(request.as_bytes())
217        .await
218        .map_err(|error| AppError::internal(format!("cannot write request: {error}")))?;
219    stream
220        .write_all(body.as_bytes())
221        .await
222        .map_err(|error| AppError::internal(format!("cannot write body: {error}")))?;
223    stream
224        .flush()
225        .await
226        .map_err(|error| AppError::internal(format!("cannot flush: {error}")))?;
227
228    let mut response = Vec::new();
229    stream
230        .read_to_end(&mut response)
231        .await
232        .map_err(|error| AppError::internal(format!("cannot read response: {error}")))?;
233
234    let text =
235        String::from_utf8(response).map_err(|_| AppError::internal("response is not utf-8"))?;
236
237    let (head, body) = text
238        .split_once("\r\n\r\n")
239        .ok_or_else(|| AppError::internal("malformed http response"))?;
240
241    let status = head
242        .lines()
243        .next()
244        .and_then(|line| line.split_whitespace().nth(1))
245        .and_then(|code| code.parse::<u16>().ok())
246        .unwrap_or(0);
247
248    // 200 carries the JSON-RPC response; 202 means the message was a notification,
249    // for which MCP expects no body.
250    if status != 200 && status != 202 {
251        return Err(AppError::internal(format!(
252            "endpoint answered with status {status}: {body}"
253        )));
254    }
255
256    Ok(body.to_owned())
257}
258
259fn split_url(url: &str) -> Result<(String, u16, String)> {
260    let rest = url
261        .strip_prefix("http://")
262        .ok_or_else(|| AppError::validation(format!("unsupported url scheme: {url}")))?;
263
264    let (authority, path) = match rest.split_once('/') {
265        Some((authority, path)) => (authority, format!("/{path}")),
266        None => (rest, "/".to_owned()),
267    };
268
269    let (host, port) = authority
270        .split_once(':')
271        .ok_or_else(|| AppError::validation(format!("url has no port: {url}")))?;
272
273    let port = port
274        .parse::<u16>()
275        .map_err(|_| AppError::validation(format!("invalid port in {url}")))?;
276
277    Ok((host.to_owned(), port, path))
278}
279
280#[cfg(test)]
281mod tests {
282    use super::*;
283
284    #[test]
285    fn a_url_splits_into_host_port_and_path() {
286        let (host, port, path) = split_url("http://127.0.0.1:54321/mcp").unwrap();
287        assert_eq!(host, "127.0.0.1");
288        assert_eq!(port, 54321);
289        assert_eq!(path, "/mcp");
290    }
291
292    #[test]
293    fn a_url_without_a_port_is_rejected() {
294        assert!(split_url("http://127.0.0.1/mcp").is_err());
295    }
296}
origin_mcp_http/transport.rs

origin_mcp_http/
transport.rs
