nat.front_ends.fastapi.response_helpers#

Functions#

`generate_streaming_response_as_str`(), result_type, ...)
`generate_streaming_response`(), result_type, output_type)
`generate_single_response`(→ Any)
`generate_streaming_response_full`(...)	Similar to generate_streaming_response but provides raw ResponseIntermediateStep objects
`generate_streaming_response_full_as_str`(...)	Similar to generate_streaming_response but converts the response to a string format.
`generate_streaming_response_atif`(...)	Stream ATIF steps by converting raw IntermediateSteps on-the-fly.
`generate_streaming_response_atif_as_str`(...)	String-serialized variant of `generate_streaming_response_atif`.

Module Contents#

async generate_streaming_response_as_str( payload: Any, *, session: nat.runtime.session.Session, streaming: bool, step_adaptor: nat.front_ends.fastapi.step_adaptor.StepAdaptor = StepAdaptor(StepAdaptorConfig()), result_type: type | None = None, output_type: type | None = None, ) → collections.abc.AsyncGenerator[str]#

async generate_streaming_response( payload: Any, *, session: nat.runtime.session.Session, streaming: bool, step_adaptor: nat.front_ends.fastapi.step_adaptor.StepAdaptor = StepAdaptor(StepAdaptorConfig()), result_type: type | None = None, output_type: type | None = None, ) → collections.abc.AsyncGenerator[nat.data_models.api_server.ResponseSerializable]#

async generate_single_response( payload: Any, session: nat.runtime.session.Session, result_type: type | None = None, ) → Any#

async generate_streaming_response_full( payload: Any, *, session: nat.runtime.session.Session, streaming: bool, result_type: type | None = None, output_type: type | None = None, filter_steps: str | None = None, ) → collections.abc.AsyncGenerator[nat.data_models.api_server.ResponseSerializable]#: Similar to generate_streaming_response but provides raw ResponseIntermediateStep objects without any step adaptor translations.

async generate_streaming_response_full_as_str( payload: Any, *, session: nat.runtime.session.Session, streaming: bool, result_type: type | None = None, output_type: type | None = None, filter_steps: str | None = None, ) → collections.abc.AsyncGenerator[str]#: Similar to generate_streaming_response but converts the response to a string format.

async generate_streaming_response_atif( payload: Any, *, session: nat.runtime.session.Session, streaming: bool, result_type: type | None = None, output_type: type | None = None, ) → collections.abc.AsyncGenerator[nat.data_models.api_server.ResponseSerializable]#

Stream ATIF steps by converting raw IntermediateSteps on-the-fly.

Each yielded item is either a ResponseATIFStep (one per completed agent turn) or a ResponsePayloadOutput (the final workflow result). A ResponseATIFTrajectory summary is emitted at the very end.

async generate_streaming_response_atif_as_str( payload: Any, *, session: nat.runtime.session.Session, streaming: bool, result_type: type | None = None, output_type: type | None = None, ) → collections.abc.AsyncGenerator[str]#: String-serialized variant of generate_streaming_response_atif.