mcp.api¶

FastAPI web interface for MCP bulk management.

This module provides a complete REST API for managing MCP servers with bulk operations, real-time progress tracking, and a clean web interface for browsing and managing the 1900+ available MCP servers.

Features:

Bulk server installation and management
Real-time progress tracking via WebSockets
Category-based server organization
Health monitoring and status reporting
Server browser and search interface
Operation history and logging

Examples

Running the FastAPI server:

# Start the server
poetry run python -m haive.mcp.api

# Or with uvicorn directly
poetry run uvicorn haive.mcp.api:app --reload --host 0.0.0.0 --port 8000

Using the API programmatically:

import httpx

# Get available categories
response = httpx.get("http://localhost:8000/api/mcp/categories")
categories = response.json()

# Install a category
response = httpx.post("http://localhost:8000/api/mcp/categories/development/install")
operation = response.json()

# Track progress
operation_id = operation["operation_id"]
progress = httpx.get(f"http://localhost:8000/api/mcp/operations/{operation_id}")

Attributes¶

`app`
`connection_manager`
`global_manager`
`logger`

Classes¶

`BulkOperationResponse`	Response for bulk operations.
`CategoryInstallRequest`	Request to install servers from a category.
`ConnectionManager`
`ServerInstallRequest`	Request to install MCP servers.
`ServerRemoveRequest`	Request to remove servers.
`ServerStatusResponse`	Response for server status.

Functions¶

`bulk_health_check`()	Perform health check on all connected servers.
`get_categories`()	Get all available server categories.
`get_operation_status`(operation_id)	Get the status of a bulk operation.
`get_status`()	Get overall MCP system status.
`install_category`(category_name[, request, ...])	Install all servers in a category.
`install_servers`(request)	Install multiple MCP servers.
`lifespan`(app)	Manage the lifecycle of the FastAPI app.
`list_servers`()	List all configured servers and their status.
`main`()	Run the FastAPI server.
`remove_servers`(request)	Remove multiple servers from the manager.
`root`()	Serve the main web interface.
`update_all_servers`()	Update all installed MCP servers to latest versions.
`websocket_endpoint`(websocket)	WebSocket endpoint for real-time progress updates.

Module Contents¶

class mcp.api.BulkOperationResponse(/, **data: Any)¶

Bases: pydantic.BaseModel

Response for bulk operations.

is_complete: bool¶

operation_id: str¶

operation_type: str¶

progress_percentage: float¶

started_at: datetime.datetime¶

status: str¶

success_rate: float¶

total_count: int¶

class mcp.api.CategoryInstallRequest(/, **data: Any)¶

Bases: pydantic.BaseModel

Request to install servers from a category.

category_name: str = None¶

max_concurrent: int = None¶

class mcp.api.ConnectionManager¶

async broadcast(message: str)¶

async connect(websocket: fastapi.WebSocket)¶

disconnect(websocket: fastapi.WebSocket)¶

async send_personal_message(message: str, websocket: fastapi.WebSocket)¶

active_connections: List[fastapi.WebSocket] = []¶

class mcp.api.ServerInstallRequest(/, **data: Any)¶

Bases: pydantic.BaseModel

Request to install MCP servers.

add_to_manager: bool = None¶

max_concurrent: int = None¶

servers: List[str] = None¶

class mcp.api.ServerRemoveRequest(/, **data: Any)¶

Bases: pydantic.BaseModel

Request to remove servers.

server_names: List[str] = None¶

class mcp.api.ServerStatusResponse(/, **data: Any)¶

Bases: pydantic.BaseModel

Response for server status.

health: Dict[str, Any] | None = None¶

server_name: str¶

status: str¶

tools_count: int¶

async mcp.api.bulk_health_check()¶: Perform health check on all connected servers.

async mcp.api.get_categories()¶: Get all available server categories.

async mcp.api.get_operation_status(operation_id: str)¶: Get the status of a bulk operation.

async mcp.api.get_status()¶: Get overall MCP system status.

async mcp.api.install_category(category_name: str, request: CategoryInstallRequest = CategoryInstallRequest(category_name='', max_concurrent=5), background_tasks: fastapi.BackgroundTasks = None)¶: Install all servers in a category.

async mcp.api.install_servers(request: ServerInstallRequest)¶: Install multiple MCP servers.

async mcp.api.lifespan(app: fastapi.FastAPI)¶: Manage the lifecycle of the FastAPI app.

async mcp.api.list_servers()¶: List all configured servers and their status.

mcp.api.main()¶: Run the FastAPI server.

async mcp.api.remove_servers(request: ServerRemoveRequest)¶: Remove multiple servers from the manager.

async mcp.api.root()¶: Serve the main web interface.

async mcp.api.update_all_servers()¶: Update all installed MCP servers to latest versions.

async mcp.api.websocket_endpoint(websocket: fastapi.WebSocket)¶: WebSocket endpoint for real-time progress updates.

mcp.api.app¶

mcp.api.connection_manager¶

mcp.api.global_manager: haive.mcp.manager.MCPManager | None = None¶

mcp.api.logger¶