nat.data_models.api_server#

Attributes#

`FINISH_REASONS`
`UserContent`
`Choice`
`AuthPayload`

Classes#

`UserMessageContentRoleType`	Enum representing chat message roles in API requests and responses.
`Request`	Request is a data model that represents HTTP request and WebSocket attributes.
`ChatContentType`	ChatContentType is an Enum that represents the type of Chat content.
`InputAudio`
`AudioContent`
`ImageUrl`
`ImageContent`
`TextContent`
`Message`
`ChatRequest`	ChatRequest is a data model that represents a request to the NAT chat API.
`ChatRequestOrMessage`	`ChatRequestOrMessage` is a data model that represents either a conversation or a string input.
`ChoiceMessage`
`ChoiceDeltaToolCallFunction`	Function details within a streamed tool call delta (OpenAI-compatible).
`ChoiceDeltaToolCall`	Tool call delta for streaming responses (OpenAI-compatible).
`ChoiceDelta`	Delta object for streaming responses (OpenAI-compatible)
`ChoiceBase`	Base choice model with common fields for both streaming and non-streaming responses
`ChatResponseChoice`	Choice model for non-streaming responses - contains message field
`ChatResponseChunkChoice`	Choice model for streaming responses - contains delta field
`Usage`
`ResponseSerializable`	ResponseSerializable is an abstract class that defines the interface for serializing output for the NAT
`ResponseBaseModelOutput`
`ResponseBaseModelIntermediate`
`ChatResponse`	ChatResponse is a data model that represents a response from the NAT chat API.
`ChatResponseChunk`	ChatResponseChunk is a data model that represents a response chunk from the NAT chat streaming API.
`ResponseIntermediateStep`	ResponseSerializedStep is a data model that represents a serialized step in the NAT chat streaming API.
`ResponseObservabilityTrace`	ResponseObservabilityTrace is a data model that represents an observability trace event
`ResponsePayloadOutput`
`ResponseATIFStep`	An ATIF step emitted during streaming on the `/v1/workflow/atif` endpoint.
`ResponseATIFTrajectory`	Final ATIF trajectory summary emitted at the end of an ATIF stream.
`GenerateResponse`
`WebSocketMessageType`	WebSocketMessageType is an Enum that represents WebSocket Message types.
`WorkflowSchemaType`	WorkflowSchemaType is an Enum that represents Workkflow response types.
`WebSocketMessageStatus`	WebSocketMessageStatus is an Enum that represents the status of a WebSocket message.
`UserMessages`
`UserMessageContent`
`User`
`ErrorTypes`	Enum where members are also (and must be) strings
`Error`
`WebSocketUserMessage`	For more details, refer to the API documentation:
`WebSocketUserInteractionResponseMessage`	For more details, refer to the API documentation:
`AuthMethod`	Supported authentication methods for WebSocket auth messages.
`JwtAuthPayload`	JWT Bearer token authentication payload.
`ApiKeyAuthPayload`	API key authentication payload.
`BasicAuthPayload`	Username/password authentication payload.
`WebSocketAuthMessage`	WebSocket authentication message for payload-based auth when header or cookie auth is not feasible.
`AuthMessageStatus`	Outcome of a WebSocket authentication attempt.
`WebSocketAuthResponseMessage`	Server response to a WebSocket `auth_message`.
`SystemIntermediateStepContent`
`WebSocketSystemIntermediateStepMessage`	For more details, refer to the API documentation:
`SystemResponseContent`
`WebSocketSystemResponseTokenMessage`	For more details, refer to the API documentation:
`WebSocketSystemInteractionMessage`	For more details, refer to the API documentation:
`ObservabilityTraceContent`
`WebSocketObservabilityTraceMessage`	WebSocket message for observability trace ID.

Functions#

