nat.data_models.api_server#

Attributes#

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

Classes#

`UserMessageContentRoleType`	Enum representing chat message roles in API requests and responses.
`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.
`ChatRequestOrMessage`	`ChatRequestOrMessage` is a data model that represents either a conversation or a string input.
`ChoiceMessage`
`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.
`ResponsePayloadOutput`
`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`	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)
`_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: str, enum.Enum

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 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: 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 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#

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, ) → 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 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 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 ‘utf-8’. 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'#

parent_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 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#

_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

AIQChatRequest#

AIQChoiceMessage#

AIQChoiceDelta#

AIQChoice#

AIQUsage#

AIQResponseSerializable#

AIQResponseBaseModelOutput#

AIQResponseBaseModelIntermediate#

AIQChatResponse#

AIQChatResponseChunk#

AIQResponseIntermediateStep#

AIQResponsePayloadOutput#

AIQGenerateResponse#