Skip to content

[http-client-csharp] Add JSONL and SSE streaming support #11462

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Draft
JoshLove-msft wants to merge 4 commits into
base: main
Choose a base branch
from
Draft

[http-client-csharp] Add JSONL and SSE streaming support #11462

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
71d3227
feat(http-client-csharp): add streaming client results
JoshLove-msft Jul 29, 2026
8e7a11c
Merge upstream main into streaming client results
JoshLove-msft Jul 29, 2026
f3a42e1
fix(http-client-csharp): return streaming protocol results
JoshLove-msft Jul 29, 2026
976537f
fix(http-client-csharp): restore optional npm packages
JoshLove-msft Jul 29, 2026
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
111 changes: 107 additions & 4 deletions packages/http-client-csharp/emitter/src/lib/operation-converter.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -24,6 +24,7 @@ import {
SdkQueryParameter,
SdkServiceMethod,
SdkServiceResponseHeader,
SdkStreamMetadata,
SdkType,
shouldGenerateConvenient,
shouldGenerateProtocol,
Expand All @@ -34,9 +35,12 @@ import {
getDeprecated,
isErrorModel,
NoTarget,
Union,
} from "@typespec/compiler";
import { unsafe_getEventDefinitions } from "@typespec/events/experimental";
import { HttpStatusCodeRange } from "@typespec/http";
import { getResourceOperation } from "@typespec/rest";
import { isTerminalEvent } from "@typespec/sse";
import { CSharpEmitterContext } from "../sdk-context.js";
import { collectionFormatToDelimMap } from "../type/collection-format.js";
import { HttpResponseHeader } from "../type/http-response-header.js";
Expand All @@ -62,6 +66,7 @@ import {
InputMethodParameter,
InputPathParameter,
InputQueryParameter,
InputStreamingType,
InputType,
} from "../type/input-type.js";
import { convertLroFinalStateVia } from "../type/operation-final-state-via.js";
Expand Down Expand Up @@ -232,7 +237,9 @@ export function fromSdkServiceMethodOperation(
path: method.operation.path,
externalDocsUrl: getExternalDocs(sdkContext, method.operation.__raw.operation)?.url,
requestMediaTypes: requestMediaTypes,
bufferResponse: true,
bufferResponse: !method.operation.responses.some((response) =>
isSupportedStream(response.streamMetadata),
),
generateProtocolMethod: shouldGenerateProtocol(sdkContext, method.operation.__raw.operation),
generateConvenienceMethod: generateConvenience,
crossLanguageDefinitionId: method.crossLanguageDefinitionId,
Expand Down Expand Up @@ -349,8 +356,23 @@ function fromSdkServiceMethodParameters(
for (const p of method.parameters) {
const methodInputParameter = diagnostics.pipe(fromMethodParameter(sdkContext, p, namespace));
const operationHttpParameter = getHttpOperationParameter(method, p);
const streamMetadata = method.operation.bodyParam?.streamMetadata;
const isStreamingBodyParameter =
isJsonLinesStream(streamMetadata) &&
method.operation.bodyParam!.methodParameterSegments.some((segments) =>
segments.some(
(segment) =>
segment === p || segment.crossLanguageDefinitionId === p.crossLanguageDefinitionId,
),
);

if (!operationHttpParameter) {
if (isStreamingBodyParameter) {
methodInputParameter.type = diagnostics.pipe(
fromSdkStreamMetadata(sdkContext, streamMetadata),
);
methodInputParameter.location = RequestLocation.Body;
}
parameters.push(methodInputParameter);
continue;
}
Expand All @@ -363,6 +385,12 @@ function fromSdkServiceMethodParameters(
rootApiVersions,
diagnostics,
);
if (isStreamingBodyParameter) {
methodInputParameter.type = diagnostics.pipe(
fromSdkStreamMetadata(sdkContext, streamMetadata),
);
methodInputParameter.location = RequestLocation.Body;
}
parameters.push(methodInputParameter);
}

Expand All @@ -376,6 +404,15 @@ function updateMethodParameter(
rootApiVersions: string[],
diagnostics: ReturnType<typeof createDiagnosticCollector>,
): void {
if (
operationHttpParameter.kind === "body" &&
isJsonLinesStream(operationHttpParameter.streamMetadata)
) {
methodParameter.type = diagnostics.pipe(
fromSdkStreamMetadata(sdkContext, operationHttpParameter.streamMetadata),
);
}

// for content type parameter
if (isContentType(operationHttpParameter)) {
methodParameter.type = diagnostics.pipe(
Expand Down Expand Up @@ -406,7 +443,9 @@ function fromSdkServiceMethodResponse(
const diagnostics = createDiagnosticCollector();

return diagnostics.wrap({
type: diagnostics.pipe(getResponseType(sdkContext, methodResponse.type)),
type: isSupportedStream(methodResponse.streamMetadata)
? diagnostics.pipe(fromSdkStreamMetadata(sdkContext, methodResponse.streamMetadata))
: diagnostics.pipe(getResponseType(sdkContext, methodResponse.type)),
resultSegments: methodResponse.resultSegments?.map((segment) =>
getResponseSegmentName(segment),
),
Expand Down Expand Up @@ -605,7 +644,9 @@ function fromBodyParameter(
rootApiVersions: string[],
): [InputBodyParameter, readonly Diagnostic[]] {
const diagnostics = createDiagnosticCollector();
const parameterType = diagnostics.pipe(fromSdkType(sdkContext, p.type, p));
const parameterType = isJsonLinesStream(p.streamMetadata)
? diagnostics.pipe(fromSdkStreamMetadata(sdkContext, p.streamMetadata))
: diagnostics.pipe(fromSdkType(sdkContext, p.type, p));

const retVar: InputBodyParameter = {
kind: "body",
Expand Down Expand Up @@ -723,7 +764,9 @@ export function fromSdkHttpOperationResponse(
const range = sdkResponse.statusCodes;
retVar = {
statusCodes: toStatusCodesArray(range),
bodyType: diagnostics.pipe(getResponseType(sdkContext, sdkResponse.type)),
bodyType: isSupportedStream(sdkResponse.streamMetadata)
? diagnostics.pipe(fromSdkStreamMetadata(sdkContext, sdkResponse.streamMetadata))
: diagnostics.pipe(getResponseType(sdkContext, sdkResponse.type)),
headers: diagnostics.pipe(fromSdkServiceResponseHeaders(sdkContext, sdkResponse.headers)),
isErrorResponse:
sdkResponse.type !== undefined && isErrorModel(sdkContext.program, sdkResponse.type.__raw!),
Expand All @@ -735,6 +778,66 @@ export function fromSdkHttpOperationResponse(
return diagnostics.wrap(retVar);
}

function fromSdkStreamMetadata(
sdkContext: CSharpEmitterContext,
streamMetadata: SdkStreamMetadata,
): [InputStreamingType, readonly Diagnostic[]] {
const diagnostics = createDiagnosticCollector();
const originalType = streamMetadata.originalType;
const streamKind = getStreamKind(streamMetadata)!;

@jorgerangel-msft jorgerangel-msft Jul 30, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

is streamkind guaranteed to not be undefined here ?

let terminalEventType: string | undefined;
let terminalEventValue: string | undefined;

if (streamKind === "sse" && streamMetadata.streamType.__raw?.kind === "Union") {
const eventDefinitions = diagnostics.pipe(
unsafe_getEventDefinitions(sdkContext.program, streamMetadata.streamType.__raw as Union),
);
const terminalDefinition = eventDefinitions.find((definition) =>
isTerminalEvent(sdkContext.program, definition.root),
);
if (terminalDefinition?.payloadType.kind === "String") {
terminalEventType = terminalDefinition.eventType;
terminalEventValue = terminalDefinition.payloadType.value;
}
}

return diagnostics.wrap({
kind: "streaming",
name: originalType.kind === "model" ? originalType.name : "Stream",
valueType: diagnostics.pipe(fromSdkType(sdkContext, streamMetadata.streamType)),
streamKind,
contentTypes: streamMetadata.contentTypes,
terminalEventType,
terminalEventValue,
crossLanguageDefinitionId:
originalType.kind === "model" ? originalType.crossLanguageDefinitionId : "",
});
}

function getStreamKind(
streamMetadata: SdkStreamMetadata | undefined,
): InputStreamingType["streamKind"] | undefined {
if (streamMetadata?.contentTypes.includes("application/jsonl")) {
return "jsonl";
}
if (streamMetadata?.contentTypes.includes("text/event-stream")) {
return "sse";
}
return undefined;
}

function isSupportedStream(
streamMetadata: SdkStreamMetadata | undefined,
): streamMetadata is SdkStreamMetadata {
return getStreamKind(streamMetadata) !== undefined;
}

function isJsonLinesStream(
streamMetadata: SdkStreamMetadata | undefined,
): streamMetadata is SdkStreamMetadata {
return getStreamKind(streamMetadata) === "jsonl";
}

function fromSdkServiceResponseHeaders(
sdkContext: CSharpEmitterContext,
headers: SdkServiceResponseHeader[],
Expand Down
12 changes: 12 additions & 0 deletions packages/http-client-csharp/emitter/src/type/input-type.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -79,6 +79,7 @@ export type InputType =
| InputEnumType
| InputEnumValueType
| InputArrayType
| InputStreamingType
| InputDictionaryType
| InputNullableType;

Expand Down Expand Up @@ -305,6 +306,17 @@ export interface InputArrayType extends InputTypeBase {
crossLanguageDefinitionId: string;
}

export interface InputStreamingType extends InputTypeBase {
kind: "streaming";
name: string;
valueType: InputType;
streamKind: "jsonl" | "sse";
contentTypes: string[];
terminalEventType?: string;
terminalEventValue?: string;
crossLanguageDefinitionId: string;
}

export interface InputDictionaryType extends InputTypeBase {
kind: "dict";
keyType: InputType;
Expand Down
89 changes: 89 additions & 0 deletions packages/http-client-csharp/emitter/test/Unit/operation-converter.test.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -215,6 +215,95 @@ describe("Operation Converter", () => {
});

describe("Operation response type conversion", () => {
it("preserves JSONL stream item types", async () => {
const program = await typeSpecCompile(
`
model Info {
desc: string;
}

@post op send(stream: JsonlStream<Info>): void;
op receive(): JsonlStream<Info>;
`,
runner,
);
const context = createEmitterContext(program);
const sdkContext = await createCSharpSdkContext(context);
const [root] = createModel(sdkContext);

const send = root.clients[0].methods.find((method) => method.name === "send");
ok(send);
const sendParameter = send.parameters.find((parameter) => parameter.name === "stream");
ok(sendParameter);
ok(sendParameter.type.kind === "streaming");
ok(sendParameter.type.valueType.kind === "model");
strictEqual(sendParameter.type.valueType.name, "Info");

const bodyParameter = send.operation.parameters.find(
(parameter) => parameter.kind === "body",
);
ok(bodyParameter);
ok(bodyParameter.type.kind === "streaming");
ok(bodyParameter.type.valueType.kind === "model");
strictEqual(bodyParameter.type.valueType.name, "Info");

const receive = root.clients[0].methods.find((method) => method.name === "receive");
ok(receive);
ok(receive.response.type?.kind === "streaming");
ok(receive.response.type.valueType.kind === "model");
strictEqual(receive.response.type.valueType.name, "Info");
ok(receive.operation.responses[0].bodyType?.kind === "streaming");
ok(receive.operation.responses[0].bodyType.valueType.kind === "model");
strictEqual(receive.operation.responses[0].bodyType.valueType.name, "Info");
strictEqual(receive.operation.bufferResponse, false);
});

it("preserves SSE event unions and terminal events", async () => {
const program = await typeSpecCompile(
`
model ResponseCreated {
id: string;
}

model ResponseDelta {
delta: string;
}

@events
union ResponseEvents {
@Events.contentType("application/json")
responseCreated: ResponseCreated,

@Events.contentType("application/json")
responseDelta: ResponseDelta,

@Events.contentType("text/plain")
@terminalEvent
"[DONE]",
}

op receive(): SSEStream<ResponseEvents>;
`,
runner,
{ IsSseNeeded: true },
);
const context = createEmitterContext(program);
const sdkContext = await createCSharpSdkContext(context);
const [root] = createModel(sdkContext);

const receive = root.clients[0].methods.find((method) => method.name === "receive");
ok(receive);
ok(receive.response.type?.kind === "streaming");
strictEqual(receive.response.type.streamKind, "sse");
ok(receive.response.type.valueType.kind === "union");
strictEqual(receive.response.type.valueType.name, "ResponseEvents");
strictEqual(receive.response.type.terminalEventType, undefined);
strictEqual(receive.response.type.terminalEventValue, "[DONE]");
ok(receive.operation.responses[0].bodyType?.kind === "streaming");
strictEqual(receive.operation.responses[0].bodyType.streamKind, "sse");
strictEqual(receive.operation.bufferResponse, false);
});

describe("With anonymous union enum response type", () => {
it("should convert anonymous union enum response type to value type", async () => {
const program = await typeSpecCompile(
Expand Down
12 changes: 12 additions & 0 deletions packages/http-client-csharp/emitter/test/Unit/utils/test-util.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -3,8 +3,11 @@ import type { CreateSdkContextOptions } from "@azure-tools/typespec-client-gener
import { SdkTestLibrary } from "@azure-tools/typespec-client-generator-core/testing";
import { CompilerOptions, EmitContext, Program } from "@typespec/compiler";
import { createTestHost, TestHost } from "@typespec/compiler/testing";
import { EventsTestLibrary } from "@typespec/events/testing";
import { HttpTestLibrary } from "@typespec/http/testing";
import { RestTestLibrary } from "@typespec/rest/testing";
import { SSETestLibrary } from "@typespec/sse/testing";
import { StreamsTestLibrary } from "@typespec/streams/testing";
import { VersioningTestLibrary } from "@typespec/versioning/testing";
import { XmlTestLibrary } from "@typespec/xml/testing";
import { LoggerLevel } from "../../../src/lib/logger-level.js";
Expand All @@ -16,6 +19,9 @@ export async function createEmitterTestHost(): Promise<TestHost> {
libraries: [
RestTestLibrary,
HttpTestLibrary,
EventsTestLibrary,
SSETestLibrary,
StreamsTestLibrary,
VersioningTestLibrary,
AzureCoreTestLibrary,
SdkTestLibrary,
Expand Down Expand Up @@ -43,6 +49,7 @@ export interface TypeSpecCompileOptions {
AuthDecorator?: string;
NoEmit?: boolean;
IsVersionNeeded?: boolean;
IsSseNeeded?: boolean;
}

export async function typeSpecCompile(
Expand All @@ -55,6 +62,7 @@ export async function typeSpecCompile(
const needTCGC = options?.IsTCGCNeeded ?? false;
const needXml = options?.IsXmlNeeded ?? false;
const needVersion = options?.IsVersionNeeded ?? true;
const needSse = options?.IsSseNeeded ?? false;
const authDecorator =
options?.AuthDecorator ?? `@useAuth(ApiKeyAuth<ApiKeyLocation.header, "api-key">)`;
const versions = `enum Versions {
Expand All @@ -75,12 +83,16 @@ export async function typeSpecCompile(
const fileContent = `
import "@typespec/rest";
import "@typespec/http";
import "@typespec/http/streams";
${needSse ? 'import "@typespec/events";\nimport "@typespec/sse";' : ""}
import "@typespec/versioning";
${needXml ? 'import "@typespec/xml";' : ""}
${needAzureCore ? 'import "@azure-tools/typespec-azure-core";' : ""}
${needTCGC ? 'import "@azure-tools/typespec-client-generator-core";' : ""}
using TypeSpec.Rest;
using TypeSpec.Http;
using TypeSpec.Http.Streams;
${needSse ? "using TypeSpec.Events;\nusing TypeSpec.SSE;" : ""}
using TypeSpec.Versioning;
${needXml ? "using TypeSpec.Xml;" : ""}
${needAzureCore ? "using Azure.Core;\nusing Azure.Core.Traits;" : ""}
Expand Down
1 change: 0 additions & 1 deletion packages/http-client-csharp/eng/scripts/Spector-Helper.psm1
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -5,7 +5,6 @@ $failingSpecs = @(
Join-Path 'http' 'type' 'model' 'templated'
Join-Path 'http' 'type' 'file'
Join-Path 'http' 'client' 'naming' # pending until https://github.com/microsoft/typespec/issues/5653 is resolved
Join-Path 'http' 'streaming' 'jsonl'
Join-Path 'http' 'type' 'union' 'discriminated' # pending design
Join-Path 'http' 'authentication' 'noauth' 'union' # pending design
)
Expand Down
Loading
Loading