haive.mcp.downloader

MCP Server Downloader - Download, install, and manage MCP servers.

class haive.mcp.downloader.BinaryInstaller[source]

Installer for binary executable MCP servers.

Downloads and installs pre-compiled binary executables.

Examples

Binary installation:

async can_handle(server_config, template)[source]

Check if this is a binary installation.

Parameters:
Return type:

bool

async install(server_config, template, install_dir)[source]

Download and install binary executable.

Parameters:
Returns:

Installation result dictionary

Return type:

dict[str, Any]

async verify(server_config, template, install_dir)[source]

Verify binary installation.

Checks if the binary exists and is executable.

Returns:

True if binary exists and is executable

Parameters:
Return type:

bool

class haive.mcp.downloader.CurlInstaller[source]

Installer for direct HTTP downloads.

Downloads files directly via HTTP/HTTPS without package managers.

Examples

Curl installation:

async can_handle(server_config, template)[source]

Check if this is a curl/HTTP download.

Parameters:
Return type:

bool

async install(server_config, template, install_dir)[source]

Download files via HTTP.

Parameters:
Returns:

Installation result dictionary

Return type:

dict[str, Any]

async verify(server_config, template, install_dir)[source]

Verify curl installation.

Checks if the expected files exist.

Returns:

True if installation directory exists with files

Parameters:
Return type:

bool

class haive.mcp.downloader.DiscoveredServer(*, name, source, source_url=None, description=None, package_name=None, repo_url=None, author=None, version=None, stars=None, tags=<factory>, metadata=<factory>)[source]

Information about a discovered MCP server.

Parameters:

  • data (Any)

  • name (str)

  • source (str)

  • source_url (str | None)

  • description (str | None)

  • package_name (str | None)

  • repo_url (str | None)

  • author (str | None)

  • version (str | None)

  • stars (int | None)

  • tags (set[str])

  • metadata (dict[str, Any])

name

Server name

Type:

str

source

Where it was discovered (npm, github, etc.)

Type:

str

source_url

URL where it was found

Type:

str | None

description

Server description

Type:

str | None

package_name

Package name (for npm/pypi)

Type:

str | None

repo_url

Repository URL (for git)

Type:

str | None

author

Author/owner name

Type:

str | None

version

Latest version

Type:

str | None

stars

GitHub stars or similar metric

Type:

int | None

tags

Extracted tags

Type:

set[str]

metadata

Additional metadata

Type:

dict[str, Any]

Examples

Discovered server info:

author: str | None
description: str | None
metadata: dict[str, Any]
model_config: ClassVar[ConfigDict] = {}

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

name: str
package_name: str | None
repo_url: str | None
source: str
source_url: str | None
stars: int | None
tags: set[str]
version: str | None
class haive.mcp.downloader.DiscoveryConfig(*, sources=<factory>, patterns=<factory>, auto_discover=False, discovery_interval=24, max_servers=100)[source]

Configuration for server discovery.

Parameters:
sources

List of discovery source URLs

Type:

list[str]

patterns

Package naming patterns by registry

Type:

dict[str, list[str]]

auto_discover

Enable automatic discovery

Type:

bool

discovery_interval

Hours between discovery runs

Type:

int

max_servers

Maximum servers to discover per source

Type:

int

auto_discover: bool
discovery_interval: int
max_servers: int
model_config: ClassVar[ConfigDict] = {}

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

patterns: dict[str, list[str]]
sources: list[str]
class haive.mcp.downloader.DockerInstaller[source]

Installer for Docker-based MCP servers.

Handles pulling Docker images for containerized MCP servers.

Examples

Docker installation:

async can_handle(server_config, template)[source]

Check if this is a Docker installation.

Parameters:
Return type:

bool

async install(server_config, template, install_dir)[source]

Pull Docker image.

Parameters:
Returns:

Installation result dictionary

Return type:

dict[str, Any]

async verify(server_config, template, install_dir)[source]

Verify Docker installation.

Checks if the Docker image exists locally.

Returns:

True if image is available

Parameters:
Return type:

bool

