vLLM MCP Server

Instructions

Implementation Reference

• src/vllm_mcp_server/tools/server_control.py:467-526 (handler)

  Main handler implementation for stop_vllm tool. This async function stops and optionally removes a vLLM Docker/Podman container. It validates that the container runtime is running, checks if the container exists and is running, executes the stop command with timeout, and optionally removes the container.

  async def stop_vllm(arguments: dict[str, Any]) -> list[TextContent]:
      """
      Stop a running vLLM container.
  
      Args:
          arguments: Dictionary containing:
              - container_name: Name of container to stop (default: from settings)
              - remove: Whether to remove the container (default: True)
              - timeout: Seconds to wait before killing (default: 10)
  
      Returns:
          List of TextContent with the result.
      """
      settings = get_settings()
      
      platform_info = await get_platform_info()
      if not platform_info.runtime_running:
          runtime_name = platform_info.container_runtime.value.capitalize() if platform_info.container_runtime != ContainerRuntime.NONE else "Container runtime"
          return [TextContent(type="text", text=f"❌ Error: {runtime_name} is not running.")]
  
      runtime_cmd = _get_runtime_cmd(platform_info.container_runtime)
      container_name = arguments.get("container_name", settings.container_name)
      remove = arguments.get("remove", True)
      timeout = arguments.get("timeout", 10)
  
      # Check if running
      is_running = await _is_container_running(container_name, platform_info.container_runtime)
      exists = await _is_container_exists(container_name, platform_info.container_runtime)
      
      if not exists:
          return [TextContent(
              type="text",
              text=f"ℹ️ Container '{container_name}' does not exist."
          )]
      
      result_parts = []
      
      if is_running:
          # Stop container with timeout
          exit_code, _, stderr = await _run_command(
              [runtime_cmd, "stop", "-t", str(timeout), container_name]
          )
          if exit_code != 0:
              return [TextContent(
                  type="text",
                  text=f"❌ Failed to stop container: {stderr}"
              )]
          result_parts.append(f"✅ Container '{container_name}' stopped.")
      else:
          result_parts.append(f"ℹ️ Container '{container_name}' was not running.")
  
      # Remove container if requested
      if remove:
          exit_code, _, stderr = await _run_command([runtime_cmd, "rm", container_name])
          if exit_code == 0:
              result_parts.append(f"✅ Container '{container_name}' removed.")
          else:
              result_parts.append(f"⚠️ Failed to remove container: {stderr}")
  
      return [TextContent(type="text", text="\n".join(result_parts))]

• src/vllm_mcp_server/server.py:216-238 (schema)

  Tool registration and schema definition for stop_vllm. Defines the Tool object with name, description, and inputSchema specifying container_name (string), remove (boolean, default True), and timeout (integer, default 10) parameters.

  Tool(
      name="stop_vllm",
      description="Stop a running vLLM Docker container",
      inputSchema={
          "type": "object",
          "properties": {
              "container_name": {
                  "type": "string",
                  "description": "Name of the container to stop",
              },
              "remove": {
                  "type": "boolean",
                  "description": "Whether to remove the container after stopping",
                  "default": True,
              },
              "timeout": {
                  "type": "integer",
                  "description": "Seconds to wait before force killing",
                  "default": 10,
              },
          },
      },
  ),

• src/vllm_mcp_server/server.py:355-356 (registration)

  Handler invocation in the tool dispatch logic. Routes calls to stop_vllm tool to the imported handler function with the provided arguments.

  elif name == "stop_vllm":
      return await stop_vllm(arguments)

• src/vllm_mcp_server/tools/server_control.py:233-248 (helper)

  Helper functions _is_container_running and _is_container_exists used by stop_vllm to check container status before attempting to stop it.

  async def _is_container_running(container_name: str, runtime: ContainerRuntime) -> bool:
      """Check if a container is running."""
      cmd = _get_runtime_cmd(runtime)
      exit_code, stdout, _ = await _run_command([
          cmd, "ps", "--filter", f"name=^{container_name}$", "--format", "{{.Names}}"
      ])
      return exit_code == 0 and container_name in stdout.strip().split("\n")
  
  
  async def _is_container_exists(container_name: str, runtime: ContainerRuntime) -> bool:
      """Check if a container exists (running or stopped)."""
      cmd = _get_runtime_cmd(runtime)
      exit_code, stdout, _ = await _run_command([
          cmd, "ps", "-a", "--filter", f"name=^{container_name}$", "--format", "{{.Names}}"
      ])
      return exit_code == 0 and container_name in stdout.strip().split("\n")

• src/vllm_mcp_server/tools/server_control.py:68-88 (helper)

  Helper function _run_command that executes shell commands asynchronously, used by stop_vllm to run docker/podman commands for stopping and removing containers.

  async def _run_command(cmd: list[str], timeout: float = 30.0) -> tuple[int, str, str]:
      """Run a shell command and return exit code, stdout, stderr."""
      try:
          process = await asyncio.create_subprocess_exec(
              *cmd,
              stdout=asyncio.subprocess.PIPE,
              stderr=asyncio.subprocess.PIPE,
          )
          stdout, stderr = await asyncio.wait_for(
              process.communicate(),
              timeout=timeout,
          )
          return (
              process.returncode or 0,
              stdout.decode("utf-8"),
              stderr.decode("utf-8"),
          )
      except asyncio.TimeoutError:
          return (1, "", "Command timed out")
      except Exception as e:
          return (1, "", str(e))