Openai response

openai_response

OpenAIEndpointBase

OpenAIEndpointBase(endpoint_name, model_id, api_key=None, provider='openai', organization=None, project=None, base_url=None, websocket_base_url=None, timeout=None, max_retries=DEFAULT_MAX_RETRIES, default_headers=None, default_query=None, **kwargs)

Bases: Endpoint[TOpenAIResponseBase], Generic[TOpenAIResponseBase]

Base class for OpenAI Responses API endpoints (streaming and non-streaming)

Parameters:

Name	Type	Description	Default
`endpoint_name`	`str`	Name of the endpoint	required
`model_id`	`str`	ID of the OpenAI model to use	required
`api_key`	`str \| None`	OpenAI API key (optional, uses OPENAI_API_KEY env var if not provided)	`None`
`provider`	`str`	Provider name (default: "openai")	`'openai'`
`organization`	`str \| None`	OpenAI organization ID. Defaults to None.	`None`
`project`	`str \| None`	OpenAI project ID. Defaults to None.	`None`
`base_url`	`str \| URL \| None`	Override the default base URL for the API.	`None`
`websocket_base_url`	`str \| URL \| None`	Override the default base URL for websocket connections.	`None`
`timeout`	`float \| Timeout \| dict \| None`	Request timeout in seconds, or an httpx.Timeout for granular control. If a dict is passed (as is the case after serialization), it'll be interpreted as a set of arguments to create an httpx.Timeout.	`None`
`max_retries`	`int`	Maximum number of retries for failed requests.	`DEFAULT_MAX_RETRIES`
`default_headers`	`Mapping[str, str] \| None`	Additional headers to send with every request.	`None`
`default_query`	`Mapping[str, object] \| None`	Additional query parameters to send with every request.	`None`
`**kwargs`	`Any`	Additional arguments passed to OpenAI client	`{}`

Source code in llmeter/endpoints/openai_response.py

def __init__(
    self,
    endpoint_name: str,
    model_id: str,
    api_key: str | None = None,
    provider: str = "openai",
    organization: str | None = None,
    project: str | None = None,
    base_url: str | httpx.URL | None = None,
    websocket_base_url: str | httpx.URL | None = None,
    timeout: float | httpx.Timeout | dict | None = None,
    max_retries: int = DEFAULT_MAX_RETRIES,
    default_headers: Mapping[str, str] | None = None,
    default_query: Mapping[str, object] | None = None,
    **kwargs: Any,
):
    """Initialize Response API endpoint.

    Args:
        endpoint_name: Name of the endpoint
        model_id: ID of the OpenAI model to use
        api_key: OpenAI API key (optional, uses OPENAI_API_KEY env var if not provided)
        provider: Provider name (default: "openai")
        organization: OpenAI organization ID. Defaults to None.
        project: OpenAI project ID. Defaults to None.
        base_url: Override the default base URL for the API.
        websocket_base_url: Override the default base URL for websocket connections.
        timeout: Request timeout in seconds, or an httpx.Timeout for granular control. If a
            dict is passed (as is the case after serialization), it'll be interpreted as a set
            of arguments to create an httpx.Timeout.
        max_retries: Maximum number of retries for failed requests.
        default_headers: Additional headers to send with every request.
        default_query: Additional query parameters to send with every request.
        **kwargs: Additional arguments passed to OpenAI client
    """
    super().__init__(endpoint_name, model_id, provider=provider)
    client_kwargs: dict[str, Any] = {
        "api_key": api_key,
        "organization": organization,
        "project": project,
        "max_retries": max_retries,
        **kwargs,
    }
    if base_url is not None:
        client_kwargs["base_url"] = base_url
    if websocket_base_url is not None:
        client_kwargs["websocket_base_url"] = websocket_base_url
    if timeout is not None:
        if isinstance(timeout, dict):
            timeout = httpx.Timeout(**timeout)
        client_kwargs["timeout"] = timeout
    if default_headers is not None:
        client_kwargs["default_headers"] = default_headers
    if default_query is not None:
        client_kwargs["default_query"] = default_query
    self._client = OpenAI(**client_kwargs)

project `property`

project

The OpenAI project ID configured on the client.

create_payload `staticmethod`

create_payload(user_message, max_output_tokens=256, instructions=None, **kwargs)

Create a payload for the Responses API request.

This is a convenience helper. You can also build the payload directly using openai.types.responses.ResponseCreateParams.

Parameters:

Name	Type	Description	Default
`user_message`	`str \| Sequence[str]`	User message(s) to send (can be string or array of messages)	required
`max_output_tokens`	`int`	Maximum tokens in response (default: 256)	`256`
`instructions`	`str \| None`	Optional system-level instructions	`None`
`**kwargs`	`Any`	Additional payload parameters (temperature, top_p, text.format, etc.)	`{}`

