dataflow.platform.models¶

Platform Models - Pydantic-First Architecture with Intelligent Inheritance

This module provides the complete set of platform models for the unified Haive ecosystem. All models use pure Pydantic architecture with no __init__ methods and intelligent inheritance patterns.

Architecture Overview:¶

Base Platform Layer: - BasePlatform: Foundation for all platform models - PlatformStatus: Common status enumeration

Specialized Platforms: - MCPPlatform: MCP-specific platform (inherits BasePlatform) - PluginPlatform: Base for all plugins (inherits BasePlatform)

Server Hierarchy: - BaseServerInfo: Foundation for all servers - MCPServerInfo: MCP-specific servers (inherits BaseServerInfo) - DownloadedServerInfo: Our 63 downloaded servers (inherits MCPServerInfo)

Configuration Models: - PluginConfig: Plugin configuration - APIConfig: FastAPI configuration - DiscoveryConfig: Service discovery configuration - ConnectionConfig: Server connection configuration

Enumerations: - ServerSource: Where servers come from - ServerStatus: Server operational status - MCPTransport: MCP transport protocols - HealthStatus: Health check status

Key Features:¶

Pure Pydantic Models: - No __init__ methods anywhere - All configuration via Field definitions - Comprehensive validation through field validators - Clean inheritance patterns
Intelligent Inheritance: - BasePlatform provides core platform functionality - Specialized platforms extend with domain-specific features - Server hierarchy supports multiple server types - Consistent patterns across all models
Real-World Integration: - DownloadedServerInfo works with our 63 downloaded servers - Factory methods for creating from CSV and install report data - Connection configurations for actual server types
Comprehensive Validation: - Field validators for IDs, URLs, versions - Cross-field validation where needed - Sensible defaults and examples

Usage Examples:¶

Basic Platform Creation:

>>> from haive.dataflow.platform.models import MCPPlatform
>>> platform = MCPPlatform()  # Uses intelligent defaults
>>> print(platform.platform_name)
"Haive MCP Platform"

Plugin Configuration:

>>> from haive.dataflow.platform.models import PluginConfig
>>> plugin = PluginConfig(
...     name="mcp-browser",
...     entry_point="haive.mcp.plugins:MCPBrowserPlugin"
... )

Server from Download Data:

>>> from haive.dataflow.platform.models import DownloadedServerInfo
>>> server = DownloadedServerInfo.from_csv_and_install_report(
...     csv_data, install_report, "session-123"
... )

Inheritance Validation:

>>> isinstance(platform, BasePlatform)  # True
>>> isinstance(server, BaseServerInfo)  # True
>>> isinstance(server, MCPServerInfo)   # True

Submodules¶

Classes¶

`APIConfig`	FastAPI application configuration.
`BasePlatform`	Foundation platform model for all Haive systems.
`BaseServerInfo`	Foundation server model - all servers inherit from this.
`ConnectionConfig`	Server connection configuration.
`DiscoveryConfig`	Discovery system configuration.
`DownloadedServerInfo`	Specialized for our 63 downloaded servers - intelligent specialization.
`HealthStatus`	Health check status.
`MCPPlatform`	Specialized platform for MCP operations - inherits all base capabilities.
`MCPServerInfo`	MCP-specific server - inherits base + adds MCP capabilities.
`MCPTransport`	MCP transport protocols.
`PerformanceMetrics`	Server performance metrics.
`PlatformStatus`	Platform operational status.
`PluginConfig`	Plugin configuration model.
`PluginPlatform`	Base platform for all plugins - intelligent inheritance from BasePlatform.
`PromptInfo`	Information about a server prompt.
`ResourceInfo`	Information about a server resource.
`ServerManagementConfig`	Server management configuration.
`ServerSource`	Where the server came from.
`ServerStatus`	Server operational status.
`ToolInfo`	Information about a server tool.

Package Contents¶

class dataflow.platform.models.APIConfig(/, **data: Any)¶

Bases: pydantic.BaseModel

FastAPI application configuration.

cors_origins: List[str] = None¶

