qi(dZddlZddlmZddlmZ ddlmZddlm Z m Z ddl m Z m Z mZmZddlmZdd lmZdd lmZGddZy#e$r7Zej0d eej0d ed edZ[wwxYw)aAWS SageMaker bidirectional streaming client. This module provides a client for streaming bidirectional communication with SageMaker endpoints using the HTTP/2 protocol. Supports sending audio, text, and JSON data to SageMaker model endpoints and receiving streaming responses. N)Optional)logger)SageMakerRuntimeHTTP2Client)ConfigHTTPAuthSchemeResolver)*InvokeEndpointWithBidirectionalStreamInputRequestPayloadPartRequestStreamEventPayloadPartResponseStreamEvent)SigV4AuthScheme)EnvironmentCredentialsResolver)DuplexEventStreamz Exception: zWIn order to use SageMaker BiDi client, you need to `pip install pipecat-ai[sagemaker]`.zMissing module: c eZdZdZ ddedededefdZdZdZdd ed e efd Z d efdZ defdZ de fdZde efdZdZedefdZy )SageMakerBidiClientapClient for bidirectional streaming with AWS SageMaker endpoints. Handles low-level HTTP/2 bidirectional streaming protocol for communicating with SageMaker model endpoints. Provides methods for sending various data types (audio, text, JSON) and receiving streaming responses. This client uses AWS SigV4 authentication and supports credential resolution from environment variables, AWS CLI configuration, and instance metadata. Example:: client = SageMakerBidiClient( endpoint_name="my-deepgram-endpoint", region="us-east-2", model_invocation_path="v1/listen", model_query_string="model=nova-3&language=en" ) await client.start_session() await client.send_audio_chunk(audio_bytes) response = await client.receive_response() await client.close_session() endpoint_nameregionmodel_invocation_pathmodel_query_stringc||_||_||_||_d|d|_d|_d|_d|_d|_y)a}Initialize the SageMaker BiDi client. Args: endpoint_name: Name of the SageMaker endpoint to connect to. region: AWS region where the endpoint is deployed. model_invocation_path: API path for the model invocation (e.g., "v1/listen"). model_query_string: Query string parameters for the model (e.g., "model=nova-3"). zhttps://runtime.sagemaker.z.amazonaws.com:8443NF) rrrr bidi_endpoint_client_stream_output_stream _is_active)selfrrrrs \/opt/pipecat/venv/lib/python3.12/site-packages/pipecat/services/aws/sagemaker/bidi_client.py__init__zSageMakerBidiClient.__init__?sY+ %:""49&ATU>B   #c tjd|jtjd|jt t j dxrt j d}|stjdt|j|jttdtdi }t| |_ y ) a'Initialize the SageMaker Runtime HTTP2 client with AWS credentials. Creates and configures the SageMaker Runtime HTTP2 client with SigV4 authentication. Attempts to resolve AWS credentials from environment variables, AWS CLI configuration, or instance metadata. z/Initializing SageMaker BiDi client for region: zUsing endpoint URI: AWS_ACCESS_KEY_IDAWS_SECRET_ACCESS_KEYzAWS credentials not found in environment variables. Attempting to use EnvironmentCredentialsResolver which will check AWS CLI configuration and instance metadata.zaws.auth#sigv4 sagemaker)service) endpoint_urir!aws_credentials_identity_resolverauth_scheme_resolver auth_schemes)configN)rdebugrrboolosgetenvwarningrr rr rr)r has_env_credsr(s r_initialize_clientz&SageMakerBidiClient._initialize_clientZs  Ft{{mTU +D,>,>+?@ARYY':;b Ja@bc  NN?  ++;;.L.N!7!9*OK,PQ  3&A rcK|js|jtjd|jtjd|j tjd|j t|j|j |j } |jj|d{|_ d|_ |jjd{}|d|_ tjd|jS7]71#t$r2}tjd |d |_ td |d}~wwxYww) aStart a bidirectional streaming session with the SageMaker endpoint. Initializes the client if needed, creates the bidirectional stream, and establishes the connection to the SageMaker endpoint. Must be called before sending or receiving data. Returns: The output stream for receiving responses. Raises: RuntimeError: If client initialization or connection fails. z%Starting BiDi session with endpoint: zModel invocation path: zModel query string: )rrrNTz!BiDi session started successfullyzFailed to start BiDi session: Fz(Failed to start SageMaker BiDi session: )rr/rr)rrrr)invoke_endpoint_with_bidirectional_streamrr await_outputr Exceptionerror RuntimeError)r stream_inputoutputes r start_sessionz!SageMakerBidiClient.start_sessionwsI||  # # % #DO!I!MN N OsNB/E02D2D.-D2>D0?.D2-E0.D20D22 E-;-E((E--E0N data_bytes data_typec8K|jr |js td t||}t |}|jj j |d{y7#t$r}tjd|d}~wwxYww)aSend a chunk of data to the stream. Generic method for sending any type of data to the SageMaker endpoint. Use the convenience methods (send_audio_chunk, send_text, send_json) for common data types. Args: data_bytes: Raw bytes to send. data_type: Optional data type header. Common values are "BINARY" for audio/binary data and "UTF8" for text/JSON data. Raises: RuntimeError: If session is not active or send fails. BiDi session not active)bytes_r<)valueNzFailed to send data: ) rrr6r r input_streamsendr4rr5)rr;r<payloadeventr9s r send_datazSageMakerBidiClient.send_datasdll89 9 ( iPG1@E,,++007 7 7  LL04 5  s;$BAA0(A.)A0-B.A00 B9BBB audio_bytescFK|j|dd{y7w)aGSend a chunk of audio data to the stream. Convenience method for sending audio data. Automatically sets the data type to "BINARY". Args: audio_bytes: Raw audio bytes to send (e.g., PCM audio data). Raises: RuntimeError: If session is not active or send fails. BINARYr<N)rE)rrFs rsend_audio_chunkz$SageMakerBidiClient.send_audio_chunksnn[Hn===s !!textcdK|j|jddd{y7w)a4Send text data to the stream. Convenience method for sending text data. Automatically encodes the text as UTF-8 and sets the data type to "UTF8". Args: text: Text string to send. Raises: RuntimeError: If session is not active or send fails. utf-8UTF8rIN)rEencode)rrKs r send_textzSageMakerBidiClient.send_texts'nnT[[1VnDDDs &0.0datacKddl}|j|j|jddd{y7w)aSend JSON data to the stream. Convenience method for sending JSON-encoded messages. Useful for control messages like KeepAlive or CloseStream. Automatically serializes the dictionary to JSON, encodes as UTF-8, and sets the data type to "UTF8". Args: data: Dictionary to send as JSON (e.g., {"type": "KeepAlive"}). Raises: RuntimeError: If session is not active or send fails. rNrMrNrI)jsonrEdumpsrO)rrQrSs r send_jsonzSageMakerBidiClient.send_jsons5 nnTZZ-44W=nPPPs9AAAreturncK|jr |js td |jjd{}|S7#t$r}t j d|d}~wwxYww)aSReceive a response from the stream. Blocks until a response is available from the SageMaker endpoint. Returns None when the stream is closed. Returns: The response event containing payload data, or None if stream is closed. Raises: RuntimeError: If session is not active. r>NzFailed to receive response: )rrr6receiver4rr5)rresultr9s rreceive_responsez$SageMakerBidiClient.receive_responsesod&9&989 9 ..6688FM9  LL7s; <  s:$A7A A A A7 A A4A//A44A7cZK|jsytjdd|_ |jr,|jjj d{tjdy7#t $r"}tjd|Yd}~yd}~wwxYww)zClose the bidirectional streaming session. Gracefully closes the input stream and marks the session as inactive. Safe to call multiple times. NzClosing BiDi session...Fz BiDi session closed successfullyzError closing BiDi session: )rrr)rrAcloser4r-)rr9s r close_sessionz!SageMakerBidiClient.close_sessions   ./ ?||ll//55777 LL; <8 ? NN9!= > > ?s@*B+3A= A;!A=:B+;A== B(B#B+#B((B+c|jS)z|Check if the session is currently active. Returns: True if session is active, False otherwise. )r)rs r is_activezSageMakerBidiClient.is_activesr)r`)N)__name__ __module__ __qualname____doc__strrr/r:bytesrrErJrPdictrUr rZr]propertyr*r_rrrr's6&("$     #    6B:+OZ%HSM4 >% > EC EQDQ"1D(E,?&4rr)rdr+typingrlogurur&aws_sdk_sagemaker_runtime_http2.clientr&aws_sdk_sagemaker_runtime_http2.configrr&aws_sdk_sagemaker_runtime_http2.modelsrr r r smithy_aws_core.auth.sigv4r smithy_aws_core.identityr smithy_core.aio.eventstreamrModuleNotFoundErrorr9r5r4rrirrrss ,RU ;G=tt,FLL;qc"#FLLa &qc* ++ ,s,A B2BB