|
|
from api_data_structure import HiveComputeAPI |
|
|
from utils import logger, create_success_response, handle_exception |
|
|
from typing import Dict, Any |
|
|
from fastmcp import FastMCP |
|
|
from constant import Constants |
|
|
|
|
|
mcp = FastMCP( |
|
|
name="Compute with Hivenet MCP" |
|
|
) |
|
|
|
|
|
@mcp.tool() |
|
|
async def create_compute_instance(name: str = "default", location: str = "uae", config: str = "2x RTX 4090") -> Dict[str, Any]: |
|
|
""" |
|
|
Create a new compute instance with the specified configuration. |
|
|
|
|
|
Args: |
|
|
name: Name of the instance. Defaults to "default". |
|
|
location: Location where the instance will be deployed. Defaults to "uae". |
|
|
Valid locations: france, uae, texas, uae-2 |
|
|
config: Predefined configuration. Defaults to "1x RTX 4090". |
|
|
GPU configs: "1x RTX 4090", "2x RTX 4090", "4x RTX 4090", "8x RTX 4090", |
|
|
"1x RTX 5090", "2x RTX 5090", "4x RTX 5090", "8x RTX 5090" |
|
|
vCPU configs: "2vCPU", "4vCPU", "8vCPU", "16vCPU", "32vCPU" |
|
|
|
|
|
Returns: |
|
|
Dict containing the created instance information. |
|
|
""" |
|
|
token = Constants.HIVE_COMPUTE_DEFAULT_API_TOKEN |
|
|
api_handler = HiveComputeAPI(token=token) |
|
|
try: |
|
|
logger.info(f"Creating compute instance: name={name}, location={location}, config={config}") |
|
|
result = api_handler.create_instance(name=name, location=location, config=config) |
|
|
|
|
|
if result is None: |
|
|
logger.error(f"Failed to create instance - API returned None") |
|
|
return { |
|
|
"status": "error", |
|
|
"error": "Failed to create instance. Check API logs for details." |
|
|
} |
|
|
|
|
|
logger.info(f"Successfully created instance: {result.id if hasattr(result, 'id') else 'unknown'}") |
|
|
return create_success_response(result) |
|
|
except Exception as e: |
|
|
logger.error(f"Exception creating instance: {str(e)}") |
|
|
return handle_exception(e, "create_compute_instance") |
|
|
|
|
|
@mcp.tool() |
|
|
async def list_all_compute_instances(category: str = None) -> Dict[str, Any]: |
|
|
""" |
|
|
List all instances belonging to the user, organized into 4 categories: RUNNING, STOPPED, ERRORED, and TERMINATED. |
|
|
Shows basic information about each instance, including ID, name, status, location, spending, and resource allocation. |
|
|
Spending information includes hourly price and total in EUR spent so far. |
|
|
|
|
|
Args: |
|
|
category: Optional filter to return only instances from a specific category. |
|
|
Valid values: "RUNNING", "STOPPED", "ERRORED", "TERMINATED". |
|
|
If not provided, returns all categories. |
|
|
|
|
|
Returns: |
|
|
Dict containing instances. If category is specified, returns only instances from that category. |
|
|
If category is not specified, returns all instances organized by status categories. |
|
|
""" |
|
|
token = Constants.HIVE_COMPUTE_DEFAULT_API_TOKEN |
|
|
api_handler = HiveComputeAPI(token=token) |
|
|
try: |
|
|
logger.info(f"Listing all compute instances for token: {token}, category filter: {category}") |
|
|
all_instances = api_handler.get_all_instances() |
|
|
|
|
|
|
|
|
categorized = { |
|
|
"RUNNING": [], |
|
|
"STOPPED": [], |
|
|
"ERRORED": [], |
|
|
"TERMINATED": [] |
|
|
} |
|
|
|
|
|
for inst in all_instances: |
|
|
status = inst.status_string |
|
|
|
|
|
if status in ["RUNNING"]: |
|
|
categorized["RUNNING"].append(inst) |
|
|
elif status in ["STOPPED"]: |
|
|
categorized["STOPPED"].append(inst) |
|
|
elif status == "ERRORED": |
|
|
categorized["ERRORED"].append(inst) |
|
|
elif status in ["TERMINATED"]: |
|
|
categorized["TERMINATED"].append(inst) |
|
|
|
|
|
|
|
|
if category and category.upper() in categorized: |
|
|
return create_success_response(categorized[category.upper()]) |
|
|
|
|
|
|
|
|
return create_success_response(categorized) |
|
|
except Exception as e: |
|
|
return handle_exception(e, "list_all_compute_instances") |
|
|
|