`_generate_response_to_str`(→ str)
`_generate_response_to_chat_response`(→ ChatResponse)
`_nat_chat_request_to_string`(→ str)
`_string_to_nat_chat_request`(→ ChatRequest)
`_chat_request_or_message_to_chat_request`(→ ChatRequest)
`_chat_request_to_chat_request_or_message`(...)
`_chat_request_or_message_to_string`(→ str)
`_string_to_chat_request_or_message`(→ ChatRequestOrMessage)
`_nat_chat_response_to_string`(→ str)
`_string_to_nat_chat_response`(→ ChatResponse)	Converts a string to an ChatResponse object
`_chat_response_chunk_to_string`(→ str)
`_string_to_nat_chat_response_chunk`(→ ChatResponseChunk)	Converts a string to an ChatResponseChunk object

Module Contents#

FINISH_REASONS#

class UserMessageContentRoleType#

Bases: enum.StrEnum

Enum representing chat message roles in API requests and responses.

Initialize self. See help(type(self)) for accurate signature.

USER = 'user'#

ASSISTANT = 'assistant'#

SYSTEM = 'system'#

class Request(/, **data: Any)#

Bases: pydantic.BaseModel

Request is a data model that represents HTTP request and WebSocket attributes.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

model_config#: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

method: str | None = None#

url_path: str | None = None#

url_port: int | None = None#

url_scheme: str | None = None#

headers: Any | None = None#

query_params: Any | None = None#

path_params: dict[str, str] | None = None#

client_host: str | None = None#

client_port: int | None = None#

cookies: dict[str, str] | None = None#

payload: dict[str, Any] | None = None#

class ChatContentType#

Bases: enum.StrEnum

ChatContentType is an Enum that represents the type of Chat content.

Initialize self. See help(type(self)) for accurate signature.

TEXT = 'text'#

IMAGE_URL = 'image_url'#

INPUT_AUDIO = 'input_audio'#

class InputAudio(/, **data: Any)#

Bases: pydantic.BaseModel

data: str = 'default'#

format: str = 'default'#

class AudioContent(/, **data: Any)#

Bases: pydantic.BaseModel

model_config#: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

type: Literal[ChatContentType]#

input_audio: InputAudio#

class ImageUrl(/, **data: Any)#

Bases: pydantic.BaseModel

url: str = None#

class ImageContent(/, **data: Any)#

Bases: pydantic.BaseModel

model_config#: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

type: Literal[ChatContentType]#

image_url: ImageUrl#

class TextContent(/, **data: Any)#

Bases: pydantic.BaseModel

model_config#: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

type: Literal[ChatContentType]#

text: str = 'default'#

UserContent#

class Message(/, **data: Any)#

Bases: pydantic.BaseModel

content: str | list[UserContent]#

role: UserMessageContentRoleType#

class ChatRequest(/, **data: Any)#

Bases: pydantic.BaseModel

ChatRequest is a data model that represents a request to the NAT chat API. Fully compatible with OpenAI Chat Completions API specification.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

messages: Annotated[list[Message], conlist(Message, min_length=1)]#

model: str | None = None#

frequency_penalty: float | None = None#

logit_bias: dict[str, float] | None = None#

logprobs: bool | None = None#

top_logprobs: int | None = None#

max_tokens: int | None = None#

n: int | None = None#

presence_penalty: float | None = None#

response_format: dict[str, Any] | None = None#

seed: int | None = None#

service_tier: Literal['auto', 'default'] | None = None#

stream: bool | None = None#

stream_options: dict[str, Any] | None = None#

temperature: float | None = None#

top_p: float | None = None#

tools: list[dict[str, Any]] | None = None#

tool_choice: str | dict[str, Any] | None = None#

parallel_tool_calls: bool | None = None#

user: str | None = None#

model_config#: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

static from_string( data: str, *, model: str | None = None, temperature: float | None = None, max_tokens: int | None = None, top_p: float | None = None, ) → ChatRequest#

static from_content( content: list[UserContent], *, model: str | None = None, temperature: float | None = None, max_tokens: int | None = None, top_p: float | None = None, ) → ChatRequest#

class ChatRequestOrMessage(/, **data: Any)#

