stream

package
v0.260806.1 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Aug 6, 2026 License: MPL-2.0 Imports: 32 Imported by: 0

Documentation

Overview

Package stream — prime.go

Eager first-event start for OpenAI Responses streams. The OpenAI Go SDK returns a *Stream from NewStreaming without issuing the HTTP request — the real upstream verdict surfaces only on the first Next(). PrimeResponsesStream forces that first Next() so a pre-stream upstream error can be returned to the dispatch caller (which retries the next priority tier) before any byte hits the wire. The first event it reads is not discarded: it is replayed via firstEventReplayStream so the handler sees a complete stream.

NOTE: this is unrelated to the probe subsystem, which issues separate synthetic health-check requests. This one pulls the first event of the real business stream and replays it — nothing extra is sent.

Index

Constants

View Source 
const OpenaiFieldReasoningContent = "reasoning_content"

OpenAI extra field names that map to Anthropic content blocks

Variables

View Source 
var ErrMCPStreamContinue = errors.New("mcp stream should continue")

Functions

func AnthropicSingleBetaMessage added in v0.260414.2000 

func AnthropicSingleBetaMessage(c *gin.Context, resp *anthropic.BetaMessage, responseModel string) error

AnthropicSingleBetaMessage emits a single assembled Anthropic beta message using SSE events.

func AnthropicSingleMessage added in v0.260414.2000 

func AnthropicSingleMessage(c *gin.Context, resp *anthropic.Message, responseModel string) error

AnthropicSingleMessage emits a single assembled Anthropic v1 message using SSE events.

func AnthropicToOpenAIStream 

func AnthropicToOpenAIStream(hc *protocol.HandleContext, req *anthropic.BetaMessageNewParams, stream *anthropicstream.Stream[anthropic.BetaRawMessageStreamEventUnion], responseModel string, disableStreamUsage bool) (*protocol.TokenUsage, error)

AnthropicToOpenAIStream processes Anthropic streaming events and converts them to OpenAI format. Returns the normalized TokenUsage (input/output plus cache-read and reasoning tokens) and an error for usage tracking. Returning the full usage — rather than just input/output — keeps cache tokens out of the dropped column when the recorded usage is persisted on the conversion path.

func AnthropicToOpenAIStreamWithMCPHooks added in v0.260514.1 

func AnthropicToOpenAIStreamWithMCPHooks(hc *protocol.HandleContext, req *anthropic.BetaMessageNewParams, stream *anthropicstream.Stream[anthropic.BetaRawMessageStreamEventUnion], responseModel string, disableStreamUsage bool, hooks *AnthropicToOpenAIMCPHooks) (*protocol.TokenUsage, error)

func BuildErrorEvent 

func BuildErrorEvent(message, errorType, code string) map[string]interface{}

BuildErrorEvent builds a standard error event map

func FilterOpenAIProtocolFields 

func FilterOpenAIProtocolFields(extras map[string]interface{}) map[string]interface{}

FilterOpenAIProtocolFields removes OpenAI protocol fields that should NOT appear in Anthropic message_delta. These fields are already properly handled via content_block events and should not be duplicated.

func FilterSpecialFields 

func FilterSpecialFields(extras map[string]interface{}) map[string]interface{}

FilterSpecialFields removes special fields that have dedicated content blocks e.g., reasoning_content is handled as thinking block, not merged into text_delta

func GenerateObfuscationString 

func GenerateObfuscationString() string

GenerateObfuscationString generates a random string similar to "KOJz1A"

func HandleAnthropic added in v0.260414.2000 

func HandleAnthropic(hc *protocol.HandleContext, streamResp *anthropicstream.Stream[anthropic.MessageStreamEventUnion]) (*protocol.TokenUsage, error)

HandleAnthropic handles Anthropic v1 streaming response. Returns (UsageStat, error)

func HandleAnthropicBeta added in v0.260414.2000 

func HandleAnthropicBeta(hc *protocol.HandleContext, streamResp *anthropicstream.Stream[anthropic.BetaRawMessageStreamEventUnion]) (*protocol.TokenUsage, error)

HandleAnthropicBeta handles Anthropic v1 beta streaming response. Returns (UsageStat, error)

func HandleAnthropicBetaToOpenAIResponsesStream added in v0.260531.1 

