-module(anthropic@streaming@handler). -compile([no_auto_import, nowarn_unused_vars, nowarn_unused_function, nowarn_nomatch, inline]). -define(FILEPATH, "src/anthropic/streaming/handler.gleam"). -export([new_streaming_state/0, process_chunk/2, finalize_stream/1, is_stream_complete/1, has_stream_error/1, get_stream_error/1, get_accumulated_events/1, build_stream_result/1, parse_sse_body/1, parse_streaming_response/1, parse_sse_chunk/2, stream_message/2, stream_message_with_callback/3, get_event_text/1, get_text_deltas/1, get_full_text/1, get_message_id/1, get_model/1, is_complete/1, has_error/1, get_error/1]). -export_type([streaming_state/0, stream_result/0, stream_error/0]). -if(?OTP_RELEASE >= 27). -define(MODULEDOC(Str), -moduledoc(Str)). -define(DOC(Str), -doc(Str)). -else. -define(MODULEDOC(Str), -compile([])). -define(DOC(Str), -compile([])). -endif. ?MODULEDOC( " Streaming handler for Anthropic API\n" "\n" " This module provides sans-io functions for parsing Server-Sent Events (SSE)\n" " from the Anthropic streaming API. The design follows the sans-io pattern,\n" " allowing you to use any HTTP client that supports streaming.\n" "\n" " ## Recommended: Use `api.chat_stream`\n" "\n" " For most use cases, use `api.chat_stream` from the main API module:\n" "\n" " ```gleam\n" " import anthropic/api\n" " import anthropic/client\n" " import anthropic/types/request\n" " import anthropic/types/message.{user_message}\n" "\n" " let assert Ok(client) = client.init()\n" " let req = request.new(\"claude-sonnet-4-20250514\", [user_message(\"Hello!\")], 1024)\n" "\n" " case api.chat_stream(client, req) {\n" " Ok(result) -> io.println(api.stream_text(result))\n" " Error(err) -> handle_error(err)\n" " }\n" " ```\n" "\n" " ## True Real-Time Streaming (Sans-IO)\n" "\n" " For true real-time streaming where you process events as they arrive,\n" " use the incremental parsing functions with your own streaming HTTP client:\n" "\n" " ```gleam\n" " import anthropic/http\n" " import anthropic/streaming/handler.{\n" " new_streaming_state, process_chunk, finalize_stream\n" " }\n" "\n" " // 1. Build the streaming request\n" " let http_request = http.build_streaming_request(api_key, base_url, request)\n" "\n" " // 2. Start streaming with your HTTP client that supports chunked responses\n" " // (e.g., using Erlang's httpc with stream option, or any other client)\n" "\n" " // 3. Initialize streaming state\n" " let state = new_streaming_state()\n" "\n" " // 4. As each chunk arrives from your HTTP client, process it:\n" " let #(events, new_state) = process_chunk(state, chunk)\n" "\n" " // 5. Handle events in real-time as they arrive\n" " list.each(events, fn(event) {\n" " case handler.get_event_text(event) {\n" " Ok(text) -> io.print(text) // Print immediately!\n" " Error(_) -> Nil\n" " }\n" " })\n" "\n" " // 6. Continue with new_state for next chunk...\n" "\n" " // 7. When stream ends, finalize to get any remaining events\n" " let final_events = finalize_stream(state)\n" " ```\n" "\n" " ## Batch Mode (Deprecated)\n" "\n" " The batch functions `stream_message` and `stream_message_with_callback` in\n" " this module are deprecated. Use `api.chat_stream` and\n" " `api.chat_stream_with_callback` instead for a unified API.\n" "\n" " **Note**: Batch mode waits for the complete response before returning.\n" " Use sans-io incremental parsing for true real-time streaming.\n" ). -type streaming_state() :: {streaming_state, anthropic@internal@sse:sse_parser_state(), list(anthropic@streaming:stream_event()), boolean(), gleam@option:option(stream_error())}. -type stream_result() :: {stream_result, list(anthropic@streaming:stream_event())}. -type stream_error() :: {http_error, anthropic@error:anthropic_error()} | {sse_parse_error, binary()} | {event_decode_error, binary()} | {api_error, integer(), binary()}. -file("src/anthropic/streaming/handler.gleam", 151). ?DOC( " Create a new streaming state for incremental parsing\n" "\n" " Use this when implementing true real-time streaming with your own HTTP client.\n" "\n" " ## Example\n" "\n" " ```gleam\n" " let state = new_streaming_state()\n" "\n" " // As chunks arrive from your streaming HTTP client:\n" " let #(events, new_state) = process_chunk(state, chunk)\n" " list.each(events, handle_event_immediately)\n" " ```\n" ). -spec new_streaming_state() -> streaming_state(). new_streaming_state() -> {streaming_state, anthropic@internal@sse:new_parser_state(), [], false, none}. -file("src/anthropic/streaming/handler.gleam", 187). ?DOC( " Process a chunk of SSE data and return parsed events\n" "\n" " This is the core function for real-time streaming. Call it each time\n" " you receive a chunk from your streaming HTTP client.\n" "\n" " Returns a tuple of:\n" " - List of events parsed from this chunk (process these immediately!)\n" " - Updated state to use for the next chunk\n" "\n" " ## Example\n" "\n" " ```gleam\n" " // In your streaming HTTP callback:\n" " fn on_chunk(state: StreamingState, chunk: String) -> StreamingState {\n" " let #(events, new_state) = process_chunk(state, chunk)\n" "\n" " // Process events immediately as they arrive\n" " list.each(events, fn(event) {\n" " case get_event_text(event) {\n" " Ok(text) -> io.print(text) // Real-time output!\n" " Error(_) -> Nil\n" " }\n" " })\n" "\n" " new_state\n" " }\n" " ```\n" ). -spec process_chunk(streaming_state(), binary()) -> {list(anthropic@streaming:stream_event()), streaming_state()}. process_chunk(State, Chunk) -> Parse_result = anthropic@internal@sse:parse_chunk( erlang:element(2, State), Chunk ), Events = begin _pipe = erlang:element(2, Parse_result), gleam@list:filter_map( _pipe, fun(Sse_event) -> case anthropic@streaming@decoder:decode_event(Sse_event) of {ok, Event} -> {ok, Event}; {error, _} -> {error, nil} end end ) end, Completed = gleam@list:any(Events, fun(Event@1) -> case Event@1 of message_stop_event -> true; _ -> false end end), Error = begin _pipe@1 = gleam@list:find_map(Events, fun(Event@2) -> case Event@2 of {error_event, Err} -> {ok, {sse_parse_error, <<<<(erlang:element(2, Err))/binary, ": "/utf8>>/binary, (erlang:element(3, Err))/binary>>}}; _ -> {error, nil} end end), gleam@option:from_result(_pipe@1) end, New_state = {streaming_state, erlang:element(3, Parse_result), lists:append(erlang:element(3, State), Events), Completed, Error}, {Events, New_state}. -file("src/anthropic/streaming/handler.gleam", 237). ?DOC( " Finalize the stream and get any remaining events\n" "\n" " Call this when your HTTP client signals the stream has ended.\n" " Returns any events that were buffered but not yet emitted.\n" ). -spec finalize_stream(streaming_state()) -> list(anthropic@streaming:stream_event()). finalize_stream(State) -> case anthropic@internal@sse:flush(erlang:element(2, State)) of {ok, Sse_event} -> case anthropic@streaming@decoder:decode_event(Sse_event) of {ok, Event} -> [Event]; {error, _} -> [] end; {error, _} -> [] end. -file("src/anthropic/streaming/handler.gleam", 250). ?DOC(" Check if the stream has completed successfully\n"). -spec is_stream_complete(streaming_state()) -> boolean(). is_stream_complete(State) -> erlang:element(4, State). -file("src/anthropic/streaming/handler.gleam", 255). ?DOC(" Check if the stream encountered an error\n"). -spec has_stream_error(streaming_state()) -> boolean(). has_stream_error(State) -> gleam@option:is_some(erlang:element(5, State)). -file("src/anthropic/streaming/handler.gleam", 260). ?DOC(" Get the error from the stream if one occurred\n"). -spec get_stream_error(streaming_state()) -> gleam@option:option(stream_error()). get_stream_error(State) -> erlang:element(5, State). -file("src/anthropic/streaming/handler.gleam", 265). ?DOC(" Get all accumulated events from the streaming state\n"). -spec get_accumulated_events(streaming_state()) -> list(anthropic@streaming:stream_event()). get_accumulated_events(State) -> erlang:element(3, State). -file("src/anthropic/streaming/handler.gleam", 270). ?DOC(" Build a StreamResult from the final streaming state\n"). -spec build_stream_result(streaming_state()) -> stream_result(). build_stream_result(State) -> {stream_result, erlang:element(3, State)}. -file("src/anthropic/streaming/handler.gleam", 305). ?DOC( " Parse SSE body text into streaming events (batch mode)\n" "\n" " Use this for batch parsing when you have the complete SSE text.\n" " For real-time streaming, use `process_chunk` instead.\n" ). -spec parse_sse_body(binary()) -> {ok, stream_result()} | {error, stream_error()}. parse_sse_body(Body) -> State = new_streaming_state(), {Events, Final_state} = process_chunk(State, Body), Remaining = finalize_stream(Final_state), {ok, {stream_result, lists:append(Events, Remaining)}}. -file("src/anthropic/streaming/handler.gleam", 292). ?DOC( " Parse an HTTP response body containing SSE events into StreamResult\n" "\n" " This is for batch parsing of a complete response. For real-time streaming,\n" " use `new_streaming_state` + `process_chunk` instead.\n" "\n" " ## Example\n" "\n" " ```gleam\n" " let http_response = HttpResponse(status: 200, headers: [], body: sse_body)\n" " case parse_streaming_response(http_response) {\n" " Ok(result) -> get_full_text(result.events)\n" " Error(err) -> handle_error(err)\n" " }\n" " ```\n" ). -spec parse_streaming_response(anthropic@http:http_response()) -> {ok, stream_result()} | {error, stream_error()}. parse_streaming_response(Response) -> case erlang:element(2, Response) of 200 -> parse_sse_body(erlang:element(4, Response)); Status -> {error, {api_error, Status, erlang:element(4, Response)}} end. -file("src/anthropic/streaming/handler.gleam", 316). ?DOC( " Parse a single SSE chunk and return events plus updated SSE state\n" "\n" " Lower-level function that works directly with SSE parser state.\n" " Consider using `process_chunk` with `StreamingState` for a higher-level API.\n" ). -spec parse_sse_chunk(anthropic@internal@sse:sse_parser_state(), binary()) -> {list(anthropic@streaming:stream_event()), anthropic@internal@sse:sse_parser_state()}. parse_sse_chunk(State, Chunk) -> Parse_result = anthropic@internal@sse:parse_chunk(State, Chunk), Events = begin _pipe = erlang:element(2, Parse_result), gleam@list:filter_map( _pipe, fun(Sse_event) -> case anthropic@streaming@decoder:decode_event(Sse_event) of {ok, Event} -> {ok, Event}; {error, _} -> {error, nil} end end ) end, {Events, erlang:element(3, Parse_result)}. -file("src/anthropic/streaming/handler.gleam", 469). ?DOC(" Convert ConnectError to string\n"). -spec connect_error_to_string(gleam@httpc:connect_error()) -> binary(). connect_error_to_string(Err) -> case Err of {posix, Code} -> <<"POSIX error: "/utf8, Code/binary>>; {tls_alert, Code@1, Detail} -> <<<<<<"TLS alert "/utf8, Code@1/binary>>/binary, ": "/utf8>>/binary, Detail/binary>> end. -file("src/anthropic/streaming/handler.gleam", 453). ?DOC(" Convert httpc error to AnthropicError\n"). -spec http_error_to_anthropic_error(gleam@httpc:http_error()) -> anthropic@error:anthropic_error(). http_error_to_anthropic_error(Err) -> case Err of invalid_utf8_response -> anthropic@error:http_error(<<"Invalid UTF-8 in response"/utf8>>); {failed_to_connect, Ip4, Ip6} -> anthropic@error:network_error( <<<<<<<<"Failed to connect to server (IPv4: "/utf8, (connect_error_to_string(Ip4))/binary>>/binary, ", IPv6: "/utf8>>/binary, (connect_error_to_string(Ip6))/binary>>/binary, ")"/utf8>> ); response_timeout -> anthropic@error:timeout_error(0) end. -file("src/anthropic/streaming/handler.gleam", 417). ?DOC(" Make a streaming HTTP request using gleam_httpc\n"). -spec make_streaming_request(anthropic@client:client(), binary()) -> {ok, gleam@http@response:response(binary())} | {error, anthropic@error:anthropic_error()}. make_streaming_request(Api_client, Body) -> Base_url = erlang:element(3, erlang:element(2, Api_client)), Full_url = <>)/binary>>, gleam@result:'try'( begin _pipe = gleam@http@request:to(Full_url), gleam@result:map_error( _pipe, fun(_) -> anthropic@error:config_error( <<"Invalid URL: "/utf8, Full_url/binary>> ) end ) end, fun(Req) -> Req@1 = begin _pipe@1 = Req, _pipe@2 = gleam@http@request:set_method(_pipe@1, post), _pipe@3 = gleam@http@request:set_header( _pipe@2, <<"content-type"/utf8>>, <<"application/json"/utf8>> ), _pipe@4 = gleam@http@request:set_header( _pipe@3, <<"x-api-key"/utf8>>, anthropic@config:api_key_to_string( erlang:element(2, erlang:element(2, Api_client)) ) ), _pipe@5 = gleam@http@request:set_header( _pipe@4, <<"anthropic-version"/utf8>>, <<"2023-06-01"/utf8>> ), _pipe@6 = gleam@http@request:set_header( _pipe@5, <<"accept"/utf8>>, <<"text/event-stream"/utf8>> ), gleam@http@request:set_body(_pipe@6, Body) end, _pipe@7 = gleam@httpc:configure(), _pipe@8 = gleam@httpc:timeout( _pipe@7, erlang:element(5, erlang:element(2, Api_client)) ), _pipe@9 = gleam@httpc:dispatch(_pipe@8, Req@1), gleam@result:map_error( _pipe@9, fun(Err) -> http_error_to_anthropic_error(Err) end ) end ). -file("src/anthropic/streaming/handler.gleam", 356). ?DOC( " Stream a message request and return all events (batch mode)\n" "\n" " @deprecated Use `api.chat_stream` instead for a unified streaming API\n" "\n" " **Note**: This function collects ALL events before returning. It does NOT\n" " provide true real-time streaming. For real-time streaming, use the sans-io\n" " functions (`new_streaming_state`, `process_chunk`) with a streaming HTTP client.\n" "\n" " ## Example\n" "\n" " ```gleam\n" " // Prefer using api.chat_stream instead:\n" " case api.chat_stream(client, request) {\n" " Ok(result) -> api.stream_text(result)\n" " Error(err) -> handle_error(err)\n" " }\n" " ```\n" ). -spec stream_message( anthropic@client:client(), anthropic@request:create_message_request() ) -> {ok, stream_result()} | {error, stream_error()}. stream_message(Api_client, Message_request) -> Streaming_request = anthropic@request:with_stream(Message_request, true), Body = anthropic@request:request_to_json_string(Streaming_request), gleam@result:'try'( begin _pipe = make_streaming_request(Api_client, Body), gleam@result:map_error(_pipe, fun(Err) -> {http_error, Err} end) end, fun(Http_response) -> case erlang:element(2, Http_response) of 200 -> parse_sse_body(erlang:element(4, Http_response)); Status -> {error, {api_error, Status, erlang:element(4, Http_response)}} end end ). -file("src/anthropic/streaming/handler.gleam", 399). ?DOC( " Stream a message request with a callback for each event (batch mode)\n" "\n" " @deprecated Use `api.chat_stream_with_callback` instead for a unified streaming API\n" "\n" " **Note**: Despite the callback, this function collects ALL events before\n" " calling the callbacks. It does NOT provide true real-time streaming.\n" " For real-time streaming, use the sans-io functions with a streaming HTTP client.\n" "\n" " ## Example\n" "\n" " ```gleam\n" " // Prefer using api.chat_stream_with_callback instead:\n" " api.chat_stream_with_callback(client, request, fn(event) {\n" " case api.event_text(event) {\n" " Ok(text) -> io.print(text)\n" " Error(_) -> Nil\n" " }\n" " })\n" " ```\n" ). -spec stream_message_with_callback( anthropic@client:client(), anthropic@request:create_message_request(), fun((anthropic@streaming:stream_event()) -> nil) ) -> {ok, stream_result()} | {error, stream_error()}. stream_message_with_callback(Api_client, Message_request, Callback) -> gleam@result:'try'( stream_message(Api_client, Message_request), fun(Stream_result) -> gleam@list:each(erlang:element(2, Stream_result), Callback), {ok, Stream_result} end ). -file("src/anthropic/streaming/handler.gleam", 494). ?DOC( " Extract text from a single event if it contains text content\n" "\n" " Useful for real-time streaming to print text as it arrives.\n" "\n" " ## Example\n" "\n" " ```gleam\n" " list.each(events, fn(event) {\n" " case get_event_text(event) {\n" " Ok(text) -> io.print(text)\n" " Error(_) -> Nil\n" " }\n" " })\n" " ```\n" ). -spec get_event_text(anthropic@streaming:stream_event()) -> {ok, binary()} | {error, nil}. get_event_text(Event) -> case Event of {content_block_delta_event_variant, Delta_event} -> case erlang:element(3, Delta_event) of {text_content_delta, Text_delta} -> {ok, erlang:element(2, Text_delta)}; _ -> {error, nil} end; _ -> {error, nil} end. -file("src/anthropic/streaming/handler.gleam", 507). ?DOC(" Filter events to only text deltas\n"). -spec get_text_deltas(list(anthropic@streaming:stream_event())) -> list(binary()). get_text_deltas(Events) -> _pipe = Events, gleam@list:filter_map(_pipe, fun get_event_text/1). -file("src/anthropic/streaming/handler.gleam", 513). ?DOC(" Get the full text from a stream of events\n"). -spec get_full_text(list(anthropic@streaming:stream_event())) -> binary(). get_full_text(Events) -> _pipe = get_text_deltas(Events), gleam@string:join(_pipe, <<""/utf8>>). -file("src/anthropic/streaming/handler.gleam", 519). ?DOC(" Get the message ID from events\n"). -spec get_message_id(list(anthropic@streaming:stream_event())) -> {ok, binary()} | {error, nil}. get_message_id(Events) -> _pipe = Events, gleam@list:find_map(_pipe, fun(Event) -> case Event of {message_start_event, Msg} -> {ok, erlang:element(2, Msg)}; _ -> {error, nil} end end). -file("src/anthropic/streaming/handler.gleam", 530). ?DOC(" Get the model from events\n"). -spec get_model(list(anthropic@streaming:stream_event())) -> {ok, binary()} | {error, nil}. get_model(Events) -> _pipe = Events, gleam@list:find_map(_pipe, fun(Event) -> case Event of {message_start_event, Msg} -> {ok, erlang:element(5, Msg)}; _ -> {error, nil} end end). -file("src/anthropic/streaming/handler.gleam", 541). ?DOC(" Check if stream completed successfully (from event list)\n"). -spec is_complete(list(anthropic@streaming:stream_event())) -> boolean(). is_complete(Events) -> gleam@list:any(Events, fun(Event) -> case Event of message_stop_event -> true; _ -> false end end). -file("src/anthropic/streaming/handler.gleam", 551). ?DOC(" Check if stream ended with an error (from event list)\n"). -spec has_error(list(anthropic@streaming:stream_event())) -> boolean(). has_error(Events) -> gleam@list:any(Events, fun(Event) -> case Event of {error_event, _} -> true; _ -> false end end). -file("src/anthropic/streaming/handler.gleam", 561). ?DOC(" Get error from events if present\n"). -spec get_error(list(anthropic@streaming:stream_event())) -> {ok, anthropic@streaming:stream_error()} | {error, nil}. get_error(Events) -> _pipe = Events, gleam@list:find_map(_pipe, fun(Event) -> case Event of {error_event, Err} -> {ok, Err}; _ -> {error, nil} end end).