Bases: pydantic.BaseModel

ChatRequestOrMessage is a data model that represents either a conversation or a string input. This is useful for functions that can handle either type of input.

messages is compatible with the OpenAI Chat Completions API specification.
input_message is a string input that can be used for functions that do not require a conversation.

Note: When messages is provided, extra fields are allowed to enable lossless round-trip conversion with ChatRequest. When input_message is provided, no extra fields are permitted.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

model_config#: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

messages: Annotated[list[Message] | None, conlist(Message, min_length=1)] = None#

input_message: str | None = None#

property is_string: bool#

property is_conversation: bool#

validate_model()#

class ChoiceMessage(/, **data: Any)#

Bases: pydantic.BaseModel

content: str | None = None#

role: UserMessageContentRoleType | None = None#

class ChoiceDeltaToolCallFunction(/, **data: Any)#

Bases: pydantic.BaseModel

Function details within a streamed tool call delta (OpenAI-compatible).

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

name: str | None = None#

arguments: str | None = None#

class ChoiceDeltaToolCall(/, **data: Any)#

Bases: pydantic.BaseModel

Tool call delta for streaming responses (OpenAI-compatible).

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

index: int#

id: str | None = None#

type: str | None = None#

function: ChoiceDeltaToolCallFunction | None = None#

class ChoiceDelta(/, **data: Any)#

Bases: pydantic.BaseModel

Delta object for streaming responses (OpenAI-compatible)

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

content: str | None = None#

role: UserMessageContentRoleType | None = None#

tool_calls: list[ChoiceDeltaToolCall] | None = None#

class ChoiceBase(/, **data: Any)#

Bases: pydantic.BaseModel

Base choice model with common fields for both streaming and non-streaming responses

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

model_config#: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

finish_reason: Literal['stop', 'length', 'tool_calls', 'content_filter', 'function_call'] | None = None#

index: int#

class ChatResponseChoice(/, **data: Any)#

Bases: ChoiceBase

Choice model for non-streaming responses - contains message field

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

message: ChoiceMessage#

class ChatResponseChunkChoice(/, **data: Any)#

Bases: ChoiceBase

Choice model for streaming responses - contains delta field

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

delta: ChoiceDelta#

Choice#

class Usage(/, **data: Any)#

Bases: pydantic.BaseModel

prompt_tokens: int | None = None#

completion_tokens: int | None = None#

total_tokens: int | None = None#

class ResponseSerializable#

Bases: abc.ABC

ResponseSerializable is an abstract class that defines the interface for serializing output for the NAT Toolkit chat streaming API.

abstractmethod get_stream_data() → str#

class ResponseBaseModelOutput(/, **data: Any)#

Bases: pydantic.BaseModel, ResponseSerializable

get_stream_data() → str#

class ResponseBaseModelIntermediate(/, **data: Any)#

Bases: pydantic.BaseModel, ResponseSerializable

get_stream_data() → str#

class ChatResponse(/, **data: Any)#

Bases: ResponseBaseModelOutput

ChatResponse is a data model that represents a response from the NAT chat API. Fully compatible with OpenAI Chat Completions API specification.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

model_config#: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

id: str#

object: str = 'chat.completion'#

model: str = 'unknown-model'#

created: datetime.datetime#

choices: list[ChatResponseChoice]#

usage: Usage#

system_fingerprint: str | None = None#

service_tier: Literal['scale', 'default'] | None = None#

serialize_created(created: datetime.datetime) → int#: Serialize datetime to Unix timestamp for OpenAI compatibility

static from_string( data: str, *, id_: str | None = None, object_: str | None = None, model: str | None = None, created: datetime.datetime | None = None, usage: Usage, ) → ChatResponse#

class ChatResponseChunk(/, **data: Any)#

Bases: ResponseBaseModelOutput

ChatResponseChunk is a data model that represents a response chunk from the NAT chat streaming API. Fully compatible with OpenAI Chat Completions API specification.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

model_config#: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

