boos/BoosAPI
2
0
Fork 2
Code Issues Pull Requests Packages Projects Releases Wiki Activity

feat: add workspace and claude messages support

Browse Source
This commit is contained in:
boos
2026-05-22 05:57:37 +08:00
parent c1074ddf71
commit 7cdb02ef1d
24 changed files with 3282 additions and 479 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/routes/proxy/mod.rs
+233
View File
@@ -51,6 +51,7 @@ pub enum DownstreamSource {
pub fn routes() -> Router<AppState> {
Router::new()
.route("/v1/chat/completions", post(chat_completions))
.route("/v1/messages", post(messages_handler))
.route("/v1/responses", post(responses_handler))
.route("/v1/completions", post(completions_handler))
.route("/v1/embeddings", post(embeddings_handler))
@@ -88,6 +89,31 @@ pub fn routes() -> Router<AppState> {
// ── Handlers ───────────────────────────────────────────────────────────
/// POST /v1/messages — Anthropic Claude Messages API compatibility.
async fn messages_handler(
State(state): State<AppState>,
headers: HeaderMap,
Json(body): Json<Value>,
) -> Result<Response, AppError> {
info!("POST /v1/messages");
let key_id = check_auth(&state, &headers).await?;
let canonical = crate::transformers::anthropic::messages::parse_messages_request(&body)
.map_err(AppError::BadRequest)?;
let model = canonical.model.clone();
if canonical.stream {
let chunks = execute_messages_stream_with_conductor(state, key_id, model, canonical).await?;
let sse = Sse::new(stream::iter(chunks).map(|chunk| {
Ok::<_, Infallible>(axum::response::sse::Event::default().data(chunk))
}));
Ok(sse.into_response())
} else {
let result = execute_messages_with_conductor(state, key_id, model, canonical).await?;
Ok(Json(result).into_response())
}
}
/// POST /v1/responses — OpenAI Responses API (non-stream + stream).
///
/// Behaves like /v1/chat/completions: routes through ProxyConductor for
@@ -722,6 +748,213 @@ fn upstream_status_from_error(err: &AppError) -> Option<u16> {
.and_then(|value| value.parse::<u16>().ok())
}
async fn execute_messages_with_conductor(
state: AppState,
key_id: Option<i64>,
model: String,
request: crate::transformers::canonical::CanonicalRequest,
) -> AppResult<Value> {
let started_at = Instant::now();
let conductor = build_conductor(&state);
let billing_state = state.clone();
let billing_model = model.clone();
let result = conductor
.execute(ExecuteInput {
requested_model: model,
downstream_policy: None,
attempt_fn: Box::new(move |ctx| {
let mut request = request.clone();
request.model = ctx.selected.actual_model.clone();
async move {
match executors::execute_anthropic_messages(
selected_channel_to_route(ctx.selected),
request,
)
.await
{
Ok(response) => AttemptResult::Success {
response,
latency_ms: None,
cost: None,
},
Err(err) => AttemptResult::Failure {
action: crate::proxy::conductor::AttemptFailureAction::Failover,
status: upstream_status_from_error(&err),
raw_error_text: Some(err.to_string()),
},
}
}
.boxed()
}),
on_terminal_failure: None,
})
.await;
match result {
ExecuteResult::Success {
selected,
response,
attempts,
} => {
let (prompt, completion) = extract_usage(&response);
record_proxy_log_success(
&billing_state,
key_id,
&selected,
&billing_model,
Some(elapsed_ms_since(started_at)),
prompt,
completion,
0.0,
attempts,
false,
)
.await;
record_billing(
&billing_state,
key_id,
&billing_model,
prompt,
completion,
0.0,
)
.await;
Ok(response)
}
ExecuteResult::Failure {
reason,
selected,
status,
raw_error_text,
attempts,
} => {
record_proxy_log_failure(
&billing_state,
key_id,
selected.as_ref(),
&billing_model,
status,
Some(elapsed_ms_since(started_at)),
raw_error_text.as_deref(),
attempts,
false,
)
.await;
Err(AppError::UpstreamError(format!(
"Proxy conductor failed after {attempts} attempts ({reason:?}, status: {:?}): {}",
status,
raw_error_text.unwrap_or_else(|| "no upstream error text".to_string())
)))
}
}
}
async fn execute_messages_stream_with_conductor(
state: AppState,
key_id: Option<i64>,
model: String,
request: crate::transformers::canonical::CanonicalRequest,
) -> AppResult<Vec<String>> {
let started_at = Instant::now();
let conductor = build_conductor(&state);
let billing_state = state.clone();
let billing_model = model.clone();
let result = conductor
.execute(ExecuteInput {
requested_model: model,
downstream_policy: None,
attempt_fn: Box::new(move |ctx| {
let mut request = request.clone();
request.model = ctx.selected.actual_model.clone();
async move {
match executors::execute_anthropic_messages_stream(
selected_channel_to_route(ctx.selected),
request,
)
.await
{
Ok(chunks) => AttemptResult::Success {
response: Value::Array(chunks.into_iter().map(Value::String).collect()),
latency_ms: None,
cost: None,
},
Err(err) => AttemptResult::Failure {
action: crate::proxy::conductor::AttemptFailureAction::Failover,
status: upstream_status_from_error(&err),
raw_error_text: Some(err.to_string()),
},
}
}
.boxed()
}),
on_terminal_failure: None,
})
.await;
let (selected, attempts, chunks) = match result {
ExecuteResult::Success {
selected,
response,
attempts,
} => match response {
Value::Array(chunks) => (
selected,
attempts,
chunks
.into_iter()
.filter_map(|chunk| chunk.as_str().map(str::to_string))
.collect::<Vec<_>>(),
),
_ => {
return Err(AppError::UpstreamError(
"Invalid streaming response from conductor".into(),
))
}
},
ExecuteResult::Failure {
reason,
selected,
status,
raw_error_text,
attempts,
} => {
record_proxy_log_failure(
&billing_state,
key_id,
selected.as_ref(),
&billing_model,
status,
Some(elapsed_ms_since(started_at)),
raw_error_text.as_deref(),
attempts,
true,
)
.await;
return Err(AppError::UpstreamError(format!(
"Proxy conductor failed after {attempts} attempts ({reason:?}, status: {:?}): {}",
status,
raw_error_text.unwrap_or_else(|| "no upstream error text".to_string())
)));
}
};
record_proxy_log_success(
&billing_state,
key_id,
&selected,
&billing_model,
Some(elapsed_ms_since(started_at)),
0,
0,
0.0,
attempts,
true,
)
.await;
record_billing(&billing_state, key_id, &billing_model, 0, 0, 0.0).await;
Ok(chunks)
}
async fn execute_chat_completions_with_conductor(
state: AppState,
key_id: Option<i64>,