class haive.mcp.downloader.DownloadResult(*, total, successful, failed, success_rate, successful_servers=<factory>, failed_servers=<factory>, config_file=None, duration=None)[source]

Result of a download operation.

Parameters:
total

Total servers attempted

Type:

int

successful

Number of successful installations

Type:

int

failed

Number of failed installations

Type:

int

success_rate

Percentage success rate

Type:

float

successful_servers

List of successful server details

Type:

list[dict[str, Any]]

failed_servers

List of failed server details

Type:

list[dict[str, Any]]

config_file

Path to generated configuration

Type:

str | None

duration

Operation duration in seconds

Type:

float | None

Example

Checking results:

config_file: str | None
duration: float | None
failed: int
failed_servers: list[dict[str, Any]]
model_config: ClassVar[ConfigDict] = {}

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

success_rate: float
successful: int
successful_servers: list[dict[str, Any]]
total: int
class haive.mcp.downloader.DownloaderConfig(*, install_dir=<factory>, config_dir=<factory>, log_dir=<factory>, max_concurrent=5, retry_attempts=3, retry_delay=5, health_check_enabled=True, health_check_timeout=30, backup_enabled=True, backup_dir=<factory>, templates=<factory>, servers=<factory>, discovery=<factory>)[source]

Overall configuration for the MCP downloader.

Parameters:
install_dir

Directory for server installations

Type:

pathlib.Path

config_dir

Directory for configuration files

Type:

pathlib.Path

log_dir

Directory for log files

Type:

pathlib.Path

max_concurrent

Maximum concurrent downloads

Type:

int

retry_attempts

Number of retry attempts

Type:

int

retry_delay

Delay between retries in seconds

Type:

int

health_check_enabled

Enable health checks after install

Type:

bool

health_check_timeout

Health check timeout in seconds

Type:

int

backup_enabled

Enable configuration backups

Type:

bool

backup_dir

Directory for backups

Type:

pathlib.Path

templates

List of server templates

Type:

list[haive.mcp.downloader.config.ServerTemplate]

servers

List of server configurations

Type:

list[haive.mcp.downloader.config.ServerConfig]

discovery

Discovery configuration

Type:

haive.mcp.downloader.config.DiscoveryConfig

Example

Full configuration:

backup_dir: Path
backup_enabled: bool
config_dir: Path
discovery: DiscoveryConfig
health_check_enabled: bool
health_check_timeout: int
install_dir: Path
log_dir: Path
max_concurrent: int
model_config: ClassVar[ConfigDict] = {}

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

retry_attempts: int
retry_delay: int
servers: list[ServerConfig]
templates: list[ServerTemplate]
class haive.mcp.downloader.GeneralMCPDownloader(config_file=None, install_dir=None)[source]

General MCP Server Downloader with configurable patterns and installers.

This is the main orchestrator class that manages the downloading, installation, and configuration of MCP servers from various sources using a plugin-based architecture.

Parameters:

  • config_file (str | None)

  • install_dir (str | None)

config

Downloader configuration

installers

List of available installers

discovery

Server discovery instance

status_tracker

Server status tracking

Example

Creating and using downloader:

Note

The downloader automatically creates necessary directories and default configuration if not provided.

__init__(config_file=None, install_dir=None)[source]

Initialize the General MCP Downloader.

Parameters:

  • config_file (str | None) – Path to YAML configuration file. If not provided, creates a default configuration.

  • install_dir (str | None) – Directory for server installations. Defaults to ~/.mcp/servers if not specified.

Example

Initialization:

add_custom_server(server)[source]

Add a custom server configuration.

Parameters:

server (ServerConfig) – ServerConfig to add

Return type:

None

Example

Adding custom server:

add_custom_template(template)[source]

Add a custom template.

Parameters:

template (ServerTemplate) – ServerTemplate to add

Return type:

None

Example

Adding custom template:

async auto_discover_and_download(limit=None, auto_install=True)[source]

Auto-discover servers from registries and optionally download them.

This method discovers MCP servers from configured sources and can automatically install them.

Parameters:

  • limit (int | None) – Maximum number of servers to discover per source

  • auto_install (bool) – Whether to automatically install discovered servers

Returns:

