dataflow.server_management¶

Server management framework for Haive.

This module provides a generic, type-safe server management framework using Pydantic models and validation. It serves as the foundation for managing various types of servers (MCP, HAP, Docker, etc.) across the Haive ecosystem.

Key Features: - Generic base classes with full type safety - Pydantic validation for all configurations - Extensible architecture for different server types - Smart defaults and auto-configuration - Comprehensive error handling with clear messages

Example

Basic usage with MCP servers:

from haive.dataflow.server_management import BaseServerManager
from haive.mcp.servers import MCPServerConfig, MCPServerInfo

class MCPServerManager(BaseServerManager[MCPServerConfig, MCPServerInfo]):
    config_class = Field(default=MCPServerConfig, exclude=True)
    info_class = Field(default=MCPServerInfo, exclude=True)

    async def start_server(self, name: str, config: Optional[MCPServerConfig] = None) -> MCPServerInfo:
        # Implementation
        pass

Submodules¶

Classes¶

`BaseServerConfig`	Base configuration all servers must have.
`BaseServerInfo`	Base runtime info all servers have.
`BaseServerManager`	Generic base class for all server managers.
`ConfigLoaderProtocol`	Protocol for configuration loading capabilities.
`HealthMonitorProtocol`	Protocol for health monitoring capabilities.
`MetricsProtocol`	Protocol for metrics collection capabilities.
`ServerDiscoveryProtocol`	Protocol for server discovery capabilities.
`ServerManagerProtocol`	Protocol defining the server manager interface.
`ServerStatus`	Universal server status enum.

Package Contents¶

class dataflow.server_management.BaseServerConfig(/, **data: Any)¶

Bases: pydantic.BaseModel

Base configuration all servers must have.

This model defines the minimum configuration required for any server type. Subclasses should extend this with type-specific fields.

name¶: Unique server identifier

command¶: Command to execute (executable + args)

description¶: Human-readable description

working_directory¶: Optional working directory for process

environment¶: Additional environment variables

timeout_seconds¶: Startup timeout in seconds

auto_restart¶: Whether to restart on failure

health_check_enabled¶: Whether to perform health checks

Example

Basic server configuration:

config = BaseServerConfig(
    name="my-server",
    command=["python", "-m", "server"],
    description="My custom server"
)

classmethod validate_command_not_empty(v: List[str]) → List[str]¶: Validate command has executable and no empty strings.

classmethod validate_environment_vars(v: Dict[str, str]) → Dict[str, str]¶: Validate environment variable names and values.

classmethod validate_name_format(v: str) → str¶: Validate name follows naming convention.

classmethod validate_working_directory(v: str | None) → str | None¶: Validate working directory if provided.

auto_restart: bool = None¶

command: List[str] = None¶

description: str = None¶

environment: Dict[str, str] = None¶

health_check_enabled: bool = None¶

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

name: str = None¶

timeout_seconds: int = None¶

working_directory: str | None = None¶

class dataflow.server_management.BaseServerInfo(/, **data: Any)¶

Bases: pydantic.BaseModel

Base runtime info all servers have.

This model represents the runtime state of a server process. It excludes the actual process handle from serialization.

name¶: Server name

pid¶: Process ID

status¶: Current server status

started_at¶: When the server was started

config_snapshot¶: Configuration used to start server

error_message¶: Last error message if any

restart_count¶: Number of times server has been restarted

last_health_check¶: Last successful health check time

uptime_seconds¶: Computed uptime in seconds

record_health_check(success: bool) → None¶: Record health check result.

update_status(new_status: ServerStatus, error: str | None = None) → None¶: Update server status with optional error message.

config_snapshot: Dict[str, Any] = None¶

error_message: str | None = None¶

property is_running: bool¶: Check if server process is currently running.

last_health_check: datetime.datetime | None = None¶

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

name: str = None¶

pid: int = None¶

process_handle: Any | None = None¶

restart_count: int = None¶

started_at: datetime.datetime = None¶

status: ServerStatus = None¶

