| 90 | } |
| 91 | |
| 92 | async fn receive(&self) -> Result<Option<JsonRpcMessage>, McpClientError> { |
| 93 | let mut process_guard = self.process.lock().await; |
| 94 | let child = process_guard |
| 95 | .as_mut() |
| 96 | .ok_or_else(|| McpClientError::TransportError("Process not running".to_string()))?; |
| 97 | |
| 98 | let stdout = child |
| 99 | .stdout |
| 100 | .as_mut() |
| 101 | .ok_or_else(|| McpClientError::TransportError("No stdout".to_string()))?; |
| 102 | |
| 103 | let mut reader = BufReader::new(stdout); |
| 104 | let mut header_line = String::new(); |
| 105 | loop { |
| 106 | header_line.clear(); |
| 107 | let bytes_read = reader.read_line(&mut header_line).await.map_err(|e| { |
| 108 | McpClientError::TransportError(format!("Failed to read header: {}", e)) |
| 109 | })?; |
| 110 | |
| 111 | if bytes_read == 0 { |
| 112 | return Ok(None); |
| 113 | } |
| 114 | |
| 115 | let trimmed = header_line.trim(); |
| 116 | |
| 117 | if trimmed.is_empty() { |
| 118 | break; |
| 119 | } |
| 120 | |
| 121 | if trimmed.starts_with("Content-Length:") { |
| 122 | let content_length: usize = trimmed[15..].trim().parse().map_err(|e| { |
| 123 | McpClientError::ProtocolError(format!("Invalid content length: {}", e)) |
| 124 | })?; |
| 125 | |
| 126 | let mut content_buf = vec![0u8; content_length]; |
| 127 | reader.read_exact(&mut content_buf).await.map_err(|e| { |
| 128 | McpClientError::TransportError(format!("Failed to read content: {}", e)) |
| 129 | })?; |
| 130 | |
| 131 | let content = String::from_utf8_lossy(&content_buf); |
| 132 | let message = JsonRpcMessage::from_str(&content).map_err(|e| { |
| 133 | McpClientError::ProtocolError(format!("Failed to parse message: {}", e)) |
| 134 | })?; |
| 135 | |
| 136 | return Ok(Some(message)); |
| 137 | } |
| 138 | } |
| 139 | |
| 140 | Ok(None) |
| 141 | } |
| 142 | |
| 143 | async fn close(&self) -> Result<(), McpClientError> { |
| 144 | let mut process_guard = self.process.lock().await; |