claude_codes/
client_sync.rs

1//! Synchronous client for Claude communication
2
3use crate::cli::ClaudeCliBuilder;
4use crate::error::{Error, Result};
5use crate::io::{
6    ClaudeInput, ClaudeOutput, ContentBlock, ControlRequestMessage, ControlResponse,
7    ControlResponseMessage,
8};
9use crate::protocol::Protocol;
10use log::{debug, warn};
11use serde::{Deserialize, Serialize};
12use std::io::{BufRead, BufReader, Write};
13use std::process::{Child, ChildStdin, ChildStdout};
14use uuid::Uuid;
15
16/// Synchronous client for communicating with Claude
17pub struct SyncClient {
18    child: Child,
19    stdin: ChildStdin,
20    stdout: BufReader<ChildStdout>,
21    session_uuid: Option<Uuid>,
22    /// Whether tool approval protocol has been initialized
23    tool_approval_enabled: bool,
24}
25
26/// Buffer size for reading Claude's stdout (10MB).
27const STDOUT_BUFFER_SIZE: usize = 10 * 1024 * 1024;
28
29impl SyncClient {
30    /// Create a new synchronous client from an existing child process
31    pub fn new(mut child: Child) -> Result<Self> {
32        let stdin = child
33            .stdin
34            .take()
35            .ok_or_else(|| Error::Protocol("Failed to get stdin".to_string()))?;
36        let stdout = child
37            .stdout
38            .take()
39            .ok_or_else(|| Error::Protocol("Failed to get stdout".to_string()))?;
40
41        Ok(Self {
42            child,
43            stdin,
44            stdout: BufReader::with_capacity(STDOUT_BUFFER_SIZE, stdout),
45            session_uuid: None,
46            tool_approval_enabled: false,
47        })
48    }
49
50    /// Create a new synchronous client with default settings
51    pub fn with_defaults() -> Result<Self> {
52        // Check Claude version (only warns once per session)
53        // NOTE: The claude-codes API is in high flux. If you wish to work around
54        // this version check, you can use SyncClient::new() directly with:
55        //   let child = ClaudeCliBuilder::new().spawn_sync()?;
56        //   SyncClient::new(child)
57        crate::version::check_claude_version()?;
58        let child = ClaudeCliBuilder::new().spawn_sync().map_err(Error::Io)?;
59        Self::new(child)
60    }
61
62    /// Resume a previous session by UUID
63    /// This creates a new client that resumes an existing session
64    pub fn resume_session(session_uuid: Uuid) -> Result<Self> {
65        let child = ClaudeCliBuilder::new()
66            .resume(Some(session_uuid.to_string()))
67            .spawn_sync()
68            .map_err(Error::Io)?;
69
70        debug!("Resuming Claude session with UUID: {}", session_uuid);
71        let mut client = Self::new(child)?;
72        // Pre-populate the session UUID since we're resuming
73        client.session_uuid = Some(session_uuid);
74        Ok(client)
75    }
76
77    /// Resume a previous session with a specific model
78    pub fn resume_session_with_model(session_uuid: Uuid, model: &str) -> Result<Self> {
79        let child = ClaudeCliBuilder::new()
80            .model(model)
81            .resume(Some(session_uuid.to_string()))
82            .spawn_sync()
83            .map_err(Error::Io)?;
84
85        debug!(
86            "Resuming Claude session with UUID: {} and model: {}",
87            session_uuid, model
88        );
89        let mut client = Self::new(child)?;
90        // Pre-populate the session UUID since we're resuming
91        client.session_uuid = Some(session_uuid);
92        Ok(client)
93    }
94
95    /// Send a query and collect all responses
96    pub fn query(&mut self, input: ClaudeInput) -> Result<Vec<ClaudeOutput>> {
97        let mut responses = Vec::new();
98        for response in self.query_stream(input)? {
99            responses.push(response?);
100        }
101        Ok(responses)
102    }
103
104    /// Send a query and return an iterator over responses
105    pub fn query_stream(&mut self, input: ClaudeInput) -> Result<ResponseIterator<'_>> {
106        // Send the input
107        Protocol::write_sync(&mut self.stdin, &input)?;
108
109        Ok(ResponseIterator {
110            client: self,
111            finished: false,
112        })
113    }
114
115    /// Read the next response from Claude
116    fn read_next(&mut self) -> Result<Option<ClaudeOutput>> {
117        let mut line = String::new();
118        match self.stdout.read_line(&mut line) {
119            Ok(0) => {
120                debug!("[CLIENT] Stream closed");
121                Ok(None)
122            }
123            Ok(_) => {
124                let trimmed = line.trim();
125                if trimmed.is_empty() {
126                    debug!("[CLIENT] Skipping empty line");
127                    return self.read_next();
128                }
129
130                debug!("[CLIENT] Received: {}", trimmed);
131                match ClaudeOutput::parse_json_tolerant(trimmed) {
132                    Ok(output) => {
133                        // Capture UUID from first response if not already set
134                        if self.session_uuid.is_none() {
135                            if let ClaudeOutput::Assistant(ref msg) = output {
136                                if let Some(ref uuid_str) = msg.uuid {
137                                    if let Ok(uuid) = Uuid::parse_str(uuid_str) {
138                                        debug!("[CLIENT] Captured session UUID: {}", uuid);
139                                        self.session_uuid = Some(uuid);
140                                    }
141                                }
142                            } else if let ClaudeOutput::Result(ref msg) = output {
143                                if let Some(ref uuid_str) = msg.uuid {
144                                    if let Ok(uuid) = Uuid::parse_str(uuid_str) {
145                                        debug!("[CLIENT] Captured session UUID: {}", uuid);
146                                        self.session_uuid = Some(uuid);
147                                    }
148                                }
149                            }
150                        }
151
152                        // Check if this is a result message
153                        if matches!(output, ClaudeOutput::Result(_)) {
154                            debug!("[CLIENT] Received result message, stream complete");
155                            Ok(Some(output))
156                        } else {
157                            Ok(Some(output))
158                        }
159                    }
160                    Err(parse_error) => {
161                        warn!("[CLIENT] Failed to deserialize message from Claude CLI. Please report this at https://github.com/meawoppl/rust-claude-codes/issues with the raw message below.");
162                        warn!("[CLIENT] Parse error: {}", parse_error.error_message);
163                        warn!("[CLIENT] Raw message: {}", trimmed);
164                        Err(parse_error.into())
165                    }
166                }
167            }
168            Err(e) => {
169                debug!("[CLIENT] Error reading from stdout: {}", e);
170                Err(Error::Io(e))
171            }
172        }
173    }
174
175    /// Shutdown the client and wait for the process to exit
176    pub fn shutdown(&mut self) -> Result<()> {
177        debug!("[CLIENT] Shutting down client");
178        self.child.kill().map_err(Error::Io)?;
179        self.child.wait().map_err(Error::Io)?;
180        Ok(())
181    }
182
183    /// Get the session UUID if available
184    /// Returns an error if no response has been received yet
185    pub fn session_uuid(&self) -> Result<Uuid> {
186        self.session_uuid.ok_or(Error::SessionNotInitialized)
187    }
188
189    /// Test if the Claude connection is working by sending a ping message
190    /// Returns true if Claude responds with "pong", false otherwise
191    pub fn ping(&mut self) -> bool {
192        // Send a simple ping request
193        let ping_input = ClaudeInput::user_message(
194            "ping - respond with just the word 'pong' and nothing else",
195            self.session_uuid.unwrap_or_else(Uuid::new_v4),
196        );
197
198        // Try to send the ping and get responses
199        match self.query(ping_input) {
200            Ok(responses) => {
201                // Check all responses for "pong"
202                for output in responses {
203                    if let ClaudeOutput::Assistant(msg) = &output {
204                        for content in &msg.message.content {
205                            if let ContentBlock::Text(text) = content {
206                                if text.text.to_lowercase().contains("pong") {
207                                    return true;
208                                }
209                            }
210                        }
211                    }
212                }
213                false
214            }
215            Err(e) => {
216                debug!("Ping failed: {}", e);
217                false
218            }
219        }
220    }
221
222    // =========================================================================
223    // Tool Approval Protocol
224    // =========================================================================
225
226    /// Enable the tool approval protocol by performing the initialization handshake.
227    ///
228    /// After calling this method, the CLI will send `ControlRequest` messages when
229    /// Claude wants to use a tool. You must handle these by calling
230    /// `send_control_response()` with an appropriate response.
231    ///
232    /// **Important**: The client must have been created with
233    /// `ClaudeCliBuilder::permission_prompt_tool("stdio")` for this to work.
234    ///
235    /// # Example
236    ///
237    /// ```no_run
238    /// use claude_codes::{SyncClient, ClaudeCliBuilder, ClaudeOutput, ControlRequestPayload};
239    ///
240    /// # fn example() -> Result<(), Box<dyn std::error::Error>> {
241    /// let child = ClaudeCliBuilder::new()
242    ///     .model("sonnet")
243    ///     .permission_prompt_tool("stdio")
244    ///     .spawn_sync()?;
245    ///
246    /// let mut client = SyncClient::new(child)?;
247    /// client.enable_tool_approval()?;
248    ///
249    /// // Now when you receive messages, you may get ControlRequest messages
250    /// // that need responses
251    /// # Ok(())
252    /// # }
253    /// ```
254    pub fn enable_tool_approval(&mut self) -> Result<()> {
255        if self.tool_approval_enabled {
256            debug!("[TOOL_APPROVAL] Already enabled, skipping initialization");
257            return Ok(());
258        }
259
260        let request_id = format!("init-{}", Uuid::new_v4());
261        let init_request = ControlRequestMessage::initialize(&request_id);
262
263        debug!("[TOOL_APPROVAL] Sending initialization handshake");
264        Protocol::write_sync(&mut self.stdin, &init_request)?;
265
266        // Wait for the initialization response
267        loop {
268            let mut line = String::new();
269            let bytes_read = self.stdout.read_line(&mut line).map_err(Error::Io)?;
270
271            if bytes_read == 0 {
272                return Err(Error::ConnectionClosed);
273            }
274
275            let trimmed = line.trim();
276            if trimmed.is_empty() {
277                continue;
278            }
279
280            debug!("[TOOL_APPROVAL] Received: {}", trimmed);
281
282            // Try to parse as ClaudeOutput
283            match ClaudeOutput::parse_json_tolerant(trimmed) {
284                Ok(ClaudeOutput::ControlResponse(resp)) => {
285                    use crate::io::ControlResponsePayload;
286                    match &resp.response {
287                        ControlResponsePayload::Success {
288                            request_id: rid, ..
289                        } if rid == &request_id => {
290                            debug!("[TOOL_APPROVAL] Initialization successful");
291                            self.tool_approval_enabled = true;
292                            return Ok(());
293                        }
294                        ControlResponsePayload::Error { error, .. } => {
295                            return Err(Error::Protocol(format!(
296                                "Tool approval initialization failed: {}",
297                                error
298                            )));
299                        }
300                        _ => {
301                            // Different request_id, keep waiting
302                            continue;
303                        }
304                    }
305                }
306                Ok(_) => {
307                    // Got a different message type (system, etc.), keep waiting
308                    continue;
309                }
310                Err(e) => {
311                    return Err(e.into());
312                }
313            }
314        }
315    }
316
317    /// Send a control response back to the CLI.
318    ///
319    /// Use this to respond to `ControlRequest` messages received during tool approval.
320    /// The easiest way to create responses is using the helper methods on
321    /// `ToolPermissionRequest`:
322    ///
323    /// # Example
324    ///
325    /// ```ignore
326    /// use claude_codes::{SyncClient, ControlRequestPayload, ControlResponse, ToolPermissionRequest};
327    ///
328    /// fn handle_permission(
329    ///     client: &mut SyncClient,
330    ///     perm_req: &ToolPermissionRequest,
331    ///     request_id: &str,
332    /// ) -> claude_codes::Result<()> {
333    ///     let response = if perm_req.tool_name == "Bash" {
334    ///         perm_req.deny("Bash commands not allowed", request_id)
335    ///     } else {
336    ///         perm_req.allow(request_id)
337    ///     };
338    ///     client.send_control_response(response)
339    /// }
340    /// ```
341    pub fn send_control_response(&mut self, response: ControlResponse) -> Result<()> {
342        let message: ControlResponseMessage = response.into();
343        debug!(
344            "[TOOL_APPROVAL] Sending control response: {:?}",
345            serde_json::to_string(&message)
346        );
347        Protocol::write_sync(&mut self.stdin, &message)?;
348        Ok(())
349    }
350
351    /// Check if tool approval protocol is enabled
352    pub fn is_tool_approval_enabled(&self) -> bool {
353        self.tool_approval_enabled
354    }
355}
356
357// Protocol extension methods for synchronous I/O
358impl Protocol {
359    /// Write a message to a synchronous writer
360    pub fn write_sync<W: Write, T: Serialize>(writer: &mut W, message: &T) -> Result<()> {
361        let line = Self::serialize(message)?;
362        debug!("[PROTOCOL] Sending: {}", line.trim());
363        writer.write_all(line.as_bytes())?;
364        writer.flush()?;
365        Ok(())
366    }
367
368    /// Read a message from a synchronous reader
369    pub fn read_sync<R: BufRead, T: for<'de> Deserialize<'de>>(reader: &mut R) -> Result<T> {
370        let mut line = String::new();
371        let bytes_read = reader.read_line(&mut line)?;
372        if bytes_read == 0 {
373            return Err(Error::ConnectionClosed);
374        }
375        debug!("[PROTOCOL] Received: {}", line.trim());
376        Self::deserialize(&line)
377    }
378}
379
380/// Iterator over responses from Claude
381pub struct ResponseIterator<'a> {
382    client: &'a mut SyncClient,
383    finished: bool,
384}
385
386impl Iterator for ResponseIterator<'_> {
387    type Item = Result<ClaudeOutput>;
388
389    fn next(&mut self) -> Option<Self::Item> {
390        if self.finished {
391            return None;
392        }
393
394        match self.client.read_next() {
395            Ok(Some(output)) => {
396                // Check if this is a result message
397                if matches!(output, ClaudeOutput::Result(_)) {
398                    self.finished = true;
399                }
400                Some(Ok(output))
401            }
402            Ok(None) => {
403                self.finished = true;
404                None
405            }
406            Err(e) => {
407                self.finished = true;
408                Some(Err(e))
409            }
410        }
411    }
412}
413
414impl Drop for SyncClient {
415    fn drop(&mut self) {
416        if let Err(e) = self.shutdown() {
417            debug!("[CLIENT] Error during shutdown: {}", e);
418        }
419    }
420}
421
422/// Stream processor for handling continuous message streams
423pub struct StreamProcessor<R> {
424    reader: BufReader<R>,
425}
426
427impl<R: std::io::Read> StreamProcessor<R> {
428    /// Create a new stream processor
429    pub fn new(reader: R) -> Self {
430        Self {
431            reader: BufReader::new(reader),
432        }
433    }
434
435    /// Process the next message from the stream
436    pub fn next_message<T: for<'de> Deserialize<'de>>(&mut self) -> Result<T> {
437        Protocol::read_sync(&mut self.reader)
438    }
439
440    /// Process all messages in the stream
441    pub fn process_all<T, F>(&mut self, mut handler: F) -> Result<()>
442    where
443        T: for<'de> Deserialize<'de>,
444        F: FnMut(T) -> Result<()>,
445    {
446        loop {
447            match self.next_message() {
448                Ok(message) => handler(message)?,
449                Err(Error::ConnectionClosed) => break,
450                Err(e) => return Err(e),
451            }
452        }
453        Ok(())
454    }
455}
claude_codes/client_sync.rs

claude_codes/
client_sync.rs