id: str#

choices: list[ChatResponseChunkChoice]#

created: datetime.datetime#

model: str = 'unknown-model'#

object: str = 'chat.completion.chunk'#

system_fingerprint: str | None = None#

service_tier: Literal['scale', 'default'] | None = None#

usage: Usage | None = None#

serialize_created(created: datetime.datetime) → int#: Serialize datetime to Unix timestamp for OpenAI compatibility

static from_string( data: str, *, id_: str | None = None, created: datetime.datetime | None = None, model: str | None = None, object_: str | None = None, finish_reason: str | None = None, ) → ChatResponseChunk#

static create_streaming_chunk( content: str, *, id_: str | None = None, created: datetime.datetime | None = None, model: str | None = None, role: UserMessageContentRoleType | None = None, finish_reason: str | None = None, usage: Usage | None = None, system_fingerprint: str | None = None, ) → ChatResponseChunk#: Create an OpenAI-compatible streaming chunk

class ResponseIntermediateStep(/, **data: Any)#

Bases: ResponseBaseModelIntermediate

ResponseSerializedStep is a data model that represents a serialized step in the NAT chat streaming API.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

model_config#: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

id: str#

parent_id: str | None = None#

type: str = 'markdown'#

name: str#

payload: str#

class ResponseObservabilityTrace(/, **data: Any)#

Bases: pydantic.BaseModel, ResponseSerializable

ResponseObservabilityTrace is a data model that represents an observability trace event sent once when the trace ID becomes available

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

observability_trace_id: str#

get_stream_data() → str#

class ResponsePayloadOutput(/, **data: Any)#

Bases: pydantic.BaseModel, ResponseSerializable

payload: Any#

get_stream_data() → str#

class ResponseATIFStep(/, **data: Any)#

Bases: pydantic.BaseModel, ResponseSerializable

An ATIF step emitted during streaming on the /v1/workflow/atif endpoint.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

step_id: int#

source: str#

message: str = ''#

timestamp: str | None = None#

model_name: str | None = None#

reasoning_content: str | None = None#

tool_calls: list[dict[str, Any]] | None = None#

observation: dict[str, Any] | None = None#

metrics: dict[str, Any] | None = None#

extra: dict[str, Any] | None = None#

get_stream_data() → str#

class ResponseATIFTrajectory(/, **data: Any)#

Bases: pydantic.BaseModel, ResponseSerializable

Final ATIF trajectory summary emitted at the end of an ATIF stream.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

schema_version: str#

session_id: str#

agent: dict[str, Any]#

final_metrics: dict[str, Any] | None = None#

get_stream_data() → str#

class GenerateResponse(/, **data: Any)#

Bases: pydantic.BaseModel

model_config#: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

intermediate_steps: list[tuple] | None = None#

output: str#

value: str | None = 'default'#

class WebSocketMessageType#

Bases: enum.StrEnum

WebSocketMessageType is an Enum that represents WebSocket Message types.

Initialize self. See help(type(self)) for accurate signature.

USER_MESSAGE = 'user_message'#

RESPONSE_MESSAGE = 'system_response_message'#

INTERMEDIATE_STEP_MESSAGE = 'system_intermediate_message'#

SYSTEM_INTERACTION_MESSAGE = 'system_interaction_message'#

USER_INTERACTION_MESSAGE = 'user_interaction_message'#

AUTH_MESSAGE = 'auth_message'#

AUTH_RESPONSE = 'auth_response_message'#

OBSERVABILITY_TRACE_MESSAGE = 'observability_trace_message'#

ERROR_MESSAGE = 'error_message'#

class WorkflowSchemaType#

Bases: enum.StrEnum

WorkflowSchemaType is an Enum that represents Workkflow response types.

Initialize self. See help(type(self)) for accurate signature.

GENERATE_STREAM = 'generate_stream'#

CHAT_STREAM = 'chat_stream'#

GENERATE = 'generate'#

CHAT = 'chat'#

class WebSocketMessageStatus#