func HandleAnthropicBetaToOpenAIResponsesStream(
	hc *protocol.HandleContext,
	stream *anthropicstream.Stream[anthropic.BetaRawMessageStreamEventUnion],
	responseModel string,
) (*protocol.TokenUsage, error)

HandleAnthropicBetaToOpenAIResponsesStream converts Anthropic Beta streaming to Responses API format using the chain pipeline architecture.

func HandleAnthropicToGoogleStreamResponse 

func HandleAnthropicToGoogleStreamResponse(c *gin.Context, stream *anthropicstream.Stream[anthropic.MessageStreamEventUnion], responseModel string) error

HandleAnthropicToGoogleStreamResponse processes Anthropic streaming events and converts them to Google format

func HandleGoogleToAnthropicBetaStreamResponse 

func HandleGoogleToAnthropicBetaStreamResponse(c *gin.Context, stream iter.Seq2[*genai.GenerateContentResponse, error], responseModel string) (*protocol.TokenUsage, error)

HandleGoogleToAnthropicBetaStreamResponse processes Google streaming events and converts them to Anthropic beta format. Returns UsageStat containing token usage information for tracking.

func HandleGoogleToAnthropicStreamResponse 

func HandleGoogleToAnthropicStreamResponse(c *gin.Context, stream iter.Seq2[*genai.GenerateContentResponse, error], responseModel string) (*protocol.TokenUsage, error)

HandleGoogleToAnthropicStreamResponse processes Google streaming events and converts them to Anthropic format. Returns UsageStat containing token usage information for tracking.

func HandleGoogleToOpenAIStreamResponse 

func HandleGoogleToOpenAIStreamResponse(c *gin.Context, stream iter.Seq2[*genai.GenerateContentResponse, error], responseModel string) error

HandleGoogleToOpenAIStreamResponse processes Google streaming events and converts them to OpenAI format

func HandleOpenAIChatStream 

func HandleOpenAIChatStream(hc *protocol.HandleContext, streamResp *openaistream.Stream[openai.ChatCompletionChunk]) (*protocol.TokenUsage, error)

HandleOpenAIChatStream handles OpenAI chat streaming response. The input-token fallback (used only when the upstream reports no usage) comes from hc.EstimatedInputTokens, so the handler no longer takes the request.

func HandleOpenAIChatToResponsesStream 

func HandleOpenAIChatToResponsesStream(hc *protocol.HandleContext, stream *openaistream.Stream[openai.ChatCompletionChunk], responseModel string) (*protocol.TokenUsage, error)

HandleOpenAIChatToResponsesStream converts OpenAI Chat Completions streaming to Responses API format using the chain pipeline architecture.

func HandleOpenAIResponsesStream 

func HandleOpenAIResponsesStream(hc *protocol.HandleContext, stream ResponsesStreamIter, responseModel string) (*protocol.TokenUsage, error)

HandleOpenAIResponsesStream handles OpenAI Responses API streaming response. Returns (UsageStat, error)

func HandleOpenAIResponsesStreamToAnthropic 

func HandleOpenAIResponsesStreamToAnthropic(c *gin.Context, stream ResponsesStreamIter, responseModel string) (*protocol.TokenUsage, error)

HandleOpenAIResponsesStreamToAnthropic handles OpenAI Responses API streaming response and transforms it to Anthropic message format. This is used for ChatGPT backend API providers when the original request was in Anthropic format. Returns (TokenUsage, error)

func HandleOpenAIToAnthropicBetaStream 

func HandleOpenAIToAnthropicBetaStream(hc *protocol.HandleContext, req *openai.ChatCompletionNewParams, stream *openaistream.Stream[openai.ChatCompletionChunk], responseModel string) (*protocol.TokenUsage, error)

HandleOpenAIToAnthropicBetaStream processes OpenAI streaming events and converts them to Anthropic beta format. Returns UsageStat containing token usage information for tracking.

func HandleOpenAIToAnthropicBetaStreamWithMCPHooks added in v0.260507.1 

func HandleOpenAIToAnthropicBetaStreamWithMCPHooks(
	hc *protocol.HandleContext,
	req *openai.ChatCompletionNewParams,
	stream *openaistream.Stream[openai.ChatCompletionChunk],
	responseModel string,
	hooks *OpenAIToAnthropicMCPHooks,
) (*protocol.TokenUsage, error)