Returns:

Type	Description
`ResponseCreateParams`	ResponseCreateParams formatted for Responses API

Source code in llmeter/endpoints/openai_response.py

@staticmethod
def create_payload(
    user_message: str | Sequence[str],
    max_output_tokens: int = 256,
    instructions: str | None = None,
    **kwargs: Any,
) -> ResponseCreateParams:
    """Create a payload for the Responses API request.

    This is a convenience helper. You can also build the payload directly
    using ``openai.types.responses.ResponseCreateParams``.

    Args:
        user_message: User message(s) to send (can be string or array of messages)
        max_output_tokens: Maximum tokens in response (default: 256)
        instructions: Optional system-level instructions
        **kwargs: Additional payload parameters (temperature, top_p, text.format, etc.)

    Returns:
        ResponseCreateParams formatted for Responses API
    """
    if isinstance(user_message, str):
        input_value: str | list[dict] = user_message
    else:
        input_value = [{"role": "user", "content": msg} for msg in user_message]

    payload: dict = {
        "input": input_value,
        "max_output_tokens": max_output_tokens,
    }

    if instructions:
        payload["instructions"] = instructions

    payload.update(kwargs)
    return cast(ResponseCreateParams, payload)

OpenAIResponseEndpoint

OpenAIResponseEndpoint(model_id, endpoint_name='openai-response', api_key=None, provider='openai', organization=None, project=None, base_url=None, websocket_base_url=None, timeout=None, max_retries=DEFAULT_MAX_RETRIES, default_headers=None, default_query=None, **kwargs)

Bases: OpenAIEndpointBase[Response]

Endpoint for OpenAI Responses API (non-streaming).

This endpoint provides access to OpenAI's newer Responses API which offers structured outputs, better response format control, and improved multi-turn conversation handling.

Parameters:

Name	Type	Description	Default
`model_id`	`str`	ID of the OpenAI model to use	required
`endpoint_name`	`str`	Name of the endpoint (default: "openai-response")	`'openai-response'`
`api_key`	`str \| None`	OpenAI API key (optional, uses OPENAI_API_KEY env var if not provided)	`None`
`provider`	`str`	Provider name (default: "openai")	`'openai'`
`organization`	`str \| None`	OpenAI organization ID. Defaults to None.	`None`
`project`	`str \| None`	OpenAI project ID. Defaults to None.	`None`
`base_url`	`str \| URL \| None`	Override the default base URL for the API.	`None`
`websocket_base_url`	`str \| URL \| None`	Override the default base URL for websocket connections.	`None`
`timeout`	`float \| Timeout \| dict \| None`	Request timeout in seconds, or an httpx.Timeout for granular control.	`None`
`max_retries`	`int`	Maximum number of retries for failed requests.	`DEFAULT_MAX_RETRIES`
`default_headers`	`Mapping[str, str] \| None`	Additional headers to send with every request.	`None`
`default_query`	`Mapping[str, object] \| None`	Additional query parameters to send with every request.	`None`
`**kwargs`	`Any`	Additional arguments passed to OpenAI client	`{}`

Source code in llmeter/endpoints/openai_response.py

def __init__(
    self,
    model_id: str,
    endpoint_name: str = "openai-response",
    api_key: str | None = None,
    provider: str = "openai",
    organization: str | None = None,
    project: str | None = None,
    base_url: str | httpx.URL | None = None,
    websocket_base_url: str | httpx.URL | None = None,
    timeout: float | httpx.Timeout | dict | None = None,
    max_retries: int = DEFAULT_MAX_RETRIES,
    default_headers: Mapping[str, str] | None = None,
    default_query: Mapping[str, object] | None = None,
    **kwargs: Any,
):
    """Initialize Response API endpoint.

    Args:
        model_id: ID of the OpenAI model to use
        endpoint_name: Name of the endpoint (default: "openai-response")
        api_key: OpenAI API key (optional, uses OPENAI_API_KEY env var if not provided)
        provider: Provider name (default: "openai")
        organization: OpenAI organization ID. Defaults to None.
        project: OpenAI project ID. Defaults to None.
        base_url: Override the default base URL for the API.
        websocket_base_url: Override the default base URL for websocket connections.
        timeout: Request timeout in seconds, or an httpx.Timeout for granular control.
        max_retries: Maximum number of retries for failed requests.
        default_headers: Additional headers to send with every request.
        default_query: Additional query parameters to send with every request.
        **kwargs: Additional arguments passed to OpenAI client
    """
    super().__init__(
        endpoint_name,
        model_id,
        api_key=api_key,
        provider=provider,
        organization=organization,
        project=project,
        base_url=base_url,
        websocket_base_url=websocket_base_url,
        timeout=timeout,
        max_retries=max_retries,
        default_headers=default_headers,
        default_query=default_query,
        **kwargs,
    )

