linode_tools.py•7.49 kB
import json
import logging
from typing import Any, Dict, List, cast
from linode_api4 import LinodeClient
from linode_api4.objects import Region
logger = logging.getLogger(__name__)
def register_tools(mcp_server: Any, linode_client: LinodeClient):
    """
    Register all Linode tools with the MCP server
    
    Args:
        mcp_server: The MCP server instance
        linode_client: The initialized Linode client
    """
    # Store the Linode client reference for tool access
    mcp_server.linode_client = linode_client
    
    # List regions
    @mcp_server.tool()
    def list_regions() -> str:
        """List all available Linode regions"""
        try:
            # Get the regions list
            regions_list = linode_client.regions()
            
            # Process each region in the collection
            result = []
            # Type annotation to help linter
            for item in regions_list:
                # Cast to Region type to resolve linting issues
                region = cast(Region, item)
                result.append({
                    "id": region.id,
                    "country": region.country,
                    "status": region.status,
                    "capabilities": region.capabilities
                })
                
            return json.dumps(result)
        except Exception as e:
            logger.error(f"Error listing regions: {str(e)}")
            return json.dumps({"error": str(e)})
    # # List instance types with pricing
    # @mcp_server.tool(name="list_instance_types")
    # def list_instance_types(random_string: str = "") -> str:
    #     """List all available Linode instance types and their pricing"""
    #     try:
    #         types = linode_client.linode.types.all()
    #         result = [{
    #             "id": type.id,
    #             "label": type.label,
    #             "vcpus": type.vcpus,
    #             "memory": type.memory,
    #             "disk": type.disk,
    #             "transfer": type.transfer,
    #             "price_monthly": type.price.monthly,
    #             "price_hourly": type.price.hourly
    #         } for type in types]
    #         return json.dumps(result)
    #     except Exception as e:
    #         logger.error(f"Error listing instance types: {str(e)}")
    #         return json.dumps({"error": str(e)})
    # # List instances
    # @mcp_server.tool(name="list_instances")
    # def list_instances(random_string: str = "") -> str:
    #     """List all existing Linode instances"""
    #     try:
    #         instances = linode_client.linode.instances.all()
    #         result = [{
    #             "id": instance.id,
    #             "label": instance.label,
    #             "region": instance.region.id,
    #             "type": instance.type.id,
    #             "status": instance.status,
    #             "ipv4": instance.ipv4,
    #             "created": str(instance.created),
    #             "updated": str(instance.updated)
    #         } for instance in instances]
    #         return json.dumps(result)
    #     except Exception as e:
    #         logger.error(f"Error listing instances: {str(e)}")
    #         return json.dumps({"error": str(e)})
    # # Create instance
    # @mcp_server.tool(name="create_instance")
    # def create_instance(label: str, region: str, type: str, image: str, root_pass: str) -> str:
    #     """
    #     Create a new Linode instance
        
    #     Args:
    #         label: Label for the Linode
    #         region: Region ID (use list_regions to see available options)
    #         type: Instance type ID (use list_instance_types to see available options)
    #         image: Image ID (e.g., 'linode/debian11')
    #         root_pass: Strong password for the root user
    #     """
    #     try:
    #         # Input validation
    #         if len(root_pass) < 8:
    #             return json.dumps({"error": "Root password must be at least 8 characters"})
                
    #         # Create the instance
    #         instance = linode_client.linode.instances.create(
    #             type=type,
    #             region=region,
    #             image=image,
    #             label=label,
    #             root_pass=root_pass
    #         )
            
    #         result = {
    #             "id": instance.id,
    #             "label": instance.label,
    #             "status": instance.status,
    #             "ipv4": instance.ipv4,
    #             "region": instance.region.id,
    #             "type": instance.type.id
    #         }
    #         return json.dumps(result)
    #     except Exception as e:
    #         logger.error(f"Error creating instance: {str(e)}")
    #         return json.dumps({"error": str(e)})
                
    # # Get instance details
    # @mcp_server.tool(name="get_instance")
    # def get_instance(id: int) -> str:
    #     """
    #     Get details about a specific Linode instance
        
    #     Args:
    #         id: Linode instance ID
    #     """
    #     try:
    #         instance = linode_client.linode.instances.get(id)
            
    #         result = {
    #             "id": instance.id,
    #             "label": instance.label,
    #             "status": instance.status,
    #             "ipv4": instance.ipv4,
    #             "ipv6": instance.ipv6,
    #             "region": instance.region.id,
    #             "type": instance.type.id,
    #             "specs": {
    #                 "disk": instance.specs.disk,
    #                 "memory": instance.specs.memory,
    #                 "vcpus": instance.specs.vcpus,
    #                 "transfer": instance.specs.transfer
    #             },
    #             "created": str(instance.created),
    #             "updated": str(instance.updated)
    #         }
    #         return json.dumps(result)
    #     except Exception as e:
    #         logger.error(f"Error getting instance details: {str(e)}")
    #         return json.dumps({"error": str(e)})
                
    # # Delete instance
    # @mcp_server.tool(name="delete_instance")
    # def delete_instance(id: int) -> str:
    #     """
    #     Delete a Linode instance
        
    #     Args:
    #         id: Linode instance ID
    #     """
    #     try:
    #         instance = linode_client.linode.instances.get(id)
    #         instance_label = instance.label
    #         instance.delete()
            
    #         return json.dumps({
    #             "success": True,
    #             "message": f"Instance {id} ({instance_label}) has been deleted"
    #         })
    #     except Exception as e:
    #         logger.error(f"Error deleting instance: {str(e)}")
    #         return json.dumps({"error": str(e)})
                
    # # Reboot instance
    # @mcp_server.tool(name="reboot_instance")
    # def reboot_instance(id: int) -> str:
    #     """
    #     Reboot a Linode instance
        
    #     Args:
    #         id: Linode instance ID
    #     """
    #     try:
    #         instance = linode_client.linode.instances.get(id)
    #         instance.reboot()
            
    #         return json.dumps({
    #             "success": True,
    #             "message": f"Instance {id} reboot initiated"
    #         })
    #     except Exception as e:
    #         logger.error(f"Error rebooting instance: {str(e)}")
    #         return json.dumps({"error": str(e)})