DownloadResult with discovery and installation details

Return type:

DownloadResult

Example

Auto-discover and install:

async check_server_health(server_names=None)[source]

Check health status of installed servers.

Parameters:

server_names (list[str] | None) – Specific servers to check. If None, checks all installed.

Returns:

Dict with health check results

Return type:

dict[str, Any]

Example

Checking health:

async download_servers(server_names=None, categories=None, tags=None, max_concurrent=None)[source]

Download and install MCP servers.

This is the main method for downloading servers. It supports filtering by name, category, or tags, and handles concurrent downloads with retry logic.

Parameters:

  • server_names (list[str] | None) – Specific server names to download. If None, downloads all enabled servers.

  • categories (list[str] | None) – Filter servers by category (e.g., “official”, “community”)

  • tags (set[str] | None) – Filter servers by tags (e.g., {“database”, “file-operations”})

  • max_concurrent (int | None) – Maximum concurrent downloads. Uses config default if None.

Returns:

DownloadResult with details of the operation

Return type:

DownloadResult

Example

Download specific servers:

Raises:

ValueError – If no servers match the criteria

Parameters:
Return type:

DownloadResult

get_all_status()[source]

Get status for all servers.

Returns:

Dict mapping server names to ServerStatus objects

Return type:

dict[str, ServerStatus]

get_server_status(server_name)[source]

Get status for a specific server.

Parameters:

server_name (str) – Name of the server

Returns:

ServerStatus if found, None otherwise

Return type:

ServerStatus | None

Example

Checking status:

save_configuration(config_file)[source]

Save current configuration to file.

Parameters:

config_file (Path) – Path to save configuration

Return type:

None

Example

Saving config:

property servers: list[ServerConfig]

Get list of server configurations.

Returns:

List of ServerConfig objects

Return type:

list[haive.mcp.downloader.config.ServerConfig]

Example

Listing servers:

property templates: dict[str, ServerTemplate]

Get templates as a dictionary.

Returns:

Dict mapping template names to ServerTemplate objects

Return type:

dict[str, haive.mcp.downloader.config.ServerTemplate]

Example

Accessing templates:

class haive.mcp.downloader.GitHubMCPDownloader(resources_dir='agent_resources/mcp_servers')[source]

Download all MCP servers from GitHub resources.

Parameters:

resources_dir (str)

__init__(resources_dir='agent_resources/mcp_servers')[source]

Init .

Parameters:

resources_dir (str) – [TODO: Add description]

categorize_servers(servers)[source]

Categorize servers by type.

create_master_config(categorized)[source]

Create a master configuration file.

create_server_config(server_data)[source]

Create ServerConfig from server data.

async download_all_servers()[source]

Download all servers from GitHub resources.

load_all_servers()[source]

Load all server information from resources.

show_results(results)[source]

Display download results.

class haive.mcp.downloader.GitInstaller[source]

Installer for Git repository-based MCP servers.

Handles cloning Git repositories and running post-install commands.

Examples

Git installation:

async can_handle(server_config, template)[source]

Check if this is a Git installation.

Parameters:
Return type:

bool

async install(server_config, template, install_dir)[source]

Clone Git repository and run post-install commands.

Parameters:
Returns:

Installation result dictionary

Return type:

dict[str, Any]

async verify(server_config, template, install_dir)[source]

Verify Git installation.

Checks if the repository was cloned successfully.

Returns:

True if repository exists with .git directory

Parameters:
Return type:

bool

class haive.mcp.downloader.InstallationMethod(*values)[source]

Supported installation methods for MCP servers.

NPM

Node Package Manager installation

PIP

Python Package Index installation

GIT

Git repository cloning

DOCKER

Docker image pull

BINARY

Binary executable download

CURL

Direct HTTP download

MANUAL

Manual installation process

SCRIPT

Custom script execution

BINARY = 'binary'
CURL = 'curl'
DOCKER = 'docker'
GIT = 'git'
MANUAL = 'manual'
NPM = 'npm'
PIP = 'pip'
SCRIPT = 'script'
class haive.mcp.downloader.MCPInstaller[source]

Abstract base class for MCP installers.