invoke

invoke(payload)

Invoke the Responses API.

Source code in llmeter/endpoints/openai_response.py

@OpenAIEndpointBase.llmeter_invoke
def invoke(self, payload: ResponseCreateParamsNonStreaming) -> Response:
    """Invoke the Responses API."""
    client_response = self._client.responses.create(**payload)
    return client_response

prepare_payload

prepare_payload(payload)

Ensure payload specifies correct model ID and streaming disabled

Source code in llmeter/endpoints/openai_response.py

def prepare_payload(self, payload):
    """Ensure payload specifies correct model ID and streaming disabled"""
    return {
        **payload,
        "model": self.model_id,
        "stream": False,
    }

OpenAIResponseStreamEndpoint

OpenAIResponseStreamEndpoint(model_id, endpoint_name='openai-response-stream', api_key=None, provider='openai', ttft_visible_tokens_only=True, organization=None, project=None, base_url=None, websocket_base_url=None, timeout=None, max_retries=DEFAULT_MAX_RETRIES, default_headers=None, default_query=None, **kwargs)

Bases: OpenAIEndpointBase[Iterable[ResponseStreamEvent]]

Endpoint for OpenAI Responses API (streaming).

This endpoint provides streaming access to OpenAI's Responses API, enabling time-to-first-token measurements and incremental response processing.

Parameters:

Name	Type	Description	Default
`ttft_visible_tokens_only`	`bool`	Controls how `time_to_first_token` is measured for reasoning models. When `True` (default), TTFT records the time to the first visible text token (`response.output_text.delta`), ignoring reasoning events. When `False`, TTFT records the time to the first token of any kind — including reasoning summary or reasoning text deltas — giving a measure of when the model first started producing output. Has no effect for non-reasoning models.	`True`

Parameters:

Name	Type	Description	Default
`model_id`	`str`	ID of the OpenAI model to use	required
`endpoint_name`	`str`	Name of the endpoint (default: "openai-response-stream")	`'openai-response-stream'`
`api_key`	`str \| None`	OpenAI API key (optional, uses OPENAI_API_KEY env var if not provided)	`None`
`provider`	`str`	Provider name (default: "openai")	`'openai'`
`ttft_visible_tokens_only`	`bool`	When True (default), TTFT measures time to first visible text token. When False, TTFT includes reasoning token events.	`True`
`organization`	`str \| None`	OpenAI organization ID. Defaults to None.	`None`
`project`	`str \| None`	OpenAI project ID. Defaults to None.	`None`
`base_url`	`str \| URL \| None`	Override the default base URL for the API.	`None`
`websocket_base_url`	`str \| URL \| None`	Override the default base URL for websocket connections.	`None`
`timeout`	`float \| Timeout \| dict \| None`	Request timeout in seconds, or an httpx.Timeout for granular control.	`None`
`max_retries`	`int`	Maximum number of retries for failed requests.	`DEFAULT_MAX_RETRIES`
`default_headers`	`Mapping[str, str] \| None`	Additional headers to send with every request.	`None`
`default_query`	`Mapping[str, object] \| None`	Additional query parameters to send with every request.	`None`
`**kwargs`	`Any`	Additional arguments passed to OpenAI client	`{}`

Source code in llmeter/endpoints/openai_response.py

def __init__(
    self,
    model_id: str,
    endpoint_name: str = "openai-response-stream",
    api_key: str | None = None,
    provider: str = "openai",
    ttft_visible_tokens_only: bool = True,
    organization: str | None = None,
    project: str | None = None,
    base_url: str | httpx.URL | None = None,
    websocket_base_url: str | httpx.URL | None = None,
    timeout: float | httpx.Timeout | dict | None = None,
    max_retries: int = DEFAULT_MAX_RETRIES,
    default_headers: Mapping[str, str] | None = None,
    default_query: Mapping[str, object] | None = None,
    **kwargs: Any,
):
    """Initialize streaming Response API endpoint.

    Args:
        model_id: ID of the OpenAI model to use
        endpoint_name: Name of the endpoint (default: "openai-response-stream")
        api_key: OpenAI API key (optional, uses OPENAI_API_KEY env var if not provided)
        provider: Provider name (default: "openai")
        ttft_visible_tokens_only: When True (default), TTFT measures time to first visible text
            token. When False, TTFT includes reasoning token events.
        organization: OpenAI organization ID. Defaults to None.
        project: OpenAI project ID. Defaults to None.
        base_url: Override the default base URL for the API.
        websocket_base_url: Override the default base URL for websocket connections.
        timeout: Request timeout in seconds, or an httpx.Timeout for granular control.
        max_retries: Maximum number of retries for failed requests.
        default_headers: Additional headers to send with every request.
        default_query: Additional query parameters to send with every request.
        **kwargs: Additional arguments passed to OpenAI client
    """
    super().__init__(
        endpoint_name,
        model_id,
        api_key=api_key,
        provider=provider,
        organization=organization,
        project=project,
        base_url=base_url,
        websocket_base_url=websocket_base_url,
        timeout=timeout,
        max_retries=max_retries,
        default_headers=default_headers,
        default_query=default_query,
        **kwargs,
    )
    self.ttft_visible_tokens_only = ttft_visible_tokens_only

