codex_api/endpoint/
responses.rs

1use crate::auth::SharedAuthProvider;
2use crate::common::ResponseStream;
3use crate::common::ResponsesApiRequest;
4use crate::endpoint::session::EndpointSession;
5use crate::error::ApiError;
6use crate::provider::Provider;
7use crate::requests::Compression;
8use crate::requests::headers::build_session_headers;
9use crate::requests::headers::insert_header;
10use crate::requests::headers::subagent_header;
11use crate::sse::spawn_response_stream;
12use crate::telemetry::SseTelemetry;
13use codex_client::EncodedJsonBody;
14use codex_client::HttpTransport;
15use codex_client::RequestCompression;
16use codex_client::RequestTelemetry;
17use codex_protocol::protocol::SessionSource;
18use http::HeaderMap;
19use http::HeaderValue;
20use http::Method;
21use serde_json::Value;
22use std::sync::Arc;
23use std::sync::OnceLock;
24use tracing::instrument;
25
26pub struct ResponsesClient<T: HttpTransport> {
27    session: EndpointSession<T>,
28    sse_telemetry: Option<Arc<dyn SseTelemetry>>,
29}
30
31#[derive(Default)]
32pub struct ResponsesOptions {
33    pub session_id: Option<String>,
34    pub thread_id: Option<String>,
35    pub session_source: Option<SessionSource>,
36    pub extra_headers: HeaderMap,
37    pub compression: Compression,
38    pub turn_state: Option<Arc<OnceLock<String>>>,
39}
40
41impl<T: HttpTransport> ResponsesClient<T> {
42    pub fn new(transport: T, provider: Provider, auth: SharedAuthProvider) -> Self {
43        Self {
44            session: EndpointSession::new(transport, provider, auth),
45            sse_telemetry: None,
46        }
47    }
48
49    pub fn with_telemetry(
50        self,
51        request: Option<Arc<dyn RequestTelemetry>>,
52        sse: Option<Arc<dyn SseTelemetry>>,
53    ) -> Self {
54        Self {
55            session: self.session.with_request_telemetry(request),
56            sse_telemetry: sse,
57        }
58    }
59
60    #[instrument(
61        name = "responses.stream_request",
62        level = "info",
63        skip_all,
64        fields(
65            transport = "responses_http",
66            http.method = "POST",
67            api.path = "responses"
68        )
69    )]
70    pub async fn stream_request(
71        &self,
72        request: ResponsesApiRequest,
73        options: ResponsesOptions,
74    ) -> Result<ResponseStream, ApiError> {
75        let ResponsesOptions {
76            session_id,
77            thread_id,
78            session_source,
79            extra_headers,
80            compression,
81            turn_state,
82        } = options;
83
84        let body = EncodedJsonBody::encode(&request)
85            .map_err(|e| ApiError::Stream(format!("failed to encode responses request: {e}")))?;
86
87        let mut headers = extra_headers;
88        if let Some(ref thread_id) = thread_id {
89            insert_header(&mut headers, "x-client-request-id", thread_id);
90        }
91        headers.extend(build_session_headers(session_id, thread_id));
92        if let Some(subagent) = subagent_header(&session_source) {
93            insert_header(&mut headers, "x-openai-subagent", &subagent);
94        }
95
96        self.stream_encoded(body, headers, compression, turn_state)
97            .await
98    }
99
100    fn path() -> &'static str {
101        "responses"
102    }
103
104    #[instrument(
105        name = "responses.stream",
106        level = "info",
107        skip_all,
108        fields(
109            transport = "responses_http",
110            http.method = "POST",
111            api.path = "responses",
112            turn.has_state = turn_state.is_some()
113        )
114    )]
115    pub async fn stream(
116        &self,
117        body: Value,
118        extra_headers: HeaderMap,
119        compression: Compression,
120        turn_state: Option<Arc<OnceLock<String>>>,
121    ) -> Result<ResponseStream, ApiError> {
122        let body = EncodedJsonBody::encode(&body)
123            .map_err(|e| ApiError::Stream(format!("failed to encode responses request: {e}")))?;
124        self.stream_encoded(body, extra_headers, compression, turn_state)
125            .await
126    }
127
128    async fn stream_encoded(
129        &self,
130        body: EncodedJsonBody,
131        extra_headers: HeaderMap,
132        compression: Compression,
133        turn_state: Option<Arc<OnceLock<String>>>,
134    ) -> Result<ResponseStream, ApiError> {
135        let request_compression = match compression {
136            Compression::None => RequestCompression::None,
137            Compression::Zstd => RequestCompression::Zstd,
138        };
139
140        let stream_response = self
141            .session
142            .stream_encoded_json_with(
143                Method::POST,
144                Self::path(),
145                extra_headers,
146                Some(body),
147                |req| {
148                    req.headers.insert(
149                        http::header::ACCEPT,
150                        HeaderValue::from_static("text/event-stream"),
151                    );
152                    req.compression = request_compression;
153                },
154            )
155            .await?;
156
157        Ok(spawn_response_stream(
158            stream_response,
159            self.session.provider().stream_idle_timeout,
160            self.sse_telemetry.clone(),
161            turn_state,
162        ))
163    }
164}
codex_api/endpoint/responses.rs

codex_api/endpoint/
responses.rs
