1use std::collections::{HashMap, VecDeque};
25use std::sync::Arc;
26
27use agent_framework_core::client::{ChatClient, ChatStream};
28use agent_framework_core::error::{Error, Result};
29use agent_framework_core::streaming::Utf8StreamDecoder;
30use agent_framework_core::tools::ToolDefinition;
31use agent_framework_core::types::{
32 Annotation, ChatOptions, ChatResponse, ChatResponseUpdate, Content, DataContent, FinishReason,
33 FunctionApprovalRequestContent, FunctionArguments, FunctionCallContent, FunctionResultContent,
34 Message, ResponseFormat, Role, TextContent, TextReasoningContent, TextSpanRegion, ToolMode,
35 UriContent, UsageContent, UsageDetails,
36};
37use futures::StreamExt;
38use serde_json::{json, Map, Value};
39
40use crate::convert::{
41 audio_format, data_content_media_type, function_arguments_to_string, result_to_string,
42 top_level_media_type, DEFAULT_FILENAME,
43};
44use crate::{ByteStream, DEFAULT_BASE_URL};
45
46#[derive(Clone)]
48pub struct OpenAIChatClient {
49 inner: Arc<Inner>,
50}
51
52#[derive(Clone)]
53struct Inner {
54 http: reqwest::Client,
55 api_key: String,
56 base_url: String,
57 model: String,
58 organization: Option<String>,
59}
60
61impl std::fmt::Debug for OpenAIChatClient {
62 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
63 f.debug_struct("OpenAIChatClient")
64 .field("base_url", &self.inner.base_url)
65 .field("model", &self.inner.model)
66 .field("organization", &self.inner.organization)
67 .finish_non_exhaustive()
68 }
69}
70
71impl OpenAIChatClient {
72 pub fn new(api_key: impl Into<String>, model: impl Into<String>) -> Self {
74 Self {
75 inner: Arc::new(Inner {
76 http: reqwest::Client::new(),
77 api_key: api_key.into(),
78 base_url: DEFAULT_BASE_URL.to_string(),
79 model: model.into(),
80 organization: None,
81 }),
82 }
83 }
84
85 pub fn from_env(model: impl Into<String>) -> Result<Self> {
88 let key = std::env::var("OPENAI_API_KEY")
89 .map_err(|_| Error::Configuration("OPENAI_API_KEY is not set".into()))?;
90 let mut client = Self::new(key, model);
91 if let Ok(base) = std::env::var("OPENAI_BASE_URL") {
92 client = client.with_base_url(base);
93 }
94 Ok(client)
95 }
96
97 pub fn with_base_url(mut self, base_url: impl Into<String>) -> Self {
99 Arc::make_mut(&mut self.inner).base_url = base_url.into();
100 self
101 }
102
103 pub fn with_organization(mut self, org: impl Into<String>) -> Self {
105 Arc::make_mut(&mut self.inner).organization = Some(org.into());
106 self
107 }
108
109 pub fn model(&self) -> &str {
111 &self.inner.model
112 }
113
114 fn build_body(&self, messages: &[Message], options: &ChatOptions, stream: bool) -> Value {
115 let mut body = Map::new();
116 let model = options
117 .model
118 .clone()
119 .unwrap_or_else(|| self.inner.model.clone());
120 body.insert("model".into(), json!(model));
121
122 let (instructions, rest) = extract_instructions(messages, options.instructions.as_deref());
123 if let Some(instructions) = instructions {
124 body.insert("instructions".into(), json!(instructions));
125 }
126 body.insert("input".into(), json!(messages_to_input(rest)));
127
128 if let Some(conversation_id) = &options.conversation_id {
129 body.insert("previous_response_id".into(), json!(conversation_id));
130 }
131 if let Some(t) = options.temperature {
132 body.insert("temperature".into(), json!(t));
133 }
134 if let Some(t) = options.top_p {
135 body.insert("top_p".into(), json!(t));
136 }
137 if let Some(mt) = options.max_tokens {
138 body.insert("max_output_tokens".into(), json!(mt));
139 }
140 if let Some(store) = options.store {
141 body.insert("store".into(), json!(store));
142 }
143 if let Some(user) = &options.user {
144 body.insert("user".into(), json!(user));
145 }
146 if let Some(metadata) = &options.metadata {
147 body.insert("metadata".into(), json!(metadata));
148 }
149
150 if !options.tools.is_empty() {
151 let tools: Vec<Value> = options.tools.iter().map(tool_to_responses_spec).collect();
152 body.insert("tools".into(), json!(tools));
153 if let Some(allow_multi) = options.allow_multiple_tool_calls {
154 body.insert("parallel_tool_calls".into(), json!(allow_multi));
155 }
156 }
157 if let Some(tool_choice) = &options.tool_choice {
158 body.insert("tool_choice".into(), tool_choice_to_responses(tool_choice));
159 }
160 if let Some(fmt) = &options.response_format {
161 body.insert(
162 "text".into(),
163 json!({ "format": response_format_to_text(fmt) }),
164 );
165 }
166
167 for (k, v) in &options.additional_properties {
168 body.entry(k.clone()).or_insert_with(|| v.clone());
169 }
170
171 if stream {
172 body.insert("stream".into(), json!(true));
173 }
174 Value::Object(body)
175 }
176
177 async fn post(&self, body: &Value) -> Result<reqwest::Response> {
178 let url = format!("{}/responses", self.inner.base_url.trim_end_matches('/'));
179 let mut req = self
180 .inner
181 .http
182 .post(&url)
183 .bearer_auth(&self.inner.api_key)
184 .json(body);
185 if let Some(org) = &self.inner.organization {
186 req = req.header("OpenAI-Organization", org);
187 }
188 let resp = req
189 .send()
190 .await
191 .map_err(|e| Error::service(format!("request failed: {e}")))?;
192 if !resp.status().is_success() {
193 let status = resp.status();
194 let retry_after = crate::parse_retry_after(resp.headers());
195 let text = resp.text().await.unwrap_or_default();
196 return Err(crate::classify_service_error(
197 status.as_u16(),
198 &text,
199 format!("OpenAI API error {status}: {text}"),
200 retry_after,
201 ));
202 }
203 Ok(resp)
204 }
205}
206
207#[async_trait::async_trait]
208impl ChatClient for OpenAIChatClient {
209 async fn get_response(
210 &self,
211 messages: Vec<Message>,
212 options: ChatOptions,
213 ) -> Result<ChatResponse> {
214 let body = self.build_body(&messages, &options, false);
215 let resp = self.post(&body).await?;
216 let value: Value = resp
217 .json()
218 .await
219 .map_err(|e| Error::service(format!("invalid response json: {e}")))?;
220 if let Some(err) = response_failure_error(&value) {
221 return Err(err);
222 }
223 Ok(parse_response(&value, options.store))
224 }
225
226 async fn get_streaming_response(
227 &self,
228 messages: Vec<Message>,
229 options: ChatOptions,
230 ) -> Result<ChatStream> {
231 let body = self.build_body(&messages, &options, true);
232 let resp = self.post(&body).await?;
233 Ok(parse_responses_sse_stream(resp, options.store).boxed())
234 }
235
236 fn model(&self) -> Option<&str> {
237 Some(&self.inner.model)
238 }
239}
240
241pub fn extract_instructions<'a>(
252 messages: &'a [Message],
253 options_instructions: Option<&str>,
254) -> (Option<String>, &'a [Message]) {
255 let mut parts = Vec::new();
256 if let Some(instr) = options_instructions {
257 if !instr.is_empty() {
258 parts.push(instr.to_string());
259 }
260 }
261 let mut rest = messages;
262 if let Some(first) = messages.first() {
263 if first.role == Role::system() {
264 let text = first.text();
265 if !text.is_empty() {
266 parts.push(text);
267 }
268 rest = &messages[1..];
269 }
270 }
271 if parts.is_empty() {
272 (None, rest)
273 } else {
274 (Some(parts.join("\n\n")), rest)
275 }
276}
277
278pub fn messages_to_input(messages: &[Message]) -> Vec<Value> {
284 let mut out = Vec::new();
285 for msg in messages {
286 let role = msg.role.as_str();
287 if role == Role::TOOL {
288 for content in &msg.contents {
289 if let Content::FunctionResult(fr) = content {
290 out.push(function_result_to_item(fr));
291 }
292 }
293 continue;
294 }
295
296 let mut buffered: Vec<Value> = Vec::new();
297 for content in &msg.contents {
298 match content {
299 Content::Text(t) => {
300 let text_type = if role == Role::ASSISTANT {
301 "output_text"
302 } else {
303 "input_text"
304 };
305 buffered.push(json!({ "type": text_type, "text": t.text }));
306 }
307 Content::Uri(u) => {
308 if let Some(part) = content_to_input_part(&u.uri, Some(&u.media_type)) {
309 buffered.push(part);
310 }
311 }
312 Content::Data(d) => {
313 if let Some(part) =
314 content_to_input_part(&d.uri, data_content_media_type(d).as_deref())
315 {
316 buffered.push(part);
317 }
318 }
319 Content::HostedFile(h) => {
320 buffered.push(json!({ "type": "input_file", "file_id": h.file_id }));
321 }
322 Content::FunctionCall(fc) => {
323 flush_text(&mut out, &mut buffered, role);
324 out.push(json!({
325 "type": "function_call",
326 "call_id": fc.call_id,
327 "name": fc.name,
328 "arguments": function_arguments_to_string(&fc.arguments),
329 }));
330 }
331 Content::FunctionResult(fr) => {
332 flush_text(&mut out, &mut buffered, role);
333 out.push(function_result_to_item(fr));
334 }
335 Content::FunctionApprovalResponse(r) => {
336 flush_text(&mut out, &mut buffered, role);
337 out.push(json!({
338 "type": "mcp_approval_response",
339 "approval_request_id": r.id,
340 "approve": r.approved,
341 }));
342 }
343 Content::FunctionApprovalRequest(r) => {
344 flush_text(&mut out, &mut buffered, role);
345 out.push(json!({
346 "type": "mcp_approval_request",
347 "id": r.id,
348 "name": r.function_call.name,
349 "arguments": function_arguments_to_string(&r.function_call.arguments),
350 }));
351 }
352 Content::TextReasoning(tr) => {
353 if let Some(raw) = &tr.raw_representation {
358 flush_text(&mut out, &mut buffered, role);
359 out.push(raw.clone());
360 }
361 }
362 _ => {}
363 }
364 }
365 flush_text(&mut out, &mut buffered, role);
366 }
367 out
368}
369
370fn content_to_input_part(uri: &str, media_type: Option<&str>) -> Option<Value> {
375 let media_type = media_type?;
376 match top_level_media_type(media_type).as_str() {
377 "image" => Some(json!({
379 "type": "input_image",
380 "image_url": uri,
381 "detail": "auto",
382 })),
383 "audio" => {
384 let format = audio_format(media_type)?;
385 Some(json!({
388 "type": "input_audio",
389 "input_audio": {
390 "data": crate::convert::strip_data_uri_prefix(uri),
391 "format": format,
392 },
393 }))
394 }
395 "application" => Some(json!({
396 "type": "input_file",
397 "file_data": uri,
398 "filename": DEFAULT_FILENAME,
399 })),
400 _ => None,
401 }
402}
403
404fn flush_text(out: &mut Vec<Value>, buffered: &mut Vec<Value>, role: &str) {
405 if !buffered.is_empty() {
406 out.push(json!({ "type": "message", "role": role, "content": std::mem::take(buffered) }));
407 }
408}
409
410fn function_result_to_item(fr: &FunctionResultContent) -> Value {
411 json!({
412 "type": "function_call_output",
413 "call_id": fr.call_id,
414 "output": result_to_string(fr),
415 })
416}
417
418pub fn tool_to_responses_spec(tool: &ToolDefinition) -> Value {
424 use agent_framework_core::tools::ToolKind;
425 match &tool.kind {
426 ToolKind::HostedWebSearch => {
427 let mut spec = Map::new();
428 spec.insert("type".into(), json!("web_search"));
429 if let Some(loc) = tool.parameters.get("user_location") {
430 spec.insert("user_location".into(), user_location_to_responses(loc));
431 }
432 Value::Object(spec)
433 }
434 ToolKind::HostedCodeInterpreter => {
435 let mut spec = Map::new();
436 spec.insert("type".into(), json!("code_interpreter"));
437 if let Some(container) = tool.parameters.get("container") {
440 spec.insert("container".into(), container.clone());
441 } else {
442 let mut container = Map::new();
443 container.insert("type".into(), json!("auto"));
444 if let Some(file_ids) = tool.parameters.get("file_ids") {
445 container.insert("file_ids".into(), file_ids.clone());
446 }
447 spec.insert("container".into(), Value::Object(container));
448 }
449 Value::Object(spec)
450 }
451 ToolKind::HostedFileSearch { max_results } => {
452 let mut spec = Map::new();
453 spec.insert("type".into(), json!("file_search"));
454 if let Some(ids) = tool.parameters.get("vector_store_ids") {
458 spec.insert("vector_store_ids".into(), ids.clone());
459 }
460 let max = (*max_results)
462 .map(|n| json!(n))
463 .or_else(|| tool.parameters.get("max_results").cloned());
464 if let Some(n) = max {
465 spec.insert("max_num_results".into(), n);
466 }
467 Value::Object(spec)
468 }
469 ToolKind::HostedMcp { url, allowed_tools } => {
470 let mut spec = Map::new();
471 spec.insert("type".into(), json!("mcp"));
472 spec.insert("server_label".into(), json!(tool.name.replace(' ', "_")));
473 spec.insert("server_url".into(), json!(url));
474 if !tool.description.is_empty() {
475 spec.insert("server_description".into(), json!(tool.description));
476 }
477 if let Some(headers) = tool.parameters.get("headers") {
478 spec.insert("headers".into(), headers.clone());
479 }
480 if let Some(allowed) = allowed_tools {
481 spec.insert("allowed_tools".into(), json!(allowed));
482 }
483 spec.insert("require_approval".into(), mcp_require_approval(tool));
484 Value::Object(spec)
485 }
486 ToolKind::HostedImageGeneration => {
487 let mut spec = Map::new();
488 spec.insert("type".into(), json!("image_generation"));
489 if let Value::Object(params) = &tool.parameters {
492 for (k, v) in params {
493 if k != "type" && k != "properties" {
494 spec.insert(k.clone(), v.clone());
495 }
496 }
497 }
498 Value::Object(spec)
499 }
500 ToolKind::Function => json!({
501 "type": "function",
502 "name": tool.name,
503 "description": tool.description,
504 "parameters": tool.parameters,
505 }),
506 }
507}
508
509fn user_location_to_responses(location: &Value) -> Value {
512 let mut loc = Map::new();
513 loc.insert("type".into(), json!("approximate"));
514 for key in ["city", "country", "region", "timezone"] {
515 if let Some(v) = location.get(key) {
516 loc.insert(key.into(), v.clone());
517 }
518 }
519 Value::Object(loc)
520}
521
522fn mcp_require_approval(tool: &ToolDefinition) -> Value {
528 use agent_framework_core::tools::ApprovalMode;
529 match tool.parameters.get("approval_mode") {
530 Some(Value::String(s)) => {
531 return json!(if s == "always_require" {
532 "always"
533 } else {
534 "never"
535 });
536 }
537 Some(Value::Object(modes)) => {
538 let mut req = Map::new();
539 if let Some(always) = modes.get("always") {
540 req.insert("always".into(), json!({ "tool_names": always }));
541 }
542 if let Some(never) = modes.get("never") {
543 req.insert("never".into(), json!({ "tool_names": never }));
544 }
545 if !req.is_empty() {
546 return Value::Object(req);
547 }
548 }
549 _ => {}
550 }
551 json!(match tool.approval_mode {
552 ApprovalMode::AlwaysRequire => "always",
553 ApprovalMode::NeverRequire => "never",
554 })
555}
556
557pub fn tool_choice_to_responses(mode: &ToolMode) -> Value {
560 match mode {
561 ToolMode::Auto => json!("auto"),
562 ToolMode::None => json!("none"),
563 ToolMode::Required(Some(name)) => json!({ "type": "function", "name": name }),
564 ToolMode::Required(None) => json!("required"),
565 }
566}
567
568pub fn response_format_to_text(format: &ResponseFormat) -> Value {
575 match format {
576 ResponseFormat::Text => json!({ "type": "text" }),
577 ResponseFormat::JsonObject => json!({ "type": "json_object" }),
578 ResponseFormat::JsonSchema {
579 name,
580 description,
581 schema,
582 strict,
583 } => {
584 let mut obj = Map::new();
585 obj.insert("type".into(), json!("json_schema"));
586 obj.insert("name".into(), json!(name));
587 if let Some(d) = description {
588 obj.insert("description".into(), json!(d));
589 }
590 obj.insert("schema".into(), schema.clone());
591 if let Some(st) = strict {
592 obj.insert("strict".into(), json!(st));
593 }
594 Value::Object(obj)
595 }
596 }
597}
598
599pub fn response_failure_error(value: &Value) -> Option<Error> {
615 if value.get("status").and_then(Value::as_str) != Some("failed") {
616 return None;
617 }
618 let error = value.get("error");
619 let msg = error
620 .and_then(|e| e.get("message"))
621 .and_then(Value::as_str)
622 .unwrap_or("response failed")
623 .to_string();
624 let code = error.and_then(|e| e.get("code")).and_then(Value::as_str);
625 Some(match code {
626 Some("content_filter") => Error::service_content_filter(msg),
627 _ => Error::service(msg),
628 })
629}
630
631pub fn parse_response(value: &Value, store: Option<bool>) -> ChatResponse {
636 let mut response = ChatResponse {
637 response_id: value.get("id").and_then(Value::as_str).map(String::from),
638 model: value.get("model").and_then(Value::as_str).map(String::from),
639 ..Default::default()
640 };
641
642 let mut contents: Vec<Content> = Vec::new();
643 if let Some(items) = value.get("output").and_then(Value::as_array) {
644 for item in items {
645 parse_output_item(item, &mut contents);
646 }
647 }
648
649 let mut message = Message::with_contents(Role::assistant(), contents);
650 message.message_id = response.response_id.clone();
651 response.messages.push(message);
652
653 response.finish_reason = finish_reason_from_response(value);
654
655 if let Some(usage) = value.get("usage") {
656 response.usage_details = Some(parse_responses_usage(usage));
657 }
658 if store != Some(false) {
659 response.conversation_id = response.response_id.clone();
660 }
661 response
662}
663
664fn parse_output_item(item: &Value, contents: &mut Vec<Content>) {
665 match item.get("type").and_then(Value::as_str) {
666 Some("message") => {
667 if let Some(parts) = item.get("content").and_then(Value::as_array) {
668 for part in parts {
669 match part.get("type").and_then(Value::as_str) {
670 Some("output_text") => {
671 if let Some(text) = part.get("text").and_then(Value::as_str) {
672 let mut tc = TextContent::new(text);
673 tc.annotations = parse_annotations(part);
674 contents.push(Content::Text(tc));
675 }
676 }
677 Some("refusal") => {
678 if let Some(text) = part.get("refusal").and_then(Value::as_str) {
679 contents.push(Content::Text(TextContent::new(text)));
680 }
681 }
682 _ => {}
683 }
684 }
685 }
686 }
687 Some("function_call") => {
688 let call_id = item
689 .get("call_id")
690 .and_then(Value::as_str)
691 .unwrap_or_default()
692 .to_string();
693 let name = item
694 .get("name")
695 .and_then(Value::as_str)
696 .unwrap_or_default()
697 .to_string();
698 let args = item
699 .get("arguments")
700 .and_then(Value::as_str)
701 .unwrap_or("{}")
702 .to_string();
703 contents.push(Content::FunctionCall(FunctionCallContent::new(
704 call_id,
705 name,
706 Some(FunctionArguments::Raw(args)),
707 )));
708 }
709 Some("reasoning") => {
710 let summaries: Vec<String> = item
711 .get("summary")
712 .and_then(Value::as_array)
713 .map(|arr| {
714 arr.iter()
715 .filter_map(|s| s.get("text").and_then(Value::as_str))
716 .map(str::to_string)
717 .collect()
718 })
719 .unwrap_or_default();
720 let raw = Some(item.clone());
727 if summaries.is_empty() {
728 contents.push(Content::TextReasoning(TextReasoningContent {
730 text: String::new(),
731 annotations: None,
732 raw_representation: raw,
733 protected_data: None,
734 }));
735 } else {
736 let n = summaries.len();
737 for (i, text) in summaries.into_iter().enumerate() {
738 contents.push(Content::TextReasoning(TextReasoningContent {
739 text,
740 annotations: None,
741 raw_representation: (i == n - 1).then(|| raw.clone()).flatten(),
742 protected_data: None,
743 }));
744 }
745 }
746 }
747 Some("code_interpreter_call") => {
751 let outputs = item
752 .get("outputs")
753 .and_then(Value::as_array)
754 .filter(|a| !a.is_empty());
755 if let Some(outputs) = outputs {
756 for output in outputs {
757 match output.get("type").and_then(Value::as_str) {
758 Some("logs") => {
759 if let Some(logs) = output.get("logs").and_then(Value::as_str) {
760 contents.push(Content::Text(TextContent::new(logs)));
761 }
762 }
763 Some("image") => {
764 if let Some(url) = output.get("url").and_then(Value::as_str) {
765 contents.push(Content::Uri(UriContent {
766 uri: url.to_string(),
767 media_type: "image".to_string(),
768 }));
769 }
770 }
771 _ => {}
772 }
773 }
774 } else if let Some(code) = item.get("code").and_then(Value::as_str) {
775 contents.push(Content::Text(TextContent::new(code)));
776 }
777 }
778 Some("image_generation_call") => {
782 if let Some(result) = item.get("result").and_then(Value::as_str) {
783 let (uri, media_type) = if result.starts_with("data:") {
784 let media_type = if result.contains(';') {
785 result
786 .strip_prefix("data:")
787 .and_then(|r| r.split(';').next())
788 .filter(|s| !s.is_empty())
789 .map(String::from)
790 } else {
791 None
792 };
793 (result.to_string(), media_type)
794 } else {
795 (
796 format!("data:image/png;base64,{result}"),
797 Some("image/png".to_string()),
798 )
799 };
800 contents.push(Content::Data(DataContent { uri, media_type }));
801 }
802 }
803 Some("mcp_approval_request") => {
807 let id = item
808 .get("id")
809 .and_then(Value::as_str)
810 .unwrap_or_default()
811 .to_string();
812 let name = item
813 .get("name")
814 .and_then(Value::as_str)
815 .unwrap_or_default()
816 .to_string();
817 let args = item
818 .get("arguments")
819 .and_then(Value::as_str)
820 .unwrap_or("{}")
821 .to_string();
822 contents.push(Content::FunctionApprovalRequest(
823 FunctionApprovalRequestContent {
824 id: id.clone(),
825 function_call: FunctionCallContent::new(
826 id,
827 name,
828 Some(FunctionArguments::Raw(args)),
829 ),
830 },
831 ));
832 }
833 _ => {}
834 }
835}
836
837fn parse_annotations(part: &Value) -> Option<Vec<Annotation>> {
842 let arr = part.get("annotations").and_then(Value::as_array)?;
843 let mut out = Vec::new();
844 for ann in arr {
845 let str_field = |k: &str| ann.get(k).and_then(Value::as_str).map(String::from);
846 let regions = || {
847 Some(vec![TextSpanRegion {
848 start_index: ann.get("start_index").and_then(Value::as_i64),
849 end_index: ann.get("end_index").and_then(Value::as_i64),
850 }])
851 };
852 match ann.get("type").and_then(Value::as_str) {
853 Some("file_path") => out.push(Annotation {
854 file_id: str_field("file_id"),
855 ..Default::default()
856 }),
857 Some("file_citation") => out.push(Annotation {
858 url: str_field("filename"),
859 file_id: str_field("file_id"),
860 ..Default::default()
861 }),
862 Some("url_citation") => out.push(Annotation {
863 title: str_field("title"),
864 url: str_field("url"),
865 annotated_regions: regions(),
866 ..Default::default()
867 }),
868 Some("container_file_citation") => out.push(Annotation {
869 file_id: str_field("file_id"),
870 url: str_field("filename"),
871 annotated_regions: regions(),
872 ..Default::default()
873 }),
874 _ => {}
875 }
876 }
877 if out.is_empty() {
878 None
879 } else {
880 Some(out)
881 }
882}
883
884fn finish_reason_from_response(value: &Value) -> Option<FinishReason> {
885 let has_function_call = value
886 .get("output")
887 .and_then(Value::as_array)
888 .map(|items| {
889 items
890 .iter()
891 .any(|i| i.get("type").and_then(Value::as_str) == Some("function_call"))
892 })
893 .unwrap_or(false);
894 if has_function_call {
895 return Some(FinishReason::tool_calls());
896 }
897 let status = value.get("status").and_then(Value::as_str)?;
898 Some(match status {
899 "completed" => FinishReason::stop(),
900 "incomplete" => match value
901 .get("incomplete_details")
902 .and_then(|d| d.get("reason"))
903 .and_then(Value::as_str)
904 {
905 Some("max_output_tokens") => FinishReason::new(FinishReason::LENGTH),
906 Some("content_filter") => FinishReason::new(FinishReason::CONTENT_FILTER),
907 Some(other) => FinishReason::new(other),
908 None => FinishReason::new("incomplete"),
909 },
910 other => FinishReason::new(other),
911 })
912}
913
914fn parse_responses_usage(usage: &Value) -> UsageDetails {
915 let mut details = UsageDetails {
916 input_token_count: usage.get("input_tokens").and_then(Value::as_u64),
917 output_token_count: usage.get("output_tokens").and_then(Value::as_u64),
918 total_token_count: usage.get("total_tokens").and_then(Value::as_u64),
919 ..Default::default()
920 };
921 if let Some(cached) = usage
922 .get("input_tokens_details")
923 .and_then(|d| d.get("cached_tokens"))
924 .and_then(Value::as_u64)
925 {
926 details
927 .additional_counts
928 .insert("openai.cached_input_tokens".into(), cached);
929 details.cache_read_input_token_count = Some(cached);
931 }
932 if let Some(cache_write) = usage
935 .get("input_tokens_details")
936 .and_then(|d| d.get("cache_write_tokens"))
937 .and_then(Value::as_u64)
938 {
939 details
940 .additional_counts
941 .insert("openai.cache_write_tokens".into(), cache_write);
942 details.cache_creation_input_token_count = Some(cache_write);
943 }
944 if let Some(reasoning) = usage
945 .get("output_tokens_details")
946 .and_then(|d| d.get("reasoning_tokens"))
947 .and_then(Value::as_u64)
948 {
949 details
950 .additional_counts
951 .insert("openai.reasoning_tokens".into(), reasoning);
952 details.reasoning_output_token_count = Some(reasoning);
953 }
954 details
955}
956
957pub fn parse_responses_sse_stream(
967 resp: reqwest::Response,
968 store: Option<bool>,
969) -> impl futures::Stream<Item = Result<ChatResponseUpdate>> + Send {
970 let byte_stream: ByteStream = Box::pin(resp.bytes_stream());
971 futures::stream::unfold(
972 ResponsesSseState {
973 byte_stream,
974 buffer: String::new(),
975 utf8: Utf8StreamDecoder::new(),
976 queued: VecDeque::new(),
977 call_ids: HashMap::new(),
978 done: false,
979 store,
980 },
981 |mut state| async move {
982 loop {
983 if let Some(update) = state.queued.pop_front() {
984 return Some((Ok(update), state));
985 }
986 if state.done {
987 return None;
988 }
989 match state.byte_stream.next().await {
990 Some(Ok(bytes)) => {
991 let decoded = state.utf8.push(&bytes);
992 state.buffer.push_str(&decoded);
993 while let Some(pos) = state.buffer.find('\n') {
994 let line = state.buffer[..pos].trim().to_string();
995 state.buffer.drain(..=pos);
996 let Some(data) = line.strip_prefix("data:") else {
997 continue;
998 };
999 let data = data.trim();
1000 if data.is_empty() {
1001 continue;
1002 }
1003 let Ok(value) = serde_json::from_str::<Value>(data) else {
1004 continue;
1005 };
1006 match parse_responses_event(&value, &mut state.call_ids, state.store) {
1007 EventOutcome::Update(update) => state.queued.push_back(update),
1008 EventOutcome::Error(e) => {
1009 state.done = true;
1010 return Some((Err(e), state));
1011 }
1012 EventOutcome::None => {}
1013 }
1014 }
1015 }
1016 Some(Err(e)) => {
1017 state.done = true;
1018 return Some((Err(Error::service(format!("stream error: {e}"))), state));
1019 }
1020 None => return None,
1021 }
1022 }
1023 },
1024 )
1025}
1026
1027struct ResponsesSseState {
1028 byte_stream: ByteStream,
1029 buffer: String,
1030 utf8: Utf8StreamDecoder,
1031 queued: VecDeque<ChatResponseUpdate>,
1032 call_ids: HashMap<i64, String>,
1036 done: bool,
1037 store: Option<bool>,
1038}
1039
1040#[allow(clippy::large_enum_variant)]
1045enum EventOutcome {
1046 Update(ChatResponseUpdate),
1047 Error(Error),
1048 None,
1049}
1050
1051fn reasoning_update(text: &str) -> EventOutcome {
1054 if text.is_empty() {
1055 return EventOutcome::None;
1056 }
1057 EventOutcome::Update(ChatResponseUpdate {
1058 contents: vec![Content::TextReasoning(TextReasoningContent {
1059 text: text.to_string(),
1060 annotations: None,
1061 ..Default::default()
1062 })],
1063 role: Some(Role::assistant()),
1064 ..Default::default()
1065 })
1066}
1067
1068fn parse_responses_event(
1070 value: &Value,
1071 call_ids: &mut HashMap<i64, String>,
1072 store: Option<bool>,
1073) -> EventOutcome {
1074 let event_type = value.get("type").and_then(Value::as_str).unwrap_or("");
1075 match event_type {
1076 "response.created" => {
1077 let resp = value.get("response");
1078 let response_id = resp
1079 .and_then(|r| r.get("id"))
1080 .and_then(Value::as_str)
1081 .map(String::from);
1082 let model = resp
1083 .and_then(|r| r.get("model"))
1084 .and_then(Value::as_str)
1085 .map(String::from);
1086 if response_id.is_none() && model.is_none() {
1087 return EventOutcome::None;
1088 }
1089 EventOutcome::Update(ChatResponseUpdate {
1090 role: Some(Role::assistant()),
1091 response_id,
1092 model,
1093 ..Default::default()
1094 })
1095 }
1096 "response.output_text.delta" => {
1097 let text = value.get("delta").and_then(Value::as_str).unwrap_or("");
1098 if text.is_empty() {
1099 return EventOutcome::None;
1100 }
1101 EventOutcome::Update(ChatResponseUpdate {
1102 contents: vec![Content::Text(TextContent::new(text))],
1103 role: Some(Role::assistant()),
1104 ..Default::default()
1105 })
1106 }
1107 "response.reasoning_text.delta" | "response.reasoning_summary_text.delta" => {
1112 reasoning_update(value.get("delta").and_then(Value::as_str).unwrap_or(""))
1113 }
1114 "response.reasoning_text.done" | "response.reasoning_summary_text.done" => {
1120 EventOutcome::None
1121 }
1122 "response.output_item.added" => {
1123 let item = value.get("item");
1124 if item.and_then(|i| i.get("type")).and_then(Value::as_str) != Some("function_call") {
1125 return EventOutcome::None;
1126 }
1127 let output_index = value
1128 .get("output_index")
1129 .and_then(Value::as_i64)
1130 .unwrap_or(0);
1131 let call_id = item
1132 .and_then(|i| i.get("call_id"))
1133 .and_then(Value::as_str)
1134 .unwrap_or_default()
1135 .to_string();
1136 let name = item
1137 .and_then(|i| i.get("name"))
1138 .and_then(Value::as_str)
1139 .unwrap_or_default()
1140 .to_string();
1141 call_ids.insert(output_index, call_id.clone());
1142 EventOutcome::Update(ChatResponseUpdate {
1143 contents: vec![Content::FunctionCall(FunctionCallContent::new(
1144 call_id, name, None,
1145 ))],
1146 role: Some(Role::assistant()),
1147 ..Default::default()
1148 })
1149 }
1150 "response.function_call_arguments.delta" => {
1151 let output_index = value
1152 .get("output_index")
1153 .and_then(Value::as_i64)
1154 .unwrap_or(0);
1155 let delta = value.get("delta").and_then(Value::as_str).unwrap_or("");
1156 match call_ids.get(&output_index) {
1157 Some(call_id) => EventOutcome::Update(ChatResponseUpdate {
1158 contents: vec![Content::FunctionCall(FunctionCallContent::new(
1159 call_id.clone(),
1160 "",
1161 Some(FunctionArguments::Raw(delta.to_string())),
1162 ))],
1163 role: Some(Role::assistant()),
1164 ..Default::default()
1165 }),
1166 None => EventOutcome::None,
1167 }
1168 }
1169 "response.completed" => {
1170 let resp = value.get("response");
1171 let response_id = resp
1172 .and_then(|r| r.get("id"))
1173 .and_then(Value::as_str)
1174 .map(String::from);
1175 let model = resp
1176 .and_then(|r| r.get("model"))
1177 .and_then(Value::as_str)
1178 .map(String::from);
1179 let mut contents = Vec::new();
1180 if let Some(usage) = resp.and_then(|r| r.get("usage")) {
1181 contents.push(Content::Usage(UsageContent {
1182 details: parse_responses_usage(usage),
1183 }));
1184 }
1185 let finish_reason = resp.and_then(finish_reason_from_response);
1186 let conversation_id = if store != Some(false) {
1187 response_id.clone()
1188 } else {
1189 None
1190 };
1191 EventOutcome::Update(ChatResponseUpdate {
1192 contents,
1193 role: Some(Role::assistant()),
1194 response_id,
1195 model,
1196 conversation_id,
1197 finish_reason,
1198 ..Default::default()
1199 })
1200 }
1201 "response.failed" | "error" => {
1202 let resp = value.get("response");
1203 let err_obj = resp
1204 .and_then(|r| r.get("error"))
1205 .or_else(|| value.get("error"));
1206 let msg = err_obj
1207 .and_then(|e| e.get("message"))
1208 .and_then(Value::as_str)
1209 .unwrap_or("response failed")
1210 .to_string();
1211 let code = err_obj.and_then(|e| e.get("code")).and_then(Value::as_str);
1215 EventOutcome::Error(match code {
1216 Some("content_filter") => Error::service_content_filter(msg),
1217 _ => Error::service(msg),
1218 })
1219 }
1220 "response.function_call_arguments.done"
1224 | "response.output_item.done"
1225 | "response.content_part.added"
1226 | "response.content_part.done"
1227 | "response.in_progress" => EventOutcome::None,
1228 _ => EventOutcome::None,
1229 }
1230}
1231
1232#[cfg(test)]
1235mod tests {
1236 use super::*;
1237 use agent_framework_core::tools::{ApprovalMode, ToolDefinition, ToolKind};
1238 use agent_framework_core::types::{
1239 FunctionApprovalResponseContent, FunctionResultContent, HostedFileContent,
1240 };
1241
1242 fn user(text: &str) -> Message {
1243 Message::user(text)
1244 }
1245
1246 fn user_with(contents: Vec<Content>) -> Message {
1247 Message::with_contents(Role::user(), contents)
1248 }
1249
1250 fn parse_item(item: Value) -> Vec<Content> {
1252 let mut contents = Vec::new();
1253 parse_output_item(&item, &mut contents);
1254 contents
1255 }
1256
1257 fn client() -> OpenAIChatClient {
1258 OpenAIChatClient::new("test-key", "gpt-4o-mini")
1259 }
1260
1261 #[test]
1264 fn build_body_simple_text() {
1265 let c = client();
1266 let body = c.build_body(&[user("Hello there")], &ChatOptions::new(), false);
1267 assert_eq!(
1268 body,
1269 json!({
1270 "model": "gpt-4o-mini",
1271 "input": [
1272 { "type": "message", "role": "user", "content": [
1273 { "type": "input_text", "text": "Hello there" }
1274 ]}
1275 ],
1276 })
1277 );
1278 }
1279
1280 #[test]
1281 fn responses_usage_parses_cache_write_tokens() {
1282 let d = parse_responses_usage(&json!({
1284 "input_tokens": 2000,
1285 "output_tokens": 60,
1286 "total_tokens": 2060,
1287 "input_tokens_details": { "cached_tokens": 0, "cache_write_tokens": 1024 },
1288 }));
1289 assert_eq!(d.cache_creation_input_token_count, Some(1024));
1290 assert_eq!(d.cache_read_input_token_count, Some(0));
1291 assert_eq!(
1292 d.additional_counts.get("openai.cache_write_tokens"),
1293 Some(&1024)
1294 );
1295 }
1296
1297 #[test]
1298 fn responses_usage_omits_cache_write_tokens_when_not_reported() {
1299 let d = parse_responses_usage(&json!({
1300 "input_tokens": 100,
1301 "output_tokens": 20,
1302 "input_tokens_details": { "cached_tokens": 40 },
1303 }));
1304 assert_eq!(d.cache_creation_input_token_count, None);
1305 assert!(!d
1306 .additional_counts
1307 .contains_key("openai.cache_write_tokens"));
1308 }
1309
1310 #[test]
1311 fn build_body_extracts_leading_system_message_as_instructions() {
1312 let c = client();
1313 let messages = vec![Message::system("Be terse."), user("Hi")];
1314 let body = c.build_body(&messages, &ChatOptions::new(), false);
1315 assert_eq!(body["instructions"], json!("Be terse."));
1316 assert_eq!(
1317 body["input"],
1318 json!([
1319 { "type": "message", "role": "user", "content": [
1320 { "type": "input_text", "text": "Hi" }
1321 ]}
1322 ])
1323 );
1324 }
1325
1326 #[test]
1327 fn build_body_combines_options_instructions_and_system_message() {
1328 let c = client();
1329 let messages = vec![Message::system("Also be nice."), user("Hi")];
1330 let options = ChatOptions::new().with_instructions("Be terse.");
1331 let body = c.build_body(&messages, &options, false);
1332 assert_eq!(body["instructions"], json!("Be terse.\n\nAlso be nice."));
1333 }
1334
1335 #[test]
1336 fn build_body_assistant_text_uses_output_text_type() {
1337 let c = client();
1338 let messages = vec![user("Hi"), Message::assistant("Hello!")];
1339 let body = c.build_body(&messages, &ChatOptions::new(), false);
1340 assert_eq!(
1341 body["input"][1],
1342 json!({ "type": "message", "role": "assistant", "content": [
1343 { "type": "output_text", "text": "Hello!" }
1344 ]})
1345 );
1346 }
1347
1348 #[test]
1349 fn build_body_function_call_round_trip() {
1350 let c = client();
1351 let call = FunctionCallContent::new(
1352 "call_1",
1353 "get_weather",
1354 Some(FunctionArguments::Raw(r#"{"city":"Paris"}"#.to_string())),
1355 );
1356 let assistant_msg =
1357 Message::with_contents(Role::assistant(), vec![Content::FunctionCall(call)]);
1358 let tool_msg = Message::with_contents(
1359 Role::tool(),
1360 vec![Content::FunctionResult(FunctionResultContent::new(
1361 "call_1",
1362 Some(json!("18C and sunny")),
1363 ))],
1364 );
1365 let body = c.build_body(
1366 &[user("weather?"), assistant_msg, tool_msg],
1367 &ChatOptions::new(),
1368 false,
1369 );
1370 assert_eq!(
1371 body["input"],
1372 json!([
1373 { "type": "message", "role": "user", "content": [
1374 { "type": "input_text", "text": "weather?" }
1375 ]},
1376 { "type": "function_call", "call_id": "call_1", "name": "get_weather", "arguments": "{\"city\":\"Paris\"}" },
1377 { "type": "function_call_output", "call_id": "call_1", "output": "18C and sunny" },
1378 ])
1379 );
1380 }
1381
1382 #[test]
1383 fn build_body_tools_are_flat_not_nested() {
1384 let c = client();
1385 let tool = ToolDefinition {
1386 name: "get_weather".into(),
1387 description: "Get the weather".into(),
1388 parameters: json!({ "type": "object", "properties": {} }),
1389 kind: ToolKind::Function,
1390 approval_mode: ApprovalMode::NeverRequire,
1391 executor: None,
1392 };
1393 let options = ChatOptions::new().with_tool(tool);
1394 let body = c.build_body(&[user("hi")], &options, false);
1395 assert_eq!(
1396 body["tools"],
1397 json!([{
1398 "type": "function",
1399 "name": "get_weather",
1400 "description": "Get the weather",
1401 "parameters": { "type": "object", "properties": {} },
1402 }])
1403 );
1404 }
1405
1406 #[test]
1407 fn build_body_tool_choice_required_named() {
1408 let c = client();
1409 let options =
1410 ChatOptions::new().with_tool_choice(ToolMode::Required(Some("get_weather".into())));
1411 let body = c.build_body(&[user("hi")], &options, false);
1412 assert_eq!(
1413 body["tool_choice"],
1414 json!({ "type": "function", "name": "get_weather" })
1415 );
1416 }
1417
1418 #[test]
1419 fn build_body_conversation_id_becomes_previous_response_id() {
1420 let c = client();
1421 let mut options = ChatOptions::new();
1422 options.conversation_id = Some("resp_abc123".into());
1423 let body = c.build_body(&[user("hi")], &options, false);
1424 assert_eq!(body["previous_response_id"], json!("resp_abc123"));
1425 }
1426
1427 #[test]
1428 fn build_body_max_tokens_becomes_max_output_tokens() {
1429 let c = client();
1430 let options = ChatOptions::new().with_max_tokens(256);
1431 let body = c.build_body(&[user("hi")], &options, false);
1432 assert_eq!(body["max_output_tokens"], json!(256));
1433 assert!(body.get("max_tokens").is_none());
1434 }
1435
1436 #[test]
1437 fn build_body_response_format_becomes_text_format() {
1438 let c = client();
1439 let mut options = ChatOptions::new();
1440 options.response_format = Some(ResponseFormat::JsonSchema {
1441 name: "answer".into(),
1442 description: None,
1443 schema: json!({"type": "object"}),
1444 strict: Some(true),
1445 });
1446 let body = c.build_body(&[user("hi")], &options, false);
1447 assert_eq!(
1448 body["text"]["format"],
1449 json!({ "type": "json_schema", "name": "answer", "schema": {"type": "object"}, "strict": true })
1450 );
1451 }
1452
1453 #[test]
1454 fn build_body_stream_flag() {
1455 let c = client();
1456 let body = c.build_body(&[user("hi")], &ChatOptions::new(), true);
1457 assert_eq!(body["stream"], json!(true));
1458 }
1459
1460 #[test]
1465 fn parse_response_text_and_usage() {
1466 let value = json!({
1467 "id": "resp_123",
1468 "model": "gpt-4o-mini",
1469 "status": "completed",
1470 "output": [
1471 { "type": "message", "role": "assistant", "content": [
1472 { "type": "output_text", "text": "Hello!" }
1473 ]}
1474 ],
1475 "usage": { "input_tokens": 10, "output_tokens": 5, "total_tokens": 15 },
1476 });
1477 let resp = parse_response(&value, None);
1478 assert_eq!(resp.response_id.as_deref(), Some("resp_123"));
1479 assert_eq!(resp.conversation_id.as_deref(), Some("resp_123"));
1480 assert_eq!(resp.text(), "Hello!");
1481 assert_eq!(resp.finish_reason, Some(FinishReason::stop()));
1482 let usage = resp.usage_details.unwrap();
1483 assert_eq!(usage.input_token_count, Some(10));
1484 assert_eq!(usage.output_token_count, Some(5));
1485 assert_eq!(usage.total_token_count, Some(15));
1486 }
1487
1488 #[test]
1489 fn parse_response_store_false_omits_conversation_id() {
1490 let value = json!({
1491 "id": "resp_123",
1492 "status": "completed",
1493 "output": [],
1494 });
1495 let resp = parse_response(&value, Some(false));
1496 assert_eq!(resp.conversation_id, None);
1497 }
1498
1499 #[test]
1500 fn parse_response_function_call_sets_tool_calls_finish_reason() {
1501 let value = json!({
1502 "id": "resp_123",
1503 "status": "completed",
1504 "output": [
1505 { "type": "function_call", "call_id": "call_1", "name": "get_weather", "arguments": "{\"city\":\"Paris\"}" }
1506 ],
1507 });
1508 let resp = parse_response(&value, None);
1509 assert_eq!(resp.finish_reason, Some(FinishReason::tool_calls()));
1510 let calls = resp.function_calls();
1511 assert_eq!(calls.len(), 1);
1512 assert_eq!(calls[0].call_id, "call_1");
1513 assert_eq!(calls[0].name, "get_weather");
1514 }
1515
1516 #[test]
1517 fn parse_response_incomplete_max_output_tokens_is_length() {
1518 let value = json!({
1519 "id": "resp_123",
1520 "status": "incomplete",
1521 "incomplete_details": { "reason": "max_output_tokens" },
1522 "output": [],
1523 });
1524 let resp = parse_response(&value, None);
1525 assert_eq!(
1526 resp.finish_reason,
1527 Some(FinishReason::new(FinishReason::LENGTH))
1528 );
1529 }
1530
1531 #[test]
1532 fn parse_response_reasoning_becomes_text_reasoning() {
1533 let value = json!({
1534 "id": "resp_123",
1535 "status": "completed",
1536 "output": [
1537 { "type": "reasoning", "summary": [{ "type": "summary_text", "text": "thinking..." }] },
1538 { "type": "message", "role": "assistant", "content": [{ "type": "output_text", "text": "done" }] },
1539 ],
1540 });
1541 let resp = parse_response(&value, None);
1542 let contents = &resp.messages[0].contents;
1543 assert!(matches!(&contents[0], Content::TextReasoning(t) if t.text == "thinking..."));
1544 assert!(matches!(&contents[1], Content::Text(t) if t.text == "done"));
1545 }
1546
1547 #[test]
1548 fn reasoning_item_round_trips_through_input_for_store_false_replay() {
1549 let reasoning_item = json!({
1552 "type": "reasoning",
1553 "id": "rs_abc",
1554 "encrypted_content": "ENC",
1555 "summary": [{ "type": "summary_text", "text": "thinking..." }],
1556 });
1557 let value = json!({
1558 "id": "resp_1",
1559 "status": "completed",
1560 "output": [
1561 reasoning_item,
1562 { "type": "function_call", "call_id": "c1", "name": "f", "arguments": "{}" },
1563 ],
1564 });
1565 let resp = parse_response(&value, Some(false));
1566 let reasoning = resp.messages[0]
1568 .contents
1569 .iter()
1570 .find_map(|c| match c {
1571 Content::TextReasoning(t) => Some(t),
1572 _ => None,
1573 })
1574 .expect("reasoning content");
1575 assert_eq!(
1576 reasoning.raw_representation.as_ref().unwrap()["id"],
1577 "rs_abc"
1578 );
1579
1580 let input = messages_to_input(&resp.messages);
1583 let reasoning_pos = input
1584 .iter()
1585 .position(|i| i.get("type") == Some(&json!("reasoning")))
1586 .expect("reasoning item re-emitted");
1587 assert_eq!(input[reasoning_pos]["id"], "rs_abc");
1588 assert_eq!(input[reasoning_pos]["encrypted_content"], "ENC");
1589 let call_pos = input
1590 .iter()
1591 .position(|i| i.get("type") == Some(&json!("function_call")))
1592 .expect("function call present");
1593 assert!(reasoning_pos < call_pos, "reasoning must precede the call");
1594 }
1595
1596 #[test]
1597 fn summary_only_reasoning_is_not_re_emitted_as_input() {
1598 let msg = Message::with_contents(
1602 Role::assistant(),
1603 vec![Content::TextReasoning(TextReasoningContent {
1604 text: "just display".into(),
1605 annotations: None,
1606 raw_representation: None,
1607 protected_data: None,
1608 })],
1609 );
1610 let input = messages_to_input(&[msg]);
1611 assert!(input
1612 .iter()
1613 .all(|i| i.get("type") != Some(&json!("reasoning"))));
1614 }
1615
1616 fn sse_body(events: &[(&str, Value)]) -> String {
1621 let mut out = String::new();
1622 for (event, data) in events {
1623 out.push_str(&format!("event: {event}\ndata: {}\n\n", data));
1624 }
1625 out
1626 }
1627
1628 async fn collect_updates(text: String) -> Vec<ChatResponseUpdate> {
1629 let stream =
1634 futures::stream::once(async move { Ok::<_, reqwest::Error>(bytes::Bytes::from(text)) });
1635 let byte_stream: ByteStream = Box::pin(stream);
1636 let mut state = ResponsesSseState {
1637 byte_stream,
1638 buffer: String::new(),
1639 utf8: Utf8StreamDecoder::new(),
1640 queued: VecDeque::new(),
1641 call_ids: HashMap::new(),
1642 done: false,
1643 store: None,
1644 };
1645 let mut updates = Vec::new();
1646 if let Some(Ok(bytes)) = state.byte_stream.next().await {
1648 let decoded = state.utf8.push(&bytes);
1649 state.buffer.push_str(&decoded);
1650 while let Some(pos) = state.buffer.find('\n') {
1651 let line = state.buffer[..pos].trim().to_string();
1652 state.buffer.drain(..=pos);
1653 let Some(data) = line.strip_prefix("data:") else {
1654 continue;
1655 };
1656 let data = data.trim();
1657 if data.is_empty() {
1658 continue;
1659 }
1660 let value: Value = serde_json::from_str(data).unwrap();
1661 if let EventOutcome::Update(u) =
1662 parse_responses_event(&value, &mut state.call_ids, state.store)
1663 {
1664 updates.push(u);
1665 }
1666 }
1667 }
1668 updates
1669 }
1670
1671 #[tokio::test]
1672 async fn stream_text_only_accumulates() {
1673 let text = sse_body(&[
1674 (
1675 "response.created",
1676 json!({ "type": "response.created", "response": { "id": "resp_1", "model": "gpt-4o-mini" } }),
1677 ),
1678 (
1679 "response.output_text.delta",
1680 json!({ "type": "response.output_text.delta", "delta": "Hel" }),
1681 ),
1682 (
1683 "response.output_text.delta",
1684 json!({ "type": "response.output_text.delta", "delta": "lo!" }),
1685 ),
1686 (
1687 "response.completed",
1688 json!({ "type": "response.completed", "response": { "id": "resp_1", "model": "gpt-4o-mini", "status": "completed", "output": [], "usage": { "input_tokens": 3, "output_tokens": 2 } } }),
1689 ),
1690 ]);
1691 let updates = collect_updates(text).await;
1692 let resp = ChatResponse::from_updates(updates);
1693 assert_eq!(resp.text(), "Hello!");
1694 assert_eq!(resp.response_id.as_deref(), Some("resp_1"));
1695 assert_eq!(resp.finish_reason, Some(FinishReason::stop()));
1696 let usage = resp.usage_details.unwrap();
1697 assert_eq!(usage.input_token_count, Some(3));
1698 assert_eq!(usage.output_token_count, Some(2));
1699 }
1700
1701 #[tokio::test]
1702 async fn stream_tool_call_accumulates_arguments() {
1703 let text = sse_body(&[
1704 (
1705 "response.output_item.added",
1706 json!({ "type": "response.output_item.added", "output_index": 0, "item": { "type": "function_call", "call_id": "call_1", "name": "get_weather" } }),
1707 ),
1708 (
1709 "response.function_call_arguments.delta",
1710 json!({ "type": "response.function_call_arguments.delta", "output_index": 0, "delta": "{\"city\":" }),
1711 ),
1712 (
1713 "response.function_call_arguments.delta",
1714 json!({ "type": "response.function_call_arguments.delta", "output_index": 0, "delta": "\"Paris\"}" }),
1715 ),
1716 (
1717 "response.completed",
1718 json!({ "type": "response.completed", "response": { "id": "resp_2", "status": "completed", "output": [{"type":"function_call","call_id":"call_1","name":"get_weather","arguments":"{\"city\":\"Paris\"}"}] } }),
1719 ),
1720 ]);
1721 let updates = collect_updates(text).await;
1722 let resp = ChatResponse::from_updates(updates);
1723 let calls = resp.function_calls();
1724 assert_eq!(calls.len(), 1);
1725 assert_eq!(calls[0].call_id, "call_1");
1726 assert_eq!(calls[0].name, "get_weather");
1727 assert_eq!(
1728 calls[0].parse_arguments().unwrap().get("city").unwrap(),
1729 &json!("Paris")
1730 );
1731 assert_eq!(resp.finish_reason, Some(FinishReason::tool_calls()));
1732 }
1733
1734 #[tokio::test]
1735 async fn stream_failed_event_is_error() {
1736 let text = sse_body(&[(
1737 "response.failed",
1738 json!({ "type": "response.failed", "response": { "error": { "message": "boom" } } }),
1739 )]);
1740 let stream =
1741 futures::stream::once(async move { Ok::<_, reqwest::Error>(bytes::Bytes::from(text)) });
1742 let byte_stream: ByteStream = Box::pin(stream);
1743 let mut state = ResponsesSseState {
1744 byte_stream,
1745 buffer: String::new(),
1746 utf8: Utf8StreamDecoder::new(),
1747 queued: VecDeque::new(),
1748 call_ids: HashMap::new(),
1749 done: false,
1750 store: None,
1751 };
1752 let bytes = state.byte_stream.next().await.unwrap().unwrap();
1753 let decoded = state.utf8.push(&bytes);
1754 state.buffer.push_str(&decoded);
1755 let mut saw_error = false;
1756 while let Some(pos) = state.buffer.find('\n') {
1757 let line = state.buffer[..pos].trim().to_string();
1758 state.buffer.drain(..=pos);
1759 let Some(data) = line.strip_prefix("data:") else {
1760 continue;
1761 };
1762 let data = data.trim();
1763 if data.is_empty() {
1764 continue;
1765 }
1766 let value: Value = serde_json::from_str(data).unwrap();
1767 if let EventOutcome::Error(e) =
1768 parse_responses_event(&value, &mut state.call_ids, state.store)
1769 {
1770 assert!(e.to_string().contains("boom"));
1771 saw_error = true;
1772 }
1773 }
1774 assert!(
1775 saw_error,
1776 "expected a response.failed event to surface an error"
1777 );
1778 }
1779
1780 #[test]
1781 fn stream_failed_event_classifies_content_filter() {
1782 let mut call_ids = HashMap::new();
1785 let filtered = parse_responses_event(
1786 &json!({
1787 "type": "response.failed",
1788 "response": { "error": { "code": "content_filter", "message": "blocked" } }
1789 }),
1790 &mut call_ids,
1791 None,
1792 );
1793 let EventOutcome::Error(err) = filtered else {
1794 panic!("expected an error outcome");
1795 };
1796 assert!(matches!(err, Error::ServiceContentFilter { .. }));
1797
1798 let generic = parse_responses_event(
1799 &json!({
1800 "type": "response.failed",
1801 "response": { "error": { "code": "server_error", "message": "boom" } }
1802 }),
1803 &mut call_ids,
1804 None,
1805 );
1806 let EventOutcome::Error(err) = generic else {
1807 panic!("expected an error outcome");
1808 };
1809 assert!(matches!(err, Error::Service(_)));
1810 }
1811
1812 static ENV_MUTEX: std::sync::Mutex<()> = std::sync::Mutex::new(());
1821
1822 #[test]
1823 fn from_env_reads_api_key_and_base_url() {
1824 let _guard = ENV_MUTEX.lock().unwrap();
1825 unsafe {
1828 std::env::set_var("OPENAI_API_KEY", "sk-test-123");
1829 std::env::set_var("OPENAI_BASE_URL", "https://example.test/v1");
1830 }
1831 let client = OpenAIChatClient::from_env("gpt-4o-mini").unwrap();
1832 assert_eq!(client.inner.api_key, "sk-test-123");
1833 assert_eq!(client.inner.base_url, "https://example.test/v1");
1834 unsafe {
1835 std::env::remove_var("OPENAI_API_KEY");
1836 std::env::remove_var("OPENAI_BASE_URL");
1837 }
1838 }
1839
1840 #[test]
1841 fn from_env_errors_when_api_key_missing() {
1842 let _guard = ENV_MUTEX.lock().unwrap();
1843 unsafe {
1845 std::env::remove_var("OPENAI_API_KEY");
1846 std::env::remove_var("OPENAI_BASE_URL");
1847 }
1848 let result = OpenAIChatClient::from_env("gpt-4o-mini");
1849 assert!(result.is_err());
1850 }
1851 #[test]
1852 fn build_body_maps_hosted_tools_to_responses_types() {
1853 use agent_framework_core::tools::{
1854 hosted_code_interpreter, hosted_file_search, hosted_mcp, hosted_web_search,
1855 };
1856 let c = client();
1857 let mut options = ChatOptions::new();
1858 options.tools = vec![
1859 hosted_web_search(),
1860 hosted_code_interpreter(),
1861 hosted_file_search(Some(7)),
1862 hosted_mcp(
1863 "docs",
1864 "https://mcp.example/sse",
1865 Some(vec!["search".into()]),
1866 ),
1867 ];
1868 let body = c.build_body(&[user("hi")], &options, false);
1869 let tools = body["tools"].as_array().unwrap();
1870 assert_eq!(tools[0], json!({ "type": "web_search" }));
1871 assert_eq!(
1872 tools[1],
1873 json!({ "type": "code_interpreter", "container": { "type": "auto" } })
1874 );
1875 assert_eq!(tools[2]["type"], "file_search");
1876 assert_eq!(tools[2]["max_num_results"], json!(7));
1877 assert_eq!(tools[3]["type"], "mcp");
1878 assert_eq!(tools[3]["server_url"], "https://mcp.example/sse");
1879 assert_eq!(tools[3]["allowed_tools"], json!(["search"]));
1880 assert_eq!(tools[3]["require_approval"], "never");
1881 }
1882
1883 #[test]
1888 fn input_image_uri_becomes_input_image_part() {
1889 let msg = user_with(vec![Content::Uri(UriContent {
1890 uri: "https://example.com/cat.png".into(),
1891 media_type: "image/png".into(),
1892 })]);
1893 let input = messages_to_input(&[msg]);
1894 assert_eq!(
1895 input[0],
1896 json!({ "type": "message", "role": "user", "content": [
1897 { "type": "input_image", "image_url": "https://example.com/cat.png", "detail": "auto" }
1898 ]})
1899 );
1900 }
1901
1902 #[test]
1903 fn input_audio_file_and_hosted_file_parts() {
1904 let msg = user_with(vec![
1905 Content::Data(DataContent {
1906 uri: "data:audio/wav;base64,QQ".into(),
1907 media_type: Some("audio/wav".into()),
1908 }),
1909 Content::Data(DataContent {
1910 uri: "data:application/pdf;base64,JV".into(),
1911 media_type: Some("application/pdf".into()),
1912 }),
1913 Content::HostedFile(HostedFileContent {
1914 file_id: "file-123".into(),
1915 }),
1916 ]);
1917 let input = messages_to_input(&[msg]);
1918 assert_eq!(
1922 input[0]["content"],
1923 json!([
1924 { "type": "input_audio", "input_audio": { "data": "QQ", "format": "wav" } },
1925 { "type": "input_file", "file_data": "data:application/pdf;base64,JV", "filename": "file" },
1926 { "type": "input_file", "file_id": "file-123" },
1927 ])
1928 );
1929 }
1930
1931 #[test]
1932 fn approval_response_becomes_mcp_approval_response_item() {
1933 let resp = FunctionApprovalResponseContent {
1934 approved: true,
1935 id: "appr_1".into(),
1936 function_call: FunctionCallContent::new("appr_1", "search", None),
1937 };
1938 let msg = user_with(vec![Content::FunctionApprovalResponse(resp)]);
1939 let input = messages_to_input(&[msg]);
1940 assert_eq!(
1941 input[0],
1942 json!({
1943 "type": "mcp_approval_response",
1944 "approval_request_id": "appr_1",
1945 "approve": true,
1946 })
1947 );
1948 }
1949
1950 #[test]
1951 fn approval_request_becomes_mcp_approval_request_item() {
1952 let req = FunctionApprovalRequestContent {
1953 id: "appr_1".into(),
1954 function_call: FunctionCallContent::new(
1955 "appr_1",
1956 "search",
1957 Some(FunctionArguments::Raw(r#"{"q":"x"}"#.into())),
1958 ),
1959 };
1960 let msg = Message::with_contents(
1961 Role::assistant(),
1962 vec![Content::FunctionApprovalRequest(req)],
1963 );
1964 let input = messages_to_input(&[msg]);
1965 assert_eq!(
1966 input[0],
1967 json!({
1968 "type": "mcp_approval_request",
1969 "id": "appr_1",
1970 "name": "search",
1971 "arguments": r#"{"q":"x"}"#,
1972 })
1973 );
1974 }
1975
1976 #[test]
1981 fn output_text_url_citation_annotation() {
1982 let contents = parse_item(json!({
1983 "type": "message", "role": "assistant", "content": [{
1984 "type": "output_text", "text": "See source.",
1985 "annotations": [{
1986 "type": "url_citation", "title": "Src", "url": "https://ex.com",
1987 "start_index": 0, "end_index": 3,
1988 }],
1989 }],
1990 }));
1991 let Content::Text(t) = &contents[0] else {
1992 panic!("expected text content");
1993 };
1994 let ann = t.annotations.as_ref().unwrap();
1995 assert_eq!(ann[0].title.as_deref(), Some("Src"));
1996 assert_eq!(ann[0].url.as_deref(), Some("https://ex.com"));
1997 let region = &ann[0].annotated_regions.as_ref().unwrap()[0];
1998 assert_eq!(region.start_index, Some(0));
1999 assert_eq!(region.end_index, Some(3));
2000 }
2001
2002 #[test]
2003 fn output_text_file_and_container_citations() {
2004 let contents = parse_item(json!({
2005 "type": "message", "role": "assistant", "content": [{
2006 "type": "output_text", "text": "x",
2007 "annotations": [
2008 { "type": "file_citation", "filename": "doc.pdf", "file_id": "file-1", "index": 2 },
2009 { "type": "file_path", "file_id": "file-2", "index": 0 },
2010 { "type": "container_file_citation", "filename": "c.txt", "file_id": "file-3",
2011 "container_id": "cont-1", "start_index": 1, "end_index": 4 },
2012 ],
2013 }],
2014 }));
2015 let Content::Text(t) = &contents[0] else {
2016 panic!("expected text content");
2017 };
2018 let ann = t.annotations.as_ref().unwrap();
2019 assert_eq!(ann[0].url.as_deref(), Some("doc.pdf"));
2020 assert_eq!(ann[0].file_id.as_deref(), Some("file-1"));
2021 assert_eq!(ann[1].file_id.as_deref(), Some("file-2"));
2022 assert_eq!(ann[2].file_id.as_deref(), Some("file-3"));
2023 assert_eq!(ann[2].url.as_deref(), Some("c.txt"));
2024 assert_eq!(
2025 ann[2].annotated_regions.as_ref().unwrap()[0].end_index,
2026 Some(4)
2027 );
2028 }
2029
2030 #[test]
2031 fn code_interpreter_outputs_become_text_and_uri() {
2032 let contents = parse_item(json!({
2033 "type": "code_interpreter_call",
2034 "outputs": [
2035 { "type": "logs", "logs": "hello stdout" },
2036 { "type": "image", "url": "https://ex.com/plot.png" },
2037 ],
2038 }));
2039 assert!(matches!(&contents[0], Content::Text(t) if t.text == "hello stdout"));
2040 assert!(
2041 matches!(&contents[1], Content::Uri(u) if u.uri == "https://ex.com/plot.png" && u.media_type == "image")
2042 );
2043 }
2044
2045 #[test]
2046 fn code_interpreter_without_outputs_falls_back_to_code() {
2047 let contents = parse_item(json!({
2048 "type": "code_interpreter_call", "code": "print(1)",
2049 }));
2050 assert!(matches!(&contents[0], Content::Text(t) if t.text == "print(1)"));
2051 }
2052
2053 #[test]
2054 fn image_generation_raw_base64_becomes_png_data() {
2055 let contents = parse_item(json!({
2056 "type": "image_generation_call", "result": "AAAABBBB",
2057 }));
2058 let Content::Data(d) = &contents[0] else {
2059 panic!("expected data content");
2060 };
2061 assert_eq!(d.uri, "data:image/png;base64,AAAABBBB");
2062 assert_eq!(d.media_type.as_deref(), Some("image/png"));
2063 }
2064
2065 #[test]
2066 fn image_generation_data_uri_keeps_stated_media_type() {
2067 let contents = parse_item(json!({
2068 "type": "image_generation_call", "result": "data:image/webp;base64,ZZZ",
2069 }));
2070 let Content::Data(d) = &contents[0] else {
2071 panic!("expected data content");
2072 };
2073 assert_eq!(d.uri, "data:image/webp;base64,ZZZ");
2074 assert_eq!(d.media_type.as_deref(), Some("image/webp"));
2075 }
2076
2077 #[test]
2078 fn mcp_approval_request_output_round_trips_into_response() {
2079 let contents = parse_item(json!({
2080 "type": "mcp_approval_request",
2081 "id": "appr_9", "name": "search", "arguments": r#"{"q":"rust"}"#,
2082 "server_label": "docs",
2083 }));
2084 let Content::FunctionApprovalRequest(req) = &contents[0] else {
2085 panic!("expected approval request");
2086 };
2087 assert_eq!(req.id, "appr_9");
2088 assert_eq!(req.function_call.call_id, "appr_9");
2089 assert_eq!(req.function_call.name, "search");
2090
2091 let msg = user_with(vec![Content::FunctionApprovalResponse(
2093 req.create_response(true),
2094 )]);
2095 let input = messages_to_input(&[msg]);
2096 assert_eq!(input[0]["type"], json!("mcp_approval_response"));
2097 assert_eq!(input[0]["approval_request_id"], json!("appr_9"));
2098 assert_eq!(input[0]["approve"], json!(true));
2099 }
2100
2101 fn reasoning_event(value: Value) -> EventOutcome {
2106 let mut ids = HashMap::new();
2107 parse_responses_event(&value, &mut ids, None)
2108 }
2109
2110 #[test]
2111 fn reasoning_text_delta_streams_and_done_is_terminal_metadata() {
2112 let EventOutcome::Update(delta) =
2113 reasoning_event(json!({ "type": "response.reasoning_text.delta", "delta": "Th" }))
2114 else {
2115 panic!("expected update");
2116 };
2117 assert!(matches!(&delta.contents[0], Content::TextReasoning(t) if t.text == "Th"));
2118
2119 let done =
2122 reasoning_event(json!({ "type": "response.reasoning_text.done", "text": "Think" }));
2123 assert!(matches!(done, EventOutcome::None));
2124 }
2125
2126 #[test]
2127 fn reasoning_summary_text_events_map_to_reasoning_content() {
2128 let EventOutcome::Update(delta) = reasoning_event(
2129 json!({ "type": "response.reasoning_summary_text.delta", "delta": "sum" }),
2130 ) else {
2131 panic!("expected update");
2132 };
2133 assert!(matches!(&delta.contents[0], Content::TextReasoning(t) if t.text == "sum"));
2134
2135 let done = reasoning_event(
2136 json!({ "type": "response.reasoning_summary_text.done", "text": "summary" }),
2137 );
2138 assert!(matches!(done, EventOutcome::None));
2139 }
2140
2141 #[test]
2142 fn response_failure_error_classifies_content_filter() {
2143 let filtered = json!({
2144 "status": "failed",
2145 "error": { "code": "content_filter", "message": "blocked" }
2146 });
2147 assert!(matches!(
2148 response_failure_error(&filtered),
2149 Some(Error::ServiceContentFilter { .. })
2150 ));
2151
2152 let generic = json!({
2153 "status": "failed",
2154 "error": { "code": "server_error", "message": "boom" }
2155 });
2156 assert!(matches!(
2157 response_failure_error(&generic),
2158 Some(Error::Service(_))
2159 ));
2160
2161 assert!(response_failure_error(&json!({ "status": "completed" })).is_none());
2162 }
2163
2164 #[test]
2165 fn input_audio_data_strips_data_uri_prefix() {
2166 let part = content_to_input_part("data:audio/wav;base64,QUJD", Some("audio/wav"))
2167 .expect("audio part");
2168 assert_eq!(part["input_audio"]["data"], "QUJD");
2169 assert_eq!(part["input_audio"]["format"], "wav");
2170 }
2171
2172 fn hosted(kind: ToolKind, name: &str, params: Value) -> ToolDefinition {
2177 ToolDefinition {
2178 name: name.into(),
2179 description: String::new(),
2180 parameters: params,
2181 kind,
2182 approval_mode: ApprovalMode::NeverRequire,
2183 executor: None,
2184 }
2185 }
2186
2187 #[test]
2188 fn web_search_passes_through_user_location() {
2189 let tool = hosted(
2190 ToolKind::HostedWebSearch,
2191 "web_search",
2192 json!({ "user_location": { "city": "Paris", "country": "FR" } }),
2193 );
2194 assert_eq!(
2195 tool_to_responses_spec(&tool),
2196 json!({
2197 "type": "web_search",
2198 "user_location": { "type": "approximate", "city": "Paris", "country": "FR" },
2199 })
2200 );
2201 }
2202
2203 #[test]
2204 fn file_search_passes_vector_store_ids_and_max_results_param() {
2205 let tool = hosted(
2206 ToolKind::HostedFileSearch { max_results: None },
2207 "file_search",
2208 json!({ "vector_store_ids": ["vs_1"], "max_results": 12 }),
2209 );
2210 let spec = tool_to_responses_spec(&tool);
2211 assert_eq!(spec["vector_store_ids"], json!(["vs_1"]));
2212 assert_eq!(spec["max_num_results"], json!(12));
2213 }
2214
2215 #[test]
2216 fn image_generation_maps_to_responses_tool_with_passthrough_params() {
2217 let tool = hosted(
2218 ToolKind::HostedImageGeneration,
2219 "image_generation",
2220 json!({ "size": "1024x1024", "quality": "high" }),
2221 );
2222 let spec = tool_to_responses_spec(&tool);
2223 assert_eq!(spec["type"], json!("image_generation"));
2224 assert_eq!(spec["size"], json!("1024x1024"));
2225 assert_eq!(spec["quality"], json!("high"));
2226 }
2227
2228 #[test]
2229 fn code_interpreter_passes_file_ids_and_container_override() {
2230 let with_files = hosted(
2231 ToolKind::HostedCodeInterpreter,
2232 "ci",
2233 json!({ "file_ids": ["file-1", "file-2"] }),
2234 );
2235 assert_eq!(
2236 tool_to_responses_spec(&with_files)["container"],
2237 json!({ "type": "auto", "file_ids": ["file-1", "file-2"] })
2238 );
2239 let with_container = hosted(
2240 ToolKind::HostedCodeInterpreter,
2241 "ci",
2242 json!({ "container": { "type": "secure", "id": "c1" } }),
2243 );
2244 assert_eq!(
2245 tool_to_responses_spec(&with_container)["container"],
2246 json!({ "type": "secure", "id": "c1" })
2247 );
2248 }
2249
2250 #[test]
2251 fn mcp_passes_headers_and_string_approval_mode_override() {
2252 let tool = hosted(
2253 ToolKind::HostedMcp {
2254 url: "https://mcp/sse".into(),
2255 allowed_tools: None,
2256 },
2257 "docs",
2258 json!({ "headers": { "Authorization": "Bearer x" }, "approval_mode": "always_require" }),
2259 );
2260 let spec = tool_to_responses_spec(&tool);
2262 assert_eq!(spec["headers"], json!({ "Authorization": "Bearer x" }));
2263 assert_eq!(spec["require_approval"], json!("always"));
2264 }
2265
2266 #[test]
2267 fn mcp_object_approval_mode_maps_to_tool_name_lists() {
2268 let tool = hosted(
2269 ToolKind::HostedMcp {
2270 url: "https://mcp/sse".into(),
2271 allowed_tools: None,
2272 },
2273 "docs",
2274 json!({ "approval_mode": { "always": ["delete"], "never": ["read"] } }),
2275 );
2276 assert_eq!(
2277 tool_to_responses_spec(&tool)["require_approval"],
2278 json!({ "always": { "tool_names": ["delete"] }, "never": { "tool_names": ["read"] } })
2279 );
2280 }
2281
2282 }