description: str = None¶

docs_url: str = None¶

host: str = None¶

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

openapi_url: str = None¶

port: int = None¶

redoc_url: str = None¶

title: str = None¶

version: str = None¶

class dataflow.platform.models.BasePlatform(/, **data: Any)¶

Bases: pydantic.BaseModel

Foundation platform model for all Haive systems.

This is the base class that all platform models inherit from, providing: - Core platform identification and metadata - Standard configuration and state management - Lifecycle tracking (created_at, updated_at, status) - Capability flags for different platform features - Comprehensive validation for platform fields

Design Philosophy: - Pure Pydantic model (no __init__ method) - All configuration via Field definitions - Validation through field validators - Extensible through inheritance

Examples

Basic platform creation:

platform = BasePlatform(
    platform_id="my-platform",
    platform_name="My Platform",
    description="A sample platform"
)

With capabilities:

platform = BasePlatform(
    platform_id="advanced-platform",
    platform_name="Advanced Platform",
    description="Platform with enhanced capabilities",
    supports_discovery=True,
    supports_health_monitoring=True,
    supports_authentication=True
)

add_metadata(key: str, value: Any) → None¶

Add metadata entry to platform.

Parameters:

key – Metadata key
value – Metadata value

get_capability_summary() → Dict[str, bool]¶

Get summary of all platform capabilities.

Returns:: Dictionary mapping capability names to their status

update_status(new_status: PlatformStatus, update_timestamp: bool = True) → None¶

Update platform status and optionally timestamp.

Parameters:

new_status – New platform status
update_timestamp – Whether to update the updated_at timestamp

classmethod validate_platform_id(v: str) → str¶

Validate platform ID format.

Platform IDs must be: - Lowercase only - Alphanumeric characters - Hyphens (-) and underscores (_) allowed - No spaces or special characters

Parameters:: v – Platform ID to validate
Returns:: Validated platform ID
Raises:: ValueError – If platform ID format is invalid

classmethod validate_version(v: str) → str¶

Validate semantic version format.

Versions must follow semantic versioning (semver) format: - MAJOR.MINOR.PATCH - Optional pre-release: 1.0.0-beta.1 - Optional build metadata: 1.0.0+20210101

Parameters:: v – Version string to validate
Returns:: Validated version string
Raises:: ValueError – If version format is invalid

config: Dict[str, Any] = None¶

created_at: datetime.datetime = None¶

description: str = None¶

metadata: Dict[str, Any] = None¶

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

platform_id: str = None¶

platform_name: str = None¶

status: PlatformStatus = None¶

supports_authentication: bool = None¶

supports_caching: bool = None¶

supports_discovery: bool = None¶

supports_health_monitoring: bool = None¶

updated_at: datetime.datetime | None = None¶

version: str = None¶

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

Bases: pydantic.BaseModel

Foundation server model - all servers inherit from this.

This is the base class for all server models in the Haive ecosystem. It provides core identification, operational status, and metadata management that all server types need, regardless of their specific protocol or purpose.

Design Philosophy: - Pure Pydantic model (no __init__ method) - Foundation for intelligent inheritance - Common fields that ALL servers need - Extensible through inheritance - Comprehensive validation

Inheritance Strategy: - BaseServerInfo (this class): Core server identity and status - MCPServerInfo: Adds MCP-specific fields and capabilities - DownloadedServerInfo: Specialized for our 63 downloaded servers - [Future]: HAPServerInfo, CustomServerInfo, etc.

Examples

Basic server:

server = BaseServerInfo(
    server_id="my-server",
    server_name="My Test Server",
    description="A basic server for testing"
)

Server with status:

server = BaseServerInfo(
    server_id="prod-server",
    server_name="Production Server",
    description="Production service",
    status=ServerStatus.ACTIVE,
    health_status=HealthStatus.HEALTHY
)

get_server_summary() → Dict[str, Any]¶

Get basic server information summary.

Returns:: Dictionary with core server information

update_health_status(new_health_status: dataflow.platform.models.mcp.HealthStatus) → None¶

