nemo_evaluator.adapters.server

Main server which serves on a local port and holds chain of interceptors

Module Contents

Classes

AdapterServer	Adapter server with registry-based interceptor support
AdapterServerProcess

Functions

is_port_open	Check if the given port is open on the host.
spawn_adapter_server	Spawn an AdapterServer in a separate process.
wait_for_server	Wait for server to be ready with timeout.

Data

logger

API

class nemo_evaluator.adapters.server.AdapterServer(

api_url: str,

output_dir: str,

adapter_config: nemo_evaluator.adapters.adapter_config.AdapterConfig,

)

Adapter server with registry-based interceptor support

Initialization

Initialize the adapter server.

Args: api_url: The upstream API URL to forward requests to output_dir: Directory for output files adapter_config: Adapter configuration including interceptors and discovery

DEFAULT_ADAPTER_HOST: str

‘localhost’

DEFAULT_ADAPTER_PORT: int

3825

generate_report() → None

Generate HTML report of cached requests and responses.

run() → None

Start the Flask server.

run_post_eval_hooks() → None

Run all configured post-evaluation hooks.

class nemo_evaluator.adapters.server.AdapterServerProcess(

evaluation: nemo_evaluator.api.api_dataclasses.Evaluation,

)

Initialization

nemo_evaluator.adapters.server.is_port_open(host: str, port: int, timeout: float = 0.5) → bool

Check if the given port is open on the host.

Args: host: The host to check port: The port to check timeout: Socket timeout in seconds

Returns: bool: True if port is open, False otherwise

nemo_evaluator.adapters.server.logger

‘get_logger(…)’

nemo_evaluator.adapters.server.spawn_adapter_server(

api_url: str,

output_dir: str,

adapter_config: nemo_evaluator.adapters.adapter_config.AdapterConfig,

) → multiprocessing.context.SpawnProcess | None

Spawn an AdapterServer in a separate process.

Args: api_url: The upstream API URL to forward requests to output_dir: Directory for output files adapter_config: Adapter configuration including interceptors and discovery

Returns: The multiprocessing.Process object running the adapter server, or None if no enabled interceptors or post-eval hooks are configured

nemo_evaluator.adapters.server.wait_for_server(

host: str,

port: int,

max_wait: float = 300,

interval: float = 0.2,

) → bool

Wait for server to be ready with timeout.

Args: host: The host to check port: The port to check max_wait: Maximum time to wait in seconds (default: 10) interval: Time between checks in seconds (default: 0.2)

Returns: bool: True if server is ready, False if timeout exceeded