netget/server/anthropic/
mod.rs

1//! Anthropic Messages API. Rust owns HTTP, validation and Anthropic's error envelope, the
2//! optional API key check, ids, usage estimates, the event stream, count_tokens and the models
3//! list; the handler writes every assistant message.
4pub mod actions;
5pub mod wire;
6
7use crate::llm::action_helper::call_llm;
8use crate::llm::actions::protocol_trait::ActionResult;
9use crate::logging::emit::Log;
10use crate::protocol::{Event, SpawnContext};
11use crate::server::{
12    accept_bounded::{accept_bounded, ConnectionLimiter, DEFAULT_MAX_CONNECTIONS},
13    connection::ConnectionId,
14};
15use crate::state::server::{ConnectionState, ConnectionStatus, ProtocolConnectionInfo};
16use anyhow::{ensure, Result};
17use bytes::Bytes;
18use http_body_util::{BodyExt, Full, Limited};
19use hyper::{
20    body::Incoming,
21    header::{HeaderValue, CONTENT_TYPE},
22    server::conn::http1,
23    service::service_fn,
24    Method, Request, Response, StatusCode,
25};
26use hyper_util::rt::{TokioIo, TokioTimer};
27use serde_json::{json, Value};
28use std::{convert::Infallible, net::SocketAddr, sync::Arc, time::Duration};
29
30pub const HEADER_TIMEOUT: Duration = Duration::from_secs(30);
31pub const BODY_TIMEOUT: Duration = Duration::from_secs(60);
32pub const MAX_HEADER_BYTES: usize = 64 * 1024;
33pub const MAX_HEADERS: usize = 100;
34/// Requests served on one keep-alive connection before it is closed.
35pub const MAX_REQUESTS_PER_CONNECTION: usize = 1000;
36const CAP_REFUSAL: &[u8] = b"HTTP/1.1 503 Service Unavailable\r\nContent-Length: 0\r\nConnection: close\r\nRetry-After: 5\r\n\r\n";
37
38type Reply = Response<Full<Bytes>>;
39
40struct Config {
41    api_key: Option<String>,
42    models: Vec<String>,
43}
44
45pub async fn spawn(ctx: SpawnContext) -> Result<SocketAddr> {
46    let params = ctx.startup_params.as_ref();
47    let api_key = params
48        .map(|p| p.get_optional_string("api_key"))
49        .transpose()?
50        .flatten()
51        .filter(|k| !k.is_empty());
52    let models: Vec<String> = match params
53        .map(|p| p.get_optional_array("models"))
54        .transpose()?
55        .flatten()
56    {
57        Some(list) => list
58            .iter()
59            .map(|m| m.as_str().map(str::to_string))
60            .collect::<Option<_>>()
61            .ok_or_else(|| anyhow::anyhow!("models must be an array of strings"))?,
62        None => vec![wire::DEFAULT_MODEL.to_string()],
63    };
64    ensure!(
65        !models.is_empty()
66            && models.len() <= 100
67            && models.iter().all(|m| !m.is_empty() && m.len() <= 256),
68        "models must list 1..=100 non-empty ids"
69    );
70    let config = Arc::new(Config { api_key, models });
71    let listener = tokio::net::TcpListener::bind(ctx.legacy_listen_addr()).await?;
72    let local = listener.local_addr()?;
73    Log::new(Some(&ctx.status_tx)).info(format!("Anthropic Messages API listening on {local}"));
74    let server_id = ctx.server_id;
75    let state = ctx.state.clone();
76    let accept = tokio::spawn(async move {
77        let limiter = ConnectionLimiter::new(DEFAULT_MAX_CONNECTIONS);
78        loop {
79            let (stream, peer, permit) = match accept_bounded(
80                &listener,
81                &limiter,
82                CAP_REFUSAL,
83                "Anthropic",
84                Some(&ctx.status_tx),
85            )
86            .await
87            {
88                Ok(v) => v,
89                Err(_) => break,
90            };
91            let id = ConnectionId::new(ctx.state.get_next_unified_id().await);
92            let now = crate::utils::clock::Instant::now();
93            ctx.state
94                .add_connection_to_server(
95                    server_id,
96                    ConnectionState {
97                        id,
98                        remote_addr: peer,
99                        local_addr: local,
100                        bytes_sent: 0,
101                        bytes_received: 0,
102                        packets_sent: 0,
103                        packets_received: 0,
104                        last_activity: now,
105                        status: ConnectionStatus::Active,
106                        status_changed_at: now,
107                        protocol_info: ProtocolConnectionInfo::empty(),
108                    },
109                )
110                .await;
111            let child = ctx.clone();
112            let config = config.clone();
113            ctx.state
114                .spawn_server_task(server_id, async move {
115                    let _permit = permit;
116                    let request_ctx = child.clone();
117                    let served = Arc::new(std::sync::atomic::AtomicUsize::new(0));
118                    let service = service_fn(move |request| {
119                        let ctx = request_ctx.clone();
120                        let config = config.clone();
121                        let served = served.clone();
122                        async move {
123                            let n = served.fetch_add(1, std::sync::atomic::Ordering::SeqCst) + 1;
124                            let mut r = handle(request, id, &ctx, &config).await;
125                            if n >= MAX_REQUESTS_PER_CONNECTION {
126                                r.headers_mut().insert(
127                                    hyper::header::CONNECTION,
128                                    HeaderValue::from_static("close"),
129                                );
130                            }
131                            Ok::<_, Infallible>(r)
132                        }
133                    });
134                    let mut builder = http1::Builder::new();
135                    builder
136                        .keep_alive(true)
137                        .timer(TokioTimer::new())
138                        .header_read_timeout(HEADER_TIMEOUT)
139                        .max_headers(MAX_HEADERS)
140                        .max_buf_size(MAX_HEADER_BYTES);
141                    if let Err(e) = builder
142                        .serve_connection(TokioIo::new(stream), service)
143                        .await
144                    {
145                        Log::new(Some(&child.status_tx))
146                            .warn(format!("Anthropic connection {id} HTTP error: {e}"));
147                    }
148                    child
149                        .state
150                        .update_connection_status(server_id, id, ConnectionStatus::Closed)
151                        .await;
152                    let _ = child.status_tx.send("__UPDATE_UI__".into());
153                })
154                .await;
155        }
156    });
157    state.register_server_task(server_id, accept).await;
158    Ok(local)
159}
160
161fn reply(status: u16, content_type: &'static str, body: Vec<u8>, request_id: &str) -> Reply {
162    let mut r = Response::new(Full::new(Bytes::from(body)));
163    *r.status_mut() = StatusCode::from_u16(status).unwrap_or(StatusCode::INTERNAL_SERVER_ERROR);
164    let h = r.headers_mut();
165    h.insert(CONTENT_TYPE, HeaderValue::from_static(content_type));
166    if let Ok(v) = HeaderValue::from_str(request_id) {
167        h.insert("request-id", v);
168    }
169    if content_type == "text/event-stream" {
170        h.insert(
171            hyper::header::CACHE_CONTROL,
172            HeaderValue::from_static("no-cache"),
173        );
174    }
175    r
176}
177
178fn json_reply(v: &Value, request_id: &str) -> Reply {
179    reply(
180        200,
181        "application/json",
182        serde_json::to_vec(v).unwrap_or_default(),
183        request_id,
184    )
185}
186
187fn error(error_type: &str, message: &str, request_id: &str) -> Reply {
188    let status = wire::status_of(error_type).unwrap_or(500);
189    let body = wire::error_body(error_type, message, request_id);
190    reply(
191        status,
192        "application/json",
193        serde_json::to_vec(&body).unwrap_or_default(),
194        request_id,
195    )
196}
197
198fn outcome(ctx: &SpawnContext, id: ConnectionId, decision: &str) {
199    let line = format!("Anthropic connection {id} operation=messages decision={decision}");
200    let log = Log::new(Some(&ctx.status_tx));
201    if decision.starts_with("fail_closed") || decision == "model_silent" {
202        log.error(line)
203    } else {
204        log.info(line)
205    }
206}
207
208/// The handler failed or said nothing: overloaded (529) when the backend is busy, else an API
209/// error (500). Never a fabricated message.
210fn failure(e: Option<&anyhow::Error>, request_id: &str) -> Reply {
211    let category = e
212        .map(crate::utils::WireFailure::classify)
213        .unwrap_or(crate::utils::WireFailure::Unavailable);
214    let kind = if category.is_overloaded() {
215        "overloaded_error"
216    } else {
217        "api_error"
218    };
219    error(kind, category.prefixed_text(), request_id)
220}
221
222/// Ask the handler for one answer to a messages request.
223async fn ask(
224    ctx: &SpawnContext,
225    id: ConnectionId,
226    event: Event,
227    request_id: &str,
228) -> std::result::Result<Value, Reply> {
229    let result = match call_llm(
230        &ctx.llm_client,
231        &ctx.state,
232        ctx.server_id,
233        Some(id),
234        &event,
235        &actions::AnthropicProtocol,
236    )
237    .await
238    {
239        Ok(r) => r,
240        Err(e) => {
241            outcome(ctx, id, "fail_closed_llm_error");
242            return Err(failure(Some(&e), request_id));
243        }
244    };
245    if !result.failures.is_empty() {
246        outcome(ctx, id, "fail_closed_invalid_reply");
247        return Err(failure(None, request_id));
248    }
249    let mut answers = Vec::new();
250    let mut stack = result.protocol_results;
251    while let Some(r) = stack.pop() {
252        match r {
253            ActionResult::Custom { name, data } if name.starts_with("anthropic_") => {
254                answers.push(data)
255            }
256            ActionResult::Multiple(items) => stack.extend(items),
257            _ => {}
258        }
259    }
260    match answers.len() {
261        1 => {
262            let a = answers.pop().unwrap_or_default();
263            if a["type"] == "anthropic_error" {
264                outcome(ctx, id, "model_reject");
265                return Err(error(
266                    a["error_type"].as_str().unwrap_or("api_error"),
267                    a["message"].as_str().unwrap_or_default(),
268                    request_id,
269                ));
270            }
271            outcome(ctx, id, "model_answer");
272            Ok(a)
273        }
274        0 => {
275            outcome(ctx, id, "model_silent");
276            Err(failure(None, request_id))
277        }
278        _ => {
279            outcome(ctx, id, "fail_closed_invalid_reply");
280            Err(failure(None, request_id))
281        }
282    }
283}
284
285fn key_of(parts: &hyper::http::request::Parts) -> Option<String> {
286    if let Some(k) = parts.headers.get("x-api-key").and_then(|v| v.to_str().ok()) {
287        return Some(k.to_string());
288    }
289    parts
290        .headers
291        .get(hyper::header::AUTHORIZATION)
292        .and_then(|v| v.to_str().ok())
293        .and_then(|v| v.strip_prefix("Bearer "))
294        .map(str::to_string)
295}
296
297/// Constant-time comparison, so the key cannot be found byte by byte from response timing.
298fn same(a: &str, b: &str) -> bool {
299    a.len() == b.len()
300        && a.bytes()
301            .zip(b.bytes())
302            .fold(0u8, |acc, (x, y)| acc | (x ^ y))
303            == 0
304}
305
306async fn handle(
307    request: Request<Incoming>,
308    id: ConnectionId,
309    ctx: &SpawnContext,
310    config: &Config,
311) -> Reply {
312    let request_id = wire::random_id("req_");
313    let (parts, body) = request.into_parts();
314    let bytes = match tokio::time::timeout(
315        BODY_TIMEOUT,
316        Limited::new(body, wire::MAX_BODY).collect(),
317    )
318    .await
319    {
320        Err(_) => {
321            return error(
322                "invalid_request_error",
323                "request body deadline exceeded",
324                &request_id,
325            )
326        }
327        Ok(Err(e)) => {
328            return if e
329                .downcast_ref::<http_body_util::LengthLimitError>()
330                .is_some()
331            {
332                error(
333                    "request_too_large",
334                    "Request exceeds the maximum allowed number of bytes.",
335                    &request_id,
336                )
337            } else {
338                error("invalid_request_error", "incomplete HTTP body", &request_id)
339            }
340        }
341        Ok(Ok(b)) => b.to_bytes(),
342    };
343    ctx.state
344        .update_connection_stats(
345            ctx.server_id,
346            id,
347            Some(bytes.len() as u64),
348            None,
349            Some(1),
350            None,
351        )
352        .await;
353    let key = key_of(&parts);
354    if let Some(expected) = &config.api_key {
355        if !key.as_deref().is_some_and(|k| same(k, expected)) {
356            outcome(ctx, id, "auth_reject");
357            return error("authentication_error", "invalid x-api-key", &request_id);
358        }
359    }
360    let path = parts.uri.path();
361    let parsed = || serde_json::from_slice::<Value>(&bytes);
362    let unparsable = |e: serde_json::Error| {
363        error(
364            "invalid_request_error",
365            &format!("There was an issue with your request body: {e}"),
366            &request_id,
367        )
368    };
369    let r = match (&parts.method, path) {
370        (&Method::POST, "/v1/messages") => {
371            let body = match parsed() {
372                Ok(b) => b,
373                Err(e) => return unparsable(e),
374            };
375            messages(ctx, id, &body, key.is_some(), &request_id).await
376        }
377        (&Method::POST, "/v1/messages/count_tokens") => {
378            let body = match parsed() {
379                Ok(b) => b,
380                Err(e) => return unparsable(e),
381            };
382            match wire::request_summary(&body, true) {
383                Ok(_) => json_reply(
384                    &json!({"input_tokens": wire::estimate_tokens(&wire::request_text(&body))}),
385                    &request_id,
386                ),
387                Err(e) => error("invalid_request_error", &e.to_string(), &request_id),
388            }
389        }
390        (&Method::GET, "/v1/models") => {
391            let data: Vec<Value> = config
392                .models
393                .iter()
394                .map(|m| wire::model_object(m))
395                .collect();
396            json_reply(
397                &json!({"data": data, "has_more": false, "first_id": config.models.first(), "last_id": config.models.last()}),
398                &request_id,
399            )
400        }
401        (&Method::GET, p) if p.starts_with("/v1/models/") => {
402            let wanted = &p["/v1/models/".len()..];
403            match config.models.iter().find(|m| *m == wanted) {
404                Some(m) => json_reply(&wire::model_object(m), &request_id),
405                None => error("not_found_error", &format!("model: {wanted}"), &request_id),
406            }
407        }
408        _ => error("not_found_error", "Not found", &request_id),
409    };
410    let sent = hyper::body::Body::size_hint(r.body()).exact().unwrap_or(0);
411    ctx.state
412        .update_connection_stats(ctx.server_id, id, None, Some(sent), None, Some(1))
413        .await;
414    r
415}
416
417async fn messages(
418    ctx: &SpawnContext,
419    id: ConnectionId,
420    body: &Value,
421    has_key: bool,
422    request_id: &str,
423) -> Reply {
424    let mut summary = match wire::request_summary(body, false) {
425        Ok(s) => s,
426        Err(e) => return error("invalid_request_error", &e.to_string(), request_id),
427    };
428    summary["api_key_present"] = json!(has_key);
429    let stream = summary["stream"].as_bool().unwrap_or(false);
430    let model = summary["model"].as_str().unwrap_or_default().to_string();
431    let answer = match ask(
432        ctx,
433        id,
434        Event::new(&actions::MESSAGE_EVENT, summary),
435        request_id,
436    )
437    .await
438    {
439        Ok(a) => a,
440        Err(r) => return r,
441    };
442    // execute_action checked both; a failure here is a handler answer that changed meaning.
443    let content = match wire::reply_content(&answer) {
444        Ok(c) => c,
445        Err(_) => return failure(None, request_id),
446    };
447    let stop = match wire::stop_reason(&answer, &content) {
448        Ok(s) => s,
449        Err(_) => return failure(None, request_id),
450    };
451    let output_text: String = content
452        .iter()
453        .map(|b| match b["type"].as_str() {
454            Some("tool_use") => b["input"].to_string(),
455            _ => b["text"].as_str().unwrap_or_default().to_string(),
456        })
457        .collect();
458    let input_tokens = answer["input_tokens"]
459        .as_u64()
460        .unwrap_or_else(|| wire::estimate_tokens(&wire::request_text(body)));
461    let output_tokens = answer["output_tokens"]
462        .as_u64()
463        .unwrap_or_else(|| wire::estimate_tokens(&output_text).max(1));
464    let message = wire::message(
465        &wire::random_id("msg_"),
466        &model,
467        content,
468        &stop,
469        answer["stop_sequence"].as_str(),
470        input_tokens,
471        output_tokens,
472    );
473    if stream {
474        reply(
475            200,
476            "text/event-stream",
477            wire::event_stream(&message).into_bytes(),
478            request_id,
479        )
480    } else {
481        json_reply(&message, request_id)
482    }
483}
netget/server/anthropic/mod.rs

netget/server/anthropic/
mod.rs