Update server health status with timestamp.

Parameters:: new_health_status – New health status

update_status(new_status: dataflow.platform.models.mcp.ServerStatus, update_timestamp: bool = True) → None¶

Update server status and optionally timestamp.

Parameters:

new_status – New server status
update_timestamp – Whether to update last_updated timestamp

classmethod validate_server_id_format(v: str) → str¶

Ensure server ID is valid.

Server IDs must be: - Alphanumeric characters only - Hyphens (-) and underscores (_) allowed - No spaces or special characters - Between 3 and 100 characters

Parameters:: v – Server ID to validate
Returns:: Validated server ID
Raises:: ValueError – If server ID format is invalid

created_at: datetime.datetime = None¶

description: str | None = None¶

health_status: dataflow.platform.models.mcp.HealthStatus = None¶

last_updated: datetime.datetime | None = None¶

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

server_id: str = None¶

server_name: str = None¶

status: dataflow.platform.models.mcp.ServerStatus = None¶

version: str | None = None¶

class dataflow.platform.models.ConnectionConfig(/, **data: Any)¶

Bases: pydantic.BaseModel

Server connection configuration.

This model defines how to connect to and communicate with servers, supporting multiple transport protocols and connection methods.

get_connection_summary() → Dict[str, Any]¶: Get connection configuration summary.

is_command_based() → bool¶: Check if connection is command-based.

is_url_based() → bool¶: Check if connection is URL-based.

classmethod validate_url(v: str | None) → str | None¶: Validate URL format if provided.

args: List[str] = None¶

command: str | None = None¶

env: Dict[str, str] = None¶

max_retries: int = None¶

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

retry_delay: int = None¶

timeout: int = None¶

url: str | None = None¶

class dataflow.platform.models.DiscoveryConfig(/, **data: Any)¶

Bases: pydantic.BaseModel

Discovery system configuration.

auto_discover: bool = None¶

discovery_sources: List[str] = None¶

max_concurrent_scans: int = None¶

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

scan_intervals: Dict[str, int] = None¶

timeout_seconds: int = None¶

class dataflow.platform.models.DownloadedServerInfo(/, **data: Any)¶

Bases: MCPServerInfo

Specialized for our 63 downloaded servers - intelligent specialization.

This model is specialized for the servers we successfully downloaded using our bulk installer. It inherits all MCPServerInfo capabilities and adds specific fields and methods for managing downloaded servers.

Specialized Features: - Always has source=ServerSource.DOWNLOADED (frozen field) - Tracks download metadata (timestamp, local directory, install command) - Links to original CSV data and install reports - Factory methods for creating from real download data - Enhanced for our specific bulk download workflow

Examples

From our download data:

server = DownloadedServerInfo(
    server_id="browser-tools-mcp",
    server_name="AgentDeskAI/browser-tools-mcp",
    description="Browser monitoring and interaction tool",
    transport=MCPTransport.STDIO,
    connection_config=ConnectionConfig(
        command="npx",
        args=["-y", "browser-tools-mcp"]
    ),
    managed_by_plugin="mcp-browser",
    repository_url="https://github.com/AgentDeskAI/browser-tools-mcp",
    stars=5555,
    language="JavaScript",
    install_command_used="npx -y browser-tools-mcp",
    bulk_install_session="bulk-session-20250819"
)

classmethod from_csv_and_install_report(csv_row: Dict[str, Any], install_report_entry: Dict[str, Any], bulk_session_id: str) → DownloadedServerInfo¶

Factory method to create from our actual download data.

This factory method creates a DownloadedServerInfo instance from the real CSV data and install report that we generated during our bulk download session.

Parameters:

csv_row – Row from our mcp_servers_data.csv
install_report_entry – Entry from install report JSON
bulk_session_id – ID of the bulk install session

Returns:

DownloadedServerInfo instance configured from real data

Examples

