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