dataflow.platform.models.servers¶

Server Models - Intelligent Inheritance Hierarchy for All Server Types

This module provides a comprehensive server model hierarchy using intelligent inheritance patterns. All server models inherit from BaseServerInfo and specialize for different server types.

Inheritance Hierarchy: - BaseServerInfo (foundation)

├── MCPServerInfo (MCP-specific servers) │ └── DownloadedServerInfo (our 63 downloaded servers) └── [Future server types: HAP, Custom, etc.]

Key Features: - Pure Pydantic models with intelligent inheritance - Specialized models for different server sources - Comprehensive validation and field management - Factory methods for creating servers from real data - Connection configuration and health monitoring

Classes¶

`BaseServerInfo`	Foundation server model - all servers inherit from this.
`ConnectionConfig`	Server connection configuration.
`DownloadedServerInfo`	Specialized for our 63 downloaded servers - intelligent specialization.
`MCPServerInfo`	MCP-specific server - inherits base + adds MCP capabilities.
`PerformanceMetrics`	Server performance metrics.
`PromptInfo`	Information about a server prompt.
`ResourceInfo`	Information about a server resource.
`ToolInfo`	Information about a server tool.

Module Contents¶

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

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
)

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

Parameters:: data (Any)

get_server_summary()¶

Get basic server information summary.

Returns:: Dictionary with core server information
Return type:: Dict[str, Any]

update_health_status(new_health_status)¶

Update server health status with timestamp.

Parameters:: new_health_status (dataflow.platform.models.mcp.HealthStatus) – New health status
Return type:: None

update_status(new_status, update_timestamp=True)¶

Update server status and optionally timestamp.

Parameters:

new_status (dataflow.platform.models.mcp.ServerStatus) – New server status
update_timestamp (bool) – Whether to update last_updated timestamp

Return type:

None

classmethod validate_server_id_format(v)¶

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 (str) – Server ID to validate
Returns:: Validated server ID
Raises:: ValueError – If server ID format is invalid
Return type:: str

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

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

Bases: pydantic.BaseModel

Server connection configuration.

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

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

Parameters:: data (Any)

get_connection_summary()¶

Get connection configuration summary.

Return type:: Dict[str, Any]

is_command_based()¶

Check if connection is command-based.

Return type:: bool

is_url_based()¶

Check if connection is URL-based.

Return type:: bool

classmethod validate_url(v)¶

Validate URL format if provided.

Parameters:: v (Optional[str])
Return type:: Optional[str]

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

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

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

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

Parameters:: data (Any)

classmethod from_csv_and_install_report(csv_row, install_report_entry, bulk_session_id)¶

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 (Dict[str, Any]) – Row from our mcp_servers_data.csv
install_report_entry (Dict[str, Any]) – Entry from install report JSON
bulk_session_id (str) – ID of the bulk install session

Returns:

DownloadedServerInfo instance configured from real data

Return type:

DownloadedServerInfo

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

Get download-specific information summary.

Returns:: Dictionary with download metadata and status
Return type:: Dict[str, Any]

classmethod validate_directory_exists(v)¶

Validate local directory exists if specified.

Parameters:: v (Optional[pathlib.Path])
Return type:: Optional[pathlib.Path]

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

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")
    ]
)

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

Parameters:: data (Any)

get_mcp_capabilities_summary()¶

Get summary of MCP capabilities.

Returns:: Dictionary with MCP-specific capability information
Return type:: Dict[str, Any]

get_resource_uris()¶

Get list of resource URIs provided by this server.

Returns:: List of resource URIs
Return type:: List[str]

get_tool_names()¶

Get list of tool names provided by this server.

Returns:: List of tool names
Return type:: List[str]

record_connection_attempt(success)¶

Record a connection attempt.

Parameters:: success (bool) – Whether the connection was successful
Return type:: None

classmethod validate_stars_reasonable(v)¶

Validate star count is reasonable.

Parameters:: v (Optional[int])
Return type:: Optional[int]

property connection_success_rate: float¶

Calculate connection success rate.

Returns:: Success rate as float between 0.0 and 1.0
Return type:: float

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

Bases: pydantic.BaseModel

Server performance metrics.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

Parameters:: data (Any)

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

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

Bases: pydantic.BaseModel

Information about a server prompt.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

Parameters:: data (Any)

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

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

Bases: pydantic.BaseModel

Information about a server resource.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

Parameters:: data (Any)

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

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

Bases: pydantic.BaseModel

Information about a server tool.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

Parameters:: data (Any)

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