dataflow.platform.models.plugins

Plugin Platform Models - Base Platform for All Plugins

This module provides the plugin platform model that serves as the foundation for all plugin implementations across the Haive ecosystem, using intelligent inheritance patterns.

Key Features: - Inherits from BasePlatform for core platform capabilities - Plugin-specific extensions and lifecycle management - Provides servers, tools, resources, and discovery capabilities - Async initialization and cleanup methods - Comprehensive validation for plugin configuration

Classes

PluginPlatform

Base platform for all plugins - intelligent inheritance from BasePlatform.

Module Contents

class dataflow.platform.models.plugins.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)

Check if plugin dependencies are satisfied.

Parameters:

available_plugins (List[str]) – List of available plugin names

Returns:

Tuple of (satisfied, missing_dependencies)

Return type:

tuple[bool, List[str]]

Examples

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

async cleanup()

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()

Return type:

None

get_load_priority_info()

Get plugin loading priority information.

Returns:

Dictionary with priority info and loading recommendations

Return type:

Dict[str, Any]

get_plugin_info()

Get comprehensive plugin information.

Returns:

Dictionary containing plugin metadata, capabilities, and status

Return type:

Dict[str, Any]

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()

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

Return type:

None

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)

Ensure routes prefix starts with / and is properly formatted.

Parameters:

v (str) – Routes prefix to normalize

Returns:

Normalized routes prefix (always starts with /)

Return type:

str

Examples

• “mcp” -> “/mcp”

• “/api/v1” -> “/api/v1”

• “tools/” -> “/tools”

classmethod validate_entry_point_format(v)

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 (str) – Entry point string to validate

Returns:

Validated entry point string

Raises:

ValueError – If entry point format is invalid

Return type:

str

Examples

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