qcli/provider/
endpoint.rs

1//! Where requests go and how they are carried there.
2
3use std::io::BufReader;
4use std::time::Duration;
5
6use serde_json::Value;
7
8use super::reply::{self, Delta, Reply};
9use super::{Key, ProviderError, error};
10
11/// How long connecting may take. A server that does not answer within this is not there.
12const CONNECT: Duration = Duration::from_secs(30);
13
14/// How long the server may think before it starts answering. Generous rather than tight: a large
15/// conversation on a busy server takes a while before the first byte, and a short limit would
16/// call a working server broken. Finite, because a reply that waits forever is a stuck screen.
17const FIRST_BYTE: Duration = Duration::from_secs(300);
18
19/// The version of the Messages API this module speaks.
20const API_VERSION: &str = "2023-06-01";
21
22/// A provider speaking the Anthropic Messages shape.
23#[derive(Debug, Clone, PartialEq, Eq)]
24pub struct Endpoint {
25    /// The address the API lives under, such as `https://api.example.com/anthropic`; requests go
26    /// to `{address}/v1/messages`.
27    pub address: String,
28    /// The model asked for.
29    pub model: String,
30    /// The header the key goes in: `x-api-key` for most, another name for some servers. For
31    /// `authorization` the key goes in as a bearer token.
32    pub key_header: String,
33    /// The key.
34    pub key: Key,
35}
36
37impl Endpoint {
38    /// Where a message request goes. The key is never part of it, so it is safe to show.
39    #[must_use]
40    pub fn url(&self) -> String {
41        format!("{}/v1/messages", self.address.trim().trim_end_matches('/'))
42    }
43
44    /// Sends `body` (a Messages request without `model` and `stream`, which are added here) and
45    /// reads the streamed reply, handing each piece to `on_delta` as it arrives. `cancelled` is
46    /// asked between events.
47    ///
48    /// # Errors
49    ///
50    /// When the server cannot be reached, refuses, or answers with something that is not a
51    /// Messages reply. No error carries the key.
52    pub fn stream(
53        &self,
54        body: &Value,
55        on_delta: &mut dyn FnMut(Delta),
56        cancelled: &dyn Fn() -> bool,
57    ) -> Result<Reply, ProviderError> {
58        let url = self.url();
59        let mut body = body.clone();
60        if let Some(object) = body.as_object_mut() {
61            object.insert("model".to_owned(), Value::String(self.model.clone()));
62            object.insert("stream".to_owned(), Value::Bool(true));
63        }
64        let config = ureq::Agent::config_builder()
65            .timeout_connect(Some(CONNECT))
66            .timeout_recv_response(Some(FIRST_BYTE))
67            // A refusal is an answer, and its body says what the server objected to; it would be
68            // lost if the status alone became an error.
69            .http_status_as_error(false)
70            .build();
71        let agent = ureq::Agent::new_with_config(config);
72        let header = self.key_header.trim();
73        let secret = if header.eq_ignore_ascii_case("authorization") {
74            format!("Bearer {}", self.key.expose())
75        } else {
76            self.key.expose().to_owned()
77        };
78        // What the transport says can name the address; it never names a header value, so it
79        // cannot name the key.
80        let unreachable =
81            |error: ureq::Error| ProviderError::Unreachable { url: url.clone(), reason: error.to_string() };
82        let mut answer = agent
83            .post(&url)
84            .header("anthropic-version", API_VERSION)
85            .header("accept", "text/event-stream")
86            .header(header, &secret)
87            .content_type("application/json")
88            .send(body.to_string().as_str())
89            .map_err(unreachable)?;
90        let status = answer.status().as_u16();
91        if !(200..300).contains(&status) {
92            let said = answer.body_mut().read_to_string().unwrap_or_default();
93            return Err(ProviderError::Refused { url, status, said: error::complaint(&said) });
94        }
95        let lines = BufReader::new(answer.into_body().into_reader());
96        reply::read(lines, on_delta, cancelled)
97    }
98}
qcli/provider/endpoint.rs

qcli/provider/
endpoint.rs
