qcli/provider/
endpoint.rs

1//! Where requests go and how they are carried there.
2
3use std::io::{self, BufRead, BufReader};
4use std::sync::mpsc;
5use std::thread;
6use std::time::{Duration, Instant};
7
8use serde_json::Value;
9
10use super::reply::{self, Delta, Reply};
11use super::{Key, ProviderError, error};
12
13/// How long connecting may take. A server that does not answer within this is not there.
14const CONNECT: Duration = Duration::from_secs(30);
15
16/// How long the server may think before it starts answering. Generous rather than tight: a large
17/// conversation on a busy server takes a while before the first byte, and a short limit would
18/// call a working server broken. Finite, because a reply that waits forever is a stuck screen.
19const FIRST_BYTE: Duration = Duration::from_secs(300);
20
21/// The longest one reply may take to arrive, pieces and pauses together. Far longer than any
22/// reply takes; it only bounds a reading thread left behind by a server that never closes.
23const BODY_LIMIT: Duration = Duration::from_secs(60 * 60);
24
25/// The version of the Messages API this module speaks.
26const API_VERSION: &str = "2023-06-01";
27
28/// A provider speaking the Anthropic Messages shape.
29#[derive(Debug, Clone, PartialEq, Eq)]
30pub struct Endpoint {
31    /// The address the API lives under, such as `https://api.example.com/anthropic`; requests go
32    /// to `{address}/v1/messages`.
33    pub address: String,
34    /// The model asked for.
35    pub model: String,
36    /// The header the key goes in: `x-api-key` for most, another name for some servers. For
37    /// `authorization` the key goes in as a bearer token.
38    pub key_header: String,
39    /// The key, or `None` for a relay on this machine that adds the key itself: QCode's, inside a
40    /// profile's container, which the key never enters. Without one no key header is sent at all.
41    pub key: Option<Key>,
42    /// How long the server may send nothing in the middle of a reply before it is given up on.
43    pub silence: Duration,
44}
45
46impl Endpoint {
47    /// How long a server may go quiet in the middle of a reply: generous, since a model can think
48    /// a while between two pieces, and finite, since a reply that waits for ever is a stuck screen.
49    pub const SILENCE: Duration = Duration::from_secs(120);
50
51    /// Where a message request goes. The key is never part of it, so it is safe to show.
52    #[must_use]
53    pub fn url(&self) -> String {
54        format!("{}/v1/messages", self.address.trim().trim_end_matches('/'))
55    }
56
57    /// Sends `body` (a Messages request without `model` and `stream`, which are added here) and
58    /// reads the streamed reply, handing each piece to `on_delta` as it arrives. `cancelled` is
59    /// asked between events.
60    ///
61    /// # Errors
62    ///
63    /// When the server cannot be reached, refuses, or answers with something that is not a
64    /// Messages reply. No error carries the key.
65    pub fn stream(
66        &self,
67        body: &Value,
68        on_delta: &mut dyn FnMut(Delta),
69        cancelled: &dyn Fn() -> bool,
70    ) -> Result<Reply, ProviderError> {
71        let url = self.url();
72        let mut body = body.clone();
73        if let Some(object) = body.as_object_mut() {
74            object.insert("model".to_owned(), Value::String(self.model.clone()));
75            object.insert("stream".to_owned(), Value::Bool(true));
76        }
77        let config = ureq::Agent::config_builder()
78            .timeout_connect(Some(CONNECT))
79            .timeout_recv_response(Some(FIRST_BYTE))
80            // A reply is given up on after a silence (see `Watched`); this bounds how long the
81            // thread that was reading it can still wait on a server that never closes.
82            .timeout_recv_body(Some(BODY_LIMIT))
83            // A refusal is an answer, and its body says what the server objected to; it would be
84            // lost if the status alone became an error.
85            .http_status_as_error(false)
86            .build();
87        let agent = ureq::Agent::new_with_config(config);
88        let header = self.key_header.trim();
89        let key = self.key.as_ref().map_or("", Key::expose);
90        let secret =
91            if header.eq_ignore_ascii_case("authorization") { format!("Bearer {key}") } else { key.to_owned() };
92        // What the transport says can name the address; it never names a header value, so it
93        // cannot name the key.
94        let unreachable =
95            |error: ureq::Error| ProviderError::Unreachable { url: url.clone(), reason: error.to_string() };
96        let mut request =
97            agent.post(&url).header("anthropic-version", API_VERSION).header("accept", "text/event-stream");
98        if self.key.is_some() {
99            request = request.header(header, &secret);
100        }
101        let mut answer =
102            request.content_type("application/json").send(body.to_string().as_str()).map_err(unreachable)?;
103        let status = answer.status().as_u16();
104        if !(200..300).contains(&status) {
105            let said = answer.body_mut().read_to_string().unwrap_or_default();
106            let refused = ProviderError::Refused { url, status, said: error::complaint(&said) };
107            return Err(refused.without(key));
108        }
109        let lines = Watched::start(BufReader::new(answer.into_body().into_reader()), self.silence, cancelled);
110        reply::read(lines, on_delta, cancelled).map_err(|error| error.without(key))
111    }
112}
113
114/// How often a reply that waits for the server asks whether it was stopped.
115const POLL: Duration = Duration::from_millis(100);
116
117/// The lines of a reply, read on a thread of their own so that waiting for the next one can be
118/// given up: when the person stops the reply, or when the server has sent nothing for the
119/// endpoint's silence. The reading thread ends at the next line after it is left, or when the
120/// connection ends; [`BODY_LIMIT`] bounds how long that can take.
121struct Watched<'a> {
122    lines: mpsc::Receiver<io::Result<String>>,
123    silence: Duration,
124    cancelled: &'a dyn Fn() -> bool,
125}
126
127impl<'a> Watched<'a> {
128    fn start(reader: impl BufRead + Send + 'static, silence: Duration, cancelled: &'a dyn Fn() -> bool) -> Self {
129        // A few lines ahead is enough to keep the connection moving; the channel does not grow
130        // without end when the screen is slower than the server.
131        let (sender, lines) = mpsc::sync_channel(64);
132        thread::spawn(move || {
133            for line in reader.lines() {
134                if sender.send(line).is_err() {
135                    return;
136                }
137            }
138        });
139        Self { lines, silence, cancelled }
140    }
141}
142
143impl Iterator for Watched<'_> {
144    type Item = io::Result<String>;
145
146    fn next(&mut self) -> Option<Self::Item> {
147        let waiting = Instant::now();
148        loop {
149            if (self.cancelled)() {
150                return None;
151            }
152            let left = self.silence.saturating_sub(waiting.elapsed());
153            if left.is_zero() {
154                let said =
155                    format!("the provider sent nothing for {} seconds and was given up on", self.silence.as_secs());
156                return Some(Err(io::Error::new(io::ErrorKind::TimedOut, said)));
157            }
158            match self.lines.recv_timeout(left.min(POLL)) {
159                Ok(line) => return Some(line),
160                Err(mpsc::RecvTimeoutError::Timeout) => {}
161                Err(mpsc::RecvTimeoutError::Disconnected) => return None,
162            }
163        }
164    }
165}
qcli/provider/endpoint.rs

qcli/provider/
endpoint.rs