>>> csv_row = {
...     'name': 'AgentDeskAI/browser-tools-mcp',
...     'description': 'Browser monitoring tool',
...     'repository_url': 'https://github.com/AgentDeskAI/browser-tools-mcp',
...     'stars': 5555.0,
...     'language': 'JavaScript'
... }
>>> install_entry = {
...     'name': 'AgentDeskAI/browser-tools-mcp',
...     'command': 'npx -y browser-tools-mcp',
...     'status': 'success'
... }
>>> server = DownloadedServerInfo.from_csv_and_install_report(
...     csv_row, install_entry, "bulk-session-20250819"
... )

get_download_summary() → Dict[str, Any]¶

Get download-specific information summary.

Returns:: Dictionary with download metadata and status

classmethod validate_directory_exists(v: pathlib.Path | None) → pathlib.Path | None¶: Validate local directory exists if specified.

bulk_install_session: str | None = None¶

csv_data_row: Dict[str, Any] = None¶

detected_tools: List[str] = None¶

download_timestamp: datetime.datetime = None¶

install_command_used: str | None = None¶

local_directory: pathlib.Path | None = None¶

readme_content: str | None = None¶

source: dataflow.platform.models.mcp.ServerSource = None¶

class dataflow.platform.models.HealthStatus¶

Bases: str, enum.Enum

Health check status.

HEALTHY = 'healthy'¶

UNHEALTHY = 'unhealthy'¶

UNKNOWN = 'unknown'¶

WARNING = 'warning'¶

class dataflow.platform.models.MCPPlatform¶

Bases: dataflow.platform.models.base.BasePlatform

Specialized platform for MCP operations - inherits all base capabilities.

This platform extends BasePlatform with MCP-specific functionality: - Plugin management system - Server discovery and management - API configuration - MCP transport protocol support

Inheritance Features: - Inherits all BasePlatform fields and validation - Extends capability flags with MCP-specific ones - Adds MCP-specific configuration models - Maintains platform inheritance patterns

Examples

Basic MCP platform:

platform = MCPPlatform()
# Uses all defaults, inherits platform_id from class default

With custom plugins:

platform = MCPPlatform(
    plugins=[
        PluginConfig(
            name="mcp-browser",
            entry_point="haive.mcp:MCPBrowserPlugin"
        ),
        PluginConfig(
            name="hap-agents",
            entry_point="haive.agp:HAPPlugin"
        )
    ]
)

Full configuration:

platform = MCPPlatform(
    platform_id="production-mcp",
    platform_name="Production MCP Platform",
    description="Production MCP management system",
    api_config=APIConfig(host="mcp.company.com", port=443),
    discovery_config=DiscoveryConfig(auto_discover=True)
)

add_plugin(plugin_config: PluginConfig) → None¶

Add a plugin to the platform.

Parameters:: plugin_config – Plugin configuration to add
Raises:: ValueError – If plugin name already exists

get_enabled_plugins() → List[PluginConfig]¶

Get list of enabled plugins.

Returns:: List of enabled plugin configurations

get_mcp_capability_summary() → Dict[str, bool]¶

Get summary of MCP-specific capabilities.

Returns:: Dictionary mapping MCP capability names to their status

get_plugin(plugin_name: str) → PluginConfig | None¶

Get plugin configuration by name.

Parameters:: plugin_name – Name of plugin to find
Returns:: Plugin configuration if found, None otherwise

remove_plugin(plugin_name: str) → bool¶

Remove a plugin from the platform.

Parameters:: plugin_name – Name of plugin to remove
Returns:: True if plugin was removed, False if not found

classmethod validate_plugins_unique(v: List[PluginConfig]) → List[PluginConfig]¶

Ensure plugin names are unique.

Parameters:: v – List of plugin configurations
Returns:: Validated plugin list
Raises:: ValueError – If duplicate plugin names found

api_config: APIConfig = None¶

description: str = None¶

discovery_config: DiscoveryConfig = None¶

platform_id: str = None¶

platform_name: str = None¶

plugins: List[PluginConfig] = None¶

server_management_config: ServerManagementConfig = None¶

supports_authentication: bool = None¶

supports_bulk_operations: bool = None¶

supports_discovery: bool = None¶

