Hetzner Cloud MCP Server

Instructions

Implementation Reference

• mcp_hetzner/server.py:219-227 (schema)

  Pydantic BaseModel defining the input parameters and validation for the create_volume tool.
  class CreateVolumeParams(BaseModel): name: str = Field(..., description="Name of the volume") size: int = Field(..., description="Size of the volume in GB (min 10, max 10240)") location: Optional[str] = Field(None, description="Location where the volume will be created (e.g., nbg1, fsn1)") server: Optional[int] = Field(None, description="ID of the server to attach the volume to") automount: Optional[bool] = Field(False, description="Auto-mount the volume after attaching it") format: Optional[str] = Field(None, description="Filesystem format (e.g., xfs, ext4)") labels: Optional[Dict[str, str]] = Field(None, description="User-defined labels (key-value pairs)")
• mcp_hetzner/server.py:997-1068 (handler)

  The primary handler function decorated with @mcp.tool(), implementing the create_volume tool logic: validates inputs, resolves location and server objects, calls Hetzner Cloud API to create volume, handles response and formats output including volume details, action status, and next actions.
  @mcp.tool() def create_volume(params: CreateVolumeParams) -> Dict[str, Any]: """ Create a new volume. Creates a new volume with the specified configuration. Examples: - Basic volume: {"name": "data-volume", "size": 10} - With location: {"name": "db-volume", "size": 100, "location": "fsn1"} - Attached to server: {"name": "app-volume", "size": 50, "server": 123456, "automount": true} - With format: {"name": "log-volume", "size": 20, "format": "ext4"} """ try: # Get location if provided location = None if params.location: location = client.locations.get_by_name(params.location) if not location: return {"error": f"Location '{params.location}' not found"} # Get server if provided server = None if params.server: server = client.servers.get_by_id(params.server) if not server: return {"error": f"Server with ID {params.server} not found"} # Create the volume response = client.volumes.create( name=params.name, size=params.size, location=location, server=server, automount=params.automount, format=params.format, labels=params.labels ) # Extract volume and action information volume = response.volume action = response.action next_actions = response.next_actions # Format the response return { "volume": volume_to_dict(volume), "action": { "id": action.id, "status": action.status, "command": action.command, "progress": action.progress, "error": action.error, "started": action.started.isoformat() if action.started else None, "finished": action.finished.isoformat() if action.finished else None, } if action else None, "next_actions": [ { "id": next_action.id, "status": next_action.status, "command": next_action.command, "progress": next_action.progress, "error": next_action.error, "started": next_action.started.isoformat() if next_action.started else None, "finished": next_action.finished.isoformat() if next_action.finished else None, } for next_action in next_actions ] if next_actions else None, } except Exception as e: return {"error": f"Failed to create volume: {str(e)}"}
• mcp_hetzner/server.py:82-98 (helper)

  Helper function used by create_volume (and other volume tools) to serialize Volume domain objects into JSON-friendly dictionaries for the tool response.
  def volume_to_dict(volume: Volume) -> Dict[str, Any]: """Convert a Volume object to a dictionary with relevant information.""" return { "id": volume.id, "name": volume.name, "size": volume.size, "location": volume.location.name if volume.location else None, "server": volume.server.id if volume.server else None, "linux_device": volume.linux_device, "protection": { "delete": volume.protection["delete"] if volume.protection else False, }, "labels": volume.labels, "format": volume.format, "created": volume.created.isoformat() if volume.created else None, "status": volume.status, }