openai/openai-python · error · TypeError

Expected custom parse type to be a subclass of {Stream} or {

Error message

Expected custom parse type to be a subclass of {Stream} or {AsyncStream}

What it means

When parsing an SSE streaming response with a custom cast_to type, that type must be a subclass of Stream or AsyncStream so the SDK can wrap the SSE iterator. Any other type (a plain model, dict, etc.) can't receive the stream and TypeError is raised.

Source

Thrown at src/openai/_response.py:145

        )

    def _parse(self, *, to: type[_T] | None = None) -> R | _T:
        cast_to = to if to is not None else self._cast_to

        # unwrap `TypeAlias('Name', T)` -> `T`
        if is_type_alias_type(cast_to):
            cast_to = cast_to.__value__  # type: ignore[unreachable]

        # unwrap `Annotated[T, ...]` -> `T`
        if cast_to and is_annotated_type(cast_to):
            cast_to = extract_type_arg(cast_to, 0)

        origin = get_origin(cast_to) or cast_to

        if self._is_sse_stream:
            if to:
                if not is_stream_class_type(to):
                    raise TypeError(f"Expected custom parse type to be a subclass of {Stream} or {AsyncStream}")

                return cast(
                    _T,
                    to(
                        cast_to=extract_stream_chunk_type(
                            to,
                            failure_message="Expected custom stream type to be passed with a type argument, e.g. Stream[ChunkType]",
                        ),
                        response=self.http_response,
                        client=cast(Any, self._client),
                        options=self._options,
                    ),
                )

            if self._stream_cls:
                return cast(
                    R,
                    self._stream_cls(

View on GitHub (pinned to 9917c6e28e)

Solutions

  1. Parse streams with the generated chunk type (e.g. stream = client.chat.completions.create(..., stream=True)) and iterate it
  2. If using a custom stream class, make it subclass openai.Stream (sync) or openai.AsyncStream (async)
  3. Use the non-streaming client method if you want a plain model back

Example fix

// before
resp = client.chat.completions.with_streaming_response.create(...)
result = resp.parse(ChatCompletion)  # not a Stream subclass
// after
with client.chat.completions.with_streaming_response.create(...) as resp:
    for chunk in resp.parse():  # default Stream[ChatCompletionChunk]
        ...
Defensive patterns

Strategy: type-guard

Validate before calling

from openai import Stream, AsyncStream
assert issubclass(cast_to, (Stream, AsyncStream)) or not is_streaming, 'custom parse type must subclass Stream'

Type guard

from openai import Stream, AsyncStream

def is_stream_class(t: type) -> TypeGuard[type[Stream] | type[AsyncStream]]:
    return isinstance(t, type) and issubclass(t, (Stream, AsyncStream))

Prevention

When it happens

Trigger: Calling .with_streaming_response...parse(MyModel) or .parse(some_non_stream_type) on a streaming endpoint; passing a custom class to cast_to that doesn't inherit from Stream/AsyncStream.

Common situations: Reusing a non-stream parse type with the streaming client; custom response wrappers that forget to subclass Stream/AsyncStream.

Related errors


AI-assisted analysis of openai/openai-python@9917c6e28e (2026-08-28). Data as JSON: /api/errors/644f4ab0c2edce10. Report an issue: GitHub.