nat.front_ends.fastapi.fastapi_front_end_plugin_worker#

Classes#

`FastApiFrontEndPluginWorkerBase`	Helper class that provides a standard way to create an ABC using
`RouteInfo`
`FastApiFrontEndPluginWorker`	Helper class that provides a standard way to create an ABC using

Module Contents#

class FastApiFrontEndPluginWorkerBase(config: nat.data_models.config.Config)#

Bases: abc.ABC

Helper class that provides a standard way to create an ABC using inheritance.

_config#

_front_end_config#

_dask_available = False#

_job_store = None#

_http_flow_handler: nat.front_ends.fastapi.auth_flow_handlers.http_flow_handler.HTTPAuthenticationFlowHandler | None#

_scheduler_address#

_db_url#

_config_file_path = b'.'#

_use_dask_threads#

_log_level#

property config: nat.data_models.config.Config#

property front_end_config: nat.front_ends.fastapi.fastapi_front_end_config.FastApiFrontEndConfig#

build_app() → fastapi.FastAPI#

set_cors_config(nat_app: fastapi.FastAPI) → None#: Set the cross origin resource sharing configuration.

async _suppress_authentication_logs( request: fastapi.Request, call_next: collections.abc.Callable[[fastapi.Request], collections.abc.Awaitable[fastapi.Response]], ) → fastapi.Response#: Intercepts authentication request and supreses logs that contain sensitive data.

abstractmethod configure( app: fastapi.FastAPI, builder: nat.builder.workflow_builder.WorkflowBuilder, )#

Async:

abstractmethod get_step_adaptor() → nat.front_ends.fastapi.step_adaptor.StepAdaptor#

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

Bases: pydantic.BaseModel

function_name: str | None#

class FastApiFrontEndPluginWorker(config: nat.data_models.config.Config)#

Bases: FastApiFrontEndPluginWorkerBase

Helper class that provides a standard way to create an ABC using inheritance.

_outstanding_flows: dict[str, nat.front_ends.fastapi.auth_flow_handlers.websocket_flow_handler.FlowState]#

_outstanding_flows_lock#

_session_managers: list[nat.runtime.session.SessionManager] = []#

_evaluators: dict[str, nat.builder.evaluator.EvaluatorInfo]#

_eval_builder: nat.builder.eval_builder.WorkflowEvalBuilder | None = None#

async initialize_evaluators(config: nat.data_models.config.Config)#: Initialize and store evaluators from config for single-item evaluation.

async _create_session_manager( builder: nat.builder.workflow_builder.WorkflowBuilder, entry_function: str | None = None, ) → nat.runtime.session.SessionManager#: Create and register a SessionManager.

async cleanup_session_managers()#: Clean up all SessionManager resources on shutdown.

async cleanup_evaluators()#: Clean up evaluator resources on shutdown.

get_step_adaptor() → nat.front_ends.fastapi.step_adaptor.StepAdaptor#

async configure( app: fastapi.FastAPI, builder: nat.builder.workflow_builder.WorkflowBuilder, )#

async add_routes( app: fastapi.FastAPI, builder: nat.builder.workflow_builder.WorkflowBuilder, )#

async add_default_route( app: fastapi.FastAPI, session_manager: nat.runtime.session.SessionManager, )#

async add_evaluate_route( app: fastapi.FastAPI, session_manager: nat.runtime.session.SessionManager, )#: Add the evaluate endpoint to the FastAPI app.

async add_evaluate_item_route( app: fastapi.FastAPI, session_manager: nat.runtime.session.SessionManager, )#: Add the single-item evaluation endpoint to the FastAPI app.

async add_static_files_route( app: fastapi.FastAPI, builder: nat.builder.workflow_builder.WorkflowBuilder, )#

async add_route( app: fastapi.FastAPI, endpoint: nat.front_ends.fastapi.fastapi_front_end_config.FastApiFrontEndConfig.EndpointBase, session_manager: nat.runtime.session.SessionManager, )#

async add_authorization_route(app: fastapi.FastAPI)#

async add_mcp_client_tool_list_route( app: fastapi.FastAPI, builder: nat.builder.workflow_builder.WorkflowBuilder, )#: Add the MCP client tool list endpoint to the FastAPI app.

async add_monitor_route(app: fastapi.FastAPI)#

Add the per-user monitoring endpoint to the FastAPI app.

Security Warning:: This endpoint exposes per-user identifiers and usage metrics. It should be protected by deploying behind an internal network, a reverse proxy with authentication, or similar access controls to prevent exposure to untrusted callers.

async add_health_route(app: fastapi.FastAPI) → None#: Add a health check endpoint to the FastAPI app.

async _add_flow( state: str, flow_state: nat.front_ends.fastapi.auth_flow_handlers.websocket_flow_handler.FlowState, )#

async _remove_flow(state: str)#