244 lines
10 KiB
Rust
244 lines
10 KiB
Rust
impl AnthropicRuntimeClient {
|
|
fn new(
|
|
session_id: &str,
|
|
model: String,
|
|
enable_tools: bool,
|
|
emit_output: bool,
|
|
allowed_tools: Option<AllowedToolSet>,
|
|
tool_registry: GlobalToolRegistry,
|
|
progress_reporter: Option<InternalPromptProgressReporter>,
|
|
) -> Result<Self, Box<dyn std::error::Error>> {
|
|
// Dispatch to the correct provider at construction time.
|
|
// `ApiProviderClient` (exposed by the api crate as
|
|
// `ProviderClient`) is an enum over Anthropic / xAI / OpenAI
|
|
// variants, where xAI and OpenAI both use the OpenAI-compat
|
|
// wire format under the hood. We consult
|
|
// `detect_provider_kind(&resolved_model)` so model-name prefix
|
|
// routing (`openai/`, `gpt-`, `grok`, `qwen/`) wins over
|
|
// env-var presence.
|
|
//
|
|
// For Anthropic we build the client directly instead of going
|
|
// through `ApiProviderClient::from_model_with_anthropic_auth`
|
|
// so we can explicitly apply `api::read_base_url()` — that
|
|
// reads `ANTHROPIC_BASE_URL` and is required for the local
|
|
// mock-server test harness
|
|
// (`crates/rusty-claude-cli/tests/compact_output.rs`) to point
|
|
// claw at its fake Anthropic endpoint. We also attach a
|
|
// session-scoped prompt cache on the Anthropic path; the
|
|
// prompt cache is Anthropic-only so non-Anthropic variants
|
|
// skip it.
|
|
let resolved_model = api::resolve_model_alias(&model);
|
|
let client = match detect_provider_kind(&resolved_model) {
|
|
ProviderKind::Anthropic => {
|
|
let auth = resolve_cli_auth_source()?;
|
|
let inner = AnthropicClient::from_auth(auth)
|
|
.with_base_url(api::read_base_url())
|
|
.with_prompt_cache(PromptCache::new(session_id));
|
|
ApiProviderClient::Anthropic(inner)
|
|
}
|
|
ProviderKind::Xai | ProviderKind::OpenAi => {
|
|
// The api crate's `ProviderClient::from_model_with_anthropic_auth`
|
|
// with `None` for the anthropic auth routes via
|
|
// `detect_provider_kind` and builds an
|
|
// `OpenAiCompatClient::from_env` with the matching
|
|
// `OpenAiCompatConfig` (openai / xai / dashscope).
|
|
// That reads the correct API-key env var and BASE_URL
|
|
// override internally, so this one call covers OpenAI,
|
|
// OpenRouter, xAI, DashScope, Ollama, and any other
|
|
// OpenAI-compat endpoint users configure via
|
|
// `OPENAI_BASE_URL` / `XAI_BASE_URL` / `DASHSCOPE_BASE_URL`.
|
|
ApiProviderClient::from_model_with_anthropic_auth(&resolved_model, None)?
|
|
}
|
|
};
|
|
Ok(Self {
|
|
runtime: tokio::runtime::Runtime::new()?,
|
|
client,
|
|
session_id: session_id.to_string(),
|
|
model,
|
|
enable_tools,
|
|
emit_output,
|
|
allowed_tools,
|
|
tool_registry,
|
|
progress_reporter,
|
|
reasoning_effort: None,
|
|
})
|
|
}
|
|
|
|
fn set_reasoning_effort(&mut self, effort: Option<String>) {
|
|
self.reasoning_effort = effort;
|
|
}
|
|
}
|
|
impl AnthropicRuntimeClient {
|
|
/// Consume a single streaming response, optionally applying a stall
|
|
/// timeout on the first event for post-tool continuations.
|
|
#[allow(clippy::too_many_lines)]
|
|
async fn consume_stream(
|
|
&self,
|
|
message_request: &MessageRequest,
|
|
apply_stall_timeout: bool,
|
|
) -> Result<Vec<AssistantEvent>, RuntimeError> {
|
|
let mut stream = self
|
|
.client
|
|
.stream_message(message_request)
|
|
.await
|
|
.map_err(|error| {
|
|
RuntimeError::new(format_user_visible_api_error(&self.session_id, &error))
|
|
})?;
|
|
let mut stdout = io::stdout();
|
|
let mut sink = io::sink();
|
|
let out: &mut dyn Write = if self.emit_output {
|
|
&mut stdout
|
|
} else {
|
|
&mut sink
|
|
};
|
|
let renderer = TerminalRenderer::new();
|
|
let mut markdown_stream = MarkdownStreamState::default();
|
|
let mut events = Vec::new();
|
|
let mut pending_tool: Option<(String, String, String)> = None;
|
|
let mut block_has_thinking_summary = false;
|
|
let mut saw_stop = false;
|
|
let mut received_any_event = false;
|
|
|
|
loop {
|
|
let next = if apply_stall_timeout && !received_any_event {
|
|
match tokio::time::timeout(POST_TOOL_STALL_TIMEOUT, stream.next_event()).await {
|
|
Ok(inner) => inner.map_err(|error| {
|
|
RuntimeError::new(format_user_visible_api_error(&self.session_id, &error))
|
|
})?,
|
|
Err(_elapsed) => {
|
|
return Err(RuntimeError::new(
|
|
"post-tool stall: model did not respond within timeout",
|
|
));
|
|
}
|
|
}
|
|
} else {
|
|
stream.next_event().await.map_err(|error| {
|
|
RuntimeError::new(format_user_visible_api_error(&self.session_id, &error))
|
|
})?
|
|
};
|
|
|
|
let Some(event) = next else {
|
|
break;
|
|
};
|
|
received_any_event = true;
|
|
|
|
match event {
|
|
ApiStreamEvent::MessageStart(start) => {
|
|
for block in start.message.content {
|
|
push_output_block(
|
|
block,
|
|
out,
|
|
&mut events,
|
|
&mut pending_tool,
|
|
true,
|
|
&mut block_has_thinking_summary,
|
|
)?;
|
|
}
|
|
}
|
|
ApiStreamEvent::ContentBlockStart(start) => {
|
|
push_output_block(
|
|
start.content_block,
|
|
out,
|
|
&mut events,
|
|
&mut pending_tool,
|
|
true,
|
|
&mut block_has_thinking_summary,
|
|
)?;
|
|
}
|
|
ApiStreamEvent::ContentBlockDelta(delta) => match delta.delta {
|
|
ContentBlockDelta::TextDelta { text } => {
|
|
if !text.is_empty() {
|
|
if let Some(progress_reporter) = &self.progress_reporter {
|
|
progress_reporter.mark_text_phase(&text);
|
|
}
|
|
if let Some(rendered) = markdown_stream.push(&renderer, &text) {
|
|
write!(out, "{rendered}")
|
|
.and_then(|()| out.flush())
|
|
.map_err(|error| RuntimeError::new(error.to_string()))?;
|
|
}
|
|
events.push(AssistantEvent::TextDelta(text));
|
|
}
|
|
}
|
|
ContentBlockDelta::InputJsonDelta { partial_json } => {
|
|
if let Some((_, _, input)) = &mut pending_tool {
|
|
input.push_str(&partial_json);
|
|
}
|
|
}
|
|
ContentBlockDelta::ThinkingDelta { thinking } => {
|
|
if !block_has_thinking_summary {
|
|
render_thinking_block_summary(out, None, false)?;
|
|
block_has_thinking_summary = true;
|
|
}
|
|
events.push(AssistantEvent::ThinkingDelta(thinking));
|
|
}
|
|
ContentBlockDelta::SignatureDelta { signature } => {
|
|
events.push(AssistantEvent::SignatureDelta(signature));
|
|
}
|
|
},
|
|
ApiStreamEvent::ContentBlockStop(_) => {
|
|
block_has_thinking_summary = false;
|
|
if let Some(rendered) = markdown_stream.flush(&renderer) {
|
|
write!(out, "{rendered}")
|
|
.and_then(|()| out.flush())
|
|
.map_err(|error| RuntimeError::new(error.to_string()))?;
|
|
}
|
|
if let Some((id, name, input)) = pending_tool.take() {
|
|
if let Some(progress_reporter) = &self.progress_reporter {
|
|
progress_reporter.mark_tool_phase(&name, &input);
|
|
}
|
|
// Display tool call now that input is fully accumulated
|
|
writeln!(out, "\n{}", format_tool_call_start(&name, &input))
|
|
.and_then(|()| out.flush())
|
|
.map_err(|error| RuntimeError::new(error.to_string()))?;
|
|
events.push(AssistantEvent::ToolUse { id, name, input });
|
|
}
|
|
}
|
|
ApiStreamEvent::MessageDelta(delta) => {
|
|
events.push(AssistantEvent::Usage(delta.usage.token_usage()));
|
|
}
|
|
ApiStreamEvent::MessageStop(_) => {
|
|
saw_stop = true;
|
|
if let Some(rendered) = markdown_stream.flush(&renderer) {
|
|
write!(out, "{rendered}")
|
|
.and_then(|()| out.flush())
|
|
.map_err(|error| RuntimeError::new(error.to_string()))?;
|
|
}
|
|
events.push(AssistantEvent::MessageStop);
|
|
}
|
|
}
|
|
}
|
|
|
|
push_prompt_cache_record(&self.client, &mut events);
|
|
|
|
if !saw_stop
|
|
&& events.iter().any(|event| {
|
|
matches!(event, AssistantEvent::TextDelta(text) if !text.is_empty())
|
|
|| matches!(event, AssistantEvent::ToolUse { .. })
|
|
})
|
|
{
|
|
events.push(AssistantEvent::MessageStop);
|
|
}
|
|
|
|
if events
|
|
.iter()
|
|
.any(|event| matches!(event, AssistantEvent::MessageStop))
|
|
{
|
|
return Ok(events);
|
|
}
|
|
|
|
let response = self
|
|
.client
|
|
.send_message(&MessageRequest {
|
|
stream: false,
|
|
..message_request.clone()
|
|
})
|
|
.await
|
|
.map_err(|error| {
|
|
RuntimeError::new(format_user_visible_api_error(&self.session_id, &error))
|
|
})?;
|
|
let mut events = response_to_events(response, out)?;
|
|
push_prompt_cache_record(&self.client, &mut events);
|
|
Ok(events)
|
|
}
|
|
}
|