| 154 | #[async_trait] |
| 155 | impl McpTransport for StdioTransport { |
| 156 | async fn request(&self, request: JsonRpcRequest) -> Result<JsonRpcResponse> { |
| 157 | if !self.connected.load(Ordering::SeqCst) { |
| 158 | return Err(anyhow!("Transport not connected")); |
| 159 | } |
| 160 | |
| 161 | // Create response channel |
| 162 | let (tx, rx) = oneshot::channel(); |
| 163 | let request_id = request.id; |
| 164 | |
| 165 | // Register pending request |
| 166 | { |
| 167 | let mut pending = self.pending.write().await; |
| 168 | pending.insert(request_id, tx); |
| 169 | } |
| 170 | |
| 171 | // Serialize and send request |
| 172 | let msg = serde_json::to_string(&request)? + "\n"; |
| 173 | self.stdin_tx |
| 174 | .send(msg) |
| 175 | .await |
| 176 | .map_err(|_| anyhow!("Failed to send request"))?; |
| 177 | |
| 178 | // Wait for response with timeout |
| 179 | let response = match tokio::time::timeout( |
| 180 | std::time::Duration::from_secs(self.request_timeout_secs), |
| 181 | rx, |
| 182 | ) |
| 183 | .await |
| 184 | { |
| 185 | Ok(Ok(resp)) => resp, |
| 186 | Ok(Err(_)) => { |
| 187 | // Channel closed — clean up pending entry |
| 188 | self.pending.write().await.remove(&request_id); |
| 189 | return Err(anyhow!("Response channel closed")); |
| 190 | } |
| 191 | Err(_) => { |
| 192 | // Timeout — clean up pending entry to prevent memory leak |
| 193 | self.pending.write().await.remove(&request_id); |
| 194 | return Err(anyhow!( |
| 195 | "MCP request timed out after {}s", |
| 196 | self.request_timeout_secs |
| 197 | )); |
| 198 | } |
| 199 | }; |
| 200 | |
| 201 | Ok(response) |
| 202 | } |
| 203 | |
| 204 | async fn notify(&self, notification: JsonRpcNotification) -> Result<()> { |
| 205 | if !self.connected.load(Ordering::SeqCst) { |