Skip to content
Closed
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
17 changes: 17 additions & 0 deletions .chronus/changes/structured-streaming-2026-0-0.md
Original file line number Diff line number Diff line change
@@ -0,0 +1,17 @@
---
changeKind: feature
packages:
- "@typespec/http-client-python"
---

Generate structured streaming client methods for the **Azure flavor**: operations whose HTTP response is a JSONL (`application/jsonl`) or SSE (`text/event-stream`) stream now return `Stream[T]` / `AsyncStream[T]`, yielding deserialized model instances instead of raw bytes.


The `Stream` / `AsyncStream` runtime (plus the JSONL / SSE decoders) is vendored into the generated package at `_utils/streaming_base.py` (like `_utils/model_base.py`), so it depends only on the released `azure.core.rest` — not on an unreleased `azure.core.streaming`.

```python
# For an operation returning JsonlStream<Thing> (Azure flavor):
stream = client.receive() # -> Stream[Thing]
for thing in stream: # deserialized model instances
...
```
5 changes: 5 additions & 0 deletions cspell.yaml
Original file line number Diff line number Diff line change
Expand Up @@ -7,11 +7,15 @@ dictionaries:
words:
- Ablack
- Adoptium
- aenter
- aexit
- agentic
- agentics
- aiohttp
- aiter
- alzimmer
- amqp
- anext
- AQID
- Arize
- arizeaiobservabilityeval
Expand Down Expand Up @@ -120,6 +124,7 @@ words:
- intrinsics
- ints
- IOHTTP
- isascii
- isdigit
- isinstance
- issecret
Expand Down
17 changes: 17 additions & 0 deletions packages/http-client-python/README.md
Original file line number Diff line number Diff line change
Expand Up @@ -153,3 +153,20 @@ Whether to clear the output folder before generating the code. Defaults to `fals
**Type:** `boolean`

Emit YAML code model only, without running Python generator. For batch processing.

## Structured streaming (JSONL / SSE)

For the **Azure flavor**, operations whose HTTP response is a JSONL (`application/jsonl`) or SSE (`text/event-stream`) stream generate client methods that return `Stream[T]` (sync) / `AsyncStream[T]` (async), yielding deserialized model instances instead of raw bytes. This is driven by the TCGC response stream metadata (the response stream type) — there is no opt-in emitter option. For the unbranded flavor, streaming responses keep the existing raw byte-iterator behavior (`Iterator[bytes]` / `AsyncIterator[bytes]`).

For an operation returning `JsonlStream<Thing>`, the generated method returns `Stream[Thing]` (sync) / `AsyncStream[Thing]` (async), yielding deserialized `Thing` instances as each JSONL line arrives. Similarly, `SSEStream<Events>` produces a `Stream` / `AsyncStream` over the SSE event payloads.

```python
# For an operation returning JsonlStream<Thing> (Azure flavor):
stream = client.receive() # -> Stream[Thing]
for thing in stream: # deserialized model instances
...
```

The `Stream` / `AsyncStream` runtime (plus the JSONL and SSE decoders) is **vendored** into the generated package at `_utils/streaming_base.py` (alongside `_utils/model_base.py`). It depends only on the released `azure.core.rest`, so no unreleased `azure.core.streaming` dependency is required at runtime.

SSE `@events` unions use TCGC event metadata to deserialize each named event into its corresponding generated model. Events marked with `@terminalEvent` stop iteration without being yielded.
83 changes: 83 additions & 0 deletions packages/http-client-python/emitter/src/http.ts
Original file line number Diff line number Diff line change
Expand Up @@ -42,6 +42,88 @@ export enum ReferredByOperationTypes {
NonPagingOnly = 2,
}

/**
* Determine whether a stream's payload type is "structured" (a model or union
* that can be deserialized into an item `T`), as opposed to a bare byte/string
* stream that should keep the existing raw byte-iterator behavior.
*/
export function isStructuredStreamType(type: SdkType): boolean {
switch (type.kind) {
case "model":
case "union":
return true;
case "nullable":
return isStructuredStreamType(type.type);
default:
return false;
}
}

export function getStructuredStreamKind(
response: SdkHttpResponse | SdkHttpErrorResponse,
): "jsonl" | "sse" | undefined {
if (response.sseMetadata) return "sse";

const contentTypes = response.streamMetadata?.contentTypes ?? response.contentTypes ?? [];
for (const contentType of contentTypes) {
const mediaType = contentType.split(";", 1)[0].trim().toLowerCase();
if (mediaType === "text/event-stream") return "sse";
if (mediaType === "application/jsonl") return "jsonl";
}
return undefined;
}

function getStringConstantValue(type: SdkType): string | undefined {
if (type.kind === "nullable") return getStringConstantValue(type.type);
return type.kind === "constant" && typeof type.value === "string" ? type.value : undefined;
}

/**
* Build the `streaming` block for a response YAML when the response is a JSONL/SSE
* stream with a structured payload type (driven by the TCGC stream metadata).
*
* Returns `undefined` when structured streaming should not apply, in which case
* the existing raw byte-iterator behavior is preserved.
*
* For SSE, TCGC `sseMetadata` supplies each event's wire name, payload type, and
* terminal marker. JSONL only needs the common stream item type.
*/
function getStreamingInfo(
context: PythonSdkContext,
response: SdkHttpResponse | SdkHttpErrorResponse,
): Record<string, any> | undefined {
const streamMetadata = response.streamMetadata;
if (!streamMetadata) return undefined;
if (!isStructuredStreamType(streamMetadata.streamType)) return undefined;
const kind = getStructuredStreamKind(response);
if (!kind) return undefined;

const streaming: Record<string, any> = {
kind,
itemType: getType(context, streamMetadata.streamType),
};

if (kind === "sse" && response.sseMetadata) {
const events: Record<string, any>[] = [];
let terminalEvent: string | undefined;
for (const event of response.sseMetadata.events) {
if (event.isTerminalEvent) {
terminalEvent =
getStringConstantValue(event.payloadType) ?? getStringConstantValue(event.type);
} else {
events.push({
eventType: event.eventType,
itemType: getType(context, event.payloadType),
});
}
}
if (events.length > 0) streaming.events = events;
if (terminalEvent !== undefined) streaming.terminalEvent = terminalEvent;
}

return streaming;
}

function isEtagType(type: SdkType): boolean {
if (type.kind === "nullable") return isEtagType(type.type);
const raw = type.__raw;
Expand Down Expand Up @@ -682,6 +764,7 @@ function emitHttpResponse(
"invalid-lro-result",
method,
),
streaming: isException ? undefined : getStreamingInfo(context, response),
};
}

Expand Down
45 changes: 45 additions & 0 deletions packages/http-client-python/emitter/test/streaming.test.ts
Original file line number Diff line number Diff line change
@@ -0,0 +1,45 @@
import { strictEqual } from "assert";
import { describe, it } from "vitest";
import { getStructuredStreamKind, isStructuredStreamType } from "../src/http.js";

describe("typespec-python: structured streaming", () => {
it("treats model and union payloads as structured", () => {
strictEqual(isStructuredStreamType({ kind: "model" } as any), true);
strictEqual(isStructuredStreamType({ kind: "union" } as any), true);
});

it("unwraps nullable payloads", () => {
strictEqual(isStructuredStreamType({ kind: "nullable", type: { kind: "model" } } as any), true);
strictEqual(
isStructuredStreamType({ kind: "nullable", type: { kind: "bytes" } } as any),
false,
);
});

it("treats bare byte/string payloads as unstructured", () => {
strictEqual(isStructuredStreamType({ kind: "bytes" } as any), false);
strictEqual(isStructuredStreamType({ kind: "string" } as any), false);
});

it("detects the stream protocol explicitly", () => {
strictEqual(getStructuredStreamKind({ sseMetadata: { events: [] } } as any), "sse");
strictEqual(
getStructuredStreamKind({
streamMetadata: { contentTypes: ["text/event-stream; charset=utf-8"] },
} as any),
"sse",
);
strictEqual(
getStructuredStreamKind({
streamMetadata: { contentTypes: ["application/jsonl"] },
} as any),
"jsonl",
);
strictEqual(
getStructuredStreamKind({
streamMetadata: { contentTypes: ["application/json"] },
} as any),
undefined,
);
});
});
Original file line number Diff line number Diff line change
Expand Up @@ -279,12 +279,29 @@ def need_utils_folder(self, async_mode: bool, client_namespace: str) -> bool:
self.need_utils_utils(async_mode, client_namespace)
or self.need_utils_serialization
or self.options["models-mode"] == "dpg"
or self.need_streaming_base
)

@property
def need_utils_serialization(self) -> bool:
return not self.options["client-side-validation"]

@property
def has_structured_stream(self) -> bool:
return any(
op.has_structured_stream_response
for client in self.clients
for og in client.operation_groups
for op in og.operations
)

@property
def need_streaming_base(self) -> bool:
# Whether to emit the vendored ``_utils/streaming_base.py`` (Stream / AsyncStream
# + JSONL / SSE decoders). Only needed when at least one operation returns a
# structured stream.
return self.has_structured_stream

def need_utils_utils(self, async_mode: bool, client_namespace: str) -> bool:
return (
self.need_utils_form_data(async_mode, client_namespace)
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -98,12 +98,21 @@ def exact_name_params(self) -> set[str]:

@property
def stream_value(self) -> Union[str, bool]:
# Structured streams (JSONL / SSE) must always run the pipeline with
# stream=True so the body can be consumed incrementally by Stream/AsyncStream.
if self.has_structured_stream_response:
return True
return (
f'kwargs.pop("stream", {self.has_stream_response})'
if self.expose_stream_keyword and self.has_response_body and "stream" not in self.exact_name_params
else self.has_stream_response
)

@property
def has_structured_stream_response(self) -> bool:
"""Whether any success response is a structured (JSONL / SSE) stream returning Stream[T]."""
return any(getattr(r, "is_structured_stream", False) for r in self.responses)

@property
def has_form_data_body(self):
return self.parameters.has_form_data_body
Expand Down
Loading
Loading