invoke

invoke(payload)

Invoke the Responses API with streaming.

Source code in llmeter/endpoints/openai_response.py

@OpenAIEndpointBase.llmeter_invoke
def invoke(self, payload: ResponseCreateParamsStreaming):
    """Invoke the Responses API with streaming."""
    client_response = self._client.responses.create(**payload)
    return client_response

prepare_payload

prepare_payload(payload)

Ensure payload specifies correct model ID and streaming options

Source code in llmeter/endpoints/openai_response.py

def prepare_payload(self, payload):
    """Ensure payload specifies correct model ID and streaming options"""
    payload = {**payload, "model": self.model_id}
    if not payload.get("stream"):
        payload["stream"] = True
        payload["stream_options"] = {"include_usage": True}
    return payload

process_raw_response

process_raw_response(raw_response, start_t, response)

Parse streaming Response API output into InvocationResponse.

Processes typed events from the stream:

ResponseCreatedEvent: captures response.id
ResponseTextDeltaEvent: accumulates text deltas, records TTFT
ResponseCompletedEvent: extracts usage from response.usage
ResponseFailedEvent: captures API-level errors
Reasoning events (response.reasoning_summary_text.delta, response.reasoning_text.delta): when ttft_visible_tokens_only is False, these set TTFT on the first reasoning token.

Source code in llmeter/endpoints/openai_response.py

def process_raw_response(
    self,
    raw_response: Iterable[ResponseStreamEvent],
    start_t: float,
    response: InvocationResponse,
) -> None:
    """Parse streaming Response API output into InvocationResponse.

    Processes typed events from the stream:

    - `ResponseCreatedEvent`: captures `response.id`
    - `ResponseTextDeltaEvent`: accumulates text deltas, records TTFT
    - `ResponseCompletedEvent`: extracts usage from `response.usage`
    - `ResponseFailedEvent`: captures API-level errors
    - Reasoning events (`response.reasoning_summary_text.delta`,
      `response.reasoning_text.delta`): when `ttft_visible_tokens_only` is ``False``, these set
      TTFT on the first reasoning token.
    """
    _REASONING_DELTA_TYPES = frozenset(
        (
            "response.reasoning_summary_text.delta",
            "response.reasoning_text.delta",
        )
    )

    for event in raw_response:
        now = time.perf_counter()
        if event.type == "response.created":
            response.id = event.response.id

        elif event.type == "response.output_text.delta":
            if response.response_text is None:
                response.response_text = event.delta
                if response.time_to_first_token is None:
                    response.time_to_first_token = now - start_t
            else:
                response.response_text += event.delta
            response.time_to_last_token = now - start_t

        elif (
            not self.ttft_visible_tokens_only
            and event.type in _REASONING_DELTA_TYPES
        ):
            if response.time_to_first_token is None:
                response.time_to_first_token = now - start_t

        elif event.type == "response.completed":
            usage = event.response.usage
            if usage is not None:
                response.num_tokens_input = usage.input_tokens
                response.num_tokens_output = usage.output_tokens
                details = getattr(usage, "input_tokens_details", None)
                if details:
                    response.num_tokens_input_cached = getattr(
                        details, "cached_tokens", None
                    )
                output_details = getattr(usage, "output_tokens_details", None)
                if output_details:
                    response.num_tokens_output_reasoning = getattr(
                        output_details, "reasoning_tokens", None
                    )

        elif event.type == "response.failed":
            error_obj = getattr(event.response, "error", None)
            if error_obj is not None:
                error_msg = getattr(error_obj, "message", None) or str(error_obj)
                error_code = getattr(error_obj, "code", None)
                if error_code:
                    error_msg = f"{error_code}: {error_msg}"
            else:
                error_msg = "Response API request failed"
            response.error = error_msg
            response.time_to_last_token = now - start_t

Openai response

openai_response

OpenAIEndpointBase

project property

create_payload staticmethod

OpenAIResponseEndpoint

invoke

prepare_payload

OpenAIResponseStreamEndpoint

invoke

prepare_payload

process_raw_response

project `property`

create_payload `staticmethod`