| 107 | } |
| 108 | |
| 109 | async fn complete_streaming( |
| 110 | &self, |
| 111 | _messages: &[Message], |
| 112 | _system: Option<&str>, |
| 113 | _tools: &[ToolDefinition], |
| 114 | _cancel_token: CancellationToken, |
| 115 | ) -> anyhow::Result<mpsc::Receiver<StreamEvent>> { |
| 116 | let mut responses = self.responses.lock().unwrap(); |
| 117 | if responses.is_empty() { |
| 118 | anyhow::bail!("No more mock responses"); |
| 119 | } |
| 120 | let response = responses.remove(0); |
| 121 | let (tx, rx) = mpsc::channel(10); |
| 122 | tokio::spawn(async move { |
| 123 | for block in &response.message.content { |
| 124 | if let ContentBlock::ToolUse { name, input, .. } = block { |
| 125 | tx.send(StreamEvent::ToolUseStart { |
| 126 | id: "call_001".to_string(), |
| 127 | name: name.clone(), |
| 128 | }) |
| 129 | .await |
| 130 | .ok(); |
| 131 | let json_str = serde_json::to_string(input).unwrap(); |
| 132 | for chunk in json_str.as_bytes().chunks(10) { |
| 133 | let s = String::from_utf8_lossy(chunk).to_string(); |
| 134 | tx.send(StreamEvent::ToolUseInputDelta(s)).await.ok(); |
| 135 | } |
| 136 | } else if let ContentBlock::Text { text } = block { |
| 137 | for chunk in text.as_bytes().chunks(10) { |
| 138 | let s = String::from_utf8_lossy(chunk).to_string(); |
| 139 | tx.send(StreamEvent::TextDelta(s)).await.ok(); |
| 140 | } |
| 141 | } |
| 142 | } |
| 143 | tx.send(StreamEvent::Done(response)).await.ok(); |
| 144 | }); |
| 145 | Ok(rx) |
| 146 | } |
| 147 | } |
| 148 | |
| 149 | // ======================================================================== |