Method send_streaming

core/src/llm/anthropic.rs:327–611

Execute a fully-built streaming request body (sets `stream: true`).

(
        &self,
        mut request_body: serde_json::Value,
        cancel_token: CancellationToken,
    )

impl AnthropicClient {
    /// Execute a fully-built streaming request body (sets `stream: true`).
    async fn send_streaming(
        &self,
        mut request_body: serde_json::Value,
        cancel_token: CancellationToken,
    ) -> Result<mpsc::Receiver<StreamEvent>> {
        {
            let request_started_at = Instant::now();
            request_body["stream"] = serde_json::json!(true);

            let url = format!("{}/v1/messages", self.base_url);

            let headers = vec![
                ("x-api-key", self.api_key.expose()),
                ("anthropic-version", "2023-06-01"),
                ("anthropic-beta", "prompt-caching-2024-07-31"),
            ];

            let streaming_resp = crate::retry::with_retry(&self.retry_config, |_attempt| {
                let http = &self.http;
                let url = &url;
                let headers = headers.clone();
                let request_body = &request_body;
                let cancel_token = cancel_token.clone();
                async move {
                    let resp = tokio::select! {
                        _ = cancel_token.cancelled() => {
                            return AttemptOutcome::Fatal(anyhow::anyhow!("HTTP request cancelled"));
                        }
                        result = http.post_streaming(url, headers, request_body, cancel_token.clone()) => {
                            match result {
                                Ok(r) => r,
                                Err(e) => {
                                    // A transient network error (timeout, reset,
                                    // mid-flight drop — common on throttled
                                    // endpoints) carries no HTTP status. Retry it
                                    // with backoff like 429/5xx instead of failing
                                    // the turn; a real fatal error still bails.
                                    return if crate::retry::is_transient_error(&e) {
                                        AttemptOutcome::Retryable {
                                            status: reqwest::StatusCode::SERVICE_UNAVAILABLE,
                                            body: format!("network error: {e}"),
                                            retry_after: None,
                                        }
                                    } else {
                                        AttemptOutcome::Fatal(anyhow::anyhow!(
                                            "HTTP request failed: {}",
                                            e
                                        ))
                                    };
                                }
                            }
                        }
                    };
                    let status = reqwest::StatusCode::from_u16(resp.status)
                        .unwrap_or(reqwest::StatusCode::INTERNAL_SERVER_ERROR);
                    if status.is_success() {
                        AttemptOutcome::Success(resp)
                    } else {