Bases: enum.StrEnum

WebSocketMessageStatus is an Enum that represents the status of a WebSocket message.

Initialize self. See help(type(self)) for accurate signature.

IN_PROGRESS = 'in_progress'#

COMPLETE = 'complete'#

class UserMessages(/, **data: Any)#

Bases: pydantic.BaseModel

model_config#: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

role: UserMessageContentRoleType#

content: list[UserContent]#

class UserMessageContent(/, **data: Any)#

Bases: pydantic.BaseModel

model_config#: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

messages: list[UserMessages]#

class User(/, **data: Any)#

Bases: pydantic.BaseModel

model_config#: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

name: str = 'default'#

email: str = 'default'#

class ErrorTypes#

Bases: enum.StrEnum

Enum where members are also (and must be) strings

Initialize self. See help(type(self)) for accurate signature.

UNKNOWN_ERROR = 'unknown_error'#

WORKFLOW_ERROR = 'workflow_error'#

INVALID_MESSAGE = 'invalid_message'#

INVALID_MESSAGE_TYPE = 'invalid_message_type'#

INVALID_USER_MESSAGE_CONTENT = 'invalid_user_message_content'#

INVALID_DATA_CONTENT = 'invalid_data_content'#

USER_AUTH_ERROR = 'user_auth_error'#

class Error(/, **data: Any)#

Bases: pydantic.BaseModel

model_config#: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

code: ErrorTypes#

message: str = 'default'#

details: str = 'default'#

class WebSocketUserMessage(/, **data: Any)#

Bases: pydantic.BaseModel

For more details, refer to the API documentation: docs/source/developer_guide/websockets.md

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

model_config#: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

type: Literal[WebSocketMessageType]#

schema_type: WorkflowSchemaType#

id: str = 'default'#

conversation_id: str | None = None#

content: UserMessageContent#

user: User#

error: Error#

schema_version: str = '1.0.0'#

timestamp: str = ''#

class WebSocketUserInteractionResponseMessage(/, **data: Any)#

Bases: pydantic.BaseModel

For more details, refer to the API documentation: docs/source/developer_guide/websockets.md

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

type: Literal[WebSocketMessageType]#

id: str = 'default'#

thread_id: str = 'default'#

parent_id: str = 'default'#

conversation_id: str | None = None#

content: UserMessageContent#

user: User#

error: Error#

schema_version: str = '1.0.0'#

timestamp: str = ''#

class AuthMethod#

Bases: enum.StrEnum

Supported authentication methods for WebSocket auth messages.

Initialize self. See help(type(self)) for accurate signature.

JWT = 'jwt'#

API_KEY = 'api_key'#

BASIC = 'basic'#

class JwtAuthPayload(/, **data: Any)#

Bases: pydantic.BaseModel

JWT Bearer token authentication payload.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

model_config#: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

method: Literal[AuthMethod] = None#

token: nat.data_models.common.SerializableSecretStr = None#

class ApiKeyAuthPayload(/, **data: Any)#

Bases: pydantic.BaseModel

API key authentication payload.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

model_config#: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

method: Literal[AuthMethod] = None#

token: nat.data_models.common.SerializableSecretStr = None#

class BasicAuthPayload(/, **data: Any)#

Bases: pydantic.BaseModel

Username/password authentication payload.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

model_config#: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

method: Literal[AuthMethod] = None#

username: str = None#

password: nat.data_models.common.SerializableSecretStr = None#

AuthPayload#

class WebSocketAuthMessage(/, **data: Any)#

Bases: pydantic.BaseModel

WebSocket authentication message for payload-based auth when header or cookie auth is not feasible.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

model_config#: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

type: Literal[WebSocketMessageType]#

payload: AuthPayload#

timestamp: str = None#

class AuthMessageStatus#

Bases: enum.StrEnum

Outcome of a WebSocket authentication attempt.

Initialize self. See help(type(self)) for accurate signature.

SUCCESS = 'success'#

ERROR = 'error'#