All installer implementations must inherit from this class and implement the required methods.

Examples

Creating a custom installer:

abstractmethod async can_handle(server_config, template)[source]

Check if this installer can handle the given configuration.

Parameters:
Returns:

True if this installer can handle the installation

Return type:

bool

abstractmethod async install(server_config, template, install_dir)[source]

Install the MCP server.

Parameters:
Returns:

Dict with installation results including success status

Return type:

dict[str, Any]

abstractmethod async verify(server_config, template, install_dir)[source]

Verify the installation was successful.

Parameters:
Returns:

True if installation is verified

Return type:

bool

class haive.mcp.downloader.NPMInstaller[source]

Installer for NPM-based MCP servers.

Handles installation of MCP servers distributed as NPM packages. Supports both global and local installations.

Examples

NPM installation:

async can_handle(server_config, template)[source]

Check if this is an NPM installation.

Returns:

True if template uses NPM installation method

Parameters:
Return type:

bool

async install(server_config, template, install_dir)[source]

Install NPM package.

Attempts global installation first, falls back to local if that fails.

Parameters:
Returns:

Installation result dictionary

Return type:

dict[str, Any]

async verify(server_config, template, install_dir)[source]

Verify NPM installation.

Checks if the package is accessible via npx.

Returns:

True if package is installed and accessible

Parameters:
Return type:

bool

class haive.mcp.downloader.PipInstaller[source]

Installer for Python pip-based MCP servers.

Handles installation of MCP servers distributed as Python packages.

Examples

Pip installation:

async can_handle(server_config, template)[source]

Check if this is a pip installation.

Parameters:
Return type:

bool

async install(server_config, template, install_dir)[source]

Install Python package via pip.

Parameters:
Returns:

Installation result dictionary

Return type:

dict[str, Any]

async verify(server_config, template, install_dir)[source]

Verify pip installation.

Checks if the module can be imported.

Returns:

True if module is installed and importable

Parameters:
Return type:

bool

class haive.mcp.downloader.ServerConfig(*, name, template, source, variables=<factory>, enabled=True, priority=0, tags=<factory>, env_vars=<factory>, version=None)[source]

Configuration for a specific MCP server.

Server configurations define individual servers to be installed, referencing a template and providing server-specific variables.

Parameters:
name

Server name identifier

Type:

str

template

Template name to use

Type:

str

source

Source location (npm, git URL, etc.)

Type:

str

variables

Variables to substitute in template patterns

Type:

dict[str, Any]

enabled

Whether the server is enabled for installation

Type:

bool

priority

Installation priority (lower = higher priority)

Type:

int

tags

Set of tags for categorization

Type:

set[str]

env_vars

Additional environment variables

Type:

dict[str, str]

version

Specific version to install

Type:

str | None

Example

Filesystem server config:

enabled: bool
env_vars: dict[str, str]
model_config: ClassVar[ConfigDict] = {'use_enum_values': True}

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

name: str
priority: int
source: str
tags: set[str]
template: str
variables: dict[str, Any]
version: str | None
class haive.mcp.downloader.ServerDiscovery(config)[source]

Discovers MCP servers from various sources.

This class provides methods to discover MCP servers from multiple registries and sources, with support for various search patterns and filtering options.

Parameters:

config (DiscoveryConfig)

config

Discovery configuration

discovered_cache

Cache of discovered servers

session

Aiohttp session for HTTP requests

Examples

Using discovery:

async __aenter__()[source]

Async context manager entry.

async __aexit__(exc_type, exc_val, exc_tb)[source]

Async context manager exit.

__init__(config)[source]

Initialize server discovery.

Parameters:

config (DiscoveryConfig) – Discovery configuration with sources and patterns

determine_template(server_data)[source]

Determine the appropriate template for a discovered server.

Parameters:

server_data (dict[str, Any]) – Discovered server information

Returns:

Template name to use

Return type:

str

Examples

Determining template:

async discover_all(limit_per_source=None)[source]

Discover servers from all configured sources.

Parameters:

limit_per_source (int | None) – Maximum servers to discover per source. Uses config default if not specified.

Returns:

List of discovered server dictionaries