property uptime_seconds: float¶: Calculate server uptime in seconds.

class dataflow.server_management.BaseServerManager(/, **data: Any)¶

Bases: pydantic.BaseModel, abc.ABC, Generic[ConfigT, InfoT]

Generic base class for all server managers.

This class provides a type-safe foundation for managing servers of any type. Subclasses should specify the concrete ConfigT and InfoT types.

Type Parameters:: ConfigT: Server configuration type (must extend BaseServerConfig) InfoT: Server runtime info type (must extend BaseServerInfo)

servers¶: Currently running servers mapped by name

available_configs¶: Available server configurations

config_class¶: Configuration class for type validation

info_class¶: Info class for runtime information

auto_restart¶: Whether to automatically restart failed servers

max_restart_attempts¶: Maximum restart attempts before giving up

health_check_interval¶: Seconds between health checks

restart_tracking¶: Tracks restart attempts per server

Example

Creating a concrete server manager:

class MyServerManager(BaseServerManager[MyConfig, MyInfo]):
    config_class = Field(default=MyConfig, exclude=True)
    info_class = Field(default=MyInfo, exclude=True)

    async def start_server(self, name: str, config: Optional[MyConfig] = None) -> MyInfo:
        # Implementation
        pass

add_config(name: str, config: ConfigT | Dict[str, Any]) → ConfigT¶

Add or update a server configuration.

Parameters:

name – Server name
config – Configuration object or dict

Returns:

Validated configuration object

Raises:

ValidationError – If configuration is invalid

async cleanup() → None¶: Clean up resources and stop all servers.

get_config(name: str) → ConfigT | None¶

Get server configuration by name.

Parameters:: name – Server name
Returns:: Configuration object or None if not found

get_server_info(name: str) → InfoT | None¶

Get runtime information for a server.

Parameters:: name – Server name
Returns:: Server info or None if not running

get_stats() → Dict[str, Any]¶

Get server manager statistics.

Returns:

total_configured: Number of configured servers
total_running: Number of running servers
servers_by_status: Count by status
restart_counts: Restart attempts per server

Return type:

Dictionary with stats including

abstractmethod health_check(name: str) → bool¶

Async:

Check if server is healthy.

Parameters:: name – Server name
Returns:: True if healthy, False otherwise

is_running(name: str) → bool¶

Check if server is running.

Parameters:: name – Server name
Returns:: True if server is running

list_servers(status: dataflow.server_management.models.ServerStatus | None = None) → List[str]¶

List servers by status.

Parameters:: status – Optional status filter
Returns:: List of server names

model_post_init(__context: Any) → None¶: Additional initialization after validation.

remove_config(name: str) → bool¶

Remove a server configuration.

Parameters:: name – Server name
Returns:: True if removed, False if not found
Raises:: RuntimeError – If server is currently running

abstractmethod restart_server(name: str) → InfoT¶

Async:

Restart a server.

Parameters:: name – Server name
Returns:: New server runtime information

async start_health_monitoring(name: str) → None¶

Start health monitoring for a server.

Parameters:: name – Server name

abstractmethod start_server(name: str, config: ConfigT | None = None) → InfoT¶

Async:

Start a server with given configuration.

Parameters:

name – Server name
config – Optional configuration override

Returns:

Server runtime information

Raises:

ValueError – If no configuration found
RuntimeError – If server already running

async stop_health_monitoring(name: str) → None¶

Stop health monitoring for a server.

Parameters:: name – Server name

abstractmethod stop_server(name: str, force: bool = False) → bool¶

Async:

Stop a running server.

Parameters:

name – Server name
force – Force stop if true

Returns:

True if stopped successfully

classmethod validate_config_types(v: Dict[str, Any], info: pydantic.ValidationInfo) → Dict[str, Any]¶: Ensure all configs are the correct type.

classmethod validate_info_types(v: Dict[str, Any], info: pydantic.ValidationInfo) → Dict[str, Any]¶: Ensure all server info objects are the correct type.

