nat.data_models.api_server

Attributes

FINISH_REASONS
UserContent
AIQChatRequest
AIQChoiceMessage
AIQChoiceDelta
AIQChoice
AIQUsage
AIQResponseSerializable
AIQResponseBaseModelOutput
AIQResponseBaseModelIntermediate
AIQChatResponse
AIQChatResponseChunk
AIQResponseIntermediateStep
AIQResponsePayloadOutput
AIQGenerateResponse

Classes

Request	Request is a data model that represents HTTP request attributes.
ChatContentType	ChatContentType is an Enum that represents the type of Chat content.
InputAudio
AudioContent
ImageUrl
ImageContent
TextContent
Security
Message
ChatRequest	ChatRequest is a data model that represents a request to the NAT chat API.
ChoiceMessage
ChoiceDelta	Delta object for streaming responses (OpenAI-compatible)
Choice
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.
ResponsePayloadOutput
GenerateResponse
UserMessageContentRoleType	str(object='') -> str
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	str(object='') -> str
Error
WebSocketUserMessage	For more details, refer to the API documentation:
WebSocketUserInteractionResponseMessage	For more details, refer to the API documentation:
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:

Functions

_generate_response_to_str(→ str)
_generate_response_to_chat_response(→ ChatResponse)
_nat_chat_request_to_string(→ str)
_string_to_nat_chat_request(→ ChatRequest)
_nat_chat_response_to_string(→ str)
_string_to_nat_chat_response(→ ChatResponse)	Converts a string to an ChatResponse object
_chat_response_to_chat_response_chunk(→ ChatResponseChunk)
_chat_response_chunk_to_string(→ str)
_string_to_nat_chat_response_chunk(→ ChatResponseChunk)	Converts a string to an ChatResponseChunk object
_ai_message_chunk_to_nat_chat_response_chunk(...)	Converts LangChain AINodeMessageChunk to ChatResponseChunk

Module Contents

FINISH_REASONS

class Request(/, **data: Any)

Bases: pydantic.BaseModel

Request is a data model that represents HTTP request 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

class ChatContentType

Bases: str, enum.Enum

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: pydantic.HttpUrl

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'

class Security(/, **data: Any)

Bases: pydantic.BaseModel

model_config

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

api_key: str = 'default'

token: str = 'default'

UserContent

class Message(/, **data: Any)

Bases: pydantic.BaseModel

content: str | list[UserContent]

role: str

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 ChoiceMessage(/, **data: Any)

Bases: pydantic.BaseModel

content: str | None = None

role: str | 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: str | None = None

class Choice(/, **data: Any)

Bases: pydantic.BaseModel

model_config

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

message: ChoiceMessage | None = None

delta: ChoiceDelta | None = None

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

index: int

class Usage(/, **data: Any)

Bases: pydantic.BaseModel

prompt_tokens: int

completion_tokens: int

total_tokens: int

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 = ''

created: datetime.datetime

choices: list[Choice]

usage: Usage | None = None

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 | None = None,

) → 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[Choice]

created: datetime.datetime

model: str = ''

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,

) → ChatResponseChunk

static create_streaming_chunk(

content: str,

*,

id_: str | None = None,

created: datetime.datetime | None = None,

model: str | None = None,

role: str | 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 ResponsePayloadOutput(/, **data: Any)

Bases: pydantic.BaseModel, ResponseSerializable

payload: Any

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 UserMessageContentRoleType

Bases: str, enum.Enum

str(object=’’) -> str str(bytes_or_buffer[, encoding[, errors]]) -> str

Create a new string object from the given object. If encoding or errors is specified, then the object must expose a data buffer that will be decoded using the given encoding and error handler. Otherwise, returns the result of object.__str__() (if defined) or repr(object). encoding defaults to sys.getdefaultencoding(). errors defaults to ‘strict’.

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

USER = 'user'

ASSISTANT = 'assistant'

class WebSocketMessageType

Bases: str, enum.Enum

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'

ERROR_MESSAGE = 'error_message'

class WorkflowSchemaType

Bases: str, enum.Enum

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: str, enum.Enum

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: str, enum.Enum

str(object=’’) -> str str(bytes_or_buffer[, encoding[, errors]]) -> str

Create a new string object from the given object. If encoding or errors is specified, then the object must expose a data buffer that will be decoded using the given encoding and error handler. Otherwise, returns the result of object.__str__() (if defined) or repr(object). encoding defaults to sys.getdefaultencoding(). errors defaults to ‘strict’.

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

UNKNOWN_ERROR = 'unknown_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'

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

security: Security

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'

content: UserMessageContent

user: User

security: Security

error: Error

schema_version: str = '1.0.0'

timestamp: str = ''

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 = ''

_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

_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_to_chat_response_chunk(

data: ChatResponse,

) → ChatResponseChunk

_chat_response_chunk_to_string(data: ChatResponseChunk) → str

_string_to_nat_chat_response_chunk(data: str) → ChatResponseChunk

Converts a string to an ChatResponseChunk object

_ai_message_chunk_to_nat_chat_response_chunk(

data,

) → ChatResponseChunk

Converts LangChain AINodeMessageChunk to ChatResponseChunk

AIQChatRequest

AIQChoiceMessage

AIQChoiceDelta

AIQChoice

AIQUsage

AIQResponseSerializable

AIQResponseBaseModelOutput

AIQResponseBaseModelIntermediate

AIQChatResponse

AIQChatResponseChunk

AIQResponseIntermediateStep

AIQResponsePayloadOutput

AIQGenerateResponse