-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([stream_message/2, stream_message_with_callback/3, 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([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 HTTP handler for Anthropic API\n" "\n" " This module provides the HTTP handling for streaming responses from the\n" " Anthropic Messages API. It processes Server-Sent Events (SSE) and yields\n" " parsed streaming events.\n" "\n" " Note: Currently uses synchronous HTTP with SSE parsing. True streaming\n" " with real-time chunk processing can be added via Erlang interop if needed.\n" ). -type stream_result() :: {stream_result, list(anthropic@types@streaming:stream_event())}. -type stream_error() :: {http_error, anthropic@types@error:anthropic_error()} | {sse_parse_error, binary()} | {event_decode_error, binary()} | {api_error, integer(), binary()}. -file("src/anthropic/streaming/handler.gleam", 172). ?DOC(" Parse an SSE response body into streaming events\n"). -spec parse_sse_response(binary()) -> {ok, stream_result()} | {error, stream_error()}. parse_sse_response(Body) -> State = anthropic@streaming@sse:new_parser_state(), Parse_result = anthropic@streaming@sse:parse_chunk(State, Body), 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, Final_events = case anthropic@streaming@sse:flush( erlang:element(3, Parse_result) ) of {ok, Sse_event@1} -> case anthropic@streaming@decoder:decode_event(Sse_event@1) of {ok, Event@1} -> lists:append(Events, [Event@1]); {error, _} -> Events end; {error, _} -> Events end, {ok, {stream_result, Final_events}}. -file("src/anthropic/streaming/handler.gleam", 217). ?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", 201). ?DOC(" Convert httpc error to AnthropicError\n"). -spec http_error_to_anthropic_error(gleam@httpc:http_error()) -> anthropic@types@error:anthropic_error(). http_error_to_anthropic_error(Err) -> case Err of invalid_utf8_response -> anthropic@types@error:http_error( <<"Invalid UTF-8 in response"/utf8>> ); {failed_to_connect, Ip4, Ip6} -> anthropic@types@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@types@error:timeout_error(0) end. -file("src/anthropic/streaming/handler.gleam", 139). ?DOC(" Make a streaming HTTP request\n"). -spec make_streaming_request(anthropic@client:client(), binary()) -> {ok, gleam@http@response:response(binary())} | {error, anthropic@types@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@types@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>>, 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", 76). ?DOC( " Stream a message request and return all events\n" "\n" " This function sends a streaming request to the Anthropic API,\n" " processes the SSE response, and returns all parsed events.\n" "\n" " ## Example\n" "\n" " ```gleam\n" " let request = create_request(model, messages, max_tokens)\n" " |> with_stream(True)\n" "\n" " case stream_message(client, request) {\n" " Ok(result) -> {\n" " list.each(result.events, fn(event) {\n" " io.println(event_type_string(event))\n" " })\n" " }\n" " Error(err) -> handle_error(err)\n" " }\n" " ```\n" ). -spec stream_message( anthropic@client:client(), anthropic@types@request:create_message_request() ) -> {ok, stream_result()} | {error, stream_error()}. stream_message(Api_client, Message_request) -> Streaming_request = anthropic@types@request:with_stream( Message_request, true ), Body = anthropic@types@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_response(erlang:element(4, Http_response)); Status -> {error, {api_error, Status, erlang:element(4, Http_response)}} end end ). -file("src/anthropic/streaming/handler.gleam", 121). ?DOC( " Stream a message request with a callback for each event\n" "\n" " This function is similar to `stream_message` but calls the provided\n" " callback function for each event as it is parsed.\n" "\n" " ## Example\n" "\n" " ```gleam\n" " stream_message_with_callback(client, request, fn(event) {\n" " case event {\n" " ContentBlockDeltaEventVariant(delta) -> {\n" " case delta.delta {\n" " TextContentDelta(text_delta) -> {\n" " io.print(text_delta.text)\n" " }\n" " _ -> Nil\n" " }\n" " }\n" " _ -> Nil\n" " }\n" " })\n" " ```\n" ). -spec stream_message_with_callback( anthropic@client:client(), anthropic@types@request:create_message_request(), fun((anthropic@types@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(Result) -> gleam@list:each(erlang:element(2, Result), Callback), {ok, Result} end ). -file("src/anthropic/streaming/handler.gleam", 229). ?DOC(" Filter events to only text deltas\n"). -spec get_text_deltas(list(anthropic@types@streaming:stream_event())) -> list(binary()). get_text_deltas(Events) -> _pipe = Events, gleam@list:filter_map(_pipe, fun(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 end). -file("src/anthropic/streaming/handler.gleam", 245). ?DOC(" Get the full text from a stream of events\n"). -spec get_full_text(list(anthropic@types@streaming:stream_event())) -> binary(). get_full_text(Events) -> _pipe = get_text_deltas(Events), gleam@string:join(_pipe, <<""/utf8>>). -file("src/anthropic/streaming/handler.gleam", 251). ?DOC(" Get the message ID from events\n"). -spec get_message_id(list(anthropic@types@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", 262). ?DOC(" Get the model from events\n"). -spec get_model(list(anthropic@types@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", 273). ?DOC(" Check if stream completed successfully\n"). -spec is_complete(list(anthropic@types@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", 283). ?DOC(" Check if stream ended with an error\n"). -spec has_error(list(anthropic@types@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", 293). ?DOC(" Get error from events if present\n"). -spec get_error(list(anthropic@types@streaming:stream_event())) -> {ok, anthropic@types@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).