use std::collections::HashSet;
use anyhow::{Result, anyhow};
use serde_json::{Value, json};
use crate::api::azure_base_url::{build_default_azure_base_url, normalize_azure_base_url};
use crate::api::common::{
apply_on_payload, build_http_client_for_target, finish_stream_error, invoke_on_response_from_reqwest,
is_request_aborted, merge_model_headers, send_with_abort,
};
use crate::api::openai_prompt_cache::clamp_openai_prompt_cache_key;
use crate::api::openai_responses_shared::{
ResponsesStreamState, convert_responses_messages, convert_responses_tools, process_responses_stream_event,
};
use crate::api::simple_options::build_base_options;
use crate::models::{clamp_thinking_level, thinking_level_to_str};
use crate::types::{
AssistantMessage, AssistantMessageEvent, Context, Model, ProviderStreams, SimpleStreamOptions, StopReason,
StreamOptions,
};
use crate::utils::event_stream::AssistantMessageEventStream;
use crate::utils::provider_env::get_provider_env_value;
use super::sse::for_each_sse_json_event;
const DEFAULT_AZURE_API_VERSION: &str = "v1";
const AZURE_TOOL_CALL_PROVIDERS: &[&str] = &["openai", "openai-codex", "opencode", "azure-openai-responses"];
const OPENAI_RESPONSES_MIN_OUTPUT_TOKENS: u32 = 16;
#[derive(Clone, Default)]
pub struct AzureOpenAIResponsesOptions {
pub base: StreamOptions,
pub reasoning_effort: Option<String>,
pub reasoning_summary: Option<String>,
pub azure_api_version: Option<String>,
pub azure_resource_name: Option<String>,
pub azure_base_url: Option<String>,
pub azure_deployment_name: Option<String>,
}
pub struct AzureOpenAIResponsesApi;
impl ProviderStreams for AzureOpenAIResponsesApi {
fn stream(&self, model: &Model, context: &Context, options: Option<StreamOptions>) -> AssistantMessageEventStream {
self.stream_with_options(
model,
context,
AzureOpenAIResponsesOptions {
base: options.unwrap_or_default(),
..Default::default()
},
)
}
fn stream_simple(
&self,
model: &Model,
context: &Context,
options: Option<SimpleStreamOptions>,
) -> AssistantMessageEventStream {
let opts = options.as_ref();
let base = build_base_options(model, context, opts, opts.and_then(|o| o.base.api_key.clone()));
let reasoning = opts.and_then(|o| o.reasoning).map(|r| clamp_thinking_level(model, r));
let reasoning_effort = reasoning.map(|r| thinking_level_to_str(r).to_string());
self.stream_with_options(
model,
context,
AzureOpenAIResponsesOptions {
base,
reasoning_effort,
..Default::default()
},
)
}
}
impl AzureOpenAIResponsesApi {
pub fn stream_with_options(
&self,
model: &Model,
context: &Context,
options: AzureOpenAIResponsesOptions,
) -> AssistantMessageEventStream {
let stream = AssistantMessageEventStream::new();
let model = model.clone();
let context = context.clone();
let s = stream.clone();
tokio::spawn(async move {
let mut output = AssistantMessage::empty(&model);
if let Err(e) = run_azure(&model, &context, &options, &s, &mut output).await {
let aborted = crate::api::common::is_abort_error(&e);
finish_stream_error(&s, &mut output, e, aborted);
}
});
stream
}
}
async fn run_azure(
model: &Model,
context: &Context,
options: &AzureOpenAIResponsesOptions,
stream: &AssistantMessageEventStream,
output: &mut AssistantMessage,
) -> Result<()> {
let api_key = options
.base
.api_key
.as_deref()
.ok_or_else(|| anyhow!("No API key for provider: {}", model.provider))?;
let deployment = resolve_deployment_name(model, options);
let headers = merge_model_headers(model, Some(&options.base));
let providers: HashSet<String> = AZURE_TOOL_CALL_PROVIDERS.iter().map(|s| s.to_string()).collect();
let mut params = build_params(model, context, options, &deployment, &providers)?;
params = apply_on_payload(options.base.on_payload.as_ref(), params, model).await;
let (base_url, api_version) = resolve_azure_config(model, options)?;
let url = format!("{base_url}/deployments/{deployment}/responses?api-version={api_version}");
let client = build_http_client_for_target(options.base.timeout_ms, Some(&url), options.base.env.as_ref())?;
let mut req = client.post(&url).header("api-key", api_key).json(¶ms);
for (k, v) in &headers {
req = req.header(k, v);
}
let response = send_with_abort(&options.base.signal, req).await?;
invoke_on_response_from_reqwest(options.base.on_response.as_ref(), &response, model).await;
let response = crate::api::common::check_response_ok(response).await?;
stream.push(AssistantMessageEvent::Start {
partial: output.clone(),
});
let mut responses_state = ResponsesStreamState::default();
for_each_sse_json_event(response, &options.base.signal, |event| {
process_responses_stream_event(&event, &mut responses_state, output, stream, model, None)
})
.await?;
if is_request_aborted(&options.base.signal) {
output.stop_reason = StopReason::Aborted;
} else if !responses_state.saw_terminal {
return Err(anyhow!(
"OpenAI Responses stream ended before a terminal response event"
));
}
stream.push(AssistantMessageEvent::Done {
reason: output.stop_reason,
message: output.clone(),
});
stream.end();
Ok(())
}
fn resolve_deployment_name(model: &Model, options: &AzureOpenAIResponsesOptions) -> String {
if let Some(d) = &options.azure_deployment_name {
return d.clone();
}
if let Some(map) = get_provider_env_value("AZURE_OPENAI_DEPLOYMENT_NAME_MAP", options.base.env.as_ref()) {
for entry in map.split(',') {
if let Some((id, dep)) = entry.split_once('=')
&& id.trim() == model.id
{
return dep.trim().to_string();
}
}
}
model.id.clone()
}
fn resolve_azure_config(model: &Model, options: &AzureOpenAIResponsesOptions) -> Result<(String, String)> {
let api_version = options
.azure_api_version
.clone()
.or_else(|| get_provider_env_value("AZURE_OPENAI_API_VERSION", options.base.env.as_ref()))
.unwrap_or_else(|| DEFAULT_AZURE_API_VERSION.to_string());
let base_url = options
.azure_base_url
.clone()
.or_else(|| get_provider_env_value("AZURE_OPENAI_BASE_URL", options.base.env.as_ref()))
.or_else(|| {
options
.azure_resource_name
.clone()
.or_else(|| get_provider_env_value("AZURE_OPENAI_RESOURCE_NAME", options.base.env.as_ref()))
.map(|r| build_default_azure_base_url(&r))
})
.or_else(|| Some(model.base_url.clone()))
.ok_or_else(|| anyhow!("Azure OpenAI base URL is required"))?;
Ok((normalize_azure_base_url(&base_url)?, api_version))
}
fn build_params(
model: &Model,
context: &Context,
options: &AzureOpenAIResponsesOptions,
deployment: &str,
providers: &HashSet<String>,
) -> Result<Value> {
let messages = convert_responses_messages(model, context, providers, None);
let mut params = json!({
"model": deployment,
"input": messages,
"stream": true,
"prompt_cache_key": clamp_openai_prompt_cache_key(options.base.session_id.as_deref()),
"store": false
});
if let Some(max) = options.base.max_tokens {
params["max_output_tokens"] = json!(max.max(OPENAI_RESPONSES_MIN_OUTPUT_TOKENS));
}
if let Some(temp) = options.base.temperature {
params["temperature"] = json!(temp);
}
if let Some(tools) = &context.tools
&& !tools.is_empty()
{
params["tools"] = json!(convert_responses_tools(tools, None));
}
if model.reasoning
&& let Some(effort) = &options.reasoning_effort
{
params["reasoning"] = json!({ "effort": effort, "summary": options.reasoning_summary.clone().unwrap_or_else(|| "auto".to_string()) });
params["include"] = json!(["reasoning.encrypted_content"]);
}
Ok(params)
}
#[cfg(test)]
mod tests {
use super::*;
use crate::api::openai_prompt_cache::OPENAI_PROMPT_CACHE_KEY_MAX_LENGTH;
use crate::types::{Message, UserContent};
fn sample_context() -> Context {
Context {
system_prompt: None,
messages: vec![Message::User {
content: UserContent::Text("hello".to_string()),
timestamp: 0,
}],
tools: None,
}
}
fn sample_model() -> Model {
get_builtin_model_for_test()
}
fn get_builtin_model_for_test() -> Model {
crate::get_builtin_model("azure-openai-responses", "gpt-4o-mini").expect("model")
}
#[test]
fn build_params_clamps_prompt_cache_key_and_disables_store() {
let model = sample_model();
let context = sample_context();
let long_session = "x".repeat(OPENAI_PROMPT_CACHE_KEY_MAX_LENGTH + 3);
let options = AzureOpenAIResponsesOptions {
base: StreamOptions {
session_id: Some(long_session),
..Default::default()
},
..Default::default()
};
let providers: HashSet<String> = AZURE_TOOL_CALL_PROVIDERS.iter().map(|s| s.to_string()).collect();
let params = build_params(&model, &context, &options, "gpt-4o-mini", &providers).expect("params");
assert_eq!(
params["prompt_cache_key"].as_str(),
Some("x".repeat(OPENAI_PROMPT_CACHE_KEY_MAX_LENGTH).as_str())
);
assert_eq!(params["store"], false);
}
}