HandleOpenAIToAnthropicBetaStreamWithMCPHooks enables MCP-aware tool suppression/finalization during conversion.

func HandleOpenAIToAnthropicStreamResponse 

func HandleOpenAIToAnthropicStreamResponse(hc *protocol.HandleContext, req *openai.ChatCompletionNewParams, stream *openaistream.Stream[openai.ChatCompletionChunk], responseModel string) (*protocol.TokenUsage, error)

HandleOpenAIToAnthropicStreamResponse processes OpenAI streaming events and converts them to Anthropic format. Returns UsageStat containing token usage information for tracking.

func HandleOpenAIToAnthropicStreamResponseWithMCPHooks added in v0.260507.1 

func HandleOpenAIToAnthropicStreamResponseWithMCPHooks(
	hc *protocol.HandleContext,
	req *openai.ChatCompletionNewParams,
	stream *openaistream.Stream[openai.ChatCompletionChunk],
	responseModel string,
	hooks *OpenAIToAnthropicMCPHooks,
) (*protocol.TokenUsage, error)

HandleOpenAIToAnthropicStreamResponseWithMCPHooks enables MCP-aware tool suppression/finalization during conversion.

func HandleOpenAIToGoogleStreamResponse 

func HandleOpenAIToGoogleStreamResponse(c *gin.Context, stream *openaistream.Stream[openai.ChatCompletionChunk], responseModel string) error

HandleOpenAIToGoogleStreamResponse processes OpenAI streaming events and converts them to Google format This handler writes Google-format streaming responses to the gin.Context

func HandleResponsesToAnthropicBetaAssembly 

func HandleResponsesToAnthropicBetaAssembly(c *gin.Context, stream ResponsesStreamIter, responseModel string) (*protocol.TokenUsage, error)

HandleResponsesToAnthropicBetaAssembly consumes a Responses API stream and responds with a single assembled Anthropic beta message.

func HandleResponsesToAnthropicBetaStream 

func HandleResponsesToAnthropicBetaStream(hc *protocol.HandleContext, stream ResponsesStreamIter, responseModel string) (*protocol.TokenUsage, error)

HandleResponsesToAnthropicBetaStream processes OpenAI Responses API streaming events and converts them to Anthropic beta format. Returns TokenUsage containing token usage information for tracking.

func HandleResponsesToAnthropicV1Assembly 

func HandleResponsesToAnthropicV1Assembly(c *gin.Context, stream ResponsesStreamIter, responseModel string) (*protocol.TokenUsage, error)

HandleResponsesToAnthropicV1Assembly consumes a Responses API stream and responds with a single assembled Anthropic v1 message.

func HandleResponsesToAnthropicV1Stream 

func HandleResponsesToAnthropicV1Stream(hc *protocol.HandleContext, stream ResponsesStreamIter, responseModel string) (*protocol.TokenUsage, error)

HandleResponsesToAnthropicV1Stream processes OpenAI Responses API streaming events and converts them to Anthropic v1 format. Returns TokenUsage containing token usage information for tracking.

func HandleResponsesToOpenAIChatStream added in v0.260409.1540 

func HandleResponsesToOpenAIChatStream(
	hc *protocol.HandleContext,
	stream ResponsesStreamIter,
	responseModel string,
) (*protocol.TokenUsage, error)

HandleResponsesToOpenAIChatStream converts Responses API streaming to Chat Completions format using the chain pipeline architecture.

func MarshalAndSendErrorEvent 

func MarshalAndSendErrorEvent(c *gin.Context, message, errorType, code string)

MarshalAndSendErrorEvent marshals and sends an error event

func NewChatToResponsesConverter added in v0.260611.1 

func NewChatToResponsesConverter(stream OpenAIChatStream, responseModel string) *chatToResponsesConverter

NewChatToResponsesConverter creates a converter that reads from an OpenAI Chat Completions stream and yields Responses API wire events.

func NewExampleTool 

func NewExampleTool() openai.ChatCompletionToolUnionParam

func OpenAIResponsesEvent added in v0.260514.1 

func OpenAIResponsesEvent(c *gin.Context, event string, v any)

OpenAIResponsesEvent writes one Responses API SSE event, flushes, and marks TTFT on the first content-bearing (*.delta) event. MarkFirstToken is idempotent.

func OpenAISSE added in v0.260414.2000 

func OpenAISSE(c *gin.Context, v any)

