dataflow.platform.models.mcp¶

MCP Platform Models - Specialized Platform for MCP Operations

This module provides specialized platform models for MCP (Model Context Protocol) operations, inheriting from BasePlatform and extending with MCP-specific capabilities.

Key Features: - Inherits all BasePlatform capabilities - MCP-specific configuration and capabilities - Plugin management system - API and discovery configuration - Server management capabilities

Classes¶

`APIConfig`	FastAPI application configuration.
`DiscoveryConfig`	Discovery system configuration.
`HealthStatus`	Health check status.
`MCPPlatform`	Specialized platform for MCP operations - inherits all base capabilities.
`MCPTransport`	MCP transport protocols.
`PluginConfig`	Plugin configuration model.
`ServerManagementConfig`	Server management configuration.
`ServerSource`	Where the server came from.
`ServerStatus`	Server operational status.

Module Contents¶

class dataflow.platform.models.mcp.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.mcp.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.mcp.HealthStatus¶

Bases: str, enum.Enum

Health check status.

HEALTHY = 'healthy'¶

UNHEALTHY = 'unhealthy'¶

UNKNOWN = 'unknown'¶

WARNING = 'warning'¶

class dataflow.platform.models.mcp.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.mcp.MCPTransport¶

Bases: str, enum.Enum

MCP transport protocols.

HTTP = 'http'¶

SSE = 'sse'¶

STDIO = 'stdio'¶

WEBSOCKET = 'websocket'¶

class dataflow.platform.models.mcp.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.mcp.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.mcp.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.mcp.ServerStatus¶

Bases: str, enum.Enum

Server operational status.

ACTIVE = 'active'¶

ERROR = 'error'¶

INACTIVE = 'inactive'¶

STARTING = 'starting'¶

STOPPING = 'stopping'¶

UNKNOWN = 'unknown'¶