supports_health_monitoring: bool = None¶

supports_server_management: bool = None¶

supports_tool_execution: bool = None¶

class dataflow.platform.models.MCPServerInfo(/, **data: Any)¶

Bases: BaseServerInfo

MCP-specific server - inherits base + adds MCP capabilities.

This model extends BaseServerInfo with MCP (Model Context Protocol) specific functionality while maintaining the inheritance pattern. It adds MCP transport, tools, resources, and plugin management.

Inheritance Features: - Inherits: server_id, server_name, status, timestamps from BaseServerInfo - Extends: MCP transport, connection config, tools, resources - Adds: Plugin management, performance metrics, source tracking

Examples

Basic MCP server:

server = MCPServerInfo(
    server_id="mcp-filesystem",
    server_name="MCP Filesystem Server",
    description="File system access via MCP",
    source=ServerSource.NPM_PACKAGE,
    transport=MCPTransport.STDIO,
    connection_config=ConnectionConfig(
        command="npx",
        args=["-y", "@modelcontextprotocol/server-filesystem"]
    ),
    managed_by_plugin="mcp-browser"
)

Server with tools and resources:

server = MCPServerInfo(
    server_id="mcp-web-tools",
    server_name="Web Tools MCP Server",
    source=ServerSource.DOWNLOADED,
    transport=MCPTransport.STDIO,
    connection_config=connection_config,
    managed_by_plugin="mcp-browser",
    tools=[
        ToolInfo(name="web_search", description="Search the web"),
        ToolInfo(name="web_scrape", description="Scrape web pages")
    ],
    resources=[
        ResourceInfo(uri="web://search", name="search_results")
    ]
)

get_mcp_capabilities_summary() → Dict[str, Any]¶

Get summary of MCP capabilities.

Returns:: Dictionary with MCP-specific capability information

get_resource_uris() → List[str]¶

Get list of resource URIs provided by this server.

Returns:: List of resource URIs

get_tool_names() → List[str]¶

Get list of tool names provided by this server.

Returns:: List of tool names

record_connection_attempt(success: bool) → None¶

Record a connection attempt.

Parameters:: success – Whether the connection was successful

classmethod validate_stars_reasonable(v: int | None) → int | None¶: Validate star count is reasonable.

author: str | None = None¶

connection_attempts: int = None¶

connection_config: ConnectionConfig = None¶

property connection_success_rate: float¶

Calculate connection success rate.

Returns:: Success rate as float between 0.0 and 1.0

documentation_url: str | None = None¶

language: str | None = None¶

last_health_check: datetime.datetime | None = None¶

managed_by_plugin: str = None¶

performance_metrics: PerformanceMetrics | None = None¶

plugin_specific_data: Dict[str, Any] = None¶

prompts: List[PromptInfo] = None¶

repository_url: str | None = None¶

resources: List[ResourceInfo] = None¶

source: dataflow.platform.models.mcp.ServerSource = None¶

stars: int | None = None¶

successful_connections: int = None¶

tools: List[ToolInfo] = None¶

transport: dataflow.platform.models.mcp.MCPTransport = None¶

class dataflow.platform.models.MCPTransport¶

Bases: str, enum.Enum

MCP transport protocols.

HTTP = 'http'¶

SSE = 'sse'¶

STDIO = 'stdio'¶

WEBSOCKET = 'websocket'¶

class dataflow.platform.models.PerformanceMetrics(/, **data: Any)¶

Bases: pydantic.BaseModel

Server performance metrics.

average_response_time: float | None = None¶

failed_requests: int = None¶

last_response_time: float | None = None¶

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

success_rate: float | None = None¶

total_requests: int = None¶

class dataflow.platform.models.PlatformStatus¶

Bases: str, enum.Enum

Platform operational status.

ACTIVE = 'active'¶

ERROR = 'error'¶

INACTIVE = 'inactive'¶

INITIALIZING = 'initializing'¶

MAINTENANCE = 'maintenance'¶

STARTING = 'starting'¶

STOPPING = 'stopping'¶