OpenAISSE marshals v to JSON and writes it as an OpenAI-style SSE data line, then flushes. MENTION: Must keep extra space after "data:" to match OpenAI wire format.

func OpenAISSEDone added in v0.260414.2000 

func OpenAISSEDone(c *gin.Context)

OpenAISSEDone writes the SSE [DONE] terminator and flushes.

func PrintChunk added in v0.260507.1 

func PrintChunk(chunk any)

func RunConverter added in v0.260611.1 

func RunConverter(hc *protocol.HandleContext, conv StreamConverter, writer func(event interface{}) error) (*protocol.TokenUsage, error)

RunConverter bridges a StreamConverter into ProcessStream. It sets up SSE headers, drives the converter via ProcessStream (which dispatches hooks automatically), and calls the writer for each emitted event.

func SendForwardingError 

func SendForwardingError(c *gin.Context, err error)

SendForwardingError sends an error response for request forwarding failures, propagating the upstream provider's HTTP status when the error carries one.

func SendInternalError 

func SendInternalError(c *gin.Context, errMsg string)

SendInternalError sends an error response for internal errors

func SendInvalidRequestBodyError 

func SendInvalidRequestBodyError(c *gin.Context, err error)

SendInvalidRequestBodyError sends an error response for invalid request body

func SendSSErrorEvent 

func SendSSErrorEvent(c *gin.Context, message, errorType string)

SendSSErrorEvent sends an error event through SSE

func SendSSErrorEventJSON 

func SendSSErrorEventJSON(c *gin.Context, errorJSON []byte)

SendSSErrorEventJSON sends a JSON error event through SSE

func SendStreamingError 

func SendStreamingError(c *gin.Context, err error)

SendStreamingError sends an error response for streaming request failures. When the failure occurs before any SSE frame has been written (the caller guards this with conv.MessageStarted()), the HTTP status is still settable, so we propagate the upstream provider's status (401/429/4xx) instead of flattening every pre-stream failure into a 500.

Types

type AnthropicBetaStream added in v0.260806.1 

type AnthropicBetaStream interface {
	Next() bool
	Current() anthropic.BetaRawMessageStreamEventUnion
	Err() error
}

AnthropicBetaStream is the transport-neutral iterator surface consumed by the Anthropic Beta to OpenAI Chat stream converter. The Anthropic SDK stream and dormant Stage adapters both implement this contract.

type AnthropicEvent added in v0.260806.1 

type AnthropicEvent struct {
	Type string
	Data any
}

AnthropicEvent is the transport-neutral view of an Anthropic stream event. HTTP writers may keep using the internal representation; protocol stages use this exported value to carry event name and data without taking over SSE framing.

func AsAnthropicEvent added in v0.260806.1 

func AsAnthropicEvent(event any) (AnthropicEvent, bool)

AsAnthropicEvent exposes an event emitted by an Anthropic stream converter.

func (AnthropicEvent) RawJSON added in v0.260806.1 

func (e AnthropicEvent) RawJSON() string

RawJSON returns the protocol payload rather than the transport wrapper. Protocol-owned assemblers use this to reconstruct converted streams without mistaking the Type/Data carrier itself for an Anthropic wire event.

type AnthropicToOpenAIMCPHooks added in v0.260514.1 

type AnthropicToOpenAIMCPHooks struct {
	ShouldSuppressTool func(name string) bool
	OnToolCallsFinal   func(calls []AnthropicToOpenAIToolCall) error
}

type AnthropicToOpenAIToolCall added in v0.260514.1 

type AnthropicToOpenAIToolCall struct {
	ID        string
	Name      string
	Arguments string
}

type OpenAIChatStream added in v0.260806.1 

type OpenAIChatStream interface {
	Next() bool
	Current() openai.ChatCompletionChunk
	Err() error
}

OpenAIChatStream is the minimum iterator surface required by the Chat to Anthropic state machine. The OpenAI SDK stream and stage stream adapters both implement it.

type OpenAIToAnthropicMCPHooks added in v0.260507.1 

type OpenAIToAnthropicMCPHooks struct {
	ShouldSuppressTool func(name string) bool
	OnToolCallsFinal   func(calls []OpenAIToAnthropicToolCall) error
}

OpenAIToAnthropicMCPHooks provides optional hooks for MCP-aware stream handling.