validate_restart_policy() → BaseServerManager¶: Validate restart configuration consistency.

validate_server_consistency() → BaseServerManager¶: Ensure running servers have configurations.

auto_restart: bool = None¶

available_configs: Dict[str, ConfigT] = None¶

config_class: Type[ConfigT] = None¶

health_check_interval: int = None¶

health_check_tasks: Dict[str, asyncio.Task] = None¶

info_class: Type[InfoT] = None¶

max_restart_attempts: int = None¶

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

restart_tracking: Dict[str, int] = None¶

servers: Dict[str, InfoT] = None¶

class dataflow.server_management.ConfigLoaderProtocol¶

Bases: Protocol

Protocol for configuration loading capabilities.

async load_configs_from_file(path: str) → Dict[str, dataflow.server_management.models.BaseServerConfig]¶: Load server configurations from a file.

async save_configs_to_file(path: str) → None¶: Save current configurations to a file.

class dataflow.server_management.HealthMonitorProtocol¶

Bases: Protocol

Protocol for health monitoring capabilities.

async start_health_monitoring(name: str) → None¶: Start health monitoring for a server.

async stop_health_monitoring(name: str) → None¶: Stop health monitoring for a server.

class dataflow.server_management.MetricsProtocol¶

Bases: Protocol

Protocol for metrics collection capabilities.

async export_metrics(exporter: str) → None¶: Export metrics to external system.

get_aggregate_metrics() → Dict[str, float]¶: Get aggregated metrics for all servers.

get_server_metrics(name: str) → Dict[str, float]¶: Get metrics for a specific server.

class dataflow.server_management.ServerDiscoveryProtocol¶

Bases: Protocol

Protocol for server discovery capabilities.

async adopt_server(server_info: dataflow.server_management.models.BaseServerInfo) → bool¶: Adopt an externally started server.

async discover_servers() → List[dataflow.server_management.models.BaseServerInfo]¶: Discover running servers not managed by this instance.

class dataflow.server_management.ServerManagerProtocol¶

Bases: Protocol

Protocol defining the server manager interface.

This protocol ensures all server managers implement the required methods for managing server lifecycles, regardless of the specific server type.

add_config(name: str, config: dataflow.server_management.models.BaseServerConfig | Dict[str, Any]) → dataflow.server_management.models.BaseServerConfig¶: Add or update a server configuration.

async cleanup() → None¶: Clean up resources and stop all servers.

get_config(name: str) → dataflow.server_management.models.BaseServerConfig | None¶: Get server configuration by name.

get_server_info(name: str) → dataflow.server_management.models.BaseServerInfo | None¶: Get runtime information for a server.

get_stats() → Dict[str, Any]¶: Get server manager statistics.

abstractmethod health_check(name: str) → bool¶

Async:

Check if server is healthy.

is_running(name: str) → bool¶: Check if server is running.

list_servers(status: dataflow.server_management.models.ServerStatus | None = None) → List[str]¶: List servers by status.

remove_config(name: str) → bool¶: Remove a server configuration.

abstractmethod restart_server(name: str) → dataflow.server_management.models.BaseServerInfo¶

Async:

Restart a server.

abstractmethod start_server(name: str, config: dataflow.server_management.models.BaseServerConfig | None = None) → dataflow.server_management.models.BaseServerInfo¶

Async:

Start a server with given configuration.

abstractmethod stop_server(name: str, force: bool = False) → bool¶

Async:

Stop a running server.

available_configs: Dict[str, dataflow.server_management.models.BaseServerConfig]¶

servers: Dict[str, dataflow.server_management.models.BaseServerInfo]¶

class dataflow.server_management.ServerStatus¶

Bases: str, enum.Enum

Universal server status enum.

ERROR = 'error'¶

HEALTH_CHECK_FAILED = 'health_check_failed'¶

RESTARTING = 'restarting'¶

RUNNING = 'running'¶

STARTING = 'starting'¶

STOPPED = 'stopped'¶

STOPPING = 'stopping'¶