class dataflow.platform.models.PluginConfig(/, **data: Any)¶

Bases: pydantic.BaseModel

Plugin configuration model.

classmethod validate_entry_point(v: str) → str¶: Validate entry point format.

config: Dict[str, Any] = None¶

enabled: bool = None¶

entry_point: str = None¶

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

name: str = None¶

priority: int = None¶

routes_prefix: str = None¶

class dataflow.platform.models.PluginPlatform¶

Bases: dataflow.platform.models.base.BasePlatform

Base platform for all plugins - intelligent inheritance from BasePlatform.

This platform serves as the foundation for all plugin implementations in the Haive ecosystem. It inherits all BasePlatform capabilities and extends them with plugin-specific functionality.

Inheritance Architecture: - Inherits: platform_id, platform_name, config, metadata, timestamps from BasePlatform - Extends: Plugin-specific entry points, routes, priorities, dependencies - Adds: Plugin capabilities (servers, tools, resources, discovery) - Provides: Async lifecycle methods (initialize, cleanup)

Plugin Capabilities: - provides_servers: Plugin manages MCP servers - provides_tools: Plugin provides tools for agents - provides_resources: Plugin offers resources (files, data, etc.) - provides_discovery: Plugin participates in service discovery - provides_health_checks: Plugin provides health monitoring

Lifecycle Management: - initialize(): Setup plugin resources (async) - cleanup(): Cleanup plugin resources (async) - entry_point: Module and class for plugin loading - priority: Loading order (lower numbers load first)

Examples

Basic plugin platform:

plugin = PluginPlatform(
    platform_id="my-plugin",
    platform_name="My Plugin",
    description="Sample plugin implementation",
    entry_point="mypackage.plugins:MyPlugin",
    routes_prefix="/api/my"
)

Plugin with capabilities:

plugin = PluginPlatform(
    platform_id="mcp-browser-plugin",
    platform_name="MCP Browser Plugin",
    description="Browse downloaded MCP servers",
    entry_point="haive.mcp.plugins:MCPBrowserPlugin",
    routes_prefix="/mcp",
    provides_servers=True,
    provides_discovery=True,
    provides_health_checks=True,
    priority=10  # High priority, loads early
)

Plugin with dependencies:

plugin = PluginPlatform(
    platform_id="advanced-plugin",
    platform_name="Advanced Plugin",
    description="Plugin with dependencies",
    entry_point="mypackage:AdvancedPlugin",
    routes_prefix="/advanced",
    dependencies=["base-plugin", "auth-plugin"],
    priority=200  # Lower priority, loads after dependencies
)

check_dependencies_satisfied(available_plugins: List[str]) → tuple[bool, List[str]]¶

Check if plugin dependencies are satisfied.

Parameters:: available_plugins – List of available plugin names
Returns:: Tuple of (satisfied, missing_dependencies)

Examples

>>> plugin.dependencies = ["base-plugin", "auth-plugin"]
>>> satisfied, missing = plugin.check_dependencies_satisfied(["base-plugin"])
>>> satisfied
False
>>> missing
["auth-plugin"]

async cleanup() → None¶

Cleanup plugin resources.

This method is called during plugin unloading or platform shutdown to clean up resources, close connections, and save state. Subclasses should override this method to perform their specific cleanup.

The base implementation: - Updates status to STOPPED - Sets updated_at timestamp - Adds cleanup metadata

Examples

Override in subclass:

async def cleanup(self) -> None:
    # Plugin-specific cleanup
    await self.close_database_connections()
    await self.save_state()

    # Call parent cleanup
    await super().cleanup()

get_load_priority_info() → Dict[str, Any]¶

Get plugin loading priority information.

Returns:: Dictionary with priority info and loading recommendations

get_plugin_info() → Dict[str, Any]¶

Get comprehensive plugin information.

Returns:: Dictionary containing plugin metadata, capabilities, and status

Examples

>>> plugin.get_plugin_info()
{
    "basic_info": {
        "platform_id": "mcp-browser-plugin",
        "platform_name": "MCP Browser Plugin",
        "version": "1.0.0",
        "status": "active"
    },
    "plugin_config": {
        "entry_point": "haive.mcp.plugins:MCPBrowserPlugin",
        "routes_prefix": "/mcp",
        "priority": 10
    },
    "capabilities": {
        "provides_servers": True,
        "provides_discovery": True,
        "total_capabilities": 2
    },
    "lifecycle": {
        "created_at": "2025-08-19T13:45:00Z",
        "updated_at": "2025-08-19T13:46:30Z"
    }
}

async initialize() → None¶

Initialize plugin resources.

This method is called during plugin loading to set up any resources, connections, or state that the plugin needs. Subclasses should override this method to perform their specific initialization.

The base implementation: - Updates status to ACTIVE - Sets updated_at timestamp - Adds initialization metadata

Raises:: Exception – Any initialization errors should be propagated up

Examples

Override in subclass:

async def initialize(self) -> None:
    # Plugin-specific setup
    await self.setup_database_connections()
    await self.load_configuration()

    # Call parent initialization
    await super().initialize()

    # Post-initialization tasks
    self.add_metadata("servers_loaded", len(self.get_servers()))

classmethod normalize_routes_prefix(v: str) → str¶

Ensure routes prefix starts with / and is properly formatted.

Parameters:: v – Routes prefix to normalize
Returns:: Normalized routes prefix (always starts with /)

Examples

“mcp” -> “/mcp”
“/api/v1” -> “/api/v1”
“tools/” -> “/tools”

classmethod validate_entry_point_format(v: str) → str¶

Validate entry point format.

Entry points must be in the format ‘module:class’ or ‘package.module:class’. This follows Python’s standard entry point specification.

Parameters:: v – Entry point string to validate
Returns:: Validated entry point string
Raises:: ValueError – If entry point format is invalid

Examples

Valid formats: - “mymodule:MyClass” - “package.subpackage.module:MyClass” - “haive.mcp.plugins:MCPBrowserPlugin”

dependencies: List[str] = None¶

entry_point: str = None¶

priority: int = None¶

provides_discovery: bool = None¶

provides_health_checks: bool = None¶

provides_resources: bool = None¶

provides_servers: bool = None¶

provides_tools: bool = None¶

routes_prefix: str = None¶

class dataflow.platform.models.PromptInfo(/, **data: Any)¶

Bases: pydantic.BaseModel

Information about a server prompt.

arguments: List[Dict[str, Any]] = None¶

description: str | None = None¶

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

name: str = None¶

class dataflow.platform.models.ResourceInfo(/, **data: Any)¶

Bases: pydantic.BaseModel

Information about a server resource.

description: str | None = None¶

mime_type: str | None = None¶

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

name: str | None = None¶

uri: str = None¶

class dataflow.platform.models.ServerManagementConfig(/, **data: Any)¶

Bases: pydantic.BaseModel

Server management configuration.

auto_start: bool = None¶

health_check_interval: int = None¶

max_restart_attempts: int = None¶

max_servers: int = None¶

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

restart_on_failure: bool = None¶

startup_timeout: int = None¶

class dataflow.platform.models.ServerSource¶

Bases: str, enum.Enum

Where the server came from.

DISCOVERED = 'discovered'¶

DOWNLOADED = 'downloaded'¶

HAP_AGENT = 'hap_agent'¶

LOCAL_CONFIG = 'config'¶

NPM_PACKAGE = 'npm'¶

PIP_PACKAGE = 'pip'¶

REGISTRY = 'registry'¶

class dataflow.platform.models.ServerStatus¶

Bases: str, enum.Enum

Server operational status.

ACTIVE = 'active'¶

ERROR = 'error'¶

INACTIVE = 'inactive'¶

STARTING = 'starting'¶

STOPPING = 'stopping'¶

UNKNOWN = 'unknown'¶

class dataflow.platform.models.ToolInfo(/, **data: Any)¶

Bases: pydantic.BaseModel

Information about a server tool.

description: str | None = None¶

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

name: str = None¶

schema: Dict[str, Any] | None = None¶