class WebSocketAuthResponseMessage(/, **data: Any)#

Bases: pydantic.BaseModel

Server response to a WebSocket auth_message.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

model_config#: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

type: Literal[WebSocketMessageType]#

status: AuthMessageStatus = None#

user_id: str | None = None#

payload: Error | None = None#

timestamp: str = None#

class SystemIntermediateStepContent(/, **data: Any)#

Bases: pydantic.BaseModel

model_config#: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

name: str#

payload: str#

class WebSocketSystemIntermediateStepMessage(/, **data: Any)#

Bases: pydantic.BaseModel

For more details, refer to the API documentation: docs/source/developer_guide/websockets.md

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

model_config#: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

type: Literal[WebSocketMessageType]#

id: str = 'default'#

thread_id: str | None = 'default'#

parent_id: str = 'default'#

intermediate_parent_id: str | None = 'default'#

update_message_id: str | None = 'default'#

conversation_id: str | None = None#

content: SystemIntermediateStepContent#

status: WebSocketMessageStatus#

timestamp: str = ''#

class SystemResponseContent(/, **data: Any)#

Bases: pydantic.BaseModel

model_config#: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

text: str | None = None#

class WebSocketSystemResponseTokenMessage(/, **data: Any)#

Bases: pydantic.BaseModel

For more details, refer to the API documentation: docs/source/developer_guide/websockets.md

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

model_config#: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

type: Literal[WebSocketMessageType, WebSocketMessageType]#

id: str | None = 'default'#

thread_id: str | None = 'default'#

parent_id: str = 'default'#

conversation_id: str | None = None#

content: SystemResponseContent | Error | GenerateResponse#

status: WebSocketMessageStatus#

timestamp: str = ''#

classmethod validate_content_by_type( value: SystemResponseContent | Error | GenerateResponse, info: pydantic_core.core_schema.ValidationInfo, )#

class WebSocketSystemInteractionMessage(/, **data: Any)#

Bases: pydantic.BaseModel

For more details, refer to the API documentation: docs/source/developer_guide/websockets.md

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

model_config#: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

type: Literal[WebSocketMessageType]#

id: str | None = 'default'#

thread_id: str | None = 'default'#

parent_id: str = 'default'#

conversation_id: str | None = None#

content: nat.data_models.interactive.HumanPrompt#

status: WebSocketMessageStatus#

timestamp: str = ''#

class ObservabilityTraceContent(/, **data: Any)#

Bases: pydantic.BaseModel

model_config#: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

observability_trace_id: str#

class WebSocketObservabilityTraceMessage(/, **data: Any)#

Bases: pydantic.BaseModel

WebSocket message for observability trace ID. Sent once after the workflow completes to correlate the request with observability traces.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

model_config#: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

type: Literal[WebSocketMessageType]#

id: str = 'default'#

parent_id: str = 'default'#

conversation_id: str | None = None#

content: ObservabilityTraceContent#

timestamp: str = ''#

_generate_response_to_str(response: GenerateResponse) → str#

_generate_response_to_chat_response( response: GenerateResponse, ) → ChatResponse#

_nat_chat_request_to_string(data: ChatRequest) → str#

_string_to_nat_chat_request(data: str) → ChatRequest#

_chat_request_or_message_to_chat_request( data: ChatRequestOrMessage, ) → ChatRequest#

_chat_request_to_chat_request_or_message( data: ChatRequest, ) → ChatRequestOrMessage#

_chat_request_or_message_to_string(data: ChatRequestOrMessage) → str#

_string_to_chat_request_or_message(data: str) → ChatRequestOrMessage#

_nat_chat_response_to_string(data: ChatResponse) → str#

_string_to_nat_chat_response(data: str) → ChatResponse#: Converts a string to an ChatResponse object

_chat_response_chunk_to_string(data: ChatResponseChunk) → str#

_string_to_nat_chat_response_chunk(data: str) → ChatResponseChunk#: Converts a string to an ChatResponseChunk object