Return type:

list[dict[str, Any]]

Examples

Discovering from all sources:

async discover_from_github(api_url, limit=100)[source]

Discover servers from GitHub.

Parameters:

  • api_url (str) – GitHub API search URL

  • limit (int) – Maximum results

Returns:

List of discovered servers

Return type:

list[dict[str, Any]]

Examples

GitHub discovery:

async discover_from_npm_registry(registry_url, limit=100)[source]

Discover servers from npm registry.

Parameters:

  • registry_url (str) – NPM registry search URL

  • limit (int) – Maximum results to return

Returns:

List of discovered servers

Return type:

list[dict[str, Any]]

Examples

NPM discovery:

async discover_from_pypi(search_url, limit=100)[source]

Discover servers from PyPI.

Parameters:

  • search_url (str) – PyPI search URL

  • limit (int) – Maximum results

Returns:

List of discovered servers

Return type:

list[dict[str, Any]]

async discover_from_url(url, limit=100)[source]

Discover servers from a generic URL (e.g., README).

Parameters:

  • url (str) – URL to parse for server information

  • limit (int) – Maximum results

Returns:

List of discovered servers

Return type:

list[dict[str, Any]]

async search_servers(query, sources=None, limit=50)[source]

Search for servers across sources with a specific query.

Parameters:

  • query (str) – Search query

  • sources (list[str] | None) – Specific sources to search (npm, pypi, github)

  • limit (int) – Maximum results

Returns:

List of matching servers

Return type:

list[dict[str, Any]]

Examples

Searching for servers:

class haive.mcp.downloader.ServerStatus(*, name, status='pending', last_check=None, last_success=None, install_result=None, health_status=None, error=None)[source]

Status information for an MCP server.

Parameters:
name

Server name

Type:

str

status

Current status (installed, failed, pending)

Type:

str

last_check

Timestamp of last status check

Type:

datetime.datetime | None

last_success

Timestamp of last successful operation

Type:

datetime.datetime | None

install_result

Result of installation attempt

Type:

dict[str, Any] | None

health_status

Health check status

Type:

str | None

error

Error message if failed

Type:

str | None

Example

Creating status:

error: str | None
health_status: str | None
install_result: dict[str, Any] | None
last_check: datetime | None
last_success: datetime | None
model_config: ClassVar[ConfigDict] = {}

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

name: str
status: str
class haive.mcp.downloader.ServerTemplate(*, name, installation_method, command_pattern, args_pattern=<factory>, env_vars=<factory>, capabilities=<factory>, category='general', health_check=None, prerequisites=<factory>, post_install=<factory>, timeout=300)[source]

Template for MCP server installation patterns.

Templates define reusable installation patterns that can be applied to multiple servers with similar installation requirements.

Parameters:
name

Unique template identifier

Type:

str

installation_method

Method to use for installation

Type:

haive.mcp.downloader.config.InstallationMethod

command_pattern

Pattern for the command to run the server

Type:

str

args_pattern

Default arguments for the command

Type:

list[str]

env_vars

Environment variables to set

Type:

dict[str, str]

capabilities

List of server capabilities

Type:

list[str]

category

Server category for organization

Type:

str

health_check

Command to verify server health

Type:

str | None

prerequisites

Required system dependencies

Type:

list[str]

post_install

Commands to run after installation

Type:

list[str]

timeout

Installation timeout in seconds

Type:

int

Example

NPM template:

args_pattern: list[str]
capabilities: list[str]
category: str
command_pattern: str
env_vars: dict[str, str]
health_check: str | None
installation_method: InstallationMethod
model_config: ClassVar[ConfigDict] = {'use_enum_values': True}

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

name: str
post_install: list[str]
prerequisites: list[str]
timeout: int
haive.mcp.downloader.load_config(config_file)[source]

Load configuration from YAML file.

Parameters:

config_file (Path) – Path to YAML configuration file

Returns:

Loaded configuration

Return type:

DownloaderConfig

Raises:

Example

Loading config:

haive.mcp.downloader.save_config(config, config_file)[source]

Save configuration to YAML file.

Parameters:
Return type:

None

Example

Saving config: