MCP Terminal

Instructions

Implementation Reference

• src/mcp_terminal/tools/terminal.py:118-159 (handler)

  The MCP tool handler function 'execute_command' decorated with @mcp.tool(name="execute_command"). Performs command filtering using CommandFilter, executes via self.controller.execute_command, and maps result to ExecuteCommandResponse.

  @mcp.tool(name="execute_command", description="Executes a terminal command")
  async def execute_command(
      command: str,
      wait_for_output: bool = True,
      timeout: int = 10,
  ) -> ExecuteCommandResponse:
      try:
          # Check if command is allowed
          is_allowed, reason = self.command_filter.is_command_allowed(command)
  
          if not is_allowed:
              logger.warning(
                  f"Command execution denied: {command}. Reason: {reason}"
              )
              return ExecuteCommandResponse(
                  success=False,
                  error=f"Command not allowed: {reason}",
              )
  
          # Ensure we have a controller
          if not self.controller:
              self._init_controller()
  
          # Execute the command
          result = await self.controller.execute_command(
              command, wait_for_output, timeout
          )
  
          # Convert to response model
          return ExecuteCommandResponse(
              success=result.get("success", False),
              output=result.get("output"),
              error=result.get("error"),
              return_code=result.get("return_code"),
              warning=result.get("warning"),
          )
      except Exception as e:
          logger.error(f"Error executing command: {e}")
          return ExecuteCommandResponse(
              success=False, error=f"Error executing command: {str(e)}"
          )

• src/mcp_terminal/tools/terminal.py:24-34 (schema)

  ExecuteCommandRequest Pydantic model defining the input parameters: command (str), wait_for_output (bool, default True), timeout (int, default 10).

  class ExecuteCommandRequest(BaseModel):
      """Request model for executing a terminal command."""
  
      command: str = Field(..., description="The command to execute in the terminal")
      wait_for_output: bool = Field(
          True, description="Whether to wait for the command output"
      )
      timeout: int = Field(
          10, description="Timeout in seconds for waiting for the command output"
      )

• src/mcp_terminal/tools/terminal.py:36-50 (schema)

  ExecuteCommandResponse Pydantic model defining the output: success (bool), output (str|None), error (str|None), return_code (int|None), warning (str|None).

  class ExecuteCommandResponse(BaseModel):
      """Response model for the executed terminal command."""
  
      success: bool = Field(
          ..., description="Whether the command execution was successful"
      )
      output: Optional[str] = Field(None, description="The command output if available")
      error: Optional[str] = Field(
          None, description="Error message if the command failed"
      )
      return_code: Optional[int] = Field(
          None, description="The command return code if available"
      )
      warning: Optional[str] = Field(None, description="Warning message if any")

• src/mcp_terminal/server.py:93-107 (registration)

  In MCPTerminalServer.register_tools(): instantiates TerminalTool with config, then calls terminal_tool.register_mcp(self.mcp) which defines and registers the execute_command tool.

  # Create and register the terminal tool
  terminal_tool = TerminalTool(
      self.controller_type,
      whitelist_file=self.whitelist_file,
      blacklist_file=self.blacklist_file,
      whitelist_mode=self.whitelist_mode,
  )
  file_tool = FileTool()
  terminal_tool.register_mcp(self.mcp)
  file_tool.register_mcp(self.mcp)
  self.tools["terminal"] = terminal_tool
  self.tools["file"] = file_tool
  
  self.tools_registered = True
  logger.info("Terminal tool registered with MCP server")

• src/mcp_terminal/controllers/subprocess.py:16-73 (helper)

  SubprocessTerminalController.execute_command: core implementation using asyncio.create_subprocess_shell to execute commands, capture stdout/stderr/returncode with timeout support. Default cross-platform controller.

  async def execute_command(
      self, command: str, wait_for_output: bool = True, timeout: int = 10
  ) -> Dict[str, Any]:
      """
      Execute a command using subprocess.
  
      Args:
          command: The command to execute
          wait_for_output: Whether to wait for output
          timeout: Timeout in seconds
  
      Returns:
          A dictionary with the result of the command execution
      """
      try:
          # Create subprocess
          process = await asyncio.create_subprocess_shell(
              command,
              stdout=asyncio.subprocess.PIPE,
              stderr=asyncio.subprocess.PIPE,
          )
  
          if wait_for_output:
              try:
                  # Wait for the process to complete with timeout
                  stdout, stderr = await asyncio.wait_for(
                      process.communicate(), timeout=timeout
                  )
  
                  return {
                      "success": process.returncode == 0,
                      "output": stdout.decode("utf-8", errors="replace"),
                      "error": stderr.decode("utf-8", errors="replace"),
                      "return_code": process.returncode,
                  }
              except asyncio.TimeoutError:
                  # Kill the process if it times out
                  try:
                      process.kill()
                  except ProcessLookupError:
                      pass
                  return {
                      "success": False,
                      "error": f"Command timed out after {timeout} seconds",
                  }
          else:
              # Don't wait for output
              return {
                  "success": True,
                  "output": "Command sent (output not captured)",
              }
  
      except Exception as e:
          return {
              "success": False,
              "error": f"Error executing command: {str(e)}",
          }