type OpenAIToAnthropicToolCall added in v0.260507.1 

type OpenAIToAnthropicToolCall struct {
	ID        string
	Name      string
	Arguments string
}

OpenAIToAnthropicToolCall captures a complete tool call assembled from OpenAI stream chunks.

type ResponsesStreamIter added in v0.260604.1 

type ResponsesStreamIter interface {
	Next() bool
	Current() responses.ResponseStreamEventUnion
	Err() error
	Close() error
}

ResponsesStreamIter is the iterator surface stream handlers consume. Both the SDK's *openaistream.Stream[responses.ResponseStreamEventUnion] and the replay wrapper below satisfy it.

func PrimeResponsesStream added in v0.260604.1 

func PrimeResponsesStream(stream *openaistream.Stream[responses.ResponseStreamEventUnion]) (ResponsesStreamIter, error)

PrimeResponsesStream calls Next() once on the SDK stream to force the lazy HTTP request and surface pre-stream upstream errors. Returns:

  • (iter, nil) primed; iter replays the read event first, then delegates to the SDK stream.
  • (nil, err) pre-stream failure, including a degenerate stream with zero events and no SDK error — caller retries next priority tier before any Anthropic SSE bytes are written.

type StreamConverter added in v0.260611.1 

type StreamConverter interface {
	Next() (event interface{}, done bool, err error)
	Usage() *protocol.TokenUsage
}

StreamConverter reads from an upstream stream and emits target-protocol events one at a time. It maintains internal conversion state (tool call accumulation, usage tracking, etc.) but never touches gin.Context or writes SSE directly.

Next() returns (event, false, nil) for each target event, (nil, true, nil) when the upstream is exhausted, or (nil, false, err) on error.

func NewAnthropicBetaToOpenAIChatConverter added in v0.260806.1 

func NewAnthropicBetaToOpenAIChatConverter(
	stream AnthropicBetaStream,
	responseModel string,
	disableStreamUsage bool,
) StreamConverter

NewAnthropicBetaToOpenAIChatConverter creates a hook-free, transport-neutral converter. HTTP/SSE framing, MCP hooks, and stream close ownership remain with the caller.

func NewAnthropicBetaToOpenAIResponsesConverter added in v0.260806.1 

func NewAnthropicBetaToOpenAIResponsesConverter(
	stream AnthropicBetaStream,
	responseModel string,
) StreamConverter

NewAnthropicBetaToOpenAIResponsesConverter creates a transport-neutral converter. HTTP/SSE framing and stream close ownership remain with the caller, so the same converter can serve legacy handlers and Stage Bridges.

func NewOpenAIChatToAnthropicBetaConverter added in v0.260806.1 

func NewOpenAIChatToAnthropicBetaConverter(stream OpenAIChatStream, responseModel string, req *openai.ChatCompletionNewParams) StreamConverter

NewOpenAIChatToAnthropicBetaConverter creates the transport-free beta stream state machine. The caller owns driving and closing the supplied stream.

func NewOpenAIChatToAnthropicV1Converter added in v0.260806.1 

func NewOpenAIChatToAnthropicV1Converter(stream OpenAIChatStream, responseModel string, req *openai.ChatCompletionNewParams) StreamConverter

NewOpenAIChatToAnthropicV1Converter creates the transport-free V1 stream state machine. The caller owns driving and closing the supplied stream.

func NewOpenAIResponsesToAnthropicConverter added in v0.260806.1 

func NewOpenAIResponsesToAnthropicConverter(
	ctx context.Context,
	stream ResponsesStreamIter,
	responseModel string,
) StreamConverter

NewOpenAIResponsesToAnthropicConverter creates a transport-neutral Responses-to-Anthropic converter. The caller owns stream closure and wire framing; emitted values can be normalized with AsAnthropicEvent.

func NewOpenAIResponsesToChatConverter added in v0.260806.1 

func NewOpenAIResponsesToChatConverter(
	stream ResponsesStreamIter,
	responseModel string,
	disableUsage bool,
) StreamConverter

NewOpenAIResponsesToChatConverter creates a transport-neutral Responses to Chat stream converter. The caller owns stream closure and SSE framing.

type StreamEventRecorder 

type StreamEventRecorder interface {
	RecordRawMapEvent(eventType string, event map[string]interface{})
}

StreamEventRecorder is an interface for recording stream events during protocol conversion

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.