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: 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.servers.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.servers.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.servers.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.servers.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.servers.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.servers.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.servers.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¶