nat.data_models.interactive_http#

Data models for HTTP Human-in-the-Loop (HITL) and OAuth support.

These types power the execution + polling model that enables interactive workflows over plain HTTP (no WebSocket required).

Attributes#

`ExecutionStatusResponse`
`ExecutionAcceptedResponse`

Classes#

`ExecutionStatus`	Status of an HTTP interactive execution.
`_ExecutionStatusBase`	Common fields for every execution status variant.
`ExecutionRunningStatus`	Execution is in progress, no interaction or result yet.
`ExecutionInteractionRequiredStatus`	Execution is paused waiting for a human interaction response.
`ExecutionOAuthRequiredStatus`	Execution is paused waiting for an OAuth consent flow to complete.
`ExecutionCompletedStatus`	Execution finished successfully.
`ExecutionFailedStatus`	Execution finished with an error.
`_ExecutionAcceptedBase`	Common fields for every 202 Accepted variant.
`ExecutionAcceptedInteraction`	202 response when the execution requires human interaction.
`ExecutionAcceptedOAuth`	202 response when the execution requires OAuth consent.
`InteractionResponseRequest`	Body for `POST /executions/{execution_id}/interactions/{interaction_id}/response`.
`StreamInteractionEvent`	SSE event emitted in a streaming response when the workflow requires
`StreamOAuthEvent`	SSE event emitted in a streaming response when the workflow requires

Module Contents#

class ExecutionStatus#

Bases: enum.StrEnum

Status of an HTTP interactive execution.

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

RUNNING = 'running'#

INTERACTION_REQUIRED = 'interaction_required'#

OAUTH_REQUIRED = 'oauth_required'#

COMPLETED = 'completed'#

FAILED = 'failed'#

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

Bases: pydantic.BaseModel

Common fields for every execution status variant.

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.

execution_id: str = None#

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

Bases: _ExecutionStatusBase

Execution is in progress, no interaction or result yet.

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.

status: Literal[ExecutionStatus]#

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

Bases: _ExecutionStatusBase

Execution is paused waiting for a human interaction response.

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.

status: Literal[ExecutionStatus]#

interaction_id: str = None#

prompt: nat.data_models.interactive.HumanPrompt = None#

response_url: str = None#

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

Bases: _ExecutionStatusBase

Execution is paused waiting for an OAuth consent flow to complete.

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.

status: Literal[ExecutionStatus]#

auth_url: str = None#

oauth_state: str = None#

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

Bases: _ExecutionStatusBase

Execution finished successfully.

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.

status: Literal[ExecutionStatus]#

result: Any = None#

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

Bases: _ExecutionStatusBase

Execution finished with an error.

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.

status: Literal[ExecutionStatus]#

error: str = None#

ExecutionStatusResponse#

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

Bases: _ExecutionStatusBase

Common fields for every 202 Accepted variant.

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.

status_url: str = None#

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

Bases: _ExecutionAcceptedBase

202 response when the execution requires human interaction.

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.

status: Literal[ExecutionStatus]#

interaction_id: str = None#

prompt: nat.data_models.interactive.HumanPrompt = None#

response_url: str = None#

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

Bases: _ExecutionAcceptedBase

202 response when the execution requires OAuth consent.

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.

status: Literal[ExecutionStatus]#

auth_url: str = None#

oauth_state: str = None#

ExecutionAcceptedResponse#

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

Bases: pydantic.BaseModel

Body for POST /executions/{execution_id}/interactions/{interaction_id}/response.

Uses the existing HumanResponse discriminated union so that all interaction types (text, binary, radio, checkbox, dropdown, notification) are supported without new types.

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.

response: nat.data_models.interactive.HumanResponse = None#

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

Bases: pydantic.BaseModel, nat.data_models.api_server.ResponseSerializable

SSE event emitted in a streaming response when the workflow requires human interaction (HITL).

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.

event_type: Literal['interaction_required'] = 'interaction_required'#

execution_id: str = None#

interaction_id: str = None#

prompt: nat.data_models.interactive.HumanPrompt = None#

response_url: str = None#

get_stream_data() → str#

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

Bases: pydantic.BaseModel, nat.data_models.api_server.ResponseSerializable

SSE event emitted in a streaming response when the workflow requires OAuth authentication.

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.

event_type: Literal['oauth_required'] = 'oauth_required'#

execution_id: str = None#

auth_url: str = None#

oauth_state: str = None